From ea58129ebeec91fbe3cd6ef685dc0a76c2d268db Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Tue, 28 Jul 2026 14:34:59 +0900 Subject: [PATCH] test: move workflow app service coverage to unit tests (#38946) --- .../services/test_workflow_app_service.py | 75 +--------------- .../test_workflow_app_service_metadata.py | 88 +++++++++++++++++++ 2 files changed, 90 insertions(+), 73 deletions(-) create mode 100644 api/tests/unit_tests/services/test_workflow_app_service_metadata.py diff --git a/api/tests/test_containers_integration_tests/services/test_workflow_app_service.py b/api/tests/test_containers_integration_tests/services/test_workflow_app_service.py index f553b0f72a0..71c9c81e0d2 100644 --- a/api/tests/test_containers_integration_tests/services/test_workflow_app_service.py +++ b/api/tests/test_containers_integration_tests/services/test_workflow_app_service.py @@ -3,7 +3,6 @@ from __future__ import annotations import json import uuid from datetime import UTC, datetime, timedelta -from types import SimpleNamespace from unittest.mock import patch import pytest @@ -12,13 +11,13 @@ from sqlalchemy.orm import Session from graphon.enums import WorkflowExecutionStatus from models import EndUser, Workflow, WorkflowAppLog, WorkflowArchiveLog, WorkflowRun -from models.enums import AppTriggerType, CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom +from models.enums import CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom from models.workflow import WorkflowAppLogCreatedFrom from services.account_service import AccountService, TenantService # Delay import of AppService to avoid circular dependency # from services.app_service import AppService, CreateAppParams -from services.workflow_app_service import LogView, WorkflowAppService +from services.workflow_app_service import WorkflowAppService from tests.test_containers_integration_tests.helpers import generate_valid_password @@ -1627,73 +1626,3 @@ class TestWorkflowAppService: end_user_item = next(d for d in result["data"] if d["created_by_end_user"] is not None) assert account_item["created_by_account"].id == account.id assert end_user_item["created_by_end_user"].id == end_user.id - - -class TestLogView: - def test_details_and_proxy_attributes(self): - log = SimpleNamespace(id="log-1", status="succeeded") - view = LogView(log=log, details={"trigger_metadata": {"type": "plugin"}}) - - assert view.details == {"trigger_metadata": {"type": "plugin"}} - assert view.status == "succeeded" - - -class TestHandleTriggerMetadata: - def test_returns_empty_dict_when_metadata_missing(self): - service = WorkflowAppService() - assert service.handle_trigger_metadata("tenant-1", None) == {} - - def test_enriches_plugin_icons(self): - service = WorkflowAppService() - meta = { - "type": AppTriggerType.TRIGGER_PLUGIN.value, - "icon_filename": "light.png", - "icon_dark_filename": "dark.png", - } - with patch( - "services.workflow_app_service.PluginService.get_plugin_icon_url", - side_effect=["https://cdn/light.png", "https://cdn/dark.png"], - ) as mock_icon: - result = service.handle_trigger_metadata("tenant-1", json.dumps(meta)) - - assert result["icon"] == "https://cdn/light.png" - assert result["icon_dark"] == "https://cdn/dark.png" - assert mock_icon.call_count == 2 - - def test_non_plugin_metadata_without_icon_lookup(self): - service = WorkflowAppService() - meta = {"type": AppTriggerType.TRIGGER_WEBHOOK.value} - with patch("services.workflow_app_service.PluginService.get_plugin_icon_url") as mock_icon: - result = service.handle_trigger_metadata("tenant-1", json.dumps(meta)) - - assert result["type"] == AppTriggerType.TRIGGER_WEBHOOK.value - mock_icon.assert_not_called() - - -class TestSafeJsonLoads: - @pytest.mark.parametrize( - ("value", "expected"), - [ - (None, None), - ("", None), - ('{"k":"v"}', {"k": "v"}), - ("not-json", None), - ({"raw": True}, {"raw": True}), - ], - ) - def test_handles_various_inputs(self, value, expected): - assert WorkflowAppService._safe_json_loads(value) == expected - - -class TestSafeParseUuid: - def test_returns_none_for_short_or_invalid_values(self): - service = WorkflowAppService() - assert service._safe_parse_uuid("short") is None - assert service._safe_parse_uuid("x" * 40) is None - - def test_returns_uuid_for_valid_string(self): - service = WorkflowAppService() - raw = str(uuid.uuid4()) - result = service._safe_parse_uuid(raw) - assert result is not None - assert str(result) == raw diff --git a/api/tests/unit_tests/services/test_workflow_app_service_metadata.py b/api/tests/unit_tests/services/test_workflow_app_service_metadata.py new file mode 100644 index 00000000000..ded4fab1c7a --- /dev/null +++ b/api/tests/unit_tests/services/test_workflow_app_service_metadata.py @@ -0,0 +1,88 @@ +"""Unit tests for workflow app log views and trigger metadata helpers.""" + +import json +import uuid +from unittest.mock import patch + +import pytest + +from models.enums import AppTriggerType, CreatorUserRole +from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom +from services.workflow_app_service import LogView, WorkflowAppService + + +class TestLogView: + def test_details_and_proxy_attributes(self) -> None: + log = WorkflowAppLog( + tenant_id="tenant-1", + app_id="app-1", + workflow_id="workflow-1", + workflow_run_id="run-1", + created_from=WorkflowAppLogCreatedFrom.WEB_APP, + created_by_role=CreatorUserRole.ACCOUNT, + created_by="account-1", + ) + log.id = "log-1" + + view = LogView(log=log, details={"trigger_metadata": {"type": "plugin"}}) + + assert view.details == {"trigger_metadata": {"type": "plugin"}} + assert view.id == "log-1" + + +class TestHandleTriggerMetadata: + def test_returns_empty_dict_when_metadata_missing(self) -> None: + assert WorkflowAppService().handle_trigger_metadata("tenant-1", None) == {} + + def test_enriches_plugin_icons(self) -> None: + metadata = { + "type": AppTriggerType.TRIGGER_PLUGIN.value, + "icon_filename": "light.png", + "icon_dark_filename": "dark.png", + } + with patch( + "services.workflow_app_service.PluginService.get_plugin_icon_url", + side_effect=["https://cdn/light.png", "https://cdn/dark.png"], + ) as mock_icon: + result = WorkflowAppService().handle_trigger_metadata("tenant-1", json.dumps(metadata)) + + assert result["icon"] == "https://cdn/light.png" + assert result["icon_dark"] == "https://cdn/dark.png" + assert mock_icon.call_count == 2 + + def test_non_plugin_metadata_without_icon_lookup(self) -> None: + metadata = {"type": AppTriggerType.TRIGGER_WEBHOOK.value} + with patch("services.workflow_app_service.PluginService.get_plugin_icon_url") as mock_icon: + result = WorkflowAppService().handle_trigger_metadata("tenant-1", json.dumps(metadata)) + + assert result["type"] == AppTriggerType.TRIGGER_WEBHOOK.value + mock_icon.assert_not_called() + + +class TestSafeJsonLoads: + @pytest.mark.parametrize( + ("value", "expected"), + [ + (None, None), + ("", None), + ('{"k":"v"}', {"k": "v"}), + ("not-json", None), + ({"raw": True}, {"raw": True}), + ], + ) + def test_handles_various_inputs(self, value, expected) -> None: + assert WorkflowAppService._safe_json_loads(value) == expected + + +class TestSafeParseUuid: + def test_returns_none_for_short_or_invalid_values(self) -> None: + assert WorkflowAppService._safe_parse_uuid("short") is None + assert WorkflowAppService._safe_parse_uuid("x" * 40) is None + + def test_returns_uuid_for_valid_string(self) -> None: + raw = str(uuid.uuid4()) + + result = WorkflowAppService._safe_parse_uuid(raw) + + assert result is not None + assert str(result) == raw