Commit 7ee39940 authored by lukas.burgey's avatar lukas.burgey

Modernize the tests for views.rest

parent ef96659b
Pipeline #110775 passed with stage
in 1 minute and 28 seconds
......@@ -574,6 +574,7 @@ class UserPreferences(models.Model):
DEPLOYMENT_MODE_BOTH = 'both'
DEPLOYMENT_MODE_SERVICES_ONLY = 'services-only'
DEPLOYMENT_MODE_VOS_ONLY = 'vos-only'
DEPLOYMENT_MODE_DEFAULT = DEPLOYMENT_MODE_BOTH
DEPLOYMENT_MODE_CHOICES = (
(DEPLOYMENT_MODE_BOTH, 'Pick a combination of services and VOs for deployment.'),
......@@ -584,5 +585,5 @@ class UserPreferences(models.Model):
deployment_mode = models.CharField(
max_length=20,
choices=DEPLOYMENT_MODE_CHOICES,
default=DEPLOYMENT_MODE_BOTH,
default=DEPLOYMENT_MODE_DEFAULT,
)
......@@ -3,7 +3,7 @@
import logging
from django.contrib.auth import authenticate
from django.test import Client, TestCase
from django.test import TestCase
from feudal.backend.models import Site, Service
from feudal.backend.models.users import User, SSHPublicKey
......@@ -116,20 +116,3 @@ class BaseTestCase(TestCase):
# pylint: disable=bare-except
except:
pass
# the user is logged in using session authentication
class LoggedInTest(BaseTestCase):
client = None
def setUp(self):
self.client = Client()
user = authenticate(
username=self.USER_NAME,
password=self.USER_PASSWORD,
)
self.assertIsNotNone(user)
self.client.force_login(
user=user,
)
......@@ -112,8 +112,10 @@ class ServiceView(generics.RetrieveAPIView):
serializer_class = ServiceSerializer
def get_object(self):
available_services = Service.objects.filter(vos__user=self.request.user).distinct()
LOGGER.debug('Available services: %s', available_services)
return get_object_or_404(
Service.objects.filter(vos__user=self.request.user),
available_services,
id=self.kwargs['id'],
)
......@@ -213,10 +215,8 @@ class DeploymentView(generics.RetrieveUpdateAPIView):
service = None
for s in self.request.user.services:
if s.id == int(dep_id):
if service is None:
service = s
else:
raise Exception('user has multiple services with identical id')
service = s
break
if service is None:
raise exceptions.ValidationError('You are not permitted to access a service with id "{}"'.format(dep_id))
......@@ -283,22 +283,23 @@ URLPATTERNS = [
path('user-prefs', UserPreferencesView.as_view(), name='user-prefs'),
path('services', ServiceListView.as_view(), name='user-services'),
path('services', ServiceListView.as_view(), name='user-service-list'),
re_path(r'^service/(?P<id>[0-9]+)$', ServiceView.as_view(), name='user-service'),
path('vos', VOListView.as_view(), name='user-vos'),
path('vos', VOListView.as_view(), name='user-vo-list'),
re_path(r'^vo/(?P<id>[0-9]+)$', VOView.as_view(), name='user-vo'),
path('ssh-keys', SSHPublicKeyListView.as_view(), name='user-sshkeys'),
re_path(r'^ssh-key/(?P<id>[0-9]+)$', SSHPublicKeyView.as_view(), name='user-sshkey'),
path('ssh-keys', SSHPublicKeyListView.as_view(), name='user-ssh-key-list'),
re_path(r'^ssh-key/(?P<id>[0-9]+)$', SSHPublicKeyView.as_view(), name='user-ssh-key'),
# type is either 'vo' or 'service'
re_path(r'^deployments/?(?P<type>.+)?$', DeploymentListView.as_view(), name='user-deps'),
re_path(r'^deployments/?(?P<type>.+)?$', DeploymentListView.as_view(), name='user-dep-list'),
re_path(r'^deployment/(?P<type>.+)/(?P<id>[0-9]+)$', DeploymentView.as_view(), name='user-dep'),
path('states', DeploymentStateListView.as_view(), name='user-dep-states'),
path('states', DeploymentStateListView.as_view(), name='user-dep-state-list'),
re_path(r'^state/(?P<id>[0-9]+)$', DeploymentStateView.as_view(), name='user-dep-state'),
# this catch all must be last in the list!
path('', HelpView.as_view(), name='user-help'),
re_path(r'.*$', HelpView.as_view()),
]
# pylint: disable=no-member
import logging
from feudal.backend.tests import LoggedInTest
from feudal.backend.models.users import SSHPublicKey
from django.urls import reverse
from feudal.backend.models import Service
from feudal.backend.models.users import SSHPublicKey, User, UserPreferences
from feudal.backend.models.deployments import (
Deployment, VODeployment, ServiceDeployment, DEPLOYED, NOT_DEPLOYED, QUESTIONNAIRE
VODeployment, ServiceDeployment, DEPLOYED, NOT_DEPLOYED, QUESTIONNAIRE
)
LOGGER = logging.getLogger(__name__)
def make_path(path, *args):
p = path.format(*args)
return '/user/'+p
class UserViewTest(LoggedInTest):
def test_user_deletion(self):
response = self.client.delete(
make_path('user')
)
LOGGER.debug('response: %s', response)
self.assertEqual(response.status_code, 204)
class SSHPublicKeyViewTest(LoggedInTest):
def test_list(self):
response = self.client.get(
make_path('ssh-keys')
)
LOGGER.debug('response: %s', response.json())
self.assertEqual(response.status_code, 200)
key_list = response.json()
self.assertEqual(
self.user.ssh_keys.count(),
len(key_list),
)
def test_add_success(self):
key_name = 'key_name'
key_value = 'key_value'
response = self.client.post(
make_path('ssh-keys'),
data={
'name': key_name,
'key': key_value,
},
content_type='application/json',
)
LOGGER.debug('response: %s', response)
self.assertEqual(response.status_code, 201)
try:
key = self.user.ssh_keys.get(name=key_name)
self.assertIsNotNone(key)
self.assertEqual(key.name, key_name)
self.assertEqual(key.key, key_value)
except SSHPublicKey.DoesNotExist:
self.fail('User did not receive the key')
def test_add_error(self):
# request missing key
response = self.client.post(
make_path('ssh-keys'),
data={
'name': 'key_name',
def test_user_help(user_test_client):
response = user_test_client.get(
reverse('user-help'),
)
assert response.status_code == 200
# USER
def test_user_deletion_success(user, user_test_client):
user_id = user.id
assert User.objects.filter(id=user_id).exists()
response = user_test_client.delete(
reverse('user'),
)
assert response.status_code == 204
assert not User.objects.filter(id=user_id).exists()
## USER PREFS
def test_user_prefs_fetch(user_test_client, user):
response = user_test_client.get(
reverse('user-prefs'),
)
assert response.status_code == 200
assert response.json()['deployment_mode'] == user.preferences.deployment_mode
def test_user_prefs_patch(user_test_client):
response = user_test_client.patch(
reverse('user-prefs'),
{
'deployment_mode': UserPreferences.DEPLOYMENT_MODE_SERVICES_ONLY,
},
format='json',
)
assert response.status_code == 200
## SERVICES
def test_service_list(user_test_client, user):
response = user_test_client.get(
reverse('user-service-list'),
)
assert response.status_code == 200
assert len(response.json()) == Service.objects.filter(vos__user=user).count()
def test_service(user_test_client, user, service):
_ = service # a service needs to exist
response = user_test_client.get(
reverse('user-service', kwargs={'id': Service.objects.filter(vos__user=user).first().id}),
)
assert response.status_code == 200
## SSH-KEYS
def test_ssh_key_list(user_test_client, user):
response = user_test_client.get(
reverse('user-ssh-key-list'),
)
LOGGER.debug('response: %s', response.json())
assert response.status_code == 200
key_list = response.json()
assert user.ssh_keys.count() == len(key_list)
def test_ssh_key_add_success(user_test_client, user):
key_name = 'key_name'
key_value = 'key_value'
response = user_test_client.post(
reverse('user-ssh-key-list'),
{
'name': key_name,
'key': key_value,
},
format='json',
)
LOGGER.debug('response: %s', response)
assert response.status_code == 201
key = user.ssh_keys.get(name=key_name)
assert key.key == key_value
def test_ssh_key_add_error(user_test_client):
# request missing key
response = user_test_client.post(
reverse('user-ssh-key-list'),
{
'name': 'key_name',
},
format='json',
)
assert response.status_code == 400
# request missing name
response = user_test_client.post(
reverse('user-ssh-key-list'),
{
'key': 'key_value',
},
format='json',
)
assert response.status_code == 400
def test_ssh_key_remove_success(user_test_client, user):
sshkey = SSHPublicKey(name='key_name', key='key_value', user=user)
sshkey.save()
response = user_test_client.delete(
reverse('user-ssh-key', kwargs={'id': sshkey.pk}),
)
assert response.status_code == 204
assert not user.ssh_keys.filter(name='key_name').exists()
def test_ssh_key_remove_error(user_test_client):
# no key with id 0
response = user_test_client.delete(
reverse('user-ssh-key', kwargs={'id': 0}),
)
assert response.status_code == 404
## VOS
def test_vo_list(user_test_client, user):
response = user_test_client.get(
reverse('user-vo-list'),
)
assert len(response.json()) == user.vos.count()
assert response.status_code == 200
def test_vo(user_test_client, user):
response = user_test_client.get(
reverse('user-vo', kwargs={'id': user.vos.first().id}),
)
assert response.status_code == 200
## DEPLOYMENTS
def test_dep_list(user_test_client, user):
response = user_test_client.get(
reverse('user-dep-list'),
)
assert response.status_code == 200
vo_deps = response.json()
assert user.deployments.all().count() == len(vo_deps)
def test_dep_list_vo(user_test_client, user):
response = user_test_client.get(
reverse('user-dep-list', kwargs={'type': 'vo'}),
)
assert response.status_code == 200
vo_deps = response.json()
assert user.deployments.instance_of(VODeployment).count() == len(vo_deps)
def test_dep_list_service(user_test_client, user):
response = user_test_client.get(
reverse('user-dep-list', kwargs={'type': 'service'}),
)
assert response.status_code == 200
service_deps = response.json()
assert user.deployments.instance_of(ServiceDeployment).count() == len(service_deps)
def test_service_deployment(user_test_client, user, service):
for target in [DEPLOYED, NOT_DEPLOYED]:
response = user_test_client.patch(
reverse('user-dep', kwargs={'type': 'service', 'id': service.id}),
{
'state_target': target,
},
content_type='application/json',
)
self.assertEqual(response.status_code, 400)
# request missing name
response = self.client.post(
make_path('ssh-keys'),
data={
'key': 'key_value',
},
content_type='application/json',
)
self.assertEqual(response.status_code, 400)
def test_remove_success(self):
sshkey = SSHPublicKey(name='key_name', key='key_value', user=self.user)
sshkey.save()
response = self.client.delete(
make_path('ssh-key/{}', sshkey.id),
)
self.assertEqual(response.status_code, 204)
with self.assertRaises(SSHPublicKey.DoesNotExist):
self.user.ssh_keys.get(name='key_name')
def test_remove_error(self):
# no key with id 0
response = self.client.delete(
make_path('ssh-key/{}', 0),
)
self.assertEqual(response.status_code, 404)
class VORestTest(LoggedInTest):
def test_vos(self):
response = self.client.get(
make_path('vos'),
)
self.assertEqual(len(response.json()), self.user.vos.count())
self.assertEqual(response.status_code, 200)
class DeploymentsViewTest(LoggedInTest):
def test_service_deployment(self):
service = self.service_one
for target in (DEPLOYED, NOT_DEPLOYED):
response = self.client.patch(
make_path('deployment/service/{}', service.id),
data={
'state_target': target,
},
content_type='application/json',
)
LOGGER.debug('%s', response.json())
self.assertEqual(response.status_code, 200)
try:
dep = self.user.deployments.get(servicedeployment__service=service)
self.assertEqual(dep.state_target, target)
except Deployment.DoesNotExist:
self.fail('User did not get the service deployment')
def test_vo_deployment(self):
vo = self.group_one
assert vo is not None
for target in [DEPLOYED, NOT_DEPLOYED]:
response = self.client.patch(
make_path('deployment/vo/{}', vo.id),
data={
'state_target': target,
},
content_type='application/json',
)
LOGGER.debug('%s', response.json())
self.assertEqual(response.status_code, 200)
try:
dep = self.user.deployments.get(vodeployment__vo=vo)
self.assertEqual(dep.state_target, target)
except Deployment.DoesNotExist:
self.fail('User did not get the VODeployment')
# pylint misses the reverse accessor 'deployments' on user
# pylint: disable=no-member
def test_list_vo(self):
response = self.client.get(
make_path('deployments/vo'),
format='json',
)
self.assertEqual(response.status_code, 200)
vo_deps = response.json()
self.assertEqual(
self.user.deployments.instance_of(VODeployment).count(),
len(vo_deps),
LOGGER.debug('%s', response.json())
assert response.status_code == 200
dep = user.deployments.get(servicedeployment__service=service)
assert dep.state_target == target
def test_service_deployment_failure(user_test_client):
''' no service with this id '''
target = DEPLOYED
response = user_test_client.patch(
reverse('user-dep', kwargs={'type': 'service', 'id': 0}),
{
'state_target': target,
},
format='json',
)
LOGGER.debug('%s', response.json())
assert response.status_code == 400
def test_vo_deployment(user_test_client, user, entitlement):
vo = entitlement
for target in [DEPLOYED, NOT_DEPLOYED]:
LOGGER.debug('State target: %s', target)
response = user_test_client.patch(
reverse('user-dep', kwargs={'type': 'vo', 'id': vo.id}),
{'state_target': target},
format='json',
)
# pylint misses the reverse accessor 'deployments' on user
# pylint: disable=no-member
def test_list_service(self):
response = self.client.get(
make_path('deployments/service'),
)
self.assertEqual(response.status_code, 200)
service_deps = response.json()
self.assertEqual(
self.user.deployments.instance_of(ServiceDeployment).count(),
len(service_deps),
)
class DeploymentStateViewTest(LoggedInTest):
def test_questionnaire_answers(self):
service = self.service_one
response = self.client.patch(
make_path('deployment/service/{}', service.id),
data={
'state_target': DEPLOYED,
},
content_type='application/json',
)
self.assertEqual(response.status_code, 200)
self.assertTrue('states' in response.json())
self.assertEqual(len(response.json().get('states')), 1)
state_id = response.json()['states'][0]['id']
state = self.user.states.get(id=state_id)
# mock client questionnaire response
state.state = QUESTIONNAIRE
state.questionnaire = {
'question_foo': 'description_foo',
'question_bar': 'description_bar',
}
state.state_changed()
# answer the questions as the user
response = self.client.patch(
make_path('state/{}', state.id),
data={
'answers': {
'question_foo': 'answer_foo',
'question_bar': 'answer_bar',
},
LOGGER.debug('%s', response.json())
assert response.status_code == 200
# user needs to get the deployment
dep = user.deployments.get(vodeployment__vo=vo)
assert dep.state_target == target
def test_vo_deployment_failure(user_test_client):
''' no vo with this id '''
response = user_test_client.patch(
reverse('user-dep', kwargs={'type': 'vo', 'id': 0}),
{'state_target': DEPLOYED},
format='json',
)
assert response.status_code == 400
def test_dep_failure(user_test_client):
''' no invalid type '''
response = user_test_client.patch(
reverse('user-dep', kwargs={'type': 'invalid', 'id': 0}),
{'state_target': DEPLOYED},
format='json',
)
assert response.status_code == 400
# DEP STATES
def test_dep_state_list(user_test_client, user):
response = user_test_client.get(
reverse('user-dep-state-list'),
)
assert response.status_code == 200
states = response.json()
assert user.states.all().count() == len(states)
def test_questionnaire_answers(user_test_client, user, service):
response = user_test_client.patch(
reverse('user-dep', kwargs={'type': 'service', 'id': service.id}),
{
'state_target': DEPLOYED,
},
format='json',
)
assert response.status_code == 200
assert 'states' in response.json()
assert len(response.json().get('states')) == 1
state_id = response.json()['states'][0]['id']
state = user.states.get(id=state_id)
# mock client questionnaire response
state.state = QUESTIONNAIRE
state.questionnaire = {
'question_foo': 'description_foo',
'question_bar': 'description_bar',
}
state.state_changed()
# answer the questions as the user
response = user_test_client.patch(
reverse('user-dep-state', kwargs={'id': state.id}),
{
'answers': {
'question_foo': 'answer_foo',
'question_bar': 'answer_bar',
},
content_type='application/json',
)
LOGGER.debug('Response: %s', response.json())
self.assertEqual(response.status_code, 200)
},
format='json',
)
LOGGER.debug('Response: %s', response.json())
assert response.status_code == 200
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment