From 68616271db58d5ccc43b0772c8d04de6ae4e705b Mon Sep 17 00:00:00 2001 From: -LAN- Date: Wed, 14 May 2025 15:54:31 +0800 Subject: [PATCH] test(test_sqlalchemy_repository): Add tests for model conversion methods Introduces unit tests for _to_db_model and _to_domain_model methods in the SQLAlchemyWorkflowNodeExecutionRepository. These tests ensure correct conversion between domain and database models, improving reliability and maintainability of the repository code. Also, sets default 'triggered_from' to WORKFLOW_RUN for the repository instance. Signed-off-by: -LAN- --- .../test_sqlalchemy_repository.py | 124 +++++++++++++++++- 1 file changed, 122 insertions(+), 2 deletions(-) diff --git a/api/tests/unit_tests/repositories/workflow_node_execution/test_sqlalchemy_repository.py b/api/tests/unit_tests/repositories/workflow_node_execution/test_sqlalchemy_repository.py index cecdcfd20f..64c3b66f12 100644 --- a/api/tests/unit_tests/repositories/workflow_node_execution/test_sqlalchemy_repository.py +++ b/api/tests/unit_tests/repositories/workflow_node_execution/test_sqlalchemy_repository.py @@ -2,6 +2,8 @@ Unit tests for the SQLAlchemy implementation of WorkflowNodeExecutionRepository. """ +import json +from datetime import datetime from unittest.mock import MagicMock, PropertyMock import pytest @@ -9,9 +11,12 @@ from pytest_mock import MockerFixture from sqlalchemy.orm import Session, sessionmaker from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository +from core.workflow.entities.node_entities import NodeRunMetadataKey +from core.workflow.entities.node_execution_entities import NodeExecution, NodeExecutionStatus +from core.workflow.nodes.enums import NodeType from core.workflow.repository.workflow_node_execution_repository import OrderConfig from models.account import Account, Tenant -from models.workflow import WorkflowNodeExecution +from models.workflow import WorkflowNodeExecution, WorkflowNodeExecutionStatus, WorkflowNodeExecutionTriggeredFrom def configure_mock_execution(mock_execution): @@ -67,7 +72,7 @@ def repository(session, mock_user): session_factory=session_factory, user=mock_user, app_id=app_id, - triggered_from=None, + triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN, ) @@ -306,3 +311,118 @@ def test_clear(repository, session, mocker: MockerFixture): mock_stmt.where.assert_called() session_obj.execute.assert_called_once_with(mock_stmt) session_obj.commit.assert_called_once() + + +def test_to_db_model(repository): + """Test _to_db_model method.""" + # Create a domain model + domain_model = NodeExecution( + id="test-id", + workflow_id="test-workflow-id", + node_execution_id="test-node-execution-id", + workflow_run_id="test-workflow-run-id", + index=1, + predecessor_node_id="test-predecessor-id", + node_id="test-node-id", + node_type=NodeType.START, + title="Test Node", + inputs={"input_key": "input_value"}, + process_data={"process_key": "process_value"}, + outputs={"output_key": "output_value"}, + status=NodeExecutionStatus.RUNNING, + error=None, + elapsed_time=1.5, + metadata={NodeRunMetadataKey.TOTAL_TOKENS: 100}, + created_at=datetime.now(), + finished_at=None, + ) + + # Convert to DB model + db_model = repository._to_db_model(domain_model) + + # Assert DB model has correct values + assert isinstance(db_model, WorkflowNodeExecution) + assert db_model.id == domain_model.id + assert db_model.tenant_id == repository._tenant_id + assert db_model.app_id == repository._app_id + assert db_model.workflow_id == domain_model.workflow_id + assert db_model.triggered_from == repository._triggered_from + assert db_model.workflow_run_id == domain_model.workflow_run_id + assert db_model.index == domain_model.index + assert db_model.predecessor_node_id == domain_model.predecessor_node_id + assert db_model.node_execution_id == domain_model.node_execution_id + assert db_model.node_id == domain_model.node_id + assert db_model.node_type == domain_model.node_type + assert db_model.title == domain_model.title + + assert db_model.inputs_dict == domain_model.inputs + assert db_model.process_data_dict == domain_model.process_data + assert db_model.outputs_dict == domain_model.outputs + assert db_model.execution_metadata_dict == domain_model.metadata + + assert db_model.status == domain_model.status + assert db_model.error == domain_model.error + assert db_model.elapsed_time == domain_model.elapsed_time + assert db_model.created_at == domain_model.created_at + assert db_model.created_by_role == repository._creator_user_role + assert db_model.created_by == repository._creator_user_id + assert db_model.finished_at == domain_model.finished_at + + +def test_to_domain_model(repository): + """Test _to_domain_model method.""" + # Create input dictionaries + inputs_dict = {"input_key": "input_value"} + process_data_dict = {"process_key": "process_value"} + outputs_dict = {"output_key": "output_value"} + metadata_dict = {str(NodeRunMetadataKey.TOTAL_TOKENS): 100} + + # Create a DB model using our custom subclass + db_model = WorkflowNodeExecution() + db_model.id = "test-id" + db_model.tenant_id = "test-tenant-id" + db_model.app_id = "test-app-id" + db_model.workflow_id = "test-workflow-id" + db_model.triggered_from = "workflow-run" + db_model.workflow_run_id = "test-workflow-run-id" + db_model.index = 1 + db_model.predecessor_node_id = "test-predecessor-id" + db_model.node_execution_id = "test-node-execution-id" + db_model.node_id = "test-node-id" + db_model.node_type = NodeType.START.value + db_model.title = "Test Node" + db_model.inputs = json.dumps(inputs_dict) + db_model.process_data = json.dumps(process_data_dict) + db_model.outputs = json.dumps(outputs_dict) + db_model.status = WorkflowNodeExecutionStatus.RUNNING + db_model.error = None + db_model.elapsed_time = 1.5 + db_model.execution_metadata = json.dumps(metadata_dict) + db_model.created_at = datetime.now() + db_model.created_by_role = "account" + db_model.created_by = "test-user-id" + db_model.finished_at = None + + # Convert to domain model + domain_model = repository._to_domain_model(db_model) + + # Assert domain model has correct values + assert isinstance(domain_model, NodeExecution) + assert domain_model.id == db_model.id + assert domain_model.workflow_id == db_model.workflow_id + assert domain_model.workflow_run_id == db_model.workflow_run_id + assert domain_model.index == db_model.index + assert domain_model.predecessor_node_id == db_model.predecessor_node_id + assert domain_model.node_execution_id == db_model.node_execution_id + assert domain_model.node_id == db_model.node_id + assert domain_model.node_type == NodeType(db_model.node_type) + assert domain_model.title == db_model.title + assert domain_model.inputs == inputs_dict + assert domain_model.process_data == process_data_dict + assert domain_model.outputs == outputs_dict + assert domain_model.status == NodeExecutionStatus(db_model.status) + assert domain_model.error == db_model.error + assert domain_model.elapsed_time == db_model.elapsed_time + assert domain_model.metadata == metadata_dict + assert domain_model.created_at == db_model.created_at + assert domain_model.finished_at == db_model.finished_at