Added modify_dict_settings test decorator
This decorator allows for modification of test settings that are stored as dictionaries for that tests only without needing to override the whole dictionary. It acts in a similar way to django's in built override_settings, and modify_settings. To be applied to a class the class must subclass StacktaskTestCase or StacktaskAPITestCase. In those two classes settings can also be modified using: with self.modify_dict_settings(...): # code Example Usage: @modify_dict_settings(ROLES_MAPPING=[ {'key_list': ['project_mod'], 'operation': 'remove', 'value': 'heat_stack_owner'}, {'key_list': ['project_admin'], 'operation': 'append', 'value': 'heat_stack_owner'}, ]) or @modify_dict_settings(PROJECT_QUOTA_SIZES={ 'key_list': ['small', 'nova', 'instances'], 'operations': 'override', 'value': 11 }) Available operations: Standard operations: - 'override': Either overrides or adds the value to the dictionary. - 'delete': Removes the value from the dictionary. List operations: List operations expect that the accessed value in the dictionary is a list. - 'append': Add the specified values to the end of the list - 'prepend': Add the specifed values to the start of the list - 'remove': Remove the specified values from the list Change-Id: I575c17891d14418c335b2fb8426830e6e31be515
This commit is contained in:
parent
0ff277f208
commit
bbc0f31062
@ -20,7 +20,8 @@ from stacktask.actions.v1.projects import (
|
||||
NewProjectWithUserAction, AddDefaultUsersToProjectAction)
|
||||
from stacktask.api.models import Task
|
||||
from stacktask.api.v1 import tests
|
||||
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
|
||||
from stacktask.api.v1.tests import (FakeManager, setup_temp_cache,
|
||||
modify_dict_settings)
|
||||
|
||||
|
||||
@mock.patch('stacktask.actions.user_store.IdentityManager',
|
||||
@ -449,6 +450,11 @@ class ProjectActionTests(TestCase):
|
||||
action.post_approve()
|
||||
self.assertEquals(action.valid, False)
|
||||
|
||||
@modify_dict_settings(DEFAULT_ACTION_SETTINGS={
|
||||
'key_list': ['AddDefaultUsersToProjectAction'],
|
||||
'operation': 'override',
|
||||
'value': {'default_users': ['admin', ],
|
||||
'default_roles': ['admin', ]}})
|
||||
def test_add_default_users(self):
|
||||
"""
|
||||
Base case, adds admin user with admin role to project.
|
||||
@ -512,6 +518,11 @@ class ProjectActionTests(TestCase):
|
||||
# Now the missing project should make the action invalid
|
||||
self.assertEquals(action.valid, False)
|
||||
|
||||
@modify_dict_settings(DEFAULT_ACTION_SETTINGS={
|
||||
'key_list': ['AddDefaultUsersToProjectAction'],
|
||||
'operation': 'override',
|
||||
'value': {'default_users': ['admin', ],
|
||||
'default_roles': ['admin', ]}})
|
||||
def test_add_default_users_reapprove(self):
|
||||
"""
|
||||
Ensure nothing happens or changes during rerun of approve.
|
||||
|
@ -20,7 +20,8 @@ from stacktask.actions.v1.resources import (
|
||||
NewDefaultNetworkAction, NewProjectDefaultNetworkAction,
|
||||
SetProjectQuotaAction)
|
||||
from stacktask.api.models import Task
|
||||
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
|
||||
from stacktask.api.v1.tests import (FakeManager, setup_temp_cache,
|
||||
modify_dict_settings)
|
||||
from stacktask.actions.v1.tests import (
|
||||
get_fake_neutron, get_fake_novaclient, get_fake_cinderclient,
|
||||
setup_neutron_cache, neutron_cache, cinder_cache, nova_cache,
|
||||
@ -207,6 +208,17 @@ class ProjectSetupActionTests(TestCase):
|
||||
self.assertEquals(len(
|
||||
neutron_cache['RegionOne']['test_project_id']['subnets']), 1)
|
||||
|
||||
@modify_dict_settings(DEFAULT_ACTION_SETTINGS={
|
||||
'operation': 'override',
|
||||
'key_list': ['NewDefaultNetworkAction'],
|
||||
'value': {'RegionOne': {
|
||||
'DNS_NAMESERVERS': ['193.168.1.2', '193.168.1.3'],
|
||||
'SUBNET_CIDR': '192.168.1.0/24',
|
||||
'network_name': 'somenetwork',
|
||||
'public_network': '3cb50f61-5bce-4c03-96e6-8e262e12bb35',
|
||||
'router_name': 'somerouter',
|
||||
'subnet_name': 'somesubnet'
|
||||
}}})
|
||||
def test_new_project_network_setup(self):
|
||||
"""
|
||||
Base case, setup network after a new project, no issues.
|
||||
|
@ -12,20 +12,19 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from django.test import TestCase
|
||||
|
||||
import mock
|
||||
|
||||
from stacktask.actions.v1.users import (
|
||||
EditUserRolesAction, NewUserAction, ResetUserPasswordAction)
|
||||
from stacktask.api.models import Task
|
||||
from stacktask.api.v1 import tests
|
||||
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
|
||||
from stacktask.api.v1.tests import (FakeManager, setup_temp_cache,
|
||||
modify_dict_settings, StacktaskTestCase)
|
||||
|
||||
|
||||
@mock.patch('stacktask.actions.user_store.IdentityManager',
|
||||
FakeManager)
|
||||
class UserActionTests(TestCase):
|
||||
class UserActionTests(StacktaskTestCase):
|
||||
|
||||
def test_new_user(self):
|
||||
"""
|
||||
@ -711,3 +710,125 @@ class UserActionTests(TestCase):
|
||||
|
||||
self.assertEquals(
|
||||
project.roles[user.id], ['_member_', 'project_admin'])
|
||||
|
||||
def test_edit_user_roles_modified_settings(self):
|
||||
"""
|
||||
Tests that the role mappings do come from settings and that they
|
||||
are enforced.
|
||||
"""
|
||||
|
||||
project = mock.Mock()
|
||||
project.id = 'test_project_id'
|
||||
project.name = 'test_project'
|
||||
project.domain = 'default'
|
||||
project.roles = {'user_id': ['project_mod']}
|
||||
|
||||
user = mock.Mock()
|
||||
user.id = 'user_id'
|
||||
user.name = "test@example.com"
|
||||
user.email = "test@example.com"
|
||||
user.domain = 'default'
|
||||
|
||||
setup_temp_cache({'test_project': project}, {user.id: user})
|
||||
|
||||
task = Task.objects.create(
|
||||
ip_address="0.0.0.0",
|
||||
keystone_user={
|
||||
'roles': ['project_mod'],
|
||||
'project_id': 'test_project_id',
|
||||
'project_domain_id': 'default',
|
||||
})
|
||||
|
||||
data = {
|
||||
'domain_id': 'default',
|
||||
'user_id': 'user_id',
|
||||
'project_id': 'test_project_id',
|
||||
'roles': ['heat_stack_owner'],
|
||||
'remove': False
|
||||
}
|
||||
|
||||
action = EditUserRolesAction(data, task=task, order=1)
|
||||
|
||||
action.pre_approve()
|
||||
self.assertEquals(action.valid, True)
|
||||
|
||||
# Remove role from ROLES_MAPPING
|
||||
with self.modify_dict_settings(ROLES_MAPPING={
|
||||
'key_list': ['project_mod'],
|
||||
'operation': "remove",
|
||||
'value': 'heat_stack_owner'}):
|
||||
action.post_approve()
|
||||
self.assertEquals(action.valid, False)
|
||||
|
||||
token_data = {}
|
||||
action.submit(token_data)
|
||||
self.assertEquals(action.valid, False)
|
||||
|
||||
# After Settings Reset
|
||||
action.post_approve()
|
||||
self.assertEquals(action.valid, True)
|
||||
|
||||
token_data = {}
|
||||
action.submit(token_data)
|
||||
self.assertEquals(action.valid, True)
|
||||
|
||||
self.assertEquals(len(project.roles[user.id]), 2)
|
||||
self.assertEquals(set(project.roles[user.id]),
|
||||
set(['project_mod', 'heat_stack_owner']))
|
||||
|
||||
@modify_dict_settings(ROLES_MAPPING={'key_list': ['project_mod'],
|
||||
'operation': "append", 'value': 'new_role'})
|
||||
def test_edit_user_roles_modified_settings_add(self):
|
||||
"""
|
||||
Tests that the role mappings do come from settings and a new role
|
||||
added there will be allowed.
|
||||
"""
|
||||
|
||||
project = mock.Mock()
|
||||
project.id = 'test_project_id'
|
||||
project.name = 'test_project'
|
||||
project.domain = 'default'
|
||||
project.roles = {'user_id': ['project_mod']}
|
||||
|
||||
user = mock.Mock()
|
||||
user.id = 'user_id'
|
||||
user.name = "test@example.com"
|
||||
user.email = "test@example.com"
|
||||
user.domain = 'default'
|
||||
|
||||
setup_temp_cache({'test_project': project}, {user.id: user})
|
||||
|
||||
# Add a new role to the temp cache
|
||||
tests.temp_cache['roles']['new_role'] = 'new_role'
|
||||
|
||||
task = Task.objects.create(
|
||||
ip_address="0.0.0.0",
|
||||
keystone_user={
|
||||
'roles': ['project_mod'],
|
||||
'project_id': 'test_project_id',
|
||||
'project_domain_id': 'default',
|
||||
})
|
||||
|
||||
data = {
|
||||
'domain_id': 'default',
|
||||
'user_id': 'user_id',
|
||||
'project_id': 'test_project_id',
|
||||
'roles': ['new_role'],
|
||||
'remove': False
|
||||
}
|
||||
|
||||
action = EditUserRolesAction(data, task=task, order=1)
|
||||
|
||||
action.pre_approve()
|
||||
self.assertEquals(action.valid, True)
|
||||
|
||||
action.post_approve()
|
||||
self.assertEquals(action.valid, True)
|
||||
|
||||
token_data = {}
|
||||
action.submit(token_data)
|
||||
self.assertEquals(action.valid, True)
|
||||
|
||||
self.assertEquals(len(project.roles[user.id]), 2)
|
||||
self.assertEquals(set(project.roles[user.id]),
|
||||
set(['project_mod', 'new_role']))
|
||||
|
@ -12,9 +12,15 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
import mock
|
||||
import six
|
||||
import copy
|
||||
|
||||
from django.conf import settings
|
||||
from django.test.utils import override_settings
|
||||
from django.test import TestCase
|
||||
from rest_framework.test import APITestCase
|
||||
|
||||
temp_cache = {}
|
||||
|
||||
@ -112,6 +118,7 @@ class FakeManager(object):
|
||||
r = mock.Mock()
|
||||
r.name = role
|
||||
user.roles.append(r)
|
||||
|
||||
users.append(user)
|
||||
return users
|
||||
|
||||
@ -251,3 +258,165 @@ class FakeManager(object):
|
||||
def get_region(self, region_id):
|
||||
global temp_cache
|
||||
return temp_cache['regions'].get(region_id, None)
|
||||
|
||||
|
||||
class modify_dict_settings(override_settings):
|
||||
"""
|
||||
A decorator like djangos modify_settings and override_settings, but makes
|
||||
it possible to do those same operations on dict based settings.
|
||||
|
||||
The decorator will act after both override_settings and modify_settings.
|
||||
|
||||
Can be applied to test functions or StacktaskTestCase,
|
||||
StacktaskAPITestCase classes. In those two classes settings can also
|
||||
be modified using:
|
||||
|
||||
with self.modify_dict_settings(...):
|
||||
# code
|
||||
|
||||
Example Usage:
|
||||
@modify_dict_settings(ROLES_MAPPING=[
|
||||
{'key_list': ['project_mod'],
|
||||
'operation': 'remove',
|
||||
'value': 'heat_stack_owner'},
|
||||
{'key_list': ['project_admin'],
|
||||
'operation': 'append',
|
||||
'value': 'heat_stack_owner'},
|
||||
])
|
||||
or
|
||||
@modify_dict_settings(PROJECT_QUOTA_SIZES={
|
||||
'key_list': ['small', 'nova', 'instances'],
|
||||
'operations': 'override',
|
||||
'value': 11
|
||||
})
|
||||
|
||||
Available operations:
|
||||
Standard operations:
|
||||
- 'override': Either overrides or adds the value to the dictionary.
|
||||
- 'delete': Removes the value from the dictionary.
|
||||
|
||||
List operations:
|
||||
List operations expect that the accessed value in the dictionary is a list.
|
||||
- 'append': Add the specified values to the end of the list
|
||||
- 'prepend': Add the specifed values to the start of the list
|
||||
- 'remove': Remove the specified values from the list
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
if args:
|
||||
# Hack used when instantiating from SimpleTestCase.setUpClass.
|
||||
assert not kwargs
|
||||
self.operations = args[0]
|
||||
else:
|
||||
assert not args
|
||||
self.operations = list(kwargs.items())
|
||||
super(override_settings, self).__init__()
|
||||
|
||||
def save_options(self, test_func):
|
||||
if test_func._modified_dict_settings is None:
|
||||
test_func._modified_dict_settings = self.operations
|
||||
else:
|
||||
# Duplicate list to prevent subclasses from altering their parent.
|
||||
test_func._modified_dict_settings = list(
|
||||
test_func._modified_dict_settings) + self.operations
|
||||
|
||||
def disable(self):
|
||||
self.wrapped = self._wrapped
|
||||
super(modify_dict_settings, self).disable()
|
||||
|
||||
def enable(self):
|
||||
self.options = {}
|
||||
|
||||
self._wrapped = copy.deepcopy(settings._wrapped)
|
||||
|
||||
for name, operation_list in self.operations:
|
||||
try:
|
||||
value = self.options[name]
|
||||
except KeyError:
|
||||
value = getattr(settings, name, [])
|
||||
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError("Initial setting not dictionary.")
|
||||
|
||||
if not isinstance(operation_list, list):
|
||||
operation_list = [operation_list]
|
||||
|
||||
for operation in operation_list:
|
||||
op_type = operation['operation']
|
||||
|
||||
holding_dict = value
|
||||
|
||||
# Recursively find the dict we want
|
||||
key_len = len(operation['key_list'])
|
||||
final_key = operation['key_list'][0]
|
||||
|
||||
for i in range(key_len):
|
||||
current_key = operation['key_list'][i]
|
||||
if i == (key_len - 1):
|
||||
final_key = current_key
|
||||
else:
|
||||
try:
|
||||
holding_dict = holding_dict[current_key]
|
||||
except KeyError:
|
||||
holding_dict[current_key] = {}
|
||||
holding_dict = holding_dict[current_key]
|
||||
|
||||
if op_type == "override":
|
||||
holding_dict[final_key] = operation['value']
|
||||
elif op_type == "delete":
|
||||
del holding_dict[final_key]
|
||||
else:
|
||||
val = holding_dict.get(final_key, [])
|
||||
items = operation['value']
|
||||
|
||||
if not isinstance(items, list):
|
||||
items = [items]
|
||||
|
||||
if op_type == 'append':
|
||||
holding_dict[final_key] = val + [
|
||||
item for item in items if item not in val]
|
||||
elif op_type == 'prepend':
|
||||
holding_dict[final_key] = ([item for item in items if
|
||||
item not in val] + val)
|
||||
elif op_type == 'remove':
|
||||
holding_dict[final_key] = [
|
||||
item for item in val if item not in items]
|
||||
else:
|
||||
raise ValueError("Unsupported action: %s" % op_type)
|
||||
self.options[name] = value
|
||||
super(modify_dict_settings, self).enable()
|
||||
|
||||
|
||||
class TestCaseMixin(object):
|
||||
""" Mixin to add modify_dict_settings functions to test classes """
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super(StacktaskAPITestCase, cls).setUpClass()
|
||||
if cls._modified_dict_settings:
|
||||
cls._cls_modifyied_dict_context = override_settings(
|
||||
**cls._overridden_settings)
|
||||
cls._cls_modifyied_dict_context.enable()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
if hasattr(cls, '_cls_modified_dict_context'):
|
||||
cls._cls_modified_dict_context.disable()
|
||||
delattr(cls, '_cls_modified_dict_context')
|
||||
super(StacktaskAPITestCase, cls).tearDownClass()
|
||||
|
||||
def modify_dict_settings(self, **kwargs):
|
||||
return modify_dict_settings(**kwargs)
|
||||
|
||||
|
||||
class StacktaskTestCase(TestCase, TestCaseMixin):
|
||||
"""
|
||||
TestCase override that has support for @modify_dict_settings as a
|
||||
class decorator and internal function
|
||||
"""
|
||||
|
||||
|
||||
class StacktaskAPITestCase(APITestCase, TestCaseMixin):
|
||||
"""
|
||||
APITestCase override that has support for @modify_dict_settings as a
|
||||
class decorator, and internal function
|
||||
"""
|
||||
|
@ -26,7 +26,8 @@ from rest_framework import status
|
||||
from rest_framework.test import APITestCase
|
||||
|
||||
from stacktask.api.models import Task, Token
|
||||
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
|
||||
from stacktask.api.v1.tests import (FakeManager, setup_temp_cache,
|
||||
modify_dict_settings)
|
||||
|
||||
|
||||
@mock.patch('stacktask.actions.user_store.IdentityManager',
|
||||
@ -1022,6 +1023,12 @@ class AdminAPITests(APITestCase):
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
@modify_dict_settings(TASK_SETTINGS={
|
||||
'key_list': ['reset_password', 'action_settings',
|
||||
'ResetUserPasswordAction', 'blacklisted_roles'],
|
||||
'operation': 'append',
|
||||
'value': ['admin']
|
||||
})
|
||||
def test_reset_admin(self):
|
||||
"""
|
||||
Ensure that you cannot issue a password reset for an
|
||||
|
@ -15,15 +15,15 @@
|
||||
import mock
|
||||
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APITestCase
|
||||
|
||||
from stacktask.api.models import Task, Token
|
||||
from stacktask.api.v1.tests import FakeManager, setup_temp_cache
|
||||
from stacktask.api.v1.tests import (FakeManager, setup_temp_cache,
|
||||
StacktaskAPITestCase)
|
||||
|
||||
|
||||
@mock.patch('stacktask.actions.user_store.IdentityManager',
|
||||
FakeManager)
|
||||
class TaskViewTests(APITestCase):
|
||||
class TaskViewTests(StacktaskAPITestCase):
|
||||
"""
|
||||
Tests to ensure the approval/token workflow does what is
|
||||
expected with the given TaskViews. These test don't check
|
||||
|
@ -122,9 +122,6 @@ DEFAULT_ACTION_SETTINGS = {
|
||||
'NewUserAction': {
|
||||
'allowed_roles': ['project_mod', 'project_admin', "_member_"]
|
||||
},
|
||||
'ResetUserPasswordAction': {
|
||||
'blacklisted_roles': ['admin']
|
||||
},
|
||||
'NewDefaultNetworkAction': {
|
||||
'RegionOne': {
|
||||
'DNS_NAMESERVERS': ['193.168.1.2', '193.168.1.3'],
|
||||
@ -145,14 +142,6 @@ DEFAULT_ACTION_SETTINGS = {
|
||||
'subnet_name': 'somesubnet'
|
||||
},
|
||||
},
|
||||
'AddDefaultUsersToProjectAction': {
|
||||
'default_users': [
|
||||
'admin',
|
||||
],
|
||||
'default_roles': [
|
||||
'admin',
|
||||
],
|
||||
},
|
||||
'SetProjectQuotaAction': {
|
||||
'regions': {
|
||||
'RegionOne': {
|
||||
|
Loading…
x
Reference in New Issue
Block a user