From 44542e861a577d0c5c2b9ec6ebeaee105f32ce8d Mon Sep 17 00:00:00 2001
From: Mark Jacobson <35484465+mark-saeon@users.noreply.github.com>
Date: Tue, 3 Jul 2018 11:30:16 +0200
Subject: [PATCH] Simplify call signature of main test method
---
ckanext/metadata/tests/__init__.py | 26 +++-
.../tests/test_infrastructure_actions.py | 39 ++---
.../tests/test_metadata_collection_actions.py | 53 +++----
.../tests/test_metadata_model_actions.py | 134 ++++++------------
.../tests/test_metadata_record_actions.py | 102 +++++--------
.../tests/test_metadata_schema_actions.py | 101 +++++--------
.../tests/test_metadata_validation.py | 2 +
.../tests/test_organization_actions.py | 16 +--
.../tests/test_workflow_metric_actions.py | 45 ++----
.../tests/test_workflow_rule_actions.py | 54 +++----
.../tests/test_workflow_state_actions.py | 104 +++++---------
.../tests/test_workflow_transition_actions.py | 60 +++-----
12 files changed, 251 insertions(+), 485 deletions(-)
diff --git a/ckanext/metadata/tests/__init__.py b/ckanext/metadata/tests/__init__.py
index f4f12a3..18cb8b3 100644
--- a/ckanext/metadata/tests/__init__.py
+++ b/ckanext/metadata/tests/__init__.py
@@ -10,6 +10,20 @@
import ckan.plugins.toolkit as tk
import ckan.model as ckan_model
import ckanext.metadata.model.setup as ckanext_setup
+from ckanext.metadata import model as ckanext_model
+
+_model_map = {
+ 'organization': ckan_model.Group,
+ 'infrastructure': ckan_model.Group,
+ 'metadata_collection': ckan_model.Group,
+ 'metadata_record': ckan_model.Package,
+ 'metadata_model': ckanext_model.MetadataModel,
+ 'metadata_schema': ckanext_model.MetadataSchema,
+ 'workflow_state': ckanext_model.WorkflowState,
+ 'workflow_transition': ckanext_model.WorkflowTransition,
+ 'workflow_metric': ckanext_model.WorkflowMetric,
+ 'workflow_rule': ckanext_model.WorkflowRule,
+}
def make_uuid():
@@ -120,9 +134,10 @@ def setup(self):
self.normal_user = ckan_factories.User()
self.sysadmin_user = ckan_factories.Sysadmin()
- def _test_action(self, method, model_name, model_class=None, exception_class=None, sysadmin=False, check_auth=False, **kwargs):
+ def _test_action(self, action_name, should_error=False, exception_class=tk.ValidationError,
+ sysadmin=False, check_auth=False, **kwargs):
- method_name = model_name + '_' + method
+ model, method = action_name.rsplit('_', 1)
user = self.sysadmin_user if sysadmin else self.normal_user
context = {
'user': user['name'],
@@ -131,7 +146,7 @@ def _test_action(self, method, model_name, model_class=None, exception_class=Non
obj = None
try:
- result = call_action(method_name, context, **kwargs)
+ result = call_action(action_name, context, **kwargs)
except exception_class, e:
if exception_class is tk.ValidationError:
result = e.error_dict
@@ -140,14 +155,15 @@ def _test_action(self, method, model_name, model_class=None, exception_class=Non
except Exception, e:
assert False, "Unexpected exception %s: %s" % (type(e), e)
else:
- if exception_class:
+ if should_error:
assert False, str(exception_class) + " was not raised"
finally:
# close the session to ensure that we're not just getting the obj from
# memory but are reloading it from the DB
ckan_model.Session.close_all()
- if not exception_class:
+ if not should_error:
+ model_class = _model_map[model]
if method in ('create', 'update', 'show'):
assert 'id' in result
obj = model_class.get(result['id'])
diff --git a/ckanext/metadata/tests/test_infrastructure_actions.py b/ckanext/metadata/tests/test_infrastructure_actions.py
index ed5a06f..b941c5f 100644
--- a/ckanext/metadata/tests/test_infrastructure_actions.py
+++ b/ckanext/metadata/tests/test_infrastructure_actions.py
@@ -1,7 +1,5 @@
# encoding: utf-8
-from ckan import model as ckan_model
-from ckan.plugins import toolkit as tk
from ckan.tests.helpers import call_action
from ckanext.metadata import model as ckanext_model
@@ -22,8 +20,7 @@ def test_create_valid(self):
'title': 'Test Infrastructure',
'description': 'This is a test infrastructure',
}
- result, obj = self._test_action('create', 'infrastructure',
- model_class=ckan_model.Group, **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)
@@ -33,30 +30,25 @@ def test_create_valid_sysadmin_setid(self):
'id': make_uuid(),
'name': 'test-infrastructure',
}
- result, obj = self._test_action('create', 'infrastructure',
- model_class=ckan_model.Group,
- 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('create', 'infrastructure',
- exception_class=tk.ValidationError,
+ 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('create', 'infrastructure',
- exception_class=tk.ValidationError, check_auth=True,
+ 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('create', 'infrastructure',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ 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')
@@ -68,8 +60,7 @@ def test_update_valid(self):
'title': 'Updated Test Infrastructure',
'description': 'Updated test infrastructure',
}
- result, obj = self._test_action('update', 'infrastructure',
- model_class=ckan_model.Group, **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)
@@ -80,8 +71,7 @@ def test_update_valid_partial(self):
'id': infrastructure['id'],
'title': 'Updated Test Infrastructure',
}
- result, obj = self._test_action('update', 'infrastructure',
- model_class=ckan_model.Group, **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']
@@ -95,8 +85,7 @@ def test_update_invalid_duplicate_name(self):
'id': infrastructure1['id'],
'name': infrastructure2['name'],
}
- result, obj = self._test_action('update', 'infrastructure',
- exception_class=tk.ValidationError, **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):
@@ -106,14 +95,12 @@ def test_update_invalid_hierarchy_not_allowed(self):
'id': infrastructure1['id'],
'groups': [{'name': infrastructure2['name']}],
}
- result, obj = self._test_action('update', 'infrastructure',
- exception_class=tk.ValidationError, **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('delete', 'infrastructure',
- model_class=ckan_model.Group,
+ self._test_action('infrastructure_delete',
id=infrastructure['id'])
def test_delete_with_dependencies(self):
@@ -121,14 +108,12 @@ def test_delete_with_dependencies(self):
metadata_model = ckanext_factories.MetadataModel(infrastructure_id=infrastructure['id'])
metadata_record = ckanext_factories.MetadataRecord(infrastructures=[{'id': infrastructure['id']}])
- result, obj = self._test_action('delete', 'infrastructure',
- exception_class=tk.ValidationError,
+ 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('delete', 'infrastructure',
- model_class=ckan_model.Group,
+ self._test_action('infrastructure_delete',
id=infrastructure['id'])
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'deleted'
diff --git a/ckanext/metadata/tests/test_metadata_collection_actions.py b/ckanext/metadata/tests/test_metadata_collection_actions.py
index fdada14..dad8f36 100644
--- a/ckanext/metadata/tests/test_metadata_collection_actions.py
+++ b/ckanext/metadata/tests/test_metadata_collection_actions.py
@@ -1,7 +1,5 @@
# encoding: utf-8
-from ckan import model as ckan_model
-from ckan.plugins import toolkit as tk
from ckan.tests import factories as ckan_factories
from ckan.tests.helpers import call_action
@@ -25,8 +23,7 @@ def test_create_valid(self):
'description': 'This is a test metadata collection',
'organization_id': organization['id'],
}
- result, obj = self._test_action('create', 'metadata_collection',
- model_class=ckan_model.Group, **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'])
@@ -39,8 +36,7 @@ def test_create_valid_organization_byname(self):
'name': 'test-metadata-collection',
'organization_id': organization['name'],
}
- result, obj = self._test_action('create', 'metadata_collection',
- model_class=ckan_model.Group, **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']
@@ -53,9 +49,7 @@ def test_create_valid_sysadmin_setid(self):
'name': 'test-metadata-collection',
'organization_id': organization['id'],
}
- result, obj = self._test_action('create', 'metadata_collection',
- model_class=ckan_model.Group,
- 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'])
@@ -64,35 +58,30 @@ def test_create_valid_sysadmin_setid(self):
def test_create_invalid_duplicate_name(self):
metadata_collection = ckanext_factories.MetadataCollection()
- result, obj = self._test_action('create', 'metadata_collection',
- exception_class=tk.ValidationError,
+ 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('create', 'metadata_collection',
- exception_class=tk.ValidationError, check_auth=True,
+ 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('create', 'metadata_collection',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ 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('create', 'metadata_collection',
- exception_class=tk.ValidationError,
+ 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('create', 'metadata_collection',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_collection_create', should_error=True,
organization_id=organization['id'])
assert_error(result, 'organization_id', 'Not found: Organization')
@@ -104,8 +93,7 @@ def test_update_valid(self):
'title': 'Updated Test Metadata Collection',
'description': 'Updated test metadata collection',
}
- result, obj = self._test_action('update', 'metadata_collection',
- model_class=ckan_model.Group, **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)
@@ -117,8 +105,7 @@ def test_update_valid_partial(self):
'id': metadata_collection['id'],
'title': 'Updated Test Metadata Collection',
}
- result, obj = self._test_action('update', 'metadata_collection',
- model_class=ckan_model.Group, **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']
@@ -133,8 +120,7 @@ def test_update_invalid_duplicate_name(self):
'id': metadata_collection1['id'],
'name': metadata_collection2['name'],
}
- result, obj = self._test_action('update', 'metadata_collection',
- exception_class=tk.ValidationError, **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):
@@ -144,8 +130,7 @@ def test_update_invalid_hierarchy_not_allowed(self):
'id': metadata_collection1['id'],
'groups': [{'name': metadata_collection2['name']}],
}
- result, obj = self._test_action('update', 'metadata_collection',
- exception_class=tk.ValidationError, **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):
@@ -155,14 +140,12 @@ def test_update_invalid_cannot_change_organization(self):
'id': metadata_collection['id'],
'organization_id': organization['id'],
}
- result, obj = self._test_action('update', 'metadata_collection',
- exception_class=tk.ValidationError, **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('delete', 'metadata_collection',
- model_class=ckan_model.Group,
+ self._test_action('metadata_collection_delete',
id=metadata_collection['id'])
def test_delete_with_dependencies(self):
@@ -171,12 +154,10 @@ def test_delete_with_dependencies(self):
owner_org=metadata_collection['organization_id'],
metadata_collection_id=metadata_collection['id'])
- result, obj = self._test_action('delete', 'metadata_collection',
- exception_class=tk.ValidationError,
+ 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('delete', 'metadata_collection',
- model_class=ckan_model.Group,
- id=metadata_collection['id'])
\ No newline at end of file
+ self._test_action('metadata_collection_delete',
+ id=metadata_collection['id'])
diff --git a/ckanext/metadata/tests/test_metadata_model_actions.py b/ckanext/metadata/tests/test_metadata_model_actions.py
index 159dc40..81bb303 100644
--- a/ckanext/metadata/tests/test_metadata_model_actions.py
+++ b/ckanext/metadata/tests/test_metadata_model_actions.py
@@ -2,7 +2,6 @@
import json
-from ckan.plugins import toolkit as tk
from ckan.tests import factories as ckan_factories
from ckan.tests.helpers import call_action
@@ -75,8 +74,7 @@ def test_create_valid(self):
'infrastructure_id': '',
'model_json': '{ "testkey": "testvalue" }',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.name == generate_name(metadata_schema['name'], '', '')
@@ -89,8 +87,7 @@ def test_create_valid_setname(self):
'infrastructure_id': '',
'model_json': '{ "testkey": "testvalue" }',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_with_organization_byname(self):
@@ -102,8 +99,7 @@ def test_create_valid_with_organization_byname(self):
'infrastructure_id': '',
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert obj.metadata_schema_id == metadata_schema['id']
assert obj.organization_id == organization['id']
assert obj.infrastructure_id is None
@@ -118,8 +114,7 @@ def test_create_valid_with_infrastructure_byname(self):
'infrastructure_id': infrastructure['name'],
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert obj.metadata_schema_id == metadata_schema['id']
assert obj.organization_id is None
assert obj.infrastructure_id == infrastructure['id']
@@ -134,9 +129,7 @@ def test_create_valid_sysadmin_setid(self):
'infrastructure_id': '',
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('metadata_model_create', sysadmin=True, check_auth=True, **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_same_schema_different_organization(self):
@@ -149,8 +142,7 @@ def test_create_valid_same_schema_different_organization(self):
'infrastructure_id': '',
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_same_schema_different_infrastructure(self):
@@ -163,8 +155,7 @@ def test_create_valid_same_schema_different_infrastructure(self):
'infrastructure_id': infrastructure2['id'],
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_same_organization_different_schema(self):
@@ -177,8 +168,7 @@ def test_create_valid_same_organization_different_schema(self):
'infrastructure_id': '',
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_same_infrastructure_different_schema(self):
@@ -191,8 +181,7 @@ def test_create_valid_same_infrastructure_different_schema(self):
'infrastructure_id': metadata_model['infrastructure_id'],
'model_json': '',
}
- result, obj = self._test_action('create', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_invalidate_records_matching_schema(self):
@@ -289,23 +278,20 @@ def test_create_no_invalidate_records_different_infrastructure_2(self):
def test_create_invalid_duplicate_name(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
name=metadata_model['name'])
assert_error(result, 'name', 'Duplicate name: Metadata Model')
def test_create_invalid_duplicate_schema(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
metadata_schema_id=metadata_model['metadata_schema_id'])
assert_error(result, '__after', 'Unique constraint violation')
def test_create_invalid_duplicate_schema_organization(self):
organization = ckan_factories.Organization()
metadata_model = ckanext_factories.MetadataModel(organization_id=organization['id'])
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
metadata_schema_id=metadata_model['metadata_schema_id'],
organization_id=metadata_model['organization_id'])
assert_error(result, '__after', 'Unique constraint violation')
@@ -313,8 +299,7 @@ def test_create_invalid_duplicate_schema_organization(self):
def test_create_invalid_duplicate_schema_infrastructure(self):
infrastructure = ckanext_factories.Infrastructure()
metadata_model = ckanext_factories.MetadataModel(infrastructure_id=infrastructure['id'])
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
metadata_schema_id=metadata_model['metadata_schema_id'],
infrastructure_id=metadata_model['infrastructure_id'])
assert_error(result, '__after', 'Unique constraint violation')
@@ -322,61 +307,52 @@ def test_create_invalid_duplicate_schema_infrastructure(self):
def test_create_invalid_with_organization_and_infrastructure(self):
organization = ckan_factories.Organization()
infrastructure = ckanext_factories.Infrastructure()
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
organization_id=organization['id'],
infrastructure_id=infrastructure['id'])
assert_error(result, '__after',
'A metadata model may be associated with either an organization or an infrastructure but not both.')
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('metadata_model_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_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('metadata_model_create', should_error=True, sysadmin=True, check_auth=True,
id=metadata_model['id'])
assert_error(result, 'id', 'Already exists: Metadata Model')
def test_create_invalid_not_json(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
model_json='not json')
assert_error(result, 'model_json', 'JSON decode error')
def test_create_invalid_not_json_dict(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
model_json='[1,2,3]')
assert_error(result, 'model_json', 'Expecting a JSON dictionary')
def test_create_invalid_not_json_schema(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
model_json='{"type": "foo"}')
assert_error(result, 'model_json', 'Invalid JSON schema')
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('metadata_model_create', should_error=True)
assert_error(result, 'metadata_schema_id', 'Missing parameter')
assert_error(result, 'organization_id', 'Missing parameter')
assert_error(result, 'infrastructure_id', 'Missing parameter')
assert_error(result, 'model_json', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
metadata_schema_id='')
assert_error(result, 'metadata_schema_id', 'Missing value')
def test_create_invalid_bad_references(self):
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
metadata_schema_id='a',
organization_id='b',
infrastructure_id='c')
@@ -392,8 +368,7 @@ def test_create_invalid_deleted_references(self):
call_action('organization_delete', id=organization['id'])
call_action('infrastructure_delete', id=infrastructure['id'])
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
metadata_schema_id=metadata_schema['id'],
organization_id=organization['id'],
infrastructure_id=infrastructure['id'])
@@ -413,8 +388,7 @@ def test_update_valid(self):
'infrastructure_id': '',
'model_json': '{ "testkey": "newtestvalue" }',
}
- result, obj = self._test_action('update', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.name == generate_name(metadata_schema['name'], '', '')
@@ -428,8 +402,7 @@ def test_update_valid_partial(self):
'infrastructure_id': '',
'model_json': '{ "testkey": "newtestvalue" }',
}
- result, obj = self._test_action('update', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.title == metadata_model['title']
assert obj.description == metadata_model['description']
@@ -444,8 +417,7 @@ def test_update_valid_set_organization(self):
'infrastructure_id': '',
'model_json': '',
}
- result, obj = self._test_action('update', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
metadata_schema = ckanext_model.MetadataSchema.get(metadata_model['metadata_schema_id'])
assert obj.name == generate_name(metadata_schema.name, organization['name'], '')
@@ -460,8 +432,7 @@ def test_update_valid_set_infrastructure(self):
'infrastructure_id': infrastructure['id'],
'model_json': '',
}
- result, obj = self._test_action('update', 'metadata_model',
- model_class=ckanext_model.MetadataModel, **input_dict)
+ result, obj = self._test_action('metadata_model_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
metadata_schema = ckanext_model.MetadataSchema.get(metadata_model['metadata_schema_id'])
assert obj.name == generate_name(metadata_schema.name, '', infrastructure['name'])
@@ -627,14 +598,12 @@ def test_update_invalid_duplicate_name(self):
'id': metadata_model1['id'],
'name': metadata_model2['name'],
}
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_model_update', should_error=True, **input_dict)
assert_error(result, 'name', 'Duplicate name: Metadata Model')
def test_update_invalid_missing_params(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'])
assert_error(result, 'metadata_schema_id', 'Missing parameter')
assert_error(result, 'organization_id', 'Missing parameter')
@@ -643,8 +612,7 @@ def test_update_invalid_missing_params(self):
def test_update_invalid_missing_values(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
metadata_schema_id='')
assert_error(result, 'metadata_schema_id', 'Missing value')
@@ -652,8 +620,7 @@ def test_update_invalid_missing_values(self):
def test_update_invalid_duplicate_schema(self):
metadata_model1 = ckanext_factories.MetadataModel()
metadata_model2 = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model1['id'],
metadata_schema_id=metadata_model2['metadata_schema_id'])
assert_error(result, '__after', 'Unique constraint violation')
@@ -662,8 +629,7 @@ def test_update_invalid_duplicate_schema_organization(self):
organization = ckan_factories.Organization()
metadata_model1 = ckanext_factories.MetadataModel()
metadata_model2 = ckanext_factories.MetadataModel(organization_id=organization['id'])
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model1['id'],
metadata_schema_id=metadata_model2['metadata_schema_id'],
organization_id=metadata_model2['organization_id'])
@@ -673,8 +639,7 @@ def test_update_invalid_duplicate_schema_infrastructure(self):
infrastructure = ckanext_factories.Infrastructure()
metadata_model1 = ckanext_factories.MetadataModel()
metadata_model2 = ckanext_factories.MetadataModel(infrastructure_id=infrastructure['id'])
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model1['id'],
metadata_schema_id=metadata_model2['metadata_schema_id'],
infrastructure_id=metadata_model2['infrastructure_id'])
@@ -684,8 +649,7 @@ def test_update_invalid_with_organization_set_infrastructure(self):
organization = ckan_factories.Organization()
metadata_model = ckanext_factories.MetadataModel(organization_id=organization['id'])
infrastructure = ckanext_factories.Infrastructure()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
infrastructure_id=infrastructure['id'])
assert_error(result, '__after',
@@ -695,8 +659,7 @@ def test_update_invalid_with_infrastructure_set_organization(self):
infrastructure = ckanext_factories.Infrastructure()
metadata_model = ckanext_factories.MetadataModel(infrastructure_id=infrastructure['id'])
organization = ckan_factories.Organization()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
organization_id=organization['id'])
assert_error(result, '__after',
@@ -704,32 +667,28 @@ def test_update_invalid_with_infrastructure_set_organization(self):
def test_update_invalid_not_json(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
model_json='not json')
assert_error(result, 'model_json', 'JSON decode error')
def test_update_invalid_not_json_dict(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
model_json='[1,2,3]')
assert_error(result, 'model_json', 'Expecting a JSON dictionary')
def test_update_invalid_not_json_schema(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
model_json='{"type": "foo"}')
assert_error(result, 'model_json', 'Invalid JSON schema')
def test_update_invalid_bad_references(self):
metadata_model = ckanext_factories.MetadataModel()
- result, obj = self._test_action('update', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_update', should_error=True,
id=metadata_model['id'],
metadata_schema_id='a',
organization_id='b',
@@ -747,8 +706,7 @@ def test_update_invalid_deleted_references(self):
call_action('organization_delete', id=organization['id'])
call_action('infrastructure_delete', id=infrastructure['id'])
- result, obj = self._test_action('create', 'metadata_model',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_model_create', should_error=True,
id=metadata_model['id'],
metadata_schema_id=metadata_schema['id'],
organization_id=organization['id'],
@@ -759,25 +717,21 @@ def test_update_invalid_deleted_references(self):
def test_delete_valid(self):
metadata_model = ckanext_factories.MetadataModel()
- self._test_action('delete', 'metadata_model',
- model_class=ckanext_model.MetadataModel,
+ self._test_action('metadata_model_delete',
id=metadata_model['id'])
def test_delete_invalidate_records(self):
metadata_record, metadata_model = self._generate_and_validate_metadata_record()
- self._test_action('delete', 'metadata_model',
- model_class=ckanext_model.MetadataModel,
+ self._test_action('metadata_model_delete',
id=metadata_model['id'])
assert_package_has_extra(metadata_record['id'], 'validated', False)
metadata_record, metadata_model = self._generate_and_validate_metadata_record(add_organization_to_model=True)
- self._test_action('delete', 'metadata_model',
- model_class=ckanext_model.MetadataModel,
+ self._test_action('metadata_model_delete',
id=metadata_model['id'])
assert_package_has_extra(metadata_record['id'], 'validated', False)
metadata_record, metadata_model = self._generate_and_validate_metadata_record(add_infrastructure_to_record=True, add_infrastructure_to_model=True)
- self._test_action('delete', 'metadata_model',
- model_class=ckanext_model.MetadataModel,
+ self._test_action('metadata_model_delete',
id=metadata_model['id'])
assert_package_has_extra(metadata_record['id'], 'validated', False)
diff --git a/ckanext/metadata/tests/test_metadata_record_actions.py b/ckanext/metadata/tests/test_metadata_record_actions.py
index f9762d8..0f33720 100644
--- a/ckanext/metadata/tests/test_metadata_record_actions.py
+++ b/ckanext/metadata/tests/test_metadata_record_actions.py
@@ -4,8 +4,6 @@
from ckan.tests import factories as ckan_factories
from ckan.tests.helpers import call_action
-import ckan.plugins.toolkit as tk
-import ckan.model as ckan_model
from ckanext.metadata.tests import (
ActionTestBase,
@@ -102,22 +100,19 @@ def test_create_valid(self):
input_dict['validated'] = 'ignore'
input_dict['errors'] = 'ignore'
input_dict['workflow_state_id'] = 'ignore'
- result, obj = self._test_action('create', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_create', **input_dict)
self._assert_metadata_record_ok(obj, input_dict)
def test_create_valid_setname(self):
input_dict = self._make_input_dict()
input_dict['name'] = 'test-metadata-record'
- result, obj = self._test_action('create', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_create', **input_dict)
self._assert_metadata_record_ok(obj, input_dict, name=input_dict['name'])
def test_create_valid_owner_org_byname(self):
input_dict = self._make_input_dict()
input_dict['owner_org'] = self.owner_org['name']
- result, obj = self._test_action('create', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_create', **input_dict)
self._assert_metadata_record_ok(obj, input_dict)
def test_create_valid_with_infrastructures(self):
@@ -130,8 +125,7 @@ def test_create_valid_with_infrastructures(self):
{'id': infrastructure2['name']},
],
})
- result, obj = self._test_action('create', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_create', **input_dict)
self._assert_metadata_record_ok(obj, input_dict)
assert_group_has_member(infrastructure1['id'], obj.id, 'package')
assert_group_has_member(infrastructure2['id'], obj.id, 'package')
@@ -139,27 +133,22 @@ def test_create_valid_with_infrastructures(self):
def test_create_valid_sysadmin_setid(self):
input_dict = self._make_input_dict()
input_dict['id'] = make_uuid()
- result, obj = self._test_action('create', 'metadata_record',
- model_class=ckan_model.Package,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('metadata_record_create', sysadmin=True, check_auth=True, **input_dict)
self._assert_metadata_record_ok(obj, input_dict)
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('metadata_record_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_record = ckanext_factories.MetadataRecord()
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('metadata_record_create', should_error=True, sysadmin=True, check_auth=True,
id=metadata_record['id'])
assert_error(result, 'id', 'Dataset id already exists')
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('metadata_record_create', should_error=True)
assert_error(result, 'owner_org', 'Missing parameter')
assert_error(result, 'metadata_collection_id', 'Missing parameter')
assert_error(result, 'infrastructures', 'Missing parameter')
@@ -169,8 +158,7 @@ def test_create_invalid_missing_params(self):
assert_error(result, 'metadata_url', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_create', should_error=True,
owner_org='',
metadata_collection_id='',
metadata_schema_id='')
@@ -180,26 +168,22 @@ def test_create_invalid_missing_values(self):
def test_create_invalid_duplicate_name(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_create', should_error=True,
name=metadata_record['name'])
assert_error(result, 'name', 'That URL is already in use.')
def test_create_invalid_not_json(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_create', should_error=True,
metadata_json='not json')
assert_error(result, 'metadata_json', 'JSON decode error')
def test_create_invalid_not_json_dict(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_create', should_error=True,
metadata_json='[1,2,3]')
assert_error(result, 'metadata_json', 'Expecting a JSON dictionary')
def test_create_invalid_bad_references(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_create', should_error=True,
owner_org='a',
metadata_collection_id='b',
metadata_schema_id='c',
@@ -218,8 +202,7 @@ def test_create_invalid_deleted_references(self):
input_dict = self._make_input_dict()
input_dict['infrastructures'] = [{'id': infrastructure['id']}]
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_record_create', should_error=True, **input_dict)
assert_error(result, 'owner_org', 'Not found: Organization')
assert_error(result, 'metadata_collection_id', 'Not found: Metadata Collection')
@@ -227,8 +210,7 @@ def test_create_invalid_deleted_references(self):
assert_error(result['infrastructures'][0], 'id', 'Not found: Infrastructure')
def test_create_invalid_owner_org_collection_mismatch(self):
- result, obj = self._test_action('create', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_create', should_error=True,
owner_org=self.owner_org['id'],
metadata_collection_id=self._generate_metadata_collection()['id'])
assert_error(result, '__after', 'owner_org must be the same organization that owns the metadata collection')
@@ -262,8 +244,7 @@ def test_update_valid(self):
'errors': 'ignore',
'workflow_state_id': 'ignore',
}
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
@@ -287,8 +268,7 @@ def test_update_valid_partial(self):
})
del input_dict['title']
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
@@ -302,8 +282,7 @@ def test_update_json_invalidate(self):
input_dict = self._make_input_dict_from_output_dict(metadata_record)
input_dict['metadata_json'] = '{ "newtestkey": "newtestvalue" }'
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
validated=False)
@@ -319,8 +298,7 @@ def test_update_schema_invalidate(self):
new_metadata_schema = ckanext_factories.MetadataSchema()
input_dict['metadata_schema_id'] = new_metadata_schema['id']
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
metadata_schema_id=new_metadata_schema['id'],
@@ -344,8 +322,7 @@ def test_update_owner_org_invalidate(self):
'owner_org': new_organization['id'],
'metadata_collection_id': new_metadata_collection['id'],
})
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
owner_org=new_organization['id'],
@@ -367,8 +344,7 @@ def test_update_infrastructures_invalidate(self):
assert_package_has_extra(metadata_record['id'], 'validated', True)
input_dict['infrastructures'] = [{'id': new_infrastructure['id']}]
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
validated=False)
@@ -383,8 +359,7 @@ def test_update_no_invalidate(self):
new_infrastructure = self._generate_infrastructure()
input_dict['infrastructures'] = [{'id': new_infrastructure['id']}]
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
validated=True)
@@ -397,8 +372,7 @@ def test_update_no_invalidate(self):
'owner_org': new_organization['id'],
'metadata_collection_id': new_metadata_collection['id'],
})
- result, obj = self._test_action('update', 'metadata_record',
- model_class=ckan_model.Package, **input_dict)
+ result, obj = self._test_action('metadata_record_update', **input_dict)
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
owner_org=new_organization['id'],
@@ -410,16 +384,14 @@ def test_update_no_invalidate(self):
def test_update_invalid_duplicate_name(self):
metadata_record1 = self._generate_metadata_record()
metadata_record2 = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record1['id'],
name=metadata_record2['name'])
assert_error(result, 'name', 'That URL is already in use.')
def test_update_invalid_missing_params(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'])
assert_error(result, 'owner_org', 'Missing parameter')
assert_error(result, 'metadata_collection_id', 'Missing parameter')
@@ -431,8 +403,7 @@ def test_update_invalid_missing_params(self):
def test_update_invalid_missing_values(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'],
owner_org='',
metadata_collection_id='',
@@ -443,24 +414,21 @@ def test_update_invalid_missing_values(self):
def test_update_invalid_not_json(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'],
metadata_json='not json')
assert_error(result, 'metadata_json', 'JSON decode error')
def test_update_invalid_not_json_dict(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'],
metadata_json='[1,2,3]')
assert_error(result, 'metadata_json', 'Expecting a JSON dictionary')
def test_update_invalid_bad_references(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'],
owner_org='a',
metadata_collection_id='b',
@@ -482,8 +450,7 @@ def test_update_invalid_deleted_references(self):
call_action('metadata_schema_delete', id=metadata_schema['id'])
call_action('infrastructure_delete', id=infrastructure['id'])
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'],
owner_org=organization['id'],
metadata_collection_id=metadata_collection['id'],
@@ -497,8 +464,7 @@ def test_update_invalid_deleted_references(self):
def test_update_invalid_owner_org_collection_mismatch(self):
metadata_record = self._generate_metadata_record()
- result, obj = self._test_action('update', 'metadata_record',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_record_update', should_error=True,
id=metadata_record['id'],
owner_org=self.owner_org['id'],
metadata_collection_id=self._generate_metadata_collection()['id'])
@@ -506,8 +472,7 @@ def test_update_invalid_owner_org_collection_mismatch(self):
def test_delete_valid(self):
metadata_record = self._generate_metadata_record()
- self._test_action('delete', 'metadata_record',
- model_class=ckan_model.Package,
+ self._test_action('metadata_record_delete',
id=metadata_record['id'])
def test_invalidate(self):
@@ -515,8 +480,7 @@ def test_invalidate(self):
metadata_model = self._validate_metadata_record(metadata_record)
input_dict = self._make_input_dict_from_output_dict(metadata_record)
- result, obj = self._test_action('invalidate', 'metadata_record',
- model_class=ckan_model.Package,
+ result, obj = self._test_action('metadata_record_invalidate',
id=metadata_record['id'])
self._assert_metadata_record_ok(obj, input_dict,
name=input_dict['name'],
diff --git a/ckanext/metadata/tests/test_metadata_schema_actions.py b/ckanext/metadata/tests/test_metadata_schema_actions.py
index f2d9f1f..6f9f55c 100644
--- a/ckanext/metadata/tests/test_metadata_schema_actions.py
+++ b/ckanext/metadata/tests/test_metadata_schema_actions.py
@@ -1,6 +1,5 @@
# encoding: utf-8
-from ckan.plugins import toolkit as tk
from ckan.tests.helpers import call_action
from ckanext.metadata import model as ckanext_model
@@ -25,8 +24,7 @@ def test_create_valid(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.name == generate_name(input_dict['schema_name'], input_dict['schema_version'])
@@ -38,8 +36,7 @@ def test_create_valid_setname(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_with_parent(self):
@@ -50,8 +47,7 @@ def test_create_valid_with_parent(self):
'schema_xsd': '',
'base_schema_id': metadata_schema['id'],
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_with_parent_byname(self):
@@ -62,8 +58,7 @@ def test_create_valid_with_parent_byname(self):
'schema_xsd': '',
'base_schema_id': metadata_schema['name'],
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', **input_dict)
input_dict['base_schema_id'] = metadata_schema['id']
assert_object_matches_dict(obj, input_dict)
@@ -75,9 +70,7 @@ def test_create_valid_sysadmin_setid(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', sysadmin=True, check_auth=True, **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_same_name_new_version(self):
@@ -88,8 +81,7 @@ def test_create_valid_same_name_new_version(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_same_version_different_name(self):
@@ -100,28 +92,24 @@ def test_create_valid_same_version_different_name(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('create', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_invalid_duplicate_name(self):
metadata_schema = ckanext_factories.MetadataSchema()
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_create', should_error=True,
name=metadata_schema['name'])
assert_error(result, 'name', 'Duplicate name: Metadata Schema')
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('metadata_schema_create', should_error=True)
assert_error(result, 'schema_name', 'Missing parameter')
assert_error(result, 'schema_version', 'Missing parameter')
assert_error(result, 'schema_xsd', 'Missing parameter')
assert_error(result, 'base_schema_id', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_create', should_error=True,
schema_name='')
assert_error(result, 'schema_name', 'Missing value')
@@ -131,20 +119,17 @@ def test_create_invalid_duplicate(self):
'schema_name': metadata_schema['schema_name'],
'schema_version': metadata_schema['schema_version'],
}
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_schema_create', should_error=True, **input_dict)
assert_error(result, '__after', 'Unique constraint violation')
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('metadata_schema_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_schema = ckanext_factories.MetadataSchema()
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('metadata_schema_create', should_error=True, sysadmin=True, check_auth=True,
id=metadata_schema['id'])
assert_error(result, 'id', 'Already exists: Metadata Schema')
@@ -154,22 +139,19 @@ def test_create_invalid_sysadmin_self_parent(self):
'id': new_id,
'base_schema_id': new_id,
}
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_create', should_error=True,
sysadmin=True, check_auth=True, **input_dict)
assert_error(result, 'base_schema_id', 'Not found: Metadata Schema')
def test_create_invalid_bad_parent(self):
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_create', should_error=True,
base_schema_id='foo')
assert_error(result, 'base_schema_id', 'Not found: Metadata Schema')
def test_create_invalid_deleted_parent(self):
metadata_schema = ckanext_factories.MetadataSchema()
call_action('metadata_schema_delete', id=metadata_schema['id'])
- result, obj = self._test_action('create', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_create', should_error=True,
base_schema_id=metadata_schema['id'])
assert_error(result, 'base_schema_id', 'Not found: Metadata Schema')
@@ -184,8 +166,7 @@ def test_update_valid(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('update', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.name == generate_name(input_dict['schema_name'], input_dict['schema_version'])
@@ -199,8 +180,7 @@ def test_update_valid_partial(self):
'schema_xsd': '',
'base_schema_id': '',
}
- result, obj = self._test_action('update', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.title == metadata_schema['title']
assert obj.description == metadata_schema['description']
@@ -215,8 +195,7 @@ def test_update_valid_change_parent_1(self):
'schema_xsd': metadata_schema1['schema_xsd'],
'base_schema_id': metadata_schema2['id'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_valid_change_parent_2(self):
@@ -230,8 +209,7 @@ def test_update_valid_change_parent_2(self):
'schema_xsd': metadata_schema3['schema_xsd'],
'base_schema_id': metadata_schema1['id'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_invalid_duplicate_name(self):
@@ -241,14 +219,12 @@ def test_update_invalid_duplicate_name(self):
'id': metadata_schema1['id'],
'name': metadata_schema2['name'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', should_error=True, **input_dict)
assert_error(result, 'name', 'Duplicate name: Metadata Schema')
def test_update_invalid_missing_params(self):
metadata_schema = ckanext_factories.MetadataSchema()
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_update', should_error=True,
id=metadata_schema['id'])
assert_error(result, 'schema_name', 'Missing parameter')
assert_error(result, 'schema_version', 'Missing parameter')
@@ -257,8 +233,7 @@ def test_update_invalid_missing_params(self):
def test_update_invalid_missing_values(self):
metadata_schema = ckanext_factories.MetadataSchema()
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_update', should_error=True,
id=metadata_schema['id'],
schema_name='')
assert_error(result, 'schema_name', 'Missing value')
@@ -271,8 +246,7 @@ def test_update_invalid_duplicate(self):
'schema_name': metadata_schema2['schema_name'],
'schema_version': metadata_schema2['schema_version'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', should_error=True, **input_dict)
assert_error(result, '__after', 'Unique constraint violation')
def test_update_invalid_circular_ref_1(self):
@@ -282,8 +256,7 @@ def test_update_invalid_circular_ref_1(self):
'id': metadata_schema1['id'],
'base_schema_id': metadata_schema2['id'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', should_error=True, **input_dict)
assert_error(result, 'base_schema_id', 'Loop in metadata schema hierarchy')
def test_update_invalid_circular_ref_2(self):
@@ -294,8 +267,7 @@ def test_update_invalid_circular_ref_2(self):
'id': metadata_schema1['id'],
'base_schema_id': metadata_schema3['id'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', should_error=True, **input_dict)
assert_error(result, 'base_schema_id', 'Loop in metadata schema hierarchy')
def test_update_invalid_self_parent(self):
@@ -304,14 +276,12 @@ def test_update_invalid_self_parent(self):
'id': metadata_schema['id'],
'base_schema_id': metadata_schema['id'],
}
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('metadata_schema_update', should_error=True, **input_dict)
assert_error(result, 'base_schema_id', 'Loop in metadata schema hierarchy')
def test_update_invalid_bad_parent(self):
metadata_schema = ckanext_factories.MetadataSchema()
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_update', should_error=True,
id=metadata_schema['id'],
base_schema_id='foo')
assert_error(result, 'base_schema_id', 'Not found: Metadata Schema')
@@ -320,16 +290,14 @@ def test_update_invalid_deleted_parent(self):
metadata_schema1 = ckanext_factories.MetadataSchema()
metadata_schema2 = ckanext_factories.MetadataSchema()
call_action('metadata_schema_delete', id=metadata_schema1['id'])
- result, obj = self._test_action('update', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_update', should_error=True,
id=metadata_schema2['id'],
base_schema_id=metadata_schema1['id'])
assert_error(result, 'base_schema_id', 'Not found: Metadata Schema')
def test_delete_valid(self):
metadata_schema = ckanext_factories.MetadataSchema()
- self._test_action('delete', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema,
+ self._test_action('metadata_schema_delete',
id=metadata_schema['id'])
def test_delete_with_child_schemas(self):
@@ -337,8 +305,7 @@ def test_delete_with_child_schemas(self):
metadata_schema2 = ckanext_factories.MetadataSchema(base_schema_id=metadata_schema1['id'])
assert metadata_schema2['base_schema_id'] == metadata_schema1['id']
- self._test_action('delete', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema,
+ self._test_action('metadata_schema_delete',
id=metadata_schema1['id'])
metadata_schema2['base_schema_id'] = None
del metadata_schema2['revision_id']
@@ -349,15 +316,13 @@ def test_delete_with_dependencies(self):
metadata_model = ckanext_factories.MetadataModel(metadata_schema_id=metadata_schema['id'])
metadata_record = ckanext_factories.MetadataRecord(metadata_schema_id=metadata_schema['id'])
- result, obj = self._test_action('delete', 'metadata_schema',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('metadata_schema_delete', should_error=True,
id=metadata_schema['id'])
assert_error(result, 'message', 'Metadata schema 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('delete', 'metadata_schema',
- model_class=ckanext_model.MetadataSchema,
+ self._test_action('metadata_schema_delete',
id=metadata_schema['id'])
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'deleted'
diff --git a/ckanext/metadata/tests/test_metadata_validation.py b/ckanext/metadata/tests/test_metadata_validation.py
index e69de29..2b5c637 100644
--- a/ckanext/metadata/tests/test_metadata_validation.py
+++ b/ckanext/metadata/tests/test_metadata_validation.py
@@ -0,0 +1,2 @@
+# encoding: utf-8
+
diff --git a/ckanext/metadata/tests/test_organization_actions.py b/ckanext/metadata/tests/test_organization_actions.py
index 1565b20..d530280 100644
--- a/ckanext/metadata/tests/test_organization_actions.py
+++ b/ckanext/metadata/tests/test_organization_actions.py
@@ -1,7 +1,6 @@
# encoding: utf-8
from ckan import model as ckan_model
-from ckan.plugins import toolkit as tk
from ckan.tests import factories as ckan_factories
from ckan.tests.helpers import call_action
@@ -23,16 +22,14 @@ def _generate_metadata_collection(self, **kwargs):
def test_delete_valid(self):
organization = self._generate_organization()
- self._test_action('delete', 'organization',
- model_class=ckan_model.Group,
+ self._test_action('organization_delete',
id=organization['id'])
def test_delete_valid_cascade_metadata_models(self):
organization = self._generate_organization()
metadata_model = ckanext_factories.MetadataModel(organization_id=organization['id'])
- self._test_action('delete', 'organization',
- model_class=ckan_model.Group,
+ self._test_action('organization_delete',
id=organization['id'])
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'deleted'
@@ -40,8 +37,7 @@ def test_delete_valid_cascade_metadata_collections(self):
organization = self._generate_organization()
metadata_collection = self._generate_metadata_collection(organization_id=organization['id'])
- self._test_action('delete', 'organization',
- model_class=ckan_model.Group,
+ self._test_action('organization_delete',
id=organization['id'])
assert ckan_model.Group.get(metadata_collection['id']).state == 'deleted'
@@ -52,16 +48,14 @@ def test_delete_with_dependencies(self):
metadata_record = ckanext_factories.MetadataRecord(owner_org=organization['id'],
metadata_collection_id=metadata_collection['id'])
- result, obj = self._test_action('delete', 'organization',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('organization_delete', should_error=True,
id=organization['id'])
assert_error(result, 'message', 'Organization has dependent metadata records')
assert ckan_model.Group.get(metadata_collection['id']).state == 'active'
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'active'
call_action('metadata_record_delete', id=metadata_record['id'])
- self._test_action('delete', 'organization',
- model_class=ckan_model.Group,
+ self._test_action('organization_delete',
id=organization['id'])
assert ckan_model.Group.get(metadata_collection['id']).state == 'deleted'
assert ckanext_model.MetadataModel.get(metadata_model['id']).state == 'deleted'
diff --git a/ckanext/metadata/tests/test_workflow_metric_actions.py b/ckanext/metadata/tests/test_workflow_metric_actions.py
index 19c8820..af9b718 100644
--- a/ckanext/metadata/tests/test_workflow_metric_actions.py
+++ b/ckanext/metadata/tests/test_workflow_metric_actions.py
@@ -1,7 +1,5 @@
# encoding: utf-8
-from ckan.plugins import toolkit as tk
-
from ckanext.metadata import model as ckanext_model
from ckanext.metadata.tests import (
ActionTestBase,
@@ -21,8 +19,7 @@ def test_create_valid(self):
'description': 'This is a test workflow metric',
'evaluator_url': 'http://example.net/',
}
- result, obj = self._test_action('create', 'workflow_metric',
- model_class=ckanext_model.WorkflowMetric, **input_dict)
+ result, obj = self._test_action('workflow_metric_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_sysadmin_setid(self):
@@ -31,42 +28,35 @@ def test_create_valid_sysadmin_setid(self):
'name': 'test-workflow-metric',
'evaluator_url': 'http://example.net/',
}
- result, obj = self._test_action('create', 'workflow_metric',
- model_class=ckanext_model.WorkflowMetric,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('workflow_metric_create', sysadmin=True, check_auth=True, **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_invalid_duplicate_name(self):
workflow_metric = ckanext_factories.WorkflowMetric()
- result, obj = self._test_action('create', 'workflow_metric',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_metric_create', should_error=True,
name=workflow_metric['name'])
assert_error(result, 'name', 'Duplicate name: Workflow Metric')
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'workflow_metric',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('workflow_metric_create', should_error=True)
assert_error(result, 'name', 'Missing parameter')
assert_error(result, 'evaluator_url', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'workflow_metric',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_metric_create', should_error=True,
name='',
evaluator_url='')
assert_error(result, 'name', 'Missing value')
assert_error(result, 'evaluator_url', 'Missing value')
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'workflow_metric',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('workflow_metric_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):
workflow_metric = ckanext_factories.WorkflowMetric()
- result, obj = self._test_action('create', 'workflow_metric',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('workflow_metric_create', should_error=True, sysadmin=True, check_auth=True,
id=workflow_metric['id'])
assert_error(result, 'id', 'Already exists: Workflow Metric')
@@ -79,8 +69,7 @@ def test_update_valid(self):
'description': 'Updated test workflow metric description',
'evaluator_url': 'http://updated.example.net/',
}
- result, obj = self._test_action('update', 'workflow_metric',
- model_class=ckanext_model.WorkflowMetric, **input_dict)
+ result, obj = self._test_action('workflow_metric_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_valid_partial(self):
@@ -90,8 +79,7 @@ def test_update_valid_partial(self):
'name': 'updated-test-workflow-metric',
'evaluator_url': 'http://updated.example.net/',
}
- result, obj = self._test_action('update', 'workflow_metric',
- model_class=ckanext_model.WorkflowMetric, **input_dict)
+ result, obj = self._test_action('workflow_metric_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.title == workflow_metric['title']
assert obj.description == workflow_metric['description']
@@ -103,35 +91,30 @@ def test_update_invalid_duplicate_name(self):
'id': workflow_metric1['id'],
'name': workflow_metric2['name'],
}
- result, obj = self._test_action('update', 'workflow_metric',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('workflow_metric_update', should_error=True, **input_dict)
assert_error(result, 'name', 'Duplicate name: Workflow Metric')
def test_update_invalid_missing_params(self):
workflow_metric = ckanext_factories.WorkflowMetric()
- result, obj = self._test_action('update', 'workflow_metric',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_metric_update', should_error=True,
id=workflow_metric['id'])
assert_error(result, 'evaluator_url', 'Missing parameter')
def test_update_invalid_missing_values(self):
workflow_metric = ckanext_factories.WorkflowMetric()
- result, obj = self._test_action('update', 'workflow_metric',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_metric_update', should_error=True,
id=workflow_metric['id'],
evaluator_url='')
assert_error(result, 'evaluator_url', 'Missing value')
def test_delete_valid(self):
workflow_metric = ckanext_factories.WorkflowMetric()
- self._test_action('delete', 'workflow_metric',
- model_class=ckanext_model.WorkflowMetric,
+ self._test_action('workflow_metric_delete',
id=workflow_metric['id'])
def test_delete_with_rule_references(self):
workflow_metric = ckanext_factories.WorkflowMetric()
workflow_rule = ckanext_factories.WorkflowRule(workflow_metric_id=workflow_metric['id'])
- self._test_action('delete', 'workflow_metric',
- model_class=ckanext_model.WorkflowMetric,
+ self._test_action('workflow_metric_delete',
id=workflow_metric['id'])
assert ckanext_model.WorkflowRule.get(workflow_rule['id']).state == 'deleted'
diff --git a/ckanext/metadata/tests/test_workflow_rule_actions.py b/ckanext/metadata/tests/test_workflow_rule_actions.py
index 9632c96..7eeee03 100644
--- a/ckanext/metadata/tests/test_workflow_rule_actions.py
+++ b/ckanext/metadata/tests/test_workflow_rule_actions.py
@@ -1,9 +1,7 @@
# encoding: utf-8
-from ckan.plugins import toolkit as tk
from ckan.tests.helpers import call_action
-from ckanext.metadata import model as ckanext_model
from ckanext.metadata.tests import (
ActionTestBase,
make_uuid,
@@ -23,8 +21,7 @@ def test_create_valid(self):
'workflow_metric_id': workflow_metric['id'],
'rule_json': '{ "testkey": "testvalue" }',
}
- result, obj = self._test_action('create', 'workflow_rule',
- model_class=ckanext_model.WorkflowRule, **input_dict)
+ result, obj = self._test_action('workflow_rule_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_byname(self):
@@ -35,8 +32,7 @@ def test_create_valid_byname(self):
'workflow_metric_id': workflow_metric['name'],
'rule_json': '{ "testkey": "testvalue" }',
}
- result, obj = self._test_action('create', 'workflow_rule',
- model_class=ckanext_model.WorkflowRule, **input_dict)
+ result, obj = self._test_action('workflow_rule_create', **input_dict)
input_dict = {
'workflow_state_id': workflow_state['id'],
'workflow_metric_id': workflow_metric['id'],
@@ -53,21 +49,17 @@ def test_create_valid_sysadmin_setid(self):
'workflow_metric_id': workflow_metric['id'],
'rule_json': '{ "testkey": "testvalue" }',
}
- result, obj = self._test_action('create', 'workflow_rule',
- model_class=ckanext_model.WorkflowRule,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('workflow_rule_create', sysadmin=True, check_auth=True, **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('workflow_rule_create', should_error=True)
assert_error(result, 'workflow_state_id', 'Missing parameter')
assert_error(result, 'workflow_metric_id', 'Missing parameter')
assert_error(result, 'rule_json', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_create', should_error=True,
workflow_state_id='',
workflow_metric_id='',
rule_json='')
@@ -76,21 +68,18 @@ def test_create_invalid_missing_values(self):
assert_error(result, 'rule_json', 'Missing value')
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('workflow_rule_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):
workflow_rule = ckanext_factories.WorkflowRule()
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('workflow_rule_create', should_error=True, sysadmin=True, check_auth=True,
id=workflow_rule['id'])
assert_error(result, 'id', 'Already exists: Workflow Rule')
def test_create_invalid_bad_references(self):
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_create', should_error=True,
workflow_state_id='foo',
workflow_metric_id='bar')
assert_error(result, 'workflow_state_id', 'Not found: Workflow State')
@@ -101,8 +90,7 @@ def test_create_invalid_deleted_references(self):
workflow_metric = ckanext_factories.WorkflowMetric()
call_action('workflow_state_delete', id=workflow_state['id'])
call_action('workflow_metric_delete', id=workflow_metric['id'])
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_create', should_error=True,
workflow_state_id=workflow_state['id'],
workflow_metric_id=workflow_metric['id'])
assert_error(result, 'workflow_state_id', 'Not found: Workflow State')
@@ -110,15 +98,13 @@ def test_create_invalid_deleted_references(self):
def test_create_invalid_duplicate(self):
workflow_rule = ckanext_factories.WorkflowRule()
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_create', should_error=True,
workflow_state_id=workflow_rule['workflow_state_id'],
workflow_metric_id=workflow_rule['workflow_metric_id'])
assert_error(result, '__after', 'Unique constraint violation')
def test_create_invalid_not_json(self):
- result, obj = self._test_action('create', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_create', should_error=True,
rule_json='not json')
assert_error(result, 'rule_json', 'JSON decode error')
@@ -128,14 +114,12 @@ def test_update_valid(self):
'id': workflow_rule['id'],
'rule_json': '{ "newtestkey": "newtestvalue" }',
}
- result, obj = self._test_action('update', 'workflow_rule',
- model_class=ckanext_model.WorkflowRule, **input_dict)
+ result, obj = self._test_action('workflow_rule_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_invalid_cannot_change_association(self):
workflow_rule = ckanext_factories.WorkflowRule()
- result, obj = self._test_action('update', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_update', should_error=True,
id=workflow_rule['id'],
workflow_state_id=ckanext_factories.WorkflowState(),
workflow_metric_id=ckanext_factories.WorkflowMetric())
@@ -144,29 +128,25 @@ def test_update_invalid_cannot_change_association(self):
def test_update_invalid_missing_params(self):
workflow_rule = ckanext_factories.WorkflowRule()
- result, obj = self._test_action('update', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_update', should_error=True,
id=workflow_rule['id'])
assert_error(result, 'rule_json', 'Missing parameter')
def test_update_invalid_missing_values(self):
workflow_rule = ckanext_factories.WorkflowRule()
- result, obj = self._test_action('update', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_update', should_error=True,
id=workflow_rule['id'],
rule_json='')
assert_error(result, 'rule_json', 'Missing value')
def test_update_invalid_not_json(self):
workflow_rule = ckanext_factories.WorkflowRule()
- result, obj = self._test_action('update', 'workflow_rule',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_rule_update', should_error=True,
id=workflow_rule['id'],
rule_json='not json')
assert_error(result, 'rule_json', 'JSON decode error')
def test_delete_valid(self):
workflow_rule = ckanext_factories.WorkflowRule()
- self._test_action('delete', 'workflow_rule',
- model_class=ckanext_model.WorkflowRule,
+ self._test_action('workflow_rule_delete',
id=workflow_rule['id'])
diff --git a/ckanext/metadata/tests/test_workflow_state_actions.py b/ckanext/metadata/tests/test_workflow_state_actions.py
index 3c04c1e..f085902 100644
--- a/ckanext/metadata/tests/test_workflow_state_actions.py
+++ b/ckanext/metadata/tests/test_workflow_state_actions.py
@@ -1,6 +1,5 @@
# encoding: utf-8
-from ckan.plugins import toolkit as tk
from ckan.tests.helpers import call_action
from ckanext.metadata import model as ckanext_model
@@ -23,8 +22,7 @@ def test_create_valid(self):
'description': 'This is a test workflow state',
'revert_state_id': '',
}
- result, obj = self._test_action('create', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_with_revert(self):
@@ -33,8 +31,7 @@ def test_create_valid_with_revert(self):
'name': 'test-workflow-state',
'revert_state_id': workflow_state['id'],
}
- result, obj = self._test_action('create', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_with_revert_byname(self):
@@ -43,8 +40,7 @@ def test_create_valid_with_revert_byname(self):
'name': 'test-workflow-state',
'revert_state_id': workflow_state['name'],
}
- result, obj = self._test_action('create', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_create', **input_dict)
input_dict['revert_state_id'] = workflow_state['id']
assert_object_matches_dict(obj, input_dict)
@@ -54,40 +50,33 @@ def test_create_valid_sysadmin_setid(self):
'name': 'test-workflow-state',
'revert_state_id': '',
}
- result, obj = self._test_action('create', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('workflow_state_create', sysadmin=True, check_auth=True, **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_invalid_duplicate_name(self):
workflow_state = ckanext_factories.WorkflowState()
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_create', should_error=True,
name=workflow_state['name'])
assert_error(result, 'name', 'Duplicate name: Workflow State')
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('workflow_state_create', should_error=True)
assert_error(result, 'name', 'Missing parameter')
assert_error(result, 'revert_state_id', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_create', should_error=True,
name='')
assert_error(result, 'name', 'Missing value')
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('workflow_state_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):
workflow_state = ckanext_factories.WorkflowState()
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('workflow_state_create', should_error=True, sysadmin=True, check_auth=True,
id=workflow_state['id'])
assert_error(result, 'id', 'Already exists: Workflow State')
@@ -97,22 +86,19 @@ def test_create_invalid_sysadmin_self_revert(self):
'id': new_id,
'revert_state_id': new_id,
}
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_create', should_error=True,
sysadmin=True, check_auth=True, **input_dict)
assert_error(result, 'revert_state_id', 'Not found: Workflow State')
def test_create_invalid_bad_revert(self):
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_create', should_error=True,
revert_state_id='foo')
assert_error(result, 'revert_state_id', 'Not found: Workflow State')
def test_create_invalid_deleted_revert(self):
workflow_state = ckanext_factories.WorkflowState()
call_action('workflow_state_delete', id=workflow_state['id'])
- result, obj = self._test_action('create', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_create', should_error=True,
revert_state_id=workflow_state['id'])
assert_error(result, 'revert_state_id', 'Not found: Workflow State')
@@ -125,8 +111,7 @@ def test_update_valid(self):
'description': 'Updated test workflow state description',
'revert_state_id': '',
}
- result, obj = self._test_action('update', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_valid_partial(self):
@@ -136,8 +121,7 @@ def test_update_valid_partial(self):
'name': 'updated-test-workflow-state',
'revert_state_id': '',
}
- result, obj = self._test_action('update', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
assert obj.title == workflow_state['title']
assert obj.description == workflow_state['description']
@@ -150,8 +134,7 @@ def test_update_valid_change_revert_1(self):
'name': workflow_state1['name'],
'revert_state_id': workflow_state2['id'],
}
- result, obj = self._test_action('update', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_valid_change_revert_2(self):
@@ -163,8 +146,7 @@ def test_update_valid_change_revert_2(self):
'name': workflow_state3['name'],
'revert_state_id': workflow_state1['id'],
}
- result, obj = self._test_action('update', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_valid_change_revert_3(self):
@@ -178,8 +160,7 @@ def test_update_valid_change_revert_3(self):
'name': workflow_state3['name'],
'revert_state_id': workflow_state1['id'],
}
- result, obj = self._test_action('update', 'workflow_state',
- model_class=ckanext_model.WorkflowState, **input_dict)
+ result, obj = self._test_action('workflow_state_update', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_update_invalid_duplicate_name(self):
@@ -189,14 +170,12 @@ def test_update_invalid_duplicate_name(self):
'id': workflow_state1['id'],
'name': workflow_state2['name'],
}
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('workflow_state_update', should_error=True, **input_dict)
assert_error(result, 'name', 'Duplicate name: Workflow State')
def test_update_invalid_missing_params(self):
workflow_state = ckanext_factories.WorkflowState()
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state['id'])
assert_error(result, 'revert_state_id', 'Missing parameter')
@@ -205,14 +184,12 @@ def test_update_invalid_circular_revert(self):
workflow_state2 = ckanext_factories.WorkflowState(revert_state_id=workflow_state1['id'])
workflow_state3 = ckanext_factories.WorkflowState(revert_state_id=workflow_state2['id'])
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state1['id'],
revert_state_id=workflow_state2['id'])
assert_error(result, 'revert_state_id', 'Revert loop in workflow state graph')
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state1['id'],
revert_state_id=workflow_state3['id'])
assert_error(result, 'revert_state_id', 'Revert loop in workflow state graph')
@@ -224,14 +201,12 @@ def test_update_invalid_forward_revert(self):
ckanext_factories.WorkflowTransition(from_state_id=workflow_state1['id'], to_state_id=workflow_state2['id'])
ckanext_factories.WorkflowTransition(from_state_id=workflow_state2['id'], to_state_id=workflow_state3['id'])
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state1['id'],
revert_state_id=workflow_state2['id'])
assert_error(result, 'revert_state_id', 'Forward revert in workflow state graph')
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state1['id'],
revert_state_id=workflow_state3['id'])
assert_error(result, 'revert_state_id', 'Forward revert in workflow state graph')
@@ -242,14 +217,12 @@ def test_update_invalid_self_revert(self):
'id': workflow_state['id'],
'revert_state_id': workflow_state['id'],
}
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError, **input_dict)
+ result, obj = self._test_action('workflow_state_update', should_error=True, **input_dict)
assert_error(result, 'revert_state_id', 'A workflow state cannot revert to itself')
def test_update_invalid_bad_revert(self):
workflow_state = ckanext_factories.WorkflowState()
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state['id'],
revert_state_id='foo')
assert_error(result, 'revert_state_id', 'Not found: Workflow State')
@@ -258,16 +231,14 @@ def test_update_invalid_deleted_revert(self):
workflow_state1 = ckanext_factories.WorkflowState()
workflow_state2 = ckanext_factories.WorkflowState()
call_action('workflow_state_delete', id=workflow_state1['id'])
- result, obj = self._test_action('update', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_update', should_error=True,
id=workflow_state2['id'],
revert_state_id=workflow_state1['id'])
assert_error(result, 'revert_state_id', 'Not found: Workflow State')
def test_delete_valid(self):
workflow_state = ckanext_factories.WorkflowState()
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state['id'])
def test_delete_with_dependencies(self):
@@ -278,14 +249,12 @@ def test_delete_with_dependencies(self):
workflow_state_id=workflow_state['id'])
assert_package_has_extra(metadata_record['id'], 'workflow_state_id', workflow_state['id'])
- result, obj = self._test_action('delete', 'workflow_state',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_state_delete', should_error=True,
id=workflow_state['id'])
assert_error(result, 'message', 'Workflow state has dependent metadata records')
call_action('metadata_record_delete', id=metadata_record['id'])
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state['id'])
def test_delete_with_revert_references(self):
@@ -293,8 +262,7 @@ def test_delete_with_revert_references(self):
workflow_state2 = ckanext_factories.WorkflowState(revert_state_id=workflow_state1['name'])
assert workflow_state2['revert_state_id'] == workflow_state1['id']
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state1['id'])
workflow_state2['revert_state_id'] = None
del workflow_state2['revision_id']
@@ -303,29 +271,25 @@ def test_delete_with_revert_references(self):
def test_delete_with_transition_references(self):
workflow_state = ckanext_factories.WorkflowState()
workflow_transition = ckanext_factories.WorkflowTransition(from_state_id=workflow_state['id'])
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state['id'])
assert ckanext_model.WorkflowTransition.get(workflow_transition['id']).state == 'deleted'
workflow_state = ckanext_factories.WorkflowState()
workflow_transition = ckanext_factories.WorkflowTransition(to_state_id=workflow_state['id'])
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state['id'])
assert ckanext_model.WorkflowTransition.get(workflow_transition['id']).state == 'deleted'
workflow_state = ckanext_factories.WorkflowState()
workflow_transition = ckanext_factories.WorkflowTransition(from_state_id=None, to_state_id=workflow_state['id'])
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state['id'])
assert ckanext_model.WorkflowTransition.get(workflow_transition['id']).state == 'deleted'
def test_delete_with_rule_references(self):
workflow_state = ckanext_factories.WorkflowState()
workflow_rule = ckanext_factories.WorkflowRule(workflow_state_id=workflow_state['id'])
- self._test_action('delete', 'workflow_state',
- model_class=ckanext_model.WorkflowState,
+ self._test_action('workflow_state_delete',
id=workflow_state['id'])
assert ckanext_model.WorkflowRule.get(workflow_rule['id']).state == 'deleted'
diff --git a/ckanext/metadata/tests/test_workflow_transition_actions.py b/ckanext/metadata/tests/test_workflow_transition_actions.py
index 7fd39e5..7603631 100644
--- a/ckanext/metadata/tests/test_workflow_transition_actions.py
+++ b/ckanext/metadata/tests/test_workflow_transition_actions.py
@@ -1,9 +1,7 @@
# encoding: utf-8
-from ckan.plugins import toolkit as tk
from ckan.tests.helpers import call_action
-from ckanext.metadata import model as ckanext_model
from ckanext.metadata.tests import (
ActionTestBase,
make_uuid,
@@ -22,8 +20,7 @@ def test_create_valid(self):
'from_state_id': workflow_state1['id'],
'to_state_id': workflow_state2['id'],
}
- result, obj = self._test_action('create', 'workflow_transition',
- model_class=ckanext_model.WorkflowTransition, **input_dict)
+ result, obj = self._test_action('workflow_transition_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_initial_transition(self):
@@ -32,8 +29,7 @@ def test_create_valid_initial_transition(self):
'from_state_id': '',
'to_state_id': workflow_state['id'],
}
- result, obj = self._test_action('create', 'workflow_transition',
- model_class=ckanext_model.WorkflowTransition, **input_dict)
+ result, obj = self._test_action('workflow_transition_create', **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_valid_byname(self):
@@ -43,8 +39,7 @@ def test_create_valid_byname(self):
'from_state_id': workflow_state1['name'],
'to_state_id': workflow_state2['name'],
}
- result, obj = self._test_action('create', 'workflow_transition',
- model_class=ckanext_model.WorkflowTransition, **input_dict)
+ result, obj = self._test_action('workflow_transition_create', **input_dict)
input_dict = {
'from_state_id': workflow_state1['id'],
'to_state_id': workflow_state2['id'],
@@ -58,47 +53,39 @@ def test_create_valid_sysadmin_setid(self):
'from_state_id': '',
'to_state_id': workflow_state['id'],
}
- result, obj = self._test_action('create', 'workflow_transition',
- model_class=ckanext_model.WorkflowTransition,
- sysadmin=True, check_auth=True, **input_dict)
+ result, obj = self._test_action('workflow_transition_create', sysadmin=True, check_auth=True, **input_dict)
assert_object_matches_dict(obj, input_dict)
def test_create_invalid_missing_params(self):
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError)
+ result, obj = self._test_action('workflow_transition_create', should_error=True)
assert_error(result, 'from_state_id', 'Missing parameter')
assert_error(result, 'to_state_id', 'Missing parameter')
def test_create_invalid_missing_values(self):
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
to_state_id='')
assert_error(result, 'to_state_id', 'Missing value')
def test_create_invalid_nonsysadmin_setid(self):
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError, check_auth=True,
+ result, obj = self._test_action('workflow_transition_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):
workflow_transition = ckanext_factories.WorkflowTransition()
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError, sysadmin=True, check_auth=True,
+ result, obj = self._test_action('workflow_transition_create', should_error=True, sysadmin=True, check_auth=True,
id=workflow_transition['id'])
assert_error(result, 'id', 'Already exists: Workflow Transition')
def test_create_invalid_self_transition(self):
workflow_state = ckanext_factories.WorkflowState()
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_state['id'],
to_state_id=workflow_state['id'])
assert_error(result, '__after', 'The from- and to-state of a workflow transition cannot be the same.')
def test_create_invalid_bad_states(self):
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id='foo',
to_state_id='bar')
assert_error(result, 'from_state_id', 'Not found: Workflow State')
@@ -109,8 +96,7 @@ def test_create_invalid_deleted_states(self):
workflow_state2 = ckanext_factories.WorkflowState()
call_action('workflow_state_delete', id=workflow_state1['id'])
call_action('workflow_state_delete', id=workflow_state2['id'])
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_state1['id'],
to_state_id=workflow_state2['id'])
assert_error(result, 'from_state_id', 'Not found: Workflow State')
@@ -118,15 +104,13 @@ def test_create_invalid_deleted_states(self):
def test_create_invalid_duplicate(self):
workflow_transition = ckanext_factories.WorkflowTransition()
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_transition['from_state_id'],
to_state_id=workflow_transition['to_state_id'])
assert_error(result, '__after', 'Unique constraint violation')
workflow_transition = ckanext_factories.WorkflowTransition(from_state_id='')
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id='',
to_state_id=workflow_transition['to_state_id'])
assert_error(result, '__after', 'Unique constraint violation')
@@ -138,14 +122,12 @@ def test_create_invalid_circular_transition(self):
ckanext_factories.WorkflowTransition(from_state_id=workflow_state1['id'], to_state_id=workflow_state2['id'])
ckanext_factories.WorkflowTransition(from_state_id=workflow_state2['id'], to_state_id=workflow_state3['id'])
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_state2['id'],
to_state_id=workflow_state1['id'])
assert_error(result, '__after', 'Transition loop in workflow state graph')
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_state3['id'],
to_state_id=workflow_state1['id'])
assert_error(result, '__after', 'Transition loop in workflow state graph')
@@ -155,27 +137,23 @@ def test_create_invalid_backward_transition(self):
workflow_state2 = ckanext_factories.WorkflowState(revert_state_id=workflow_state1['id'])
workflow_state3 = ckanext_factories.WorkflowState(revert_state_id=workflow_state2['id'])
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_state2['id'],
to_state_id=workflow_state1['id'])
assert_error(result, '__after', 'Backward transition in workflow state graph')
- result, obj = self._test_action('create', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_create', should_error=True,
from_state_id=workflow_state3['id'],
to_state_id=workflow_state1['id'])
assert_error(result, '__after', 'Backward transition in workflow state graph')
def test_update_invalid(self):
workflow_transition = ckanext_factories.WorkflowTransition()
- result, obj = self._test_action('update', 'workflow_transition',
- exception_class=tk.ValidationError,
+ result, obj = self._test_action('workflow_transition_update', should_error=True,
id=workflow_transition['id'])
assert_error(result, 'message', 'A workflow transition cannot be updated. Delete it and create a new one instead.')
def test_delete_valid(self):
workflow_transition = ckanext_factories.WorkflowTransition()
- self._test_action('delete', 'workflow_transition',
- model_class=ckanext_model.WorkflowTransition,
+ self._test_action('workflow_transition_delete',
id=workflow_transition['id'])