WS-290: Fix QA

schema/versions/ced8e001d262_change_sendmessage_qa_ready_action_to_.py

+"""change SendMessage qa_ready action to AnnounceQa action
+
+Revision ID: ced8e001d262
+Revises: acfdeb6777cb
+Create Date: 2021-11-10 10:16:09.070171
+
+"""
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "ced8e001d262"
+down_revision = "acfdeb6777cb"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    """
+    Modify action that gets triggered on QA workflow complete;
+    Change it to perform the AnnounceQa action instead of attempting to SendMessage with arguments qa_ready
+    """
+    op.execute(
+        """
+        UPDATE capability_state_actions SET action_type = 'AnnounceQa', arguments = null
+            WHERE action_type = 'SendMessage'
+                AND arguments = 'qa_ready'
+                AND transition_id = (
+                    SELECT transition_id FROM capability_state_transitions
+                    WHERE pattern = 'type == workflow-complete'
+                        AND capability_name = 'std_calibration'
+                )
+        """
+    )
+
+
+def downgrade():
+    """
+    Undo above change
+    """
+    op.execute(
+        """
+        UPDATE capability_state_actions SET action_type = 'SendMessage', arguments = 'qa_ready'
+            WHERE action_type = 'AnnounceQa'
+                AND arguments IS null
+                AND transition_id = (
+                    SELECT transition_id FROM capability_state_transitions
+                    WHERE pattern = 'type == workflow-complete'
+                        AND capability_name = 'std_calibration'
+                )
+        """
+    )

shared/workspaces/test/test_capability_actions.py

@@ -20,6 +20,7 @@
 Testing capability actions
 """
 from typing import Dict, List, Optional
+from unittest.mock import patch
 
 from hypothesis import given
 from hypothesis import strategies as st
@@ -30,6 +31,7 @@ from workspaces.capability.enums import (
     QueueState,
 )
 from workspaces.capability.schema import (
+    AnnounceQa,
     Capability,
     CapabilityExecution,
     CapabilityRequest,
@@ -141,7 +143,7 @@ st.register_type_strategy(
             "queue_state": st.sampled_from([name for name, _ in QueueState.__members__.items()]),
             "capability_request_id": st.integers(min_value=SQLITE_MIN_INT, max_value=SQLITE_MAX_INT),
             "version_number": st.integers(min_value=SQLITE_MIN_INT, max_value=SQLITE_MAX_INT),
-            "current_workflow_request_id": st.just(-11),
+            "current_workflow_request_id": st.integers(min_value=1, max_value=15),
             "delivery_url": st.from_regex(r"https:\/\/dl-nrao-unittest\.aoc\.nrao\.edu(?:\/[A-z0-9]*)+"),
             "created_at": st.datetimes().map(
                 lambda time: time.isoformat(),
@@ -159,7 +161,7 @@ st.register_type_strategy(
     st.from_type(Capability),
     st.integers(min_value=1, max_value=10),
 )
-def test_qapass(
+def test_qa_pass(
     mock_capability_info: CapabilityInfo,
     dummy_execution_manager: ExecutionManager,
     parameters: Optional[List[str]],
@@ -194,3 +196,17 @@ def test_qapass(
                 assert actual_version.state == CapabilityVersionState.Complete.name
             else:
                 assert actual_version.state == CapabilityVersionState.Failed.name
+
+
+@given(st.from_type(CapabilityExecution))
+def test_announce_qa(execution_json: Dict):
+    execution = CapabilityExecution.from_json(execution_json)
+    fake_workflow_url = "http://fake-workflow.edu"
+
+    with patch("workspaces.capability.schema.requests.post") as mock_post:
+        with patch("workspaces.capability.schema.CapoConfig") as mock_capo_config:
+            mock_capo_config.return_value.settings.return_value.serviceUrl = fake_workflow_url
+            AnnounceQa()(execution)
+            mock_post.assert_called_with(
+                f"{fake_workflow_url}/workflows/requests/{execution.current_workflow_request_id}/qa/qa_ready"
+            )

shared/workspaces/workspaces/capability/schema.py

@@ -28,7 +28,9 @@ import re
 from typing import Dict, List
 
 import pendulum
+import requests
 import sqlalchemy as sa
+from pycapo import CapoConfig
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import relationship
 
@@ -87,9 +89,7 @@ class Action(Base, ActionIF):
 
 
 class Noop(Action):
-    __mapper_args__ = {
-        "polymorphic_identity": "Noop",
-    }
+    __mapper_args__ = {"polymorphic_identity": "Noop"}
 
     def __call__(self, execution: CapabilityExecutionIF, manager: ExecutionManagerIF):
         # empty because this is the no-operation action
@@ -97,9 +97,7 @@ class Noop(Action):
 
 
 class ExecuteWorkflow(Action):
-    __mapper_args__ = {
-        "polymorphic_identity": "ExecuteWorkflow",
-    }
+    __mapper_args__ = {"polymorphic_identity": "ExecuteWorkflow"}
 
     def __call__(self, execution: CapabilityExecutionIF, manager: ExecutionManagerIF):
         logger.info("Executing workflow %s for execution %s", self.arguments, execution.id)
@@ -121,9 +119,7 @@ class ExecuteWorkflow(Action):
 
 
 class QueueWorkflow(Action):
-    __mapper_args__ = {
-        "polymorphic_identity": "QueueWorkflow",
-    }
+    __mapper_args__ = {"polymorphic_identity": "QueueWorkflow"}
 
     def __call__(self, execution: CapabilityExecutionIF, manager: ExecutionManagerIF):
         # message the execution manager
@@ -132,9 +128,7 @@ class QueueWorkflow(Action):
 
 
 class SendNotification(Action):
-    __mapper_args__ = {
-        "polymorphic_identity": "SendNotification",
-    }
+    __mapper_args__ = {"polymorphic_identity": "SendNotification"}
 
     def __call__(self, execution: CapabilityExecutionIF, manager: ExecutionManagerIF):
         # requests.post(f"/notifications/{self.notification}/send", users=self.user)
@@ -147,9 +141,7 @@ class SendMessage(Action):
     Action that simply sends an AMQP message with the type of the given arguments
     """
 
-    __mapper_args__ = {
-        "polymorphic_identity": "SendMessage",
-    }
+    __mapper_args__ = {"polymorphic_identity": "SendMessage"}
 
     def __call__(self, execution: CapabilityExecutionIF, manager: ExecutionManagerIF):
         message_type = self.arguments
@@ -164,15 +156,14 @@ class QaPass(Action):
     Action that performs system-level operations required by QA-passing a capability request
     """
 
-    __mapper_args__ = {
-        "polymorphic_identity": "QaPass",
-    }
+    __mapper_args__ = {"polymorphic_identity": "QaPass"}
 
     def _fail_other_versions(self, request: CapabilityRequest):
         """
-        Given a capability request, set the executions of all of its versions besides the current version to Failed
+        Given a capability request, set the state of the current version to Complete, and the state of all other
+        versions to Failed
 
-        :param request: capability request
+        :param request: Capability request
         """
         versions = request.versions
         for version in versions:
@@ -183,12 +174,36 @@ class QaPass(Action):
                 # Mark all other versions as failed
                 version.state = CapabilityVersionState.Failed.name
 
-    def __call__(self, execution: CapabilityExecutionIF, manager: ExecutionManagerIF):
+    def _abort_running_executions(self, request: CapabilityRequest):
+        """
+        Given a capability request, find all running executions associated with it and abort them
+
+        :param request: Capability request
+        """
+        raise NotImplementedError
+
+    def __call__(self, execution: CapabilityExecutionIF, *args):
         logger.info("Execution #%s has passed QA", execution.id)
         request = execution.capability_request
         self._fail_other_versions(request)
 
 
+class AnnounceQa(Action):
+    """
+    Action that makes a REST call to the announce_qa endpoint in the workflow service, which announces that an execution
+    is ready for QA
+    """
+
+    __mapper_args__ = {"polymorphic_identity": "AnnounceQa"}
+
+    def __call__(self, execution: CapabilityExecutionIF, *args):
+        workflow_request_id = execution.current_workflow_request_id
+        msg_type = "qa_ready"
+        workflow_service_url = CapoConfig().settings("edu.nrao.workspaces.WorkflowSettings").serviceUrl
+        url = f"{workflow_service_url}/workflows/requests/{workflow_request_id}/qa/{msg_type}"
+        requests.post(url)
+
+
 class State(Base):
     transitions: List[Transition]
 

shared/workspaces/workspaces/workflow/services/workflow_service.py

@@ -27,8 +27,6 @@ from pathlib import Path
 from tempfile import mkdtemp
 from typing import Dict, List, Union
 
-# pylint: disable=E0401, W1203
-
 import requests
 import transaction
 from messaging.messenger import MessageSender
@@ -45,6 +43,8 @@ from workspaces.workflow.message_architect import (
 from workspaces.workflow.schema import Workflow, WorkflowRequest, WorkflowRequestFile
 from workspaces.workflow.services.interfaces import WorkflowInfoIF, WorkflowServiceIF
 
+# pylint: disable=E0401, W1203
+
 logger = logging.getLogger(__name__)
 
 
@@ -574,7 +574,7 @@ class WorkflowService(WorkflowServiceIF):
         wf_request = self.info.lookup_workflow_request(workflow_request_id)
 
         qa_event_msg = WorkflowMessageArchitect(request=wf_request).compose_message(msg_type)
-        self.archive_messenger.send_message(**qa_event_msg)
+        self.messenger.send_message(**qa_event_msg)
 
         return qa_event_msg