From 3392e392a353074aa89556ab345e62374470248c Mon Sep 17 00:00:00 2001
From: Nathan Hertz <nhertz@nrao.edu>
Date: Fri, 12 Nov 2021 11:52:03 -0500
Subject: [PATCH] WS-290: Fix QA
---
..._change_sendmessage_qa_ready_action_to_.py | 51 +++++++++++++++++
.../test/test_capability_actions.py | 20 ++++++-
.../workspaces/capability/schema.py | 57 ++++++++++++-------
.../workflow/services/workflow_service.py | 6 +-
4 files changed, 108 insertions(+), 26 deletions(-)
create mode 100644 schema/versions/ced8e001d262_change_sendmessage_qa_ready_action_to_.py
diff --git a/schema/versions/ced8e001d262_change_sendmessage_qa_ready_action_to_.py b/schema/versions/ced8e001d262_change_sendmessage_qa_ready_action_to_.py
new file mode 100644
index 000000000..ee2b9787c
--- /dev/null
+++ b/schema/versions/ced8e001d262_change_sendmessage_qa_ready_action_to_.py
@@ -0,0 +1,51 @@
+"""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'
+ )
+ """
+ )
diff --git a/shared/workspaces/test/test_capability_actions.py b/shared/workspaces/test/test_capability_actions.py
index 7ab60b86f..81a209a0b 100644
--- a/shared/workspaces/test/test_capability_actions.py
+++ b/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"
+ )
diff --git a/shared/workspaces/workspaces/capability/schema.py b/shared/workspaces/workspaces/capability/schema.py
index 749c4f078..106b6ac53 100644
--- a/shared/workspaces/workspaces/capability/schema.py
+++ b/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]
diff --git a/shared/workspaces/workspaces/workflow/services/workflow_service.py b/shared/workspaces/workspaces/workflow/services/workflow_service.py
index e07ead320..81dbddbbc 100644
--- a/shared/workspaces/workspaces/workflow/services/workflow_service.py
+++ b/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
--
GitLab