Add tests for public API
This commit is contained in:
parent
d5e0e069d1
commit
82c0c63ff4
@ -1,5 +1,8 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
from flask import url_for
|
||||
|
||||
from capsulflask.hub_model import MockHub
|
||||
from capsulflask.db import get_model
|
||||
from capsulflask.tests_base import BaseTestCase
|
||||
|
||||
@ -14,7 +17,7 @@ class ConsoleTests(BaseTestCase):
|
||||
|
||||
ssh_key_data = {
|
||||
"name": "key2",
|
||||
"method": "POST",
|
||||
"action": "upload_ssh_key",
|
||||
"content": "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQDntq1t8Ddsa2q4p+PM7W4CLYYmxakokRRVLlf7AQlsTJFPsgBe9u0zuoOaKDMkBr0dlnuLm4Eub1Mj+BrdqAokto0YDiAnxUKRuYQKuHySKK8bLkisi2k47jGBDikx/jihgiuFTawo1mYsJJepC7PPwZGsoCImJEgq1L+ug0p3Zrj3QkUx4h25MpCSs2yvfgWjDyN8hEC76O42P+4ETezYrzrd1Kj26hdzHRnrxygvIUOtfau+5ydlaz8xQBEPrEY6/+pKDuwtXg1pBL7GmoUxBXVfHQSgq5s9jIJH+G0CR0ZoHMB25Ln4X/bsCQbLOu21+IGYKSDVM5TIMLtkKUkERQMVWvnpOp1LZKir4dC0m7SW74wpA8+2b1IsURIr9ARYGJpCEv1Q1Wz/X3yTf6Mfey7992MjUc9HcgjgU01/+kYomoXHprzolk+22Gjfgo3a4dRIoTY82GO8kkUKiaWHvDkkVURCY5dpteLA05sk3Z9aRMYsNXPLeOOPfzTlDA0="
|
||||
}
|
||||
|
||||
@ -37,15 +40,21 @@ class ConsoleTests(BaseTestCase):
|
||||
|
||||
data = self.capsul_data
|
||||
data['csrf-token'] = csrf_token
|
||||
client.post(url_for("console.create"), data=data)
|
||||
capacity_message = \
|
||||
'\n host(s) at capacity. no capsuls can be created at this time. sorry. \n '
|
||||
self.assert_message_flashed(capacity_message, category='message')
|
||||
|
||||
self.assertEqual(
|
||||
len(get_model().list_vms_for_account('test@example.com')),
|
||||
0
|
||||
)
|
||||
# Override MockHub.capacity_avaliable to always return False
|
||||
with patch.object(MockHub, 'capacity_avaliable', return_value=False) as mock_method:
|
||||
self.app.config["HUB_MODEL"] = MockHub()
|
||||
|
||||
client.post(url_for("console.create"), data=data)
|
||||
capacity_message = \
|
||||
'\n host(s) at capacity. no capsuls can be created at this time. sorry. \n '
|
||||
self.assert_message_flashed(capacity_message, category='message')
|
||||
|
||||
self.assertEqual(
|
||||
len(get_model().list_vms_for_account('test@example.com')),
|
||||
0
|
||||
)
|
||||
mock_method.assert_called_with(512 * 1024 * 1024)
|
||||
|
||||
def test_create_fails_invalid(self):
|
||||
with self.client as client:
|
||||
@ -95,15 +104,15 @@ class ConsoleTests(BaseTestCase):
|
||||
def test_keys_loads(self):
|
||||
self._login('test@example.com')
|
||||
with self.client as client:
|
||||
response = client.get(url_for("console.ssh_public_keys"))
|
||||
response = client.get(url_for("console.ssh_api_keys"))
|
||||
self.assert_200(response)
|
||||
keys = self.get_context_variable('ssh_public_keys')
|
||||
self.assertEqual(keys[0]['name'], 'key')
|
||||
|
||||
def test_keys_add_fails_invalid(self):
|
||||
def test_keys_add_ssh_fails_invalid(self):
|
||||
self._login('test@example.com')
|
||||
with self.client as client:
|
||||
client.get(url_for("console.ssh_public_keys"))
|
||||
client.get(url_for("console.ssh_api_keys"))
|
||||
csrf_token = self.get_context_variable('csrf_token')
|
||||
|
||||
data = self.ssh_key_data
|
||||
@ -112,7 +121,7 @@ class ConsoleTests(BaseTestCase):
|
||||
data_invalid_content = data
|
||||
data_invalid_content['content'] = 'foo'
|
||||
client.post(
|
||||
url_for("console.ssh_public_keys"),
|
||||
url_for("console.ssh_api_keys"),
|
||||
data=data_invalid_content
|
||||
)
|
||||
|
||||
@ -123,28 +132,37 @@ class ConsoleTests(BaseTestCase):
|
||||
|
||||
data_missing_content = data
|
||||
data_missing_content['content'] = ''
|
||||
client.post(url_for("console.ssh_public_keys"), data=data_missing_content)
|
||||
client.post(url_for("console.ssh_api_keys"), data=data_missing_content)
|
||||
|
||||
self.assert_message_flashed(
|
||||
'Content is required', category='message'
|
||||
)
|
||||
|
||||
def test_keys_add_fails_duplicate(self):
|
||||
def test_keys_add_ssh_fails_duplicate(self):
|
||||
self._login('test@example.com')
|
||||
with self.client as client:
|
||||
client.get(url_for("console.ssh_public_keys"))
|
||||
client.get(url_for("console.ssh_api_keys"))
|
||||
csrf_token = self.get_context_variable('csrf_token')
|
||||
|
||||
data = self.ssh_key_data
|
||||
data['csrf-token'] = csrf_token
|
||||
data['name'] = 'key'
|
||||
client.post(url_for("console.ssh_public_keys"), data=data)
|
||||
client.post(url_for("console.ssh_api_keys"), data=data)
|
||||
|
||||
self.assert_message_flashed(
|
||||
'A key with that name already exists',
|
||||
category='message'
|
||||
)
|
||||
|
||||
data = self.ssh_key_data
|
||||
data['csrf-token'] = csrf_token
|
||||
data['name'] = 'key'
|
||||
client.post(url_for("console.ssh_api_keys"), data=data)
|
||||
|
||||
self.assert_message_flashed(
|
||||
'A key with that name already exists',
|
||||
category='message'
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self._login('test@example.com')
|
||||
|
45
capsulflask/tests/test_publicapi.py
Normal file
45
capsulflask/tests/test_publicapi.py
Normal file
@ -0,0 +1,45 @@
|
||||
from base64 import b64encode
|
||||
import requests
|
||||
|
||||
from flask import url_for
|
||||
|
||||
from capsulflask.db import get_model
|
||||
from capsulflask.tests_base import BaseLiveServerTestCase
|
||||
|
||||
|
||||
class PublicAPITests(BaseLiveServerTestCase):
|
||||
def test_server_is_up_and_running(self):
|
||||
response = requests.get(self.get_server_url())
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_capsul_create_succeeds(self):
|
||||
response = requests.post(
|
||||
self.get_server_url() +
|
||||
url_for('publicapi.capsul_create'),
|
||||
headers={'Authorization': self.token},
|
||||
json={
|
||||
'size': 'f1-xs',
|
||||
'os': 'openbsd68',
|
||||
'ssh_key_0': 'key'
|
||||
}
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# FIXME: mock create doesn't create, see #83
|
||||
# vms = get_model().list_vms_for_account('test@example.com')
|
||||
#
|
||||
# self.assertEqual(
|
||||
# len(vms),
|
||||
# 1
|
||||
# )
|
||||
|
||||
def setUp(self):
|
||||
get_model().create_ssh_public_key('test@example.com', 'key', 'foo')
|
||||
self.token = b64encode(
|
||||
get_model().generate_api_token('test@example.com', 'apikey').encode('utf-8')
|
||||
).decode('utf-8')
|
||||
|
||||
def tearDown(self):
|
||||
get_model().delete_ssh_public_key('test@example.com', 'key')
|
||||
get_model().delete_api_token('test@example.com', 1)
|
@ -1,12 +1,13 @@
|
||||
import os
|
||||
from nanoid import generate
|
||||
|
||||
from flask_testing import TestCase
|
||||
from flask_testing import TestCase, LiveServerTestCase
|
||||
|
||||
from capsulflask import create_app
|
||||
from capsulflask.db import get_model
|
||||
|
||||
class BaseTestCase(TestCase):
|
||||
|
||||
class BaseSharedTestCase(object):
|
||||
def create_app(self):
|
||||
# Use default connection paramaters
|
||||
os.environ['POSTGRES_CONNECTION_PARAMETERS'] = "host=localhost port=5432 user=postgres password=dev dbname=capsulflask_test"
|
||||
@ -21,8 +22,14 @@ class BaseTestCase(TestCase):
|
||||
def tearDown(self):
|
||||
pass
|
||||
|
||||
|
||||
class BaseTestCase(BaseSharedTestCase, TestCase):
|
||||
def _login(self, user_email):
|
||||
get_model().login(user_email)
|
||||
with self.client.session_transaction() as session:
|
||||
session['account'] = user_email
|
||||
session['csrf-token'] = generate()
|
||||
|
||||
|
||||
class BaseLiveServerTestCase(BaseSharedTestCase, LiveServerTestCase):
|
||||
pass
|
||||
|
Loading…
Reference in New Issue
Block a user