Merge pull request #182 from neutrons/acquiring_status

Add acquiring status for before data is ready

docker-compose.yml

@@ -105,6 +105,9 @@ services:
   activemq:
     image: apache/activemq-classic
     hostname: activemq
+    ports:
+     - 8161:8161
+     - 61613:61613
     volumes:
       - ./src/workflow_app/workflow/icat_activemq.xml:/opt/apache-activemq/conf/activemq.xml
     healthcheck:

src/webmon_app/reporting/report/view_util.py

@@ -356,8 +356,7 @@ def is_acquisition_complete(run_id):
 
     :param run_id: run object
     """
-    status_items = RunStatus.objects.filter(run_id=run_id, queue_id__name="POSTPROCESS.DATA_READY")
-    return len(status_items) > 0
+    return RunStatus.objects.filter(run_id=run_id, queue_id__name="POSTPROCESS.DATA_READY").count() > 0
 
 
 def get_post_processing_status(red_timeout=0.25, yellow_timeout=120):
@@ -386,7 +385,9 @@ def get_run_status_text(run_id, show_error=False, use_element_id=False):
         else:
             element_id = ""
         s = WorkflowSummary.objects.get(run_id=run_id)
-        if s.complete is True:
+        if not is_acquisition_complete(run_id):
+            status = "<span %s>acquiring</span>" % element_id
+        elif s.complete is True:
             status = "<span %s class='green'>complete</span>" % element_id
         else:
             last_error = run_id.last_error()

src/webmon_app/reporting/templates/dasmon/run_summary.html

@@ -83,6 +83,7 @@
   <div style="float: right; margin-left:15px;">
     Status: <select id="search-status" name="status">
       <option></option>
+      <option>acquiring</option>
       <option>complete</option>
       <option>incomplete</option>
       <option>error</option>

src/webmon_app/reporting/tests/test_report/test_view_util.py

@@ -55,7 +55,7 @@ def setUpTestData(cls):
     def tearDownClass(cls):
         Instrument.objects.get(name="test_instrument").delete()
         IPTS.objects.get(expt_name="test_exp").delete()
-        StatusQueue.objects.get(name="test").delete()
+        StatusQueue.objects.all().delete()
         DataRun.objects.all().delete()
         WorkflowSummary.objects.all().delete()
 
@@ -221,7 +221,7 @@ def test_get_run_status_text(self):
         self.assertTrue("run_id_" in rst)
         # no show element id, green status
         rst = get_run_status_text(run_id)
-        self.assertTrue("green" in rst)
+        self.assertTrue("acquiring" in rst)
         # no show element, red status
         ipts = IPTS.objects.get(expt_name="test_exp")
         run = DataRun.objects.create(
@@ -231,12 +231,26 @@ def test_get_run_status_text(self):
             file="/tmp/test_65535.nxs",
         )
         run.save()
-        WorkflowSummary.objects.create(
+        workflowSummary = WorkflowSummary.objects.create(
             run_id=run,
             complete=False,
-        ).save()
+        )
+        workflowSummary.save()
+
+        rst = get_run_status_text(run)
+        self.assertTrue("acquiring" in rst)
+        dataReady = StatusQueue(name="POSTPROCESS.DATA_READY", is_workflow_input=True)
+        dataReady.save()
+        RunStatus.objects.create(run_id=run, queue_id=dataReady).save()
         rst = get_run_status_text(run)
         self.assertTrue("red" in rst)
+        self.assertTrue("incomplete" in rst)
+
+        workflowSummary.complete = True
+        workflowSummary.save()
+        rst = get_run_status_text(run)
+        self.assertTrue("green" in rst)
+        self.assertTrue("complete" in rst)
 
     def test_get_run_list_dict(self):
         from reporting.report.view_util import get_run_list_dict

tests/conftest.py

@@ -2,6 +2,7 @@
 from dotenv import dotenv_values
 import psycopg2
 import pytest
+import stomp
 
 # standard imports
 import requests
@@ -76,3 +77,14 @@ def db_connection():
     time.sleep(1)
     yield conn
     conn.close()
+
+
+@pytest.fixture(scope="session")
+def amq_connection():
+    """activemq connection with config from env files"""
+    config = dotenv_values(".env")
+    assert config
+    conn = stomp.Connection(host_and_ports=[("localhost", 61613)])
+    conn.connect(config["ICAT_USER"], config["ICAT_PASS"], wait=True)
+    yield conn
+    conn.disconnect()

tests/test_SMS_messages.py

@@ -0,0 +1,88 @@
+"""Test the status acquiring appears when a SMS message is received and before the data is ready"""
+
+import time
+import json
+import tests.utils.db as db_utils
+
+
+class TestSMSQueues:
+    instrument = "arcs"
+    IPTS = "IPTS-11111"
+    user = "InstrumentScientist"
+    pwd = "InstrumentScientist"
+
+    def create_and_send_msg(self, conn, run_number):
+        conn.send(
+            f"/topic/SNS.{self.instrument.upper()}.APP.SMS",
+            json.dumps(
+                {
+                    "instrument": self.instrument,
+                    "facility": "SNS",
+                    "ipts": self.IPTS,
+                    "run_number": run_number,
+                    "data_file": "",
+                    "reason": "SMS run started",
+                    "msg_type": "0",
+                }
+            ),
+        )
+
+    def clear_run(self, conn, run_number):
+        # remove everything for this run
+        cursor = conn.cursor()
+        cursor.execute("SELECT id FROM report_instrument where name = %s;", (self.instrument,))
+        inst_id = cursor.fetchone()
+        if inst_id is None:
+            return
+
+        cursor.execute(
+            "SELECT id FROM report_datarun WHERE instrument_id_id = %s AND run_number = %s;", (inst_id, run_number)
+        )
+        run_id = cursor.fetchone()
+        if run_id is None:
+            return
+
+        db_utils.clear_previous_runstatus(conn, run_id)
+        cursor.execute("DELETE FROM report_workflowsummary WHERE run_id_id = %s;", run_id)
+        cursor.execute("DELETE FROM report_instrumentstatus WHERE last_run_id_id = %s;", run_id)
+        cursor.execute("DELETE FROM report_datarun WHERE id = %s;", (run_id))
+        conn.commit()
+        cursor.close()
+
+    def test_acquiring(self, amq_connection, db_connection, request_page):
+        # remove data run so the tests always starts fresh
+        self.clear_run(db_connection, 100)
+        self.clear_run(db_connection, 101)
+        # send SMS message for 2 runs
+        self.create_and_send_msg(amq_connection, 100)
+        self.create_and_send_msg(amq_connection, 101)
+
+        # wait a second while things run
+        time.sleep(1)
+
+        # check IPTS page /report/arcs/experiment/IPTS-11111/ for acquiring
+        response = request_page("/report/arcs/experiment/IPTS-11111/", self.user, self.pwd)
+        assert response.status_code == 200
+        assert response.text.count("acquiring") == 2
+
+        # now send data_ready for run 100 only and check that it is no longer acquiring
+        self.create_and_send_msg(amq_connection, 100)
+
+        amq_connection.send(
+            "/queue/POSTPROCESS.DATA_READY",
+            json.dumps(
+                {
+                    "instrument": self.instrument,
+                    "facility": "SNS",
+                    "ipts": self.IPTS,
+                    "run_number": 100,
+                    "data_file": "",
+                }
+            ),
+        )
+        time.sleep(1)
+
+        # check IPTS page /report/arcs/experiment/IPTS-11111/ for acquiring
+        response = request_page("/report/arcs/experiment/IPTS-11111/", self.user, self.pwd)
+        assert response.status_code == 200
+        assert response.text.count("acquiring") == 1  # 101 is still acquiring but not 100