fix: create submission when running single job

Create a submission when running a job for a queue from a params file (JSON) so that progress can be monitored.
ga4gh · Sep 3, 2018 · 70e03b0 · 70e03b0
1 parent 8af530e
commit 70e03b0
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 14 deletions.
diff --git a/synorchestrator/orchestrator.py b/synorchestrator/orchestrator.py
@@ -23,6 +23,7 @@
 from synorchestrator.trs2wes import store_verification
 from synorchestrator.queue import get_submission_bundle
 from synorchestrator.queue import get_submissions
+from synorchestrator.queue import create_submission
 from synorchestrator.queue import update_submission
 from synorchestrator.queue import submission_queue
 
@@ -42,6 +43,9 @@ def run_job(queue_id, wes_id, wf_jsonyaml, add_attachments=None):
         wf_attachments += add_attachments
         wf_attachments = list(set(wf_attachments))
 
+    submission_id = create_submission(queue_id=queue_id, 
+                                      submission_data=wf_jsonyaml, 
+                                      wes_id=wes_id)
     wes_instance = WES(wes_id)
     request = {'workflow_url': wf_config['workflow_url'],
                'workflow_params': wf_jsonyaml,
@@ -50,6 +54,9 @@ def run_job(queue_id, wes_id, wf_jsonyaml, add_attachments=None):
     run_log['start_time'] = dt.datetime.now().ctime()
     run_status = wes_instance.get_run_status(run_log['run_id'])['state']
     run_log['status'] = run_status
+
+    update_submission(queue_id, submission_id, 'run_log', run_log)
+    update_submission(queue_id, submission_id, 'status', 'SUBMITTED')
     return run_log
 
 
@@ -131,8 +138,12 @@ def monitor_queue(queue_id):
 
         if run_log['status'] == 'COMPLETE':
             wf_config = queue_config()[queue_id]
-            store_verification(wf_config['target_queue'], submission['wes_id'])
-            update_submission(queue_id, sub_id, 'status', 'VALIDATED')
+            sub_status = run_log['status']
+            if 'target_queue' in wf_config:
+                store_verification(wf_config['target_queue'], 
+                                   submission['wes_id'])
+                sub_status = 'VALIDATED'
+            update_submission(queue_id, sub_id, 'status', sub_status)
 
         run_log['wes_id'] = submission['wes_id']
         queue_log[sub_id] = run_log

diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py
@@ -20,10 +20,12 @@ def test_run_job(mock_queue_config, mock_submission, mock_wes, monkeypatch):
                         lambda: mock_queue_config)
     monkeypatch.setattr('synorchestrator.orchestrator.fetch_queue_workflow', 
                         lambda x: mock_queue_config[x])
-    monkeypatch.setattr('synorchestrator.orchestrator.get_submission_bundle', 
-                        lambda x,y: mock_submission['mock_sub'])
+    monkeypatch.setattr('synorchestrator.testbed.create_submission', 
+                        lambda **kwargs: None)
     monkeypatch.setattr('synorchestrator.orchestrator.WES', 
                         lambda wes_id: mock_wes)
+    monkeypatch.setattr('synorchestrator.orchestrator.update_submission', 
+                        lambda w,x,y,z: None)
 
     mock_request = {'workflow_url': None,
                     'workflow_params': mock_submission['mock_sub']['data'],
@@ -144,15 +146,5 @@ def test_monitor_queue(mock_submission, mock_queue_log, mock_wes, monkeypatch):
     assert test_queue_log == mock_queue_log
 
 
-# def test_monitor(mock_queue_config, mock_queue_log, monkeypatch):
-#     monkeypatch.setattr('synorchestrator.orchestrator.queue_config', 
-#                         lambda: mock_queue_config)
-#     mock_queue_log['mock_sub']['status'] = 'COMPLETE'
-#     monkeypatch.setattr('synorchestrator.orchestrator.monitor_queue', 
-#                         lambda x: mock_queue_log)
-
-#     test_statuses = monitor()
-#     assert test_statuses == [mock_queue_log, mock_queue_log]
-
 
 
diff --git a/tests/test_testbed.py b/tests/test_testbed.py
@@ -44,6 +44,8 @@ def test_check_workflow(mock_queue_config,
                         lambda x,y: 'mock_wf_checker')
     monkeypatch.setattr('synorchestrator.testbed.add_queue', 
                         lambda **kwargs: None)
+    monkeypatch.setattr('synorchestrator.testbed.create_submission', 
+                        lambda **kwargs: None)
     mock_trs.get_workflow_tests.return_value = [{'content': '', 'url': ''}]
 
     mock_submission_log = {