Update tests for workflow changes

ckanext/metadata/logic/action/delete.py

@@ -308,14 +308,6 @@ def workflow_state_delete(context, data_dict):
     for (workflow_transition_id,) in workflow_transition_ids:
         tk.get_action('workflow_transition_delete')(cascade_context, {'id': workflow_transition_id})
 
-    # cascade delete to dependent workflow rules
-    workflow_rule_ids = session.query(ckanext_model.WorkflowRule.id) \
-        .filter(ckanext_model.WorkflowRule.workflow_state_id == workflow_state_id) \
-        .filter(ckanext_model.WorkflowRule.state != 'deleted') \
-        .all()
-    for (workflow_rule_id,) in workflow_rule_ids:
-        tk.get_action('workflow_rule_delete')(cascade_context, {'id': workflow_rule_id})
-
     rev = model.repo.new_revision()
     rev.author = user
     rev.message = _(u'REST API: Delete workflow state %s') % workflow_state_id

ckanext/metadata/tests/__init__.py

@@ -22,8 +22,7 @@
     '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,
+    'workflow_annotation': ckanext_model.WorkflowAnnotation,
 }
 
 
@@ -46,14 +45,18 @@ def generate_name(*strings):
     return re.sub('[^a-z0-9_\-]+', '-', text.lower())
 
 
-def assert_object_matches_dict(object_, dict_):
+def assert_object_matches_dict(object_, dict_, json_values=()):
     """
     Check that the object has all the items in the dict.
     """
     for key in dict_.keys():
         # any kind of empty matches any kind of empty
         dict_value = dict_[key] or None
         object_value = getattr(object_, key) or None
+        if key in json_values:
+            if isinstance(dict_value, basestring):
+                dict_value = json.loads(dict_value)
+            object_value = json.loads(object_value)
         assert dict_value == object_value
 
 

ckanext/metadata/tests/factories.py

@@ -145,7 +145,8 @@ class WorkflowState(factory.Factory):
     title = factory.LazyAttribute(lambda obj: obj.name.replace('_', ' ').title())
     description = 'A test description for this test workflow state.'
     revert_state_id = ''
-    private = False
+    metadata_records_private = False
+    workflow_rules_json = '{"type": "object"}'
 
     @classmethod
     def _build(cls, target_class, *args, **kwargs):
@@ -187,32 +188,10 @@ def _create(cls, target_class, *args, **kwargs):
                                    **kwargs)
 
 
-class WorkflowMetric(factory.Factory):
-    FACTORY_FOR = ckanext_model.WorkflowMetric
+class WorkflowAnnotation(factory.Factory):
+    FACTORY_FOR = ckanext_model.WorkflowAnnotation
 
-    name = factory.Sequence(lambda n: 'test_workflow_metric_{0:02d}'.format(n))
-    title = factory.LazyAttribute(lambda obj: obj.name.replace('_', ' ').title())
-    description = 'A test description for this test workflow metric.'
-    evaluator_url = 'http://example.net/'
-
-    @classmethod
-    def _build(cls, target_class, *args, **kwargs):
-        raise NotImplementedError(".build() isn't supported in CKAN")
-
-    @classmethod
-    def _create(cls, target_class, *args, **kwargs):
-        if args:
-            assert False, "Positional args aren't supported, use keyword args."
-
-        context = {'user': ckan_factories._get_action_user_name(kwargs)}
-
-        return helpers.call_action('workflow_metric_create', context=context, **kwargs)
-
-
-class WorkflowRule(factory.Factory):
-    FACTORY_FOR = ckanext_model.WorkflowRule
-
-    rule_json = '{ "testkey": "testvalue" }'
+    workflow_annotation_json = '{ "workflowkey": "workflowvalue" }'
 
     @classmethod
     def _build(cls, target_class, *args, **kwargs):
@@ -224,11 +203,9 @@ def _create(cls, target_class, *args, **kwargs):
             assert False, "Positional args aren't supported, use keyword args."
 
         context = {'user': ckan_factories._get_action_user_name(kwargs)}
-        workflow_state_id = kwargs.pop('workflow_state_id', None) or WorkflowState()['id']
-        workflow_metric_id = kwargs.pop('workflow_metric_id', None) or WorkflowMetric()['id']
+        metadata_record_id = kwargs.pop('metadata_record_id', None) or MetadataRecord()['id']
 
-        return helpers.call_action('workflow_rule_create',
+        return helpers.call_action('workflow_annotation_create',
                                    context=context,
-                                   workflow_state_id=workflow_state_id,
-                                   workflow_metric_id=workflow_metric_id,
+                                   metadata_record_id=metadata_record_id,
                                    **kwargs)

ckanext/metadata/tests/test_workflow_annotation_actions.py

@@ -0,0 +1,97 @@
+# encoding: utf-8
+
+from ckan.tests.helpers import call_action
+
+from ckanext.metadata.tests import (
+    ActionTestBase,
+    make_uuid,
+    assert_object_matches_dict,
+    assert_error,
+    factories as ckanext_factories,
+)
+
+
+class TestWorkflowAnnotationActions(ActionTestBase):
+
+    def test_create_valid(self):
+        metadata_record = ckanext_factories.MetadataRecord()
+        input_dict = {
+            'metadata_record_id': metadata_record['id'],
+            'workflow_annotation_json': '{ "testkey": "testvalue" }',
+        }
+        result, obj = self._test_action('workflow_annotation_create', **input_dict)
+        assert_object_matches_dict(obj, input_dict)
+
+    def test_create_valid_byname(self):
+        metadata_record = ckanext_factories.MetadataRecord()
+        input_dict = {
+            'metadata_record_id': metadata_record['name'],
+            'workflow_annotation_json': '{ "testkey": "testvalue" }',
+        }
+        result, obj = self._test_action('workflow_annotation_create', **input_dict)
+        input_dict = {
+            'metadata_record_id': metadata_record['id'],
+            'workflow_annotation_json': '{ "testkey": "testvalue" }',
+        }
+        assert_object_matches_dict(obj, input_dict)
+
+    def test_create_valid_sysadmin_setid(self):
+        metadata_record = ckanext_factories.MetadataRecord()
+        input_dict = {
+            'id': make_uuid(),
+            'metadata_record_id': metadata_record['id'],
+            'workflow_annotation_json': '{ "testkey": "testvalue" }',
+        }
+        result, obj = self._test_action('workflow_annotation_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('workflow_annotation_create', should_error=True)
+        assert_error(result, 'metadata_record_id', 'Missing parameter')
+        assert_error(result, 'workflow_annotation_json', 'Missing parameter')
+
+    def test_create_invalid_missing_values(self):
+        result, obj = self._test_action('workflow_annotation_create', should_error=True,
+                                        metadata_record_id='',
+                                        workflow_annotation_json='')
+        assert_error(result, 'metadata_record_id', 'Missing value')
+        assert_error(result, 'workflow_annotation_json', 'Missing value')
+
+    def test_create_invalid_nonsysadmin_setid(self):
+        result, obj = self._test_action('workflow_annotation_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_annotation = ckanext_factories.WorkflowAnnotation()
+        result, obj = self._test_action('workflow_annotation_create', should_error=True, sysadmin=True, check_auth=True,
+                                        id=workflow_annotation['id'])
+        assert_error(result, 'id', 'Already exists: Workflow Annotation')
+
+    def test_create_invalid_bad_references(self):
+        result, obj = self._test_action('workflow_annotation_create', should_error=True,
+                                        metadata_record_id='foo')
+        assert_error(result, 'metadata_record_id', 'Not found: Metadata Record')
+
+    def test_create_invalid_deleted_references(self):
+        metadata_record = ckanext_factories.MetadataRecord()
+        call_action('metadata_record_delete', id=metadata_record['id'])
+        result, obj = self._test_action('workflow_annotation_create', should_error=True,
+                                        metadata_record_id=metadata_record['id'])
+        assert_error(result, 'metadata_record_id', 'Not found: Metadata Record')
+
+    def test_create_invalid_not_json(self):
+        result, obj = self._test_action('workflow_annotation_create', should_error=True,
+                                        workflow_annotation_json='not json')
+        assert_error(result, 'workflow_annotation_json', 'JSON decode error')
+
+    def test_update_invalid(self):
+        workflow_annotation = ckanext_factories.WorkflowAnnotation()
+        result, obj = self._test_action('workflow_annotation_update', should_error=True,
+                                        id=workflow_annotation['id'])
+        assert_error(result, 'message', 'A workflow annotation cannot be updated. Delete it and create a new one instead.')
+
+    def test_delete_valid(self):
+        workflow_annotation = ckanext_factories.WorkflowAnnotation()
+        self._test_action('workflow_annotation_delete',
+                          id=workflow_annotation['id'])