From 47bbcb3f32c665a441c3d80d5575a2ad7719e249 Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Sun, 16 Aug 2026 22:53:58 +1000 Subject: [PATCH 1/2] Adds creator auditing for custom workflow-run states --- .../migrations/0024_state_created_by.py | 18 +++ app/workflow_manager/models/state.py | 1 + app/workflow_manager/serializers/state.py | 4 +- .../tests/test_event_bridge.py | 17 +- .../tests/test_state_viewset.py | 151 +++++++++++++++++- app/workflow_manager/viewsets/state.py | 46 +++++- .../domain/event/wrsc.py | 3 +- .../services/workflow_run.py | 9 +- .../tests/test_workflow_run.py | 23 ++- .../WorkflowRunStateChange.schema.json | 3 + .../WorkflowRunStateChange.schema.yaml | 3 + .../examples/WRSC__example1.json | 2 +- .../examples/WRSC__example2.json | 5 +- 13 files changed, 267 insertions(+), 18 deletions(-) create mode 100644 app/workflow_manager/migrations/0024_state_created_by.py diff --git a/app/workflow_manager/migrations/0024_state_created_by.py b/app/workflow_manager/migrations/0024_state_created_by.py new file mode 100644 index 0000000..181d63e --- /dev/null +++ b/app/workflow_manager/migrations/0024_state_created_by.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.12 on 2026-08-16 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("workflow_manager", "0023_alter_payload_payload_ref_id_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="state", + name="created_by", + field=models.CharField(blank=True, max_length=255, null=True), + ), + ] diff --git a/app/workflow_manager/models/state.py b/app/workflow_manager/models/state.py index dc53667..042ab08 100644 --- a/app/workflow_manager/models/state.py +++ b/app/workflow_manager/models/state.py @@ -22,6 +22,7 @@ class Meta: ) # TODO: How and where to enforce conventions? timestamp = models.DateTimeField() comment = models.CharField(max_length=255, null=True, blank=True) + created_by = models.CharField(max_length=255, null=True, blank=True) workflow_run = models.ForeignKey( WorkflowRun, related_name="states", on_delete=models.CASCADE diff --git a/app/workflow_manager/serializers/state.py b/app/workflow_manager/serializers/state.py index 32fea2f..a9678ab 100644 --- a/app/workflow_manager/serializers/state.py +++ b/app/workflow_manager/serializers/state.py @@ -8,13 +8,13 @@ class StateBaseSerializer(SerializersBase): - pass + created_by = serializers.CharField(read_only=True, allow_null=True) class StateMinSerializer(StateBaseSerializer): class Meta(OrcabusIdSerializerMetaMixin): model = State - fields = ["orcabus_id", "status", "timestamp"] + fields = ["orcabus_id", "status", "timestamp", "created_by"] class StateSerializer(StateBaseSerializer): diff --git a/app/workflow_manager/tests/test_event_bridge.py b/app/workflow_manager/tests/test_event_bridge.py index a9be306..5ab88fd 100644 --- a/app/workflow_manager/tests/test_event_bridge.py +++ b/app/workflow_manager/tests/test_event_bridge.py @@ -15,7 +15,7 @@ class WrscApiEventTestCase(SimpleTestCase): def build_event(self): return { "id": "wrsc-event-id", - "version": "1.0.0", + "version": "1.1.0", "timestamp": timezone.now().isoformat(), "orcabusId": "wfr.01J5M2JFE1JPYV62RYQEG99WR1", "portalRunId": "20260623example", @@ -46,8 +46,23 @@ def test_emit_wrsc_api_event_omits_payload(self, mock_emit_event): self.assertEqual(entry["EventBusName"], "test-event-bus") detail = json.loads(entry["Detail"]) self.assertNotIn("payload", detail) + self.assertNotIn("createdBy", detail) self.assertEqual(detail["status"], "RESOLVED") + @patch.dict(os.environ, {"EVENT_BUS_NAME": "test-event-bus"}) + @patch("workflow_manager.aws_event_bridge.event.libeb.emit_event") + def test_emit_wrsc_api_event_includes_created_by(self, mock_emit_event): + mock_emit_event.return_value = {"FailedEntryCount": 0, "Entries": [{}]} + event = self.build_event() + event["createdBy"] = "state.creator@example.com" + + emit_wrsc_api_event(event) + + entry = mock_emit_event.call_args.args[0] + detail = json.loads(entry["Detail"]) + self.assertEqual(detail["version"], "1.1.0") + self.assertEqual(detail["createdBy"], "state.creator@example.com") + @patch.dict(os.environ, {"EVENT_BUS_NAME": "test-event-bus"}) @patch("workflow_manager.aws_event_bridge.event.libeb.emit_event") def test_emit_wrsc_api_event_raises_and_logs_partial_failure(self, mock_emit_event): diff --git a/app/workflow_manager/tests/test_state_viewset.py b/app/workflow_manager/tests/test_state_viewset.py index e43ff14..757c7e0 100644 --- a/app/workflow_manager/tests/test_state_viewset.py +++ b/app/workflow_manager/tests/test_state_viewset.py @@ -2,6 +2,7 @@ from datetime import datetime, timedelta from unittest.mock import MagicMock, patch +import jwt from django.test import TestCase from django.db import DatabaseError from django.utils.timezone import make_aware @@ -21,8 +22,16 @@ class StateViewSetTestCase(TestCase): deprecate_endpoint = f"/{api_base}workflowrun/state/deprecate/" resolve_endpoint = f"/{api_base}workflowrun/state/resolve/" cancel_endpoint = f"/{api_base}workflowrun/state/cancel/" + jwt_test_secret = "state-test-secret-that-is-longer-than-32-bytes"#pragma: allowlist secret def setUp(self): + self.user_email = "state.creator@example.com" + token = jwt.encode( + {"email": self.user_email}, self.jwt_test_secret, algorithm="HS256" + ) + self.auth_header = f"Bearer {token}" + self.client.defaults["HTTP_AUTHORIZATION"] = self.auth_header + TestData().create_primary() self.wf = Workflow.objects.first() self.wfr_failed = WorkflowRun.objects.get(portal_run_id="1234") @@ -44,6 +53,21 @@ def test_list_states_returns_200(self): data = response.json() self.assertIsInstance(data, list) self.assertGreaterEqual(len(data), 1) + self.assertIn("createdBy", data[0]) + self.assertIsNone(data[0]["createdBy"]) + + def test_workflow_run_current_state_exposes_created_by(self): + StateFactory( + workflow_run=self.wfr_failed, + status="DEPRECATED", + timestamp=make_aware(datetime.now() + timedelta(days=1)), + created_by=self.user_email, + ) + + response = self.client.get(f"{self.endpoint}/{self.wfr_failed.orcabus_id}/") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["currentState"]["createdBy"], self.user_email) def test_get_states_transition_validation_map_returns_200(self): url = f"{self.endpoint}/state/get_states_transition_validation_map/" @@ -62,7 +86,7 @@ def test_state_transition_openapi_documents_all_response_shapes(self): self.assertEqual(response.status_code, 200) schema = json.loads(response.content) - expected_statuses = {"201", "207", "400", "500", "502"} + expected_statuses = {"201", "207", "400", "401", "500", "502"} transition_paths = ( self.deprecate_endpoint, self.resolve_endpoint, @@ -118,6 +142,7 @@ def test_resolve_transitions_failed_workflow_run(self, mock_emit_wrsc): data={ "workflowrunOrcabusIds": [self.wfr_failed.orcabus_id], "comment": "resolved ok", + "createdBy": "spoofed.creator@example.com", }, content_type="application/json", ) @@ -131,6 +156,7 @@ def test_resolve_transitions_failed_workflow_run(self, mock_emit_wrsc): workflow_run=self.wfr_failed, status="RESOLVED", comment="resolved ok", + created_by=self.user_email, ).exists() ) mock_emit_wrsc.assert_called_once() @@ -138,6 +164,8 @@ def test_resolve_transitions_failed_workflow_run(self, mock_emit_wrsc): self.assertEqual(wrsc_event["status"], "RESOLVED") self.assertEqual(wrsc_event["orcabusId"], self.wfr_failed.orcabus_id) self.assertEqual(wrsc_event["workflow"]["orcabusId"], self.wf.orcabus_id) + self.assertEqual(wrsc_event["createdBy"], self.user_email) + self.assertEqual(wrsc_event["version"], "1.1.0") self.assertNotIn("payload", wrsc_event) @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") @@ -159,10 +187,12 @@ def test_deprecate_transitions_succeeded_workflow_run(self, mock_emit_wrsc): workflow_run=self.wfr_succeeded, status="DEPRECATED", comment="no longer needed", + created_by=self.user_email, ).exists() ) mock_emit_wrsc.assert_called_once() self.assertEqual(mock_emit_wrsc.call_args.args[0]["status"], "DEPRECATED") + self.assertEqual(mock_emit_wrsc.call_args.args[0]["createdBy"], self.user_email) @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") def test_deprecate_rejects_failed_workflow_run(self, mock_emit_wrsc): @@ -245,6 +275,7 @@ def test_cancel_preserves_existing_allowed_source_states(self, mock_emit_wrsc): workflow_run=workflow_run, status="CANCELLED", comment="cancel transient runs", + created_by=self.user_email, ).exists() ) self.assertEqual(mock_emit_wrsc.call_count, len(source_statuses)) @@ -254,6 +285,53 @@ def test_cancel_preserves_existing_allowed_source_states(self, mock_emit_wrsc): for call in mock_emit_wrsc.call_args_list ) ) + self.assertTrue( + all( + call.args[0]["createdBy"] == self.user_email + for call in mock_emit_wrsc.call_args_list + ) + ) + + @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") + def test_state_transition_requires_valid_bearer_email(self, mock_emit_wrsc): + request_data = { + "workflowrunOrcabusIds": [self.wfr_succeeded.orcabus_id], + "comment": "no auth", + } + + response = self.client.post( + self.deprecate_endpoint, + data=request_data, + content_type="application/json", + HTTP_AUTHORIZATION="", + ) + self.assertEqual(response.status_code, 401) + + response = self.client.post( + self.deprecate_endpoint, + data=request_data, + content_type="application/json", + HTTP_AUTHORIZATION="Bearer not.a.valid.jwt", + ) + self.assertEqual(response.status_code, 401) + + token_without_email = jwt.encode( + {"sub": "state-user"}, self.jwt_test_secret, algorithm="HS256" + ) + response = self.client.post( + self.deprecate_endpoint, + data=request_data, + content_type="application/json", + HTTP_AUTHORIZATION=f"Bearer {token_without_email}", + ) + self.assertEqual(response.status_code, 401) + self.assertFalse( + State.objects.filter( + workflow_run=self.wfr_succeeded, + status="DEPRECATED", + ).exists() + ) + mock_emit_wrsc.assert_not_called() @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") def test_cancel_rejects_existing_excluded_states(self, mock_emit_wrsc): @@ -411,7 +489,9 @@ def test_update_state_comment_rejects_states_outside_validation_map(self): ) @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") - def test_update_state_comment_success(self, mock_emit_wrsc): + def test_update_legacy_state_comment_allows_any_authenticated_user( + self, mock_emit_wrsc + ): state_deprecated = StateFactory( workflow_run=self.wfr_failed, status="DEPRECATED", @@ -427,8 +507,73 @@ def test_update_state_comment_success(self, mock_emit_wrsc): self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["comment"], "updated") + self.assertIsNone(data["createdBy"]) + state_deprecated.refresh_from_db() + self.assertIsNone(state_deprecated.created_by) mock_emit_wrsc.assert_not_called() + @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") + def test_update_state_comment_allows_creator_only(self, mock_emit_wrsc): + state_deprecated = StateFactory( + workflow_run=self.wfr_failed, + status="DEPRECATED", + timestamp=make_aware(datetime.now() + timedelta(hours=11)), + comment="old", + created_by=self.user_email, + ) + url = f"{self.endpoint}/{self.wfr_failed.orcabus_id}/state/{state_deprecated.orcabus_id}/" + + response = self.client.patch( + url, + data={ + "comment": "updated by creator", + "createdBy": "spoofed.creator@example.com", + }, + content_type="application/json", + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["createdBy"], self.user_email) + state_deprecated.refresh_from_db() + self.assertEqual(state_deprecated.comment, "updated by creator") + self.assertEqual(state_deprecated.created_by, self.user_email) + + other_token = jwt.encode( + {"email": "other.user@example.com"}, + self.jwt_test_secret, + algorithm="HS256", + ) + response = self.client.patch( + url, + data={"comment": "unauthorized update"}, + content_type="application/json", + HTTP_AUTHORIZATION=f"Bearer {other_token}", + ) + self.assertEqual(response.status_code, 403) + state_deprecated.refresh_from_db() + self.assertEqual(state_deprecated.comment, "updated by creator") + self.assertEqual(state_deprecated.created_by, self.user_email) + mock_emit_wrsc.assert_not_called() + + def test_update_legacy_state_comment_requires_bearer(self): + state_deprecated = StateFactory( + workflow_run=self.wfr_failed, + status="DEPRECATED", + timestamp=make_aware(datetime.now() + timedelta(hours=12)), + comment="old", + created_by=None, + ) + url = f"{self.endpoint}/{self.wfr_failed.orcabus_id}/state/{state_deprecated.orcabus_id}/" + response = self.client.patch( + url, + data={"comment": "unauthenticated update"}, + content_type="application/json", + HTTP_AUTHORIZATION="", + ) + self.assertEqual(response.status_code, 401) + state_deprecated.refresh_from_db() + self.assertEqual(state_deprecated.comment, "old") + self.assertIsNone(state_deprecated.created_by) + def test_is_valid_next_state_current_status_none_only_allows_deprecated(self): from workflow_manager.viewsets.state import StateViewSet @@ -469,6 +614,7 @@ def test_update_prefetched_objects_cache_invalidation_runs(self): viewset = StateViewSet() request = MagicMock() request.data = {"comment": "new"} + request.META = {"HTTP_AUTHORIZATION": self.auth_header} state_deprecated = StateFactory( workflow_run=self.wfr_failed, @@ -499,6 +645,7 @@ def test_update_prefetched_objects_cache_invalidation_runs_with_patched_get_obje viewset = StateViewSet() request = MagicMock() request.data = {"comment": "new patched"} + request.META = {"HTTP_AUTHORIZATION": self.auth_header} viewset.get_success_headers = MagicMock(return_value={}) state_deprecated = StateFactory( diff --git a/app/workflow_manager/viewsets/state.py b/app/workflow_manager/viewsets/state.py index db14dcf..8277a13 100644 --- a/app/workflow_manager/viewsets/state.py +++ b/app/workflow_manager/viewsets/state.py @@ -9,6 +9,7 @@ from drf_spectacular.types import OpenApiTypes from rest_framework.decorators import action from rest_framework import mixins, status +from rest_framework.exceptions import PermissionDenied from rest_framework.response import Response from rest_framework.viewsets import GenericViewSet from django.db import DatabaseError, transaction @@ -26,6 +27,7 @@ StateTransitionResponseSerializer, StateTransitionValidationErrorSerializer, ) +from workflow_manager.viewsets.auth_utils import get_email_from_bearer_authorization logger = logging.getLogger(__name__) @@ -54,6 +56,9 @@ "transition was invalid." ), ), + status.HTTP_401_UNAUTHORIZED: OpenApiResponse( + description="A valid Bearer JWT with an email claim is required.", + ), status.HTTP_500_INTERNAL_SERVER_ERROR: OpenApiResponse( response=StateTransitionResponseSerializer, description=( @@ -160,6 +165,7 @@ def create_state_and_emit_wrsc( workflow_run: WorkflowRun, request_status: str, request_comment: str, + created_by: str, ) -> tuple[State, dict]: """Create a manual state and emit its WRSC event in the caller's transaction.""" logger.info( @@ -172,6 +178,7 @@ def create_state_and_emit_wrsc( status=request_status, timestamp=timezone.now(), comment=request_comment, + created_by=created_by, ) logger.info( "Manual workflow-run state created: workflow_run_id=%s state_id=%s status=%s", @@ -223,8 +230,19 @@ def _failure_response_status(failures: list[dict]) -> int: @extend_schema_view( partial_update=extend_schema( request=StateUpdateRequestSerializer, - responses={200: StateSerializer}, - description=("Update state comment only."), + responses={ + 200: StateSerializer, + 401: OpenApiResponse( + description="A valid Bearer JWT with an email claim is required." + ), + 403: OpenApiResponse( + description="The authenticated user did not create this state." + ), + }, + description=( + "Update the state comment only. Bearer authentication is required; " + "states with a recorded creator may only be updated by that creator." + ), ), ) class StateViewSet( @@ -245,6 +263,7 @@ def get_queryset(self): def update(self, request, *args, **kwargs): partial = kwargs.pop("partial", False) + actor = get_email_from_bearer_authorization(request) instance = self.get_object() required_fields = {"comment"} @@ -263,6 +282,12 @@ def update(self, request, *args, **kwargs): status=status.HTTP_400_BAD_REQUEST, ) + creator = (instance.created_by or "").strip().lower() + if creator and creator != actor: + raise PermissionDenied( + "You don't have permission to update this state comment." + ) + body = StateUpdateRequestSerializer(data=request.data, partial=partial) body.is_valid(raise_exception=True) vd = body.validated_data @@ -300,6 +325,7 @@ def get_states_transition_validation_map(self, request, **kwargs): return Response(self.states_transition_validation_map) def _state_transition(self, request, request_status: str): + created_by = get_email_from_bearer_authorization(request) body = StateTransitionRequestSerializer(data=request.data) body.is_valid(raise_exception=True) vd = body.validated_data @@ -371,6 +397,7 @@ def _state_transition(self, request, request_status: str): wfr, request_status, request_comment, + created_by, ) except DatabaseError: logger.exception( @@ -438,7 +465,10 @@ def _state_transition(self, request, request_status: str): request=StateTransitionRequestSerializer, responses=STATE_TRANSITION_RESPONSES, summary="Mark workflow runs as deprecated", - description="Transition workflow runs from SUCCEEDED to DEPRECATED.", + description=( + "Transition workflow runs from SUCCEEDED to DEPRECATED and record the " + "Bearer JWT email as the state creator." + ), ) @action(detail=False, methods=["post"], url_path="deprecate") def deprecate(self, request, *args, **kwargs): @@ -448,7 +478,10 @@ def deprecate(self, request, *args, **kwargs): request=StateTransitionRequestSerializer, responses=STATE_TRANSITION_RESPONSES, summary="Mark workflow runs as resolved", - description="Transition workflow runs from FAILED to RESOLVED.", + description=( + "Transition workflow runs from FAILED to RESOLVED and record the Bearer " + "JWT email as the state creator." + ), ) @action(detail=False, methods=["post"], url_path="resolve") def resolve(self, request, *args, **kwargs): @@ -458,7 +491,10 @@ def resolve(self, request, *args, **kwargs): request=StateTransitionRequestSerializer, responses=STATE_TRANSITION_RESPONSES, summary="Cancel workflow runs", - description="Transition non-terminal workflow runs to CANCELLED.", + description=( + "Transition non-terminal workflow runs to CANCELLED and record the Bearer " + "JWT email as the state creator." + ), ) @action(detail=False, methods=["post"], url_path="cancel") def cancel(self, request, *args, **kwargs): diff --git a/app/workflow_manager_proc/domain/event/wrsc.py b/app/workflow_manager_proc/domain/event/wrsc.py index b8f08c9..c17d6ab 100644 --- a/app/workflow_manager_proc/domain/event/wrsc.py +++ b/app/workflow_manager_proc/domain/event/wrsc.py @@ -1,6 +1,6 @@ # generated by datamodel-codegen: # filename: WorkflowRunStateChange.schema.json -# timestamp: 2026-05-28T08:52:26+00:00 +# timestamp: 2026-08-16T12:38:47+00:00 from __future__ import annotations @@ -56,6 +56,7 @@ class WorkflowRunStateChange(BaseModel): computeEnv: str | None = None storageEnv: str | None = None status: str + createdBy: str | None = None payload: Payload | None = None diff --git a/app/workflow_manager_proc/services/workflow_run.py b/app/workflow_manager_proc/services/workflow_run.py index a4fbe23..24f5405 100644 --- a/app/workflow_manager_proc/services/workflow_run.py +++ b/app/workflow_manager_proc/services/workflow_run.py @@ -35,7 +35,7 @@ ASSOCIATION_STATUS = "ACTIVE" EVENT_BUS_NAME = os.environ.get("EVENT_BUS_NAME") WRSC_SCHEMA_VERSION = ( - "1.0.0" # TODO: set somewhere more global (and check against schema?) + "1.1.0" # TODO: set somewhere more global (and check against schema?) ) @@ -54,7 +54,9 @@ def create_workflow_run(event: wru.WorkflowRunUpdate): emit_event( event_type=EventType.WRSC, event_bus=EVENT_BUS_NAME, - event_json=out_wrsc.model_dump_json(), + event_json=out_wrsc.model_dump_json( + exclude={"createdBy"} if out_wrsc.createdBy is None else None + ), ) else: # ignore - state has not been updated @@ -303,6 +305,7 @@ def map_workflow_run_new_state_to_wrsc( validationState=wfr.workflow.validation_state, ), status=Status.get_convention(new_state.status), # ensure we follow conventions + createdBy=new_state.created_by, ) # Set libraries @@ -399,6 +402,8 @@ def get_wrsc_hash(out_wrsc: wrsc.WorkflowRunStateChange) -> str: if out_wrsc.executionId: keywords.append(out_wrsc.executionId) keywords.append(out_wrsc.status) + if out_wrsc.createdBy: + keywords.append(out_wrsc.createdBy) keywords.append(out_wrsc.workflow.orcabusId) if out_wrsc.payload: diff --git a/app/workflow_manager_proc/tests/test_workflow_run.py b/app/workflow_manager_proc/tests/test_workflow_run.py index cfcf828..88661ea 100644 --- a/app/workflow_manager_proc/tests/test_workflow_run.py +++ b/app/workflow_manager_proc/tests/test_workflow_run.py @@ -1,3 +1,4 @@ +import json import os from unittest import mock @@ -47,6 +48,10 @@ def test_create_workflow_run(self): self.assertEqual(WorkflowRun.objects.count(), 1) self.assertEqual(State.objects.count(), 1) self.assertEqual(Payload.objects.count(), 0) + emitted_detail = json.loads( + self.mock_boto3.put_events.call_args.kwargs["Entries"][0]["Detail"] + ) + self.assertNotIn("createdBy", emitted_detail) def test_create_workflow_run_with_multiple_drafts(self): """ @@ -175,9 +180,9 @@ def test_wrsc_includes_execution_id(self): self.assertIsNotNone(out_wrsc) self.assertEqual(out_wrsc.executionId, self.mock_wru_max.executionId) - def test_get_wrsc_hash_differs_by_execution_id(self): + def test_get_wrsc_hash_differs_by_execution_id_and_created_by(self): """ - python manage.py test workflow_manager_proc.tests.test_workflow_run.WorkflowRunSrvUnitTests.test_get_wrsc_hash_differs_by_execution_id + python manage.py test workflow_manager_proc.tests.test_workflow_run.WorkflowRunSrvUnitTests.test_get_wrsc_hash_differs_by_execution_id_and_created_by """ self.load_mock_wru_max() @@ -208,6 +213,18 @@ def test_get_wrsc_hash_differs_by_execution_id(self): self.assertNotEqual(hash_a, hash_b) + wrsc_c = wrsc.WorkflowRunStateChange( + **base_kwargs, createdBy="first.user@example.com" + ) + wrsc_d = wrsc.WorkflowRunStateChange( + **base_kwargs, createdBy="second.user@example.com" + ) + + hash_c = workflow_run.get_wrsc_hash(wrsc_c) + hash_d = workflow_run.get_wrsc_hash(wrsc_d) + + self.assertNotEqual(hash_c, hash_d) + def test_create_workflow_run_state_has_not_been_updated(self): """ python manage.py test workflow_manager_proc.tests.test_workflow_run.WorkflowRunSrvUnitTests.test_create_workflow_run_state_has_not_been_updated @@ -638,6 +655,8 @@ def test_map_workflow_run_new_state_to_wrsc(self): validated_out_wrsc = wrsc.WorkflowRunStateChange.model_validate(out_wrsc) self.assertIsNotNone(validated_out_wrsc) + self.assertEqual(validated_out_wrsc.version, "1.1.0") + self.assertIsNone(validated_out_wrsc.createdBy) def test_get_wrsc_hash(self): """ diff --git a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json index 4cdf1c3..e926525 100644 --- a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json +++ b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json @@ -60,6 +60,9 @@ "status": { "type": "string" }, + "createdBy": { + "type": "string" + }, "payload": { "$ref": "#/definitions/Payload" } diff --git a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml index 01cb973..e88bd6d 100644 --- a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml +++ b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml @@ -92,6 +92,9 @@ definitions: # the current/new state of the WorkflowRun that the event announces # Note: not every event necessarily announces a new status type: string + createdBy: + # normalized email of the user who created a manual workflow-run state + type: string payload: '$ref': '#/definitions/Payload' Library: diff --git a/docs/events/WorkflowRunStateChange/examples/WRSC__example1.json b/docs/events/WorkflowRunStateChange/examples/WRSC__example1.json index 855ffe1..b6794b7 100644 --- a/docs/events/WorkflowRunStateChange/examples/WRSC__example1.json +++ b/docs/events/WorkflowRunStateChange/examples/WRSC__example1.json @@ -9,7 +9,7 @@ "resources": [], "detail": { "id": " e.g. 97534601940f17ebcfee02e9df998244", - "version": " e.g. 1.0.0", + "version": "1.1.0", "timestamp": "2025-05-01T09:25:44Z", "orcabusId": "wfr. e.g. wfr.01J5M2J44HFJ9424G7074HHHHH", "portalRunId": "202405012397gatc", diff --git a/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json b/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json index addd098..b87298d 100644 --- a/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json +++ b/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json @@ -9,7 +9,7 @@ "resources": [], "detail": { "id": " e.g. 97534601940f17ebcfee02e9df998244", - "version": " e.g. 1.0.0", + "version": "1.1.0", "timestamp": "2025-05-01T09:25:44Z", "orcabusId": "wfr. e.g. wfr.01J5M2J44HFJ9424G7074HHHHH", "portalRunId": "202405012397gatc", @@ -23,6 +23,7 @@ "executionEnginePipelineId": " e.g. 123.456.789", "validationState": "UNVALIDATED" }, - "status": "RUNNING" + "status": "DEPRECATED", + "createdBy": "user@example.com" } } From ed75a77338ec34d2348b7d58aae44de314f226c3 Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Fri, 21 Aug 2026 08:25:00 +1000 Subject: [PATCH 2/2] wrsc event update for stateCreatedBy --- app/workflow_manager/tests/test_event_bridge.py | 8 +++++--- app/workflow_manager/tests/test_state_viewset.py | 8 +++++--- app/workflow_manager_proc/domain/event/wrsc.py | 2 +- app/workflow_manager_proc/services/workflow_run.py | 10 ++++++---- app/workflow_manager_proc/tests/test_workflow_run.py | 11 ++++++----- .../WorkflowRunStateChange.schema.json | 2 +- .../WorkflowRunStateChange.schema.yaml | 5 +++-- .../examples/WRSC__example2.json | 2 +- 8 files changed, 28 insertions(+), 20 deletions(-) diff --git a/app/workflow_manager/tests/test_event_bridge.py b/app/workflow_manager/tests/test_event_bridge.py index 5ab88fd..c358398 100644 --- a/app/workflow_manager/tests/test_event_bridge.py +++ b/app/workflow_manager/tests/test_event_bridge.py @@ -47,21 +47,23 @@ def test_emit_wrsc_api_event_omits_payload(self, mock_emit_event): detail = json.loads(entry["Detail"]) self.assertNotIn("payload", detail) self.assertNotIn("createdBy", detail) + self.assertNotIn("stateCreatedBy", detail) self.assertEqual(detail["status"], "RESOLVED") @patch.dict(os.environ, {"EVENT_BUS_NAME": "test-event-bus"}) @patch("workflow_manager.aws_event_bridge.event.libeb.emit_event") - def test_emit_wrsc_api_event_includes_created_by(self, mock_emit_event): + def test_emit_wrsc_api_event_includes_state_created_by(self, mock_emit_event): mock_emit_event.return_value = {"FailedEntryCount": 0, "Entries": [{}]} event = self.build_event() - event["createdBy"] = "state.creator@example.com" + event["stateCreatedBy"] = "state.creator@example.com" emit_wrsc_api_event(event) entry = mock_emit_event.call_args.args[0] detail = json.loads(entry["Detail"]) self.assertEqual(detail["version"], "1.1.0") - self.assertEqual(detail["createdBy"], "state.creator@example.com") + self.assertNotIn("createdBy", detail) + self.assertEqual(detail["stateCreatedBy"], "state.creator@example.com") @patch.dict(os.environ, {"EVENT_BUS_NAME": "test-event-bus"}) @patch("workflow_manager.aws_event_bridge.event.libeb.emit_event") diff --git a/app/workflow_manager/tests/test_state_viewset.py b/app/workflow_manager/tests/test_state_viewset.py index 757c7e0..a3b684a 100644 --- a/app/workflow_manager/tests/test_state_viewset.py +++ b/app/workflow_manager/tests/test_state_viewset.py @@ -164,7 +164,7 @@ def test_resolve_transitions_failed_workflow_run(self, mock_emit_wrsc): self.assertEqual(wrsc_event["status"], "RESOLVED") self.assertEqual(wrsc_event["orcabusId"], self.wfr_failed.orcabus_id) self.assertEqual(wrsc_event["workflow"]["orcabusId"], self.wf.orcabus_id) - self.assertEqual(wrsc_event["createdBy"], self.user_email) + self.assertEqual(wrsc_event["stateCreatedBy"], self.user_email) self.assertEqual(wrsc_event["version"], "1.1.0") self.assertNotIn("payload", wrsc_event) @@ -192,7 +192,9 @@ def test_deprecate_transitions_succeeded_workflow_run(self, mock_emit_wrsc): ) mock_emit_wrsc.assert_called_once() self.assertEqual(mock_emit_wrsc.call_args.args[0]["status"], "DEPRECATED") - self.assertEqual(mock_emit_wrsc.call_args.args[0]["createdBy"], self.user_email) + self.assertEqual( + mock_emit_wrsc.call_args.args[0]["stateCreatedBy"], self.user_email + ) @patch("workflow_manager.viewsets.state.emit_wrsc_api_event") def test_deprecate_rejects_failed_workflow_run(self, mock_emit_wrsc): @@ -287,7 +289,7 @@ def test_cancel_preserves_existing_allowed_source_states(self, mock_emit_wrsc): ) self.assertTrue( all( - call.args[0]["createdBy"] == self.user_email + call.args[0]["stateCreatedBy"] == self.user_email for call in mock_emit_wrsc.call_args_list ) ) diff --git a/app/workflow_manager_proc/domain/event/wrsc.py b/app/workflow_manager_proc/domain/event/wrsc.py index c17d6ab..6a2c85e 100644 --- a/app/workflow_manager_proc/domain/event/wrsc.py +++ b/app/workflow_manager_proc/domain/event/wrsc.py @@ -56,7 +56,7 @@ class WorkflowRunStateChange(BaseModel): computeEnv: str | None = None storageEnv: str | None = None status: str - createdBy: str | None = None + stateCreatedBy: str | None = None payload: Payload | None = None diff --git a/app/workflow_manager_proc/services/workflow_run.py b/app/workflow_manager_proc/services/workflow_run.py index 24f5405..6770daf 100644 --- a/app/workflow_manager_proc/services/workflow_run.py +++ b/app/workflow_manager_proc/services/workflow_run.py @@ -55,7 +55,9 @@ def create_workflow_run(event: wru.WorkflowRunUpdate): event_type=EventType.WRSC, event_bus=EVENT_BUS_NAME, event_json=out_wrsc.model_dump_json( - exclude={"createdBy"} if out_wrsc.createdBy is None else None + exclude={"stateCreatedBy"} + if out_wrsc.stateCreatedBy is None + else None ), ) else: @@ -305,7 +307,7 @@ def map_workflow_run_new_state_to_wrsc( validationState=wfr.workflow.validation_state, ), status=Status.get_convention(new_state.status), # ensure we follow conventions - createdBy=new_state.created_by, + stateCreatedBy=new_state.created_by, ) # Set libraries @@ -402,8 +404,8 @@ def get_wrsc_hash(out_wrsc: wrsc.WorkflowRunStateChange) -> str: if out_wrsc.executionId: keywords.append(out_wrsc.executionId) keywords.append(out_wrsc.status) - if out_wrsc.createdBy: - keywords.append(out_wrsc.createdBy) + if out_wrsc.stateCreatedBy: + keywords.append(out_wrsc.stateCreatedBy) keywords.append(out_wrsc.workflow.orcabusId) if out_wrsc.payload: diff --git a/app/workflow_manager_proc/tests/test_workflow_run.py b/app/workflow_manager_proc/tests/test_workflow_run.py index 88661ea..1259fc5 100644 --- a/app/workflow_manager_proc/tests/test_workflow_run.py +++ b/app/workflow_manager_proc/tests/test_workflow_run.py @@ -52,6 +52,7 @@ def test_create_workflow_run(self): self.mock_boto3.put_events.call_args.kwargs["Entries"][0]["Detail"] ) self.assertNotIn("createdBy", emitted_detail) + self.assertNotIn("stateCreatedBy", emitted_detail) def test_create_workflow_run_with_multiple_drafts(self): """ @@ -180,9 +181,9 @@ def test_wrsc_includes_execution_id(self): self.assertIsNotNone(out_wrsc) self.assertEqual(out_wrsc.executionId, self.mock_wru_max.executionId) - def test_get_wrsc_hash_differs_by_execution_id_and_created_by(self): + def test_get_wrsc_hash_differs_by_execution_id_and_state_created_by(self): """ - python manage.py test workflow_manager_proc.tests.test_workflow_run.WorkflowRunSrvUnitTests.test_get_wrsc_hash_differs_by_execution_id_and_created_by + python manage.py test workflow_manager_proc.tests.test_workflow_run.WorkflowRunSrvUnitTests.test_get_wrsc_hash_differs_by_execution_id_and_state_created_by """ self.load_mock_wru_max() @@ -214,10 +215,10 @@ def test_get_wrsc_hash_differs_by_execution_id_and_created_by(self): self.assertNotEqual(hash_a, hash_b) wrsc_c = wrsc.WorkflowRunStateChange( - **base_kwargs, createdBy="first.user@example.com" + **base_kwargs, stateCreatedBy="first.user@example.com" ) wrsc_d = wrsc.WorkflowRunStateChange( - **base_kwargs, createdBy="second.user@example.com" + **base_kwargs, stateCreatedBy="second.user@example.com" ) hash_c = workflow_run.get_wrsc_hash(wrsc_c) @@ -656,7 +657,7 @@ def test_map_workflow_run_new_state_to_wrsc(self): self.assertIsNotNone(validated_out_wrsc) self.assertEqual(validated_out_wrsc.version, "1.1.0") - self.assertIsNone(validated_out_wrsc.createdBy) + self.assertIsNone(validated_out_wrsc.stateCreatedBy) def test_get_wrsc_hash(self): """ diff --git a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json index e926525..dd0903e 100644 --- a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json +++ b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.json @@ -60,7 +60,7 @@ "status": { "type": "string" }, - "createdBy": { + "stateCreatedBy": { "type": "string" }, "payload": { diff --git a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml index e88bd6d..d05ca8a 100644 --- a/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml +++ b/docs/events/WorkflowRunStateChange/WorkflowRunStateChange.schema.yaml @@ -92,8 +92,9 @@ definitions: # the current/new state of the WorkflowRun that the event announces # Note: not every event necessarily announces a new status type: string - createdBy: - # normalized email of the user who created a manual workflow-run state + stateCreatedBy: + # normalized email of the user who created this workflow-run state; + # omitted for states not created by a user type: string payload: '$ref': '#/definitions/Payload' diff --git a/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json b/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json index b87298d..e4e6762 100644 --- a/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json +++ b/docs/events/WorkflowRunStateChange/examples/WRSC__example2.json @@ -24,6 +24,6 @@ "validationState": "UNVALIDATED" }, "status": "DEPRECATED", - "createdBy": "user@example.com" + "stateCreatedBy": "user@example.com" } }