Merge pull request #290 from sartography/auto-set-primary-bpmn-142

Auto set primary bpmn #142

crc/api/file.py

@@ -6,7 +6,7 @@ from flask import send_file
 
 from crc import session
 from crc.api.common import ApiError
-from crc.models.file import FileSchema, FileModel, File, FileModelSchema, FileDataModel
+from crc.models.file import FileSchema, FileModel, File, FileModelSchema, FileDataModel, FileType
 from crc.models.workflow import WorkflowSpecModel
 from crc.services.file_service import FileService
 
@@ -47,9 +47,16 @@ def add_file(workflow_spec_id=None, workflow_id=None, form_field_key=None):
                                                    name=file.filename, content_type=file.content_type,
                                                    binary_data=file.stream.read())
     elif workflow_spec_id:
+        # check if we have a primary already
+        have_primary = FileModel.query.filter(FileModel.workflow_spec_id==workflow_spec_id, FileModel.type==FileType.bpmn, FileModel.primary==True).all()
+        # set this to primary if we don't already have one
+        if not have_primary:
+            primary = True
+        else:
+            primary = False
         workflow_spec = session.query(WorkflowSpecModel).filter_by(id=workflow_spec_id).first()
         file_model = FileService.add_workflow_spec_file(workflow_spec, file.filename, file.content_type,
-                                                        file.stream.read())
+                                                        file.stream.read(), primary=primary)
     else:
         raise ApiError("invalid_file", "You must supply either a workflow spec id or a workflow_id and form_field_key.")
 

tests/base_test.py

@@ -457,3 +457,9 @@ class BaseTest(unittest.TestCase):
 
         if 'impersonate_user' in g:
             del g.impersonate_user
+
+    def minimal_bpmn(self, content):
+        """Returns a bytesIO object of a well formed BPMN xml file with some string content of your choosing."""
+        minimal_dbpm = "<x><process id='1' isExecutable='false'><startEvent id='a'/></process>%s</x>"
+        return (minimal_dbpm % content).encode()
+

tests/files/test_files_api.py

@@ -13,11 +13,6 @@ from example_data import ExampleDataLoader
 
 class TestFilesApi(BaseTest):
 
-    def minimal_bpmn(self, content):
-        """Returns a bytesIO object of a well formed BPMN xml file with some string content of your choosing."""
-        minimal_dbpm = "<x><process id='1' isExecutable='false'><startEvent id='a'/></process>%s</x>"
-        return (minimal_dbpm % content).encode()
-
     def test_list_files_for_workflow_spec(self):
         self.load_example_data(use_crc_data=True)
         spec_id = 'core_info'

tests/test_auto_set_primary_bpmn.py

@@ -0,0 +1,39 @@
+from tests.base_test import BaseTest
+from crc import session
+from crc.models.file import FileModel, FileType
+from crc.models.workflow import WorkflowSpecModel, WorkflowSpecModelSchema, WorkflowSpecCategoryModel
+import io
+import json
+
+
+class TestAutoSetPrimaryBPMN(BaseTest):
+
+    def test_auto_set_primary_bpmn(self):
+        self.load_example_data()
+        category_id = session.query(WorkflowSpecCategoryModel).first().id
+        # Add a workflow spec
+        spec = WorkflowSpecModel(id='make_cookies', name='make_cookies', display_name='Cooooookies',
+                                 description='Om nom nom delicious cookies', category_id=category_id)
+        rv = self.app.post('/v1.0/workflow-specification',
+                           headers=self.logged_in_headers(),
+                           content_type="application/json",
+                           data=json.dumps(WorkflowSpecModelSchema().dump(spec)))
+        self.assert_success(rv)
+        # grab the spec from the db
+        db_spec = session.query(WorkflowSpecModel).filter_by(id='make_cookies').first()
+        self.assertEqual(spec.display_name, db_spec.display_name)
+        # Make sure we don't already have a primary bpmn file
+        have_primary = FileModel.query.filter(FileModel.workflow_spec_id==db_spec.id, FileModel.type==FileType.bpmn, FileModel.primary==True).all()
+        self.assertEqual(have_primary, [])
+        data = {}
+        data['file'] = io.BytesIO(self.minimal_bpmn("abcdef")), 'my_new_file.bpmn'
+
+        # Add a new BPMN file to the specification
+        rv = self.app.post('/v1.0/file?workflow_spec_id=%s' % db_spec.id, data=data, follow_redirects=True,
+                           content_type='multipart/form-data', headers=self.logged_in_headers())
+        self.assert_success(rv)
+        file_id = rv.json['id']
+        # Make sure we now have a primary bpmn
+        have_primary = FileModel.query.filter(FileModel.workflow_spec_id==db_spec.id, FileModel.type==FileType.bpmn, FileModel.primary==True).all()
+        self.assertEqual(len(have_primary), 1)
+        self.assertEqual(file_id, have_primary[0].id)