Skip to content

Commit

Permalink
Some minor neatening up of test code
Browse files Browse the repository at this point in the history
  • Loading branch information
@mark-saeon
mark-saeon committed Aug 17, 2018
1 parent 6fbd754 commit 64f62ce
Show file tree
Hide file tree
Showing 9 changed files with 616 additions and 609 deletions.
44 changes: 24 additions & 20 deletions ckanext/metadata/tests/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import pkg_resources
from collections import deque
import traceback
from nose.tools import nottest

from ckan.tests import factories as ckan_factories
from ckan.tests.helpers import FunctionalTestBase, call_action
Expand Down Expand Up @@ -125,6 +126,22 @@ def assert_group_has_member(group_id, object_id, object_table, capacity='public'
assert member.state == state


def assert_metadata_record_has_validation_models(metadata_record_id, *metadata_model_names):
"""
Check that the given record has the expected set of validation models.
"""
validation_model_list = call_action('metadata_record_validation_model_list', id=metadata_record_id)
assert set(validation_model_list) == set(metadata_model_names)


def assert_metadata_model_has_dependent_records(metadata_model_id, *metadata_record_ids):
"""
Check that the given model has the expected set of dependent records.
"""
dependent_record_list = call_action('metadata_model_dependent_record_list', id=metadata_model_id)
assert set(dependent_record_list) == set(metadata_record_ids)


def assert_error(error_dict, key, pattern):
"""
Check that the error dictionary contains the given key with the corresponding error message regex.
Expand Down Expand Up @@ -165,8 +182,9 @@ def setup(self):
self.normal_user = ckan_factories.User()
self.sysadmin_user = ckan_factories.Sysadmin()

def _test_action(self, action_name, should_error=False, exception_class=tk.ValidationError,
sysadmin=False, check_auth=False, **kwargs):
@nottest
def test_action(self, action_name, should_error=False, exception_class=tk.ValidationError,
sysadmin=False, check_auth=False, **kwargs):
"""
Test an API action.
:param action_name: action function name, e.g. 'metadata_record_create'
Expand Down Expand Up @@ -219,7 +237,7 @@ def _test_action(self, action_name, should_error=False, exception_class=tk.Valid

return result, obj

def _assert_validate_activity_logged(self, metadata_record_id, *validation_models, **validation_errors):
def assert_validate_activity_logged(self, metadata_record_id, *validation_models, **validation_errors):
"""
:param validation_models: iterable of metadata model dictionaries
:param validation_errors: dictionary mapping metadata model keys to expected error patterns (regex's)
Expand All @@ -241,7 +259,7 @@ def _assert_validate_activity_logged(self, metadata_record_id, *validation_model
for error_key, error_pattern in validation_errors.items():
assert_error(logged_errors, error_key, error_pattern)

def _assert_invalidate_activity_logged(self, metadata_record_id, trigger_action, trigger_object):
def assert_invalidate_activity_logged(self, metadata_record_id, trigger_action, trigger_object):
activity_dict = call_action('metadata_record_validation_activity_show', id=metadata_record_id)
assert activity_dict['user_id'] == self.normal_user['id']
assert activity_dict['object_id'] == metadata_record_id
Expand All @@ -252,22 +270,8 @@ def _assert_invalidate_activity_logged(self, metadata_record_id, trigger_action,
'trigger_object_id': trigger_object.id if trigger_object else None,
}

def _assert_metadata_record_has_validation_models(self, metadata_record_id, *metadata_model_names):
"""
Check that the given record has the expected set of validation models.
"""
validation_model_list = call_action('metadata_record_validation_model_list', id=metadata_record_id)
assert set(validation_model_list) == set(metadata_model_names)

def _assert_metadata_model_has_dependent_records(self, metadata_model_id, *metadata_record_ids):
"""
Check that the given model has the expected set of dependent records.
"""
dependent_record_list = call_action('metadata_model_dependent_record_list', id=metadata_model_id)
assert set(dependent_record_list) == set(metadata_record_ids)

def _assert_workflow_activity_logged(self, action_suffix, metadata_record_id, workflow_state_id,
*jsonpatch_ids, **workflow_errors):
def assert_workflow_activity_logged(self, action_suffix, metadata_record_id, workflow_state_id,
*jsonpatch_ids, **workflow_errors):
"""
:param action_suffix: 'transition' | 'revert' | 'override'
:param workflow_errors: dictionary mapping workflow annotation (flattened) keys to expected error patterns
Expand Down
36 changes: 18 additions & 18 deletions ckanext/metadata/tests/test_infrastructure_actions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ def test_create_valid(self):
'title': 'Test Infrastructure',
'description': 'This is a test infrastructure',
}
result, obj = self._test_action('infrastructure_create', **input_dict)
result, obj = self.test_action('infrastructure_create', **input_dict)
assert obj.type == 'infrastructure'
assert obj.is_organization == False
assert_object_matches_dict(obj, input_dict)
Expand All @@ -30,26 +30,26 @@ def test_create_valid_sysadmin_setid(self):
'id': make_uuid(),
'name': 'test-infrastructure',
}
result, obj = self._test_action('infrastructure_create', sysadmin=True, check_auth=True, **input_dict)
result, obj = self.test_action('infrastructure_create', sysadmin=True, check_auth=True, **input_dict)
assert obj.type == 'infrastructure'
assert obj.is_organization == False
assert_object_matches_dict(obj, input_dict)

def test_create_invalid_duplicate_name(self):
infrastructure = ckanext_factories.Infrastructure()
result, obj = self._test_action('infrastructure_create', should_error=True,
name=infrastructure['name'])
result, obj = self.test_action('infrastructure_create', should_error=True,
name=infrastructure['name'])
assert_error(result, 'name', 'Group name already exists in database')

def test_create_invalid_nonsysadmin_setid(self):
result, obj = self._test_action('infrastructure_create', should_error=True, check_auth=True,
id=make_uuid())
result, obj = self.test_action('infrastructure_create', should_error=True, check_auth=True,
id=make_uuid())
assert_error(result, 'id', 'The input field id was not expected.')

def test_create_invalid_sysadmin_duplicate_id(self):
infrastructure = ckanext_factories.Infrastructure()
result, obj = self._test_action('infrastructure_create', should_error=True, sysadmin=True, check_auth=True,
id=infrastructure['id'])
result, obj = self.test_action('infrastructure_create', should_error=True, sysadmin=True, check_auth=True,
id=infrastructure['id'])
assert_error(result, 'id', 'Already exists: Group')

def test_update_valid(self):
Expand All @@ -60,7 +60,7 @@ def test_update_valid(self):
'title': 'Updated Test Infrastructure',
'description': 'Updated test infrastructure',
}
result, obj = self._test_action('infrastructure_update', **input_dict)
result, obj = self.test_action('infrastructure_update', **input_dict)
assert obj.type == 'infrastructure'
assert obj.is_organization == False
assert_object_matches_dict(obj, input_dict)
Expand All @@ -71,7 +71,7 @@ def test_update_valid_partial(self):
'id': infrastructure['id'],
'title': 'Updated Test Infrastructure',
}
result, obj = self._test_action('infrastructure_update', **input_dict)
result, obj = self.test_action('infrastructure_update', **input_dict)
assert obj.type == 'infrastructure'
assert obj.is_organization == False
assert obj.title == input_dict['title']
Expand All @@ -85,7 +85,7 @@ def test_update_invalid_duplicate_name(self):
'id': infrastructure1['id'],
'name': infrastructure2['name'],
}
result, obj = self._test_action('infrastructure_update', should_error=True, **input_dict)
result, obj = self.test_action('infrastructure_update', should_error=True, **input_dict)
assert_error(result, 'name', 'Group name already exists in database')

def test_update_invalid_hierarchy_not_allowed(self):
Expand All @@ -95,25 +95,25 @@ def test_update_invalid_hierarchy_not_allowed(self):
'id': infrastructure1['id'],
'groups': [{'name': infrastructure2['name']}],
}
result, obj = self._test_action('infrastructure_update', should_error=True, **input_dict)
result, obj = self.test_action('infrastructure_update', should_error=True, **input_dict)
assert_error(result, '__junk', 'The input field .*groups.* was not expected.')

def test_delete_valid(self):
infrastructure = ckanext_factories.Infrastructure()
self._test_action('infrastructure_delete',
id=infrastructure['id'])
self.test_action('infrastructure_delete',
id=infrastructure['id'])

def test_delete_with_dependencies(self):
infrastructure = ckanext_factories.Infrastructure()
metadata_model = ckanext_factories.MetadataModel(infrastructure_id=infrastructure['id'])
metadata_record = ckanext_factories.MetadataRecord(infrastructures=[{'id': infrastructure['id']}])

result, obj = self._test_action('infrastructure_delete', should_error=True,
id=infrastructure['id'])
result, obj = self.test_action('infrastructure_delete', should_error=True,
id=infrastructure['id'])
assert_error(result, 'message', 'Infrastructure has dependent metadata records')
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'active'

call_action('metadata_record_delete', id=metadata_record['id'])
self._test_action('infrastructure_delete',
id=infrastructure['id'])
self.test_action('infrastructure_delete',
id=infrastructure['id'])
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'deleted'
48 changes: 24 additions & 24 deletions ckanext/metadata/tests/test_metadata_collection_actions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ def test_create_valid(self):
'description': 'This is a test metadata collection',
'organization_id': organization['id'],
}
result, obj = self._test_action('metadata_collection_create', **input_dict)
result, obj = self.test_action('metadata_collection_create', **input_dict)
assert obj.type == 'metadata_collection'
assert obj.is_organization == False
assert_group_has_extra(obj.id, 'organization_id', input_dict['organization_id'])
Expand All @@ -36,7 +36,7 @@ def test_create_valid_organization_byname(self):
'name': 'test-metadata-collection',
'organization_id': organization['name'],
}
result, obj = self._test_action('metadata_collection_create', **input_dict)
result, obj = self.test_action('metadata_collection_create', **input_dict)
assert obj.type == 'metadata_collection'
assert obj.is_organization == False
assert obj.name == input_dict['name']
Expand All @@ -49,7 +49,7 @@ def test_create_valid_sysadmin_setid(self):
'name': 'test-metadata-collection',
'organization_id': organization['id'],
}
result, obj = self._test_action('metadata_collection_create', sysadmin=True, check_auth=True, **input_dict)
result, obj = self.test_action('metadata_collection_create', sysadmin=True, check_auth=True, **input_dict)
assert obj.type == 'metadata_collection'
assert obj.is_organization == False
assert_group_has_extra(obj.id, 'organization_id', input_dict['organization_id'])
Expand All @@ -58,31 +58,31 @@ def test_create_valid_sysadmin_setid(self):

def test_create_invalid_duplicate_name(self):
metadata_collection = ckanext_factories.MetadataCollection()
result, obj = self._test_action('metadata_collection_create', should_error=True,
name=metadata_collection['name'])
result, obj = self.test_action('metadata_collection_create', should_error=True,
name=metadata_collection['name'])
assert_error(result, 'name', 'Group name already exists in database')

def test_create_invalid_nonsysadmin_setid(self):
result, obj = self._test_action('metadata_collection_create', should_error=True, check_auth=True,
id=make_uuid())
result, obj = self.test_action('metadata_collection_create', should_error=True, check_auth=True,
id=make_uuid())
assert_error(result, 'id', 'The input field id was not expected.')

def test_create_invalid_sysadmin_duplicate_id(self):
metadata_collection = ckanext_factories.MetadataCollection()
result, obj = self._test_action('metadata_collection_create', should_error=True, sysadmin=True, check_auth=True,
id=metadata_collection['id'])
result, obj = self.test_action('metadata_collection_create', should_error=True, sysadmin=True, check_auth=True,
id=metadata_collection['id'])
assert_error(result, 'id', 'Already exists: Group')

def test_create_invalid_bad_organization(self):
result, obj = self._test_action('metadata_collection_create', should_error=True,
organization_id='foo')
result, obj = self.test_action('metadata_collection_create', should_error=True,
organization_id='foo')
assert_error(result, 'organization_id', 'Not found: Organization')

def test_create_invalid_deleted_organization(self):
organization = ckan_factories.Organization()
call_action('organization_delete', id=organization['id'])
result, obj = self._test_action('metadata_collection_create', should_error=True,
organization_id=organization['id'])
result, obj = self.test_action('metadata_collection_create', should_error=True,
organization_id=organization['id'])
assert_error(result, 'organization_id', 'Not found: Organization')

def test_update_valid(self):
Expand All @@ -93,7 +93,7 @@ def test_update_valid(self):
'title': 'Updated Test Metadata Collection',
'description': 'Updated test metadata collection',
}
result, obj = self._test_action('metadata_collection_update', **input_dict)
result, obj = self.test_action('metadata_collection_update', **input_dict)
assert obj.type == 'metadata_collection'
assert obj.is_organization == False
assert_object_matches_dict(obj, input_dict)
Expand All @@ -105,7 +105,7 @@ def test_update_valid_partial(self):
'id': metadata_collection['id'],
'title': 'Updated Test Metadata Collection',
}
result, obj = self._test_action('metadata_collection_update', **input_dict)
result, obj = self.test_action('metadata_collection_update', **input_dict)
assert obj.type == 'metadata_collection'
assert obj.is_organization == False
assert obj.title == input_dict['title']
Expand All @@ -120,7 +120,7 @@ def test_update_invalid_duplicate_name(self):
'id': metadata_collection1['id'],
'name': metadata_collection2['name'],
}
result, obj = self._test_action('metadata_collection_update', should_error=True, **input_dict)
result, obj = self.test_action('metadata_collection_update', should_error=True, **input_dict)
assert_error(result, 'name', 'Group name already exists in database')

def test_update_invalid_hierarchy_not_allowed(self):
Expand All @@ -130,7 +130,7 @@ def test_update_invalid_hierarchy_not_allowed(self):
'id': metadata_collection1['id'],
'groups': [{'name': metadata_collection2['name']}],
}
result, obj = self._test_action('metadata_collection_update', should_error=True, **input_dict)
result, obj = self.test_action('metadata_collection_update', should_error=True, **input_dict)
assert_error(result, '__junk', 'The input field .*groups.* was not expected.')

def test_update_invalid_cannot_change_organization(self):
Expand All @@ -140,24 +140,24 @@ def test_update_invalid_cannot_change_organization(self):
'id': metadata_collection['id'],
'organization_id': organization['id'],
}
result, obj = self._test_action('metadata_collection_update', should_error=True, **input_dict)
result, obj = self.test_action('metadata_collection_update', should_error=True, **input_dict)
assert_error(result, 'organization_id', 'The input field organization_id was not expected.')

def test_delete_valid(self):
metadata_collection = ckanext_factories.MetadataCollection()
self._test_action('metadata_collection_delete',
id=metadata_collection['id'])
self.test_action('metadata_collection_delete',
id=metadata_collection['id'])

def test_delete_with_dependencies(self):
metadata_collection = ckanext_factories.MetadataCollection()
metadata_record = ckanext_factories.MetadataRecord(
owner_org=metadata_collection['organization_id'],
metadata_collection_id=metadata_collection['id'])

result, obj = self._test_action('metadata_collection_delete', should_error=True,
id=metadata_collection['id'])
result, obj = self.test_action('metadata_collection_delete', should_error=True,
id=metadata_collection['id'])
assert_error(result, 'message', 'Metadata collection has dependent metadata records')

call_action('metadata_record_delete', id=metadata_record['id'])
self._test_action('metadata_collection_delete',
id=metadata_collection['id'])
self.test_action('metadata_collection_delete',
id=metadata_collection['id'])
Loading

0 comments on commit 64f62ce

Please sign in to comment.