adjutant/stacktask/api/v1/tests/test_api_openstack.py

242 lines
8.8 KiB
Python

# 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
from rest_framework import status
from rest_framework.test import APITestCase
from stacktask.api.models import Token
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
@mock.patch('stacktask.actions.user_store.IdentityManager',
FakeManager)
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.
"""
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.domain = 'default'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/openstack/users"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_admin,_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)
def test_user_list(self):
"""
"""
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.domain = 'default'
project.roles = {}
setup_temp_cache({'test_project': project}, {})
url = "/v1/openstack/users"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_admin,_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)
url = "/v1/openstack/users"
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.get(url, headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['users']), 2)
def test_user_list_managable(self):
"""
Confirm that the manageable value is set correctly.
"""
user = mock.Mock()
user.id = 'user_id_1'
user.name = "test1@example.com"
user.email = "test1@example.com"
user.domain = 'default'
user2 = mock.Mock()
user2.id = 'user_id_2'
user2.name = "test2@example.com"
user2.email = "test2@example.com"
user2.domain = 'default'
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.domain = 'default'
project.roles = {
user.id: ['_member_', 'project_admin'],
user2.id: ['_member_', 'project_mod']}
setup_temp_cache(
{'test_project': project},
{user.id: user, user2.id: user2})
url = "/v1/openstack/users"
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "_member_,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/openstack/users"
response = self.client.get(url, headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['users']), 2)
for st_user in response.data['users']:
if st_user['id'] == user.id:
self.assertFalse(st_user['manageable'])
if st_user['id'] == user2.id:
self.assertTrue(st_user['manageable'])
def test_force_reset_password(self):
"""
Ensure the force password endpoint works as expected,
and only for admin.
Should also check if template can be rendered.
"""
user = mock.Mock()
user.id = 'user_id'
user.name = "test@example.com"
user.email = "test@example.com"
user.domain = 'default'
user.password = "test_password"
setup_temp_cache({}, {user.id: user})
headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "_member_",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
url = "/v1/openstack/users/password-set"
data = {'email': "test@example.com"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
headers["roles"] = "admin,_member_"
response = self.client.post(url, data, format='json', headers=headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data['notes'],
['If user with email exists, reset token will be issued.'])
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')
def test_remove_user_role(self):
""" Remove all roles on a user from our project """
project = mock.Mock()
project.id = 'test_project_id'
project.name = 'test_project'
project.domain = 'default'
project.roles = {'user_id_1': ['_member_']}
user1 = mock.Mock()
user1.id = 'user_id_1'
user1.name = 'test@example.com'
user1.password = 'testpassword'
user1.email = 'test@example.com'
setup_temp_cache(
{'test_project': project},
{'user_id_1': user1})
admin_headers = {
'project_name': "test_project",
'project_id': "test_project_id",
'roles': "project_admin,_member_,project_mod",
'username': "test@example.com",
'user_id': "test_user_id",
'authenticated': True
}
# admins removes role from the test user
url = "/v1/openstack/users/%s/roles" % user1.id
data = {'roles': ["_member_"]}
response = self.client.delete(url, data,
format='json', headers=admin_headers)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data,
{'notes': ['Task completed successfully.']})