Splitting up API tests

* Setting up better structure for api tests
* Also splitting existing tests to make use of new
  structure
* Adding module for OpenStack API tests, and first
  basic test

Change-Id: I87dec192736191cf20d8310c065fa92a5daf356f
This commit is contained in:
adriant 2015-11-27 15:05:55 +13:00
parent 29593774eb
commit bd312534b1
5 changed files with 701 additions and 572 deletions

View File

@ -0,0 +1,17 @@
# Copyright (C) 2015 Catalyst IT Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from stacktask.api.v1.tests import *
# Just pipes the tests to the tests folder to keep things clean.

View File

@ -0,0 +1,125 @@
# Copyright (C) 2015 Catalyst IT Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
temp_cache = {}
def setup_temp_cache(projects, users):
admin_user = mock.Mock()
admin_user.id = 0
admin_user.name = 'admin'
admin_user.password = 'password'
admin_user.email = 'admin@example.com'
users.update({admin_user.name: admin_user})
global temp_cache
temp_cache = {
'i': 1,
'users': users,
'projects': projects,
'roles': {
'Member': 'Member',
'_member_': '_member_',
'admin': 'admin',
'project_owner': 'project_owner',
'project_mod': 'project_mod',
'heat_stack_owner': 'heat_stack_owner'
}
}
class FakeManager(object):
def find_user(self, name):
global temp_cache
return temp_cache['users'].get(name, None)
def get_user(self, user_id):
global temp_cache
return temp_cache['users'].get(user_id, None)
def create_user(self, name, password, email, project_id):
global temp_cache
user = mock.Mock()
temp_cache['i'] += 1
user.id = temp_cache['i']
user.name = name
user.password = password
user.email = email
user.default_project = project_id
temp_cache['users'][name] = user
return user
def update_user_password(self, user, password):
global temp_cache
user = temp_cache['users'][user.name]
user.password = password
def find_role(self, name):
global temp_cache
return temp_cache['roles'].get(name, None)
def get_roles(self, user, project):
global temp_cache
try:
roles = []
for role in project.roles[user.name]:
r = mock.Mock()
r.name = role
roles.append(r)
return roles
except KeyError:
return []
def add_user_role(self, user, role, project_id):
project = self.get_project(project_id)
try:
project.roles[user.name].append(role)
except KeyError:
project.roles[user.name] = [role]
def remove_user_role(self, user, role, project_id):
project = self.get_project(project_id)
try:
project.roles[user.name].remove(role)
except KeyError:
pass
def find_project(self, project_name):
global temp_cache
return temp_cache['projects'].get(project_name, None)
def get_project(self, project_id):
global temp_cache
for project in temp_cache['projects'].values():
if project.id == project_id:
return project
def create_project(self, project_name, created_on, p_id=None):
global temp_cache
project = mock.Mock()
if p_id:
project.id = p_id
else:
temp_cache['i'] += 1
project.id = temp_cache['i']
project.name = project_name
project.roles = {}
temp_cache['projects'][project_name] = project
return project

View File

@ -19,477 +19,14 @@ import mock
from django.utils import timezone
from datetime import timedelta
import json
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
temp_cache = {}
def setup_temp_cache(projects, users):
admin_user = mock.Mock()
admin_user.id = 0
admin_user.name = 'admin'
admin_user.password = 'password'
admin_user.email = 'admin@example.com'
users.update({admin_user.name: admin_user})
global temp_cache
temp_cache = {
'i': 1,
'users': users,
'projects': projects,
'roles': {
'Member': 'Member',
'_member_': '_member_',
'admin': 'admin',
'project_owner': 'project_owner',
'project_mod': 'project_mod',
'heat_stack_owner': 'heat_stack_owner'
}
}
class FakeManager(object):
def find_user(self, name):
global temp_cache
return temp_cache['users'].get(name, None)
def get_user(self, user_id):
global temp_cache
return temp_cache['users'].get(user_id, None)
def create_user(self, name, password, email, project_id):
global temp_cache
user = mock.Mock()
temp_cache['i'] += 1
user.id = temp_cache['i']
user.name = name
user.password = password
user.email = email
user.default_project = project_id
temp_cache['users'][name] = user
return user
def update_user_password(self, user, password):
global temp_cache
user = temp_cache['users'][user.name]
user.password = password
def find_role(self, name):
global temp_cache
return temp_cache['roles'].get(name, None)
def get_roles(self, user, project):
global temp_cache
try:
roles = []
for role in project.roles[user.name]:
r = mock.Mock()
r.name = role
roles.append(r)
return roles
except KeyError:
return []
def add_user_role(self, user, role, project_id):
project = self.get_project(project_id)
try:
project.roles[user.name].append(role)
except KeyError:
project.roles[user.name] = [role]
def remove_user_role(self, user, role, project_id):
project = self.get_project(project_id)
try:
project.roles[user.name].remove(role)
except KeyError:
pass
def find_project(self, project_name):
global temp_cache
return temp_cache['projects'].get(project_name, None)
def get_project(self, project_id):
global temp_cache
for project in temp_cache['projects'].values():
if project.id == project_id:
return project
def create_project(self, project_name, created_on, p_id=None):
global temp_cache
project = mock.Mock()
if p_id:
project.id = p_id
else:
temp_cache['i'] += 1
project.id = temp_cache['i']
project.name = project_name
project.roles = {}
temp_cache['projects'][project_name] = project
return project
class APITests(APITestCase):
"""Tests to ensure the approval/token workflow does
what is expected. These test don't check final
results for actions, simply that the tasks,
action, and tokens are created/updated.
These tests also focus on authentication status
and role prermissions."""
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user(self):
"""
Ensure the new user workflow goes as expected.
Create task, create token, submit token.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'testpassword'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user_no_project(self):
"""
Can't create a user for a non-existent project.
"""
setup_temp_cache({}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data, {'errors': ['actions invalid']})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user_not_my_project(self):
"""
Can't create a user for project that isn't mine.
"""
setup_temp_cache({}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user_not_authenticated(self):
"""
Can't create a user if unauthenticated.
"""
setup_temp_cache({}, {})
url = "/v1/actions/InviteUser"
headers = {}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
self.assertEqual(
response.data,
{'errors': ["Credentials incorrect or none given."]}
)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_add_user_existing(self):
"""
Adding existing user to project.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
setup_temp_cache({'test_project': project}, {user.name: user})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'confirm': True}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_add_user_existing_with_role(self):
"""
Adding existing user to project.
Already has role.
Should 'complete' anyway but do nothing.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {user.name: ['Member']}
setup_temp_cache({'test_project': project}, {user.name: user})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data,
{'notes': 'Task completed successfully.'})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_new_project(self):
"""
Ensure the new project workflow goes as expected.
"""
setup_temp_cache({}, {})
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
new_task = Task.objects.all()[0]
url = "/v1/tasks/" + new_task.uuid
response = self.client.post(url, {'approved': True}, format='json',
headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'testpassword'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_new_project_existing(self):
"""
Test to ensure validation marks actions as invalid
if project is already present.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
new_task = Task.objects.all()[0]
url = "/v1/tasks/" + new_task.uuid
response = self.client.post(url, {'approved': True}, format='json',
headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data, {'errors': ['actions invalid']})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_new_project_existing_user(self):
"""
Project created if not present, existing user attached.
No token should be needed.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
setup_temp_cache({}, {user.name: user})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
new_task = Task.objects.all()[0]
url = "/v1/tasks/" + new_task.uuid
response = self.client.post(url, {'approved': True}, format='json',
headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data,
{'notes': 'Task completed successfully.'}
)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_reset_user(self):
"""
Ensure the reset user workflow goes as expected.
Create task + create token, submit token.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
user.password = "test_password"
setup_temp_cache({}, {user.name: user})
url = "/v1/actions/ResetPassword"
data = {'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, None)
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'new_test_password'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(user.password, 'new_test_password')
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_reset_user_no_existing(self):
"""
Actions should be successful, so usernames are not exposed.
"""
setup_temp_cache({}, {})
url = "/v1/actions/ResetPassword"
data = {'email': "test@exampleinvalid.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, None)
class AdminAPITests(APITestCase):
"""
Tests to ensure the admin api endpoints work as expected within
the context of the approval/token workflow.
"""
def test_no_token_get(self):
"""
@ -727,40 +264,6 @@ class APITests(APITestCase):
response.data,
{'notes': ['created token']})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_notification_createproject(self):
"""
CreateProject should create a notification.
We should be able to grab it.
"""
setup_temp_cache({}, {})
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
new_task = Task.objects.all()[0]
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/notifications"
response = self.client.get(url, headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data["notifications"][0]['task'],
new_task.uuid)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
@ -1086,69 +589,6 @@ class APITests(APITestCase):
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.tenant_setup.models.IdentityManager', FakeManager)
def test_duplicate_tasks_new_project(self):
"""
Ensure we can't submit duplicate tasks
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({}, {})
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'project_name': "test_project_2", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
def test_duplicate_tasks_new_user(self):
"""
Ensure we can't submit duplicate tasks
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'email': "test2@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@mock.patch('stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager', FakeManager)
def test_cancel_task_own(self):
"""
Ensure the ability to cancel your own task.
@ -1191,8 +631,10 @@ class APITests(APITestCase):
headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@mock.patch('stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.tenant_setup.models.IdentityManager', FakeManager)
def test_cancel_task_own_fail(self):
"""
Ensure the ability to cancel ONLY your own task.
@ -1227,7 +669,8 @@ class APITests(APITestCase):
headers=headers)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
@mock.patch('stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
def test_task_list(self):
"""
"""
@ -1273,7 +716,8 @@ class APITests(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['tasks']), 3)
@mock.patch('stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
def test_task_list_filter(self):
"""
"""
@ -1339,7 +783,8 @@ class APITests(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['tasks']), 2)
@mock.patch('stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
def test_task_list_filter_cross_project(self):
"""
Ensure you can't override the initial project_id filter if
@ -1395,7 +840,8 @@ class APITests(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['tasks']), 0)
@mock.patch('stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
def test_task_list_filter_formating(self):
"""
"""

View File

@ -0,0 +1,63 @@
# Copyright (C) 2015 Catalyst IT Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from rest_framework import status
from rest_framework.test import APITestCase
from stacktask.api.models import Token
import mock
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
class OpenstackAPITests(APITestCase):
"""
TaskView tests specific to the openstack style urls.
Many of the original TaskView tests are valid and need
not be repeated here, but some additional features in the
unique TaskViews need testing.
"""
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user(self):
"""
Ensure the new user workflow goes as expected.
Create task, create token, submit token.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/openstack/users"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'testpassword'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)

View File

@ -0,0 +1,478 @@
# Copyright (C) 2015 Catalyst IT Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from rest_framework import status
from rest_framework.test import APITestCase
from stacktask.api.models import Task, Token
import mock
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
class TaskViewTests(APITestCase):
"""
Tests to ensure the approval/token workflow does what is
expected with the given TaskViews. These test don't check
final results for actions, simply that the tasks, action,
and tokens are created/updated.
"""
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user(self):
"""
Ensure the new user workflow goes as expected.
Create task, create token, submit token.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'testpassword'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user_no_project(self):
"""
Can't create a user for a non-existent project.
"""
setup_temp_cache({}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data, {'errors': ['actions invalid']})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user_not_my_project(self):
"""
Can't create a user for project that isn't mine.
"""
setup_temp_cache({}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_new_user_not_authenticated(self):
"""
Can't create a user if unauthenticated.
"""
setup_temp_cache({}, {})
url = "/v1/actions/InviteUser"
headers = {}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
self.assertEqual(
response.data,
{'errors': ["Credentials incorrect or none given."]}
)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_add_user_existing(self):
"""
Adding existing user to project.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
setup_temp_cache({'test_project': project}, {user.name: user})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'confirm': True}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_add_user_existing_with_role(self):
"""
Adding existing user to project.
Already has role.
Should 'complete' anyway but do nothing.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {user.name: ['Member']}
setup_temp_cache({'test_project': project}, {user.name: user})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data,
{'notes': 'Task completed successfully.'})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_new_project(self):
"""
Ensure the new project workflow goes as expected.
"""
setup_temp_cache({}, {})
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
new_task = Task.objects.all()[0]
url = "/v1/tasks/" + new_task.uuid
response = self.client.post(url, {'approved': True}, format='json',
headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'testpassword'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_new_project_existing(self):
"""
Test to ensure validation marks actions as invalid
if project is already present.
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
new_task = Task.objects.all()[0]
url = "/v1/tasks/" + new_task.uuid
response = self.client.post(url, {'approved': True}, format='json',
headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data, {'errors': ['actions invalid']})
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_new_project_existing_user(self):
"""
Project created if not present, existing user attached.
No token should be needed.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
setup_temp_cache({}, {user.name: user})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
new_task = Task.objects.all()[0]
url = "/v1/tasks/" + new_task.uuid
response = self.client.post(url, {'approved': True}, format='json',
headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data,
{'notes': 'Task completed successfully.'}
)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_reset_user(self):
"""
Ensure the reset user workflow goes as expected.
Create task + create token, submit token.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
user.password = "test_password"
setup_temp_cache({}, {user.name: user})
url = "/v1/actions/ResetPassword"
data = {'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, None)
new_token = Token.objects.all()[0]
url = "/v1/tokens/" + new_token.token
data = {'password': 'new_test_password'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(user.password, 'new_test_password')
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
def test_reset_user_no_existing(self):
"""
Actions should be successful, so usernames are not exposed.
"""
setup_temp_cache({}, {})
url = "/v1/actions/ResetPassword"
data = {'email': "test@exampleinvalid.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, None)
@mock.patch('stacktask.actions.models.user_store.IdentityManager',
FakeManager)
@mock.patch('stacktask.actions.tenant_setup.models.IdentityManager',
FakeManager)
def test_notification_createproject(self):
"""
CreateProject should create a notification.
We should be able to grab it.
"""
setup_temp_cache({}, {})
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
new_task = Task.objects.all()[0]
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "admin,Member",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/notifications"
response = self.client.get(url, headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data['notifications'][0]['task'],
new_task.uuid)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
@mock.patch(
'stacktask.actions.tenant_setup.models.IdentityManager', FakeManager)
def test_duplicate_tasks_new_project(self):
"""
Ensure we can't submit duplicate tasks
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({}, {})
url = "/v1/actions/CreateProject"
data = {'project_name': "test_project", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'project_name': "test_project_2", 'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
@mock.patch(
'stacktask.actions.models.user_store.IdentityManager', FakeManager)
def test_duplicate_tasks_new_user(self):
"""
Ensure we can't submit duplicate tasks
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/actions/InviteUser"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_owner,Member,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
data = {'email': "test@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'email': "test2@example.com", 'roles': ["Member"],
'project_id': 'test_project_id'}
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, {'notes': ['created token']})
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)