From 326b2b573b838ef3e2b6dc5d1bdcb4e7cbfa82e0 Mon Sep 17 00:00:00 2001 From: -LAN- Date: Fri, 18 Jul 2025 07:38:12 +0800 Subject: [PATCH] test(test_sqlalchemy_repository): Remove outdated test cases Signed-off-by: -LAN- --- .../test_sqlalchemy_repository.py | 113 ------------------ 1 file changed, 113 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 643efb0a0c..c60800c493 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 @@ -137,37 +137,6 @@ def test_save_with_existing_tenant_id(repository, session): session_obj.merge.assert_called_once_with(modified_execution) -def test_get_by_node_execution_id(repository, session, mocker: MockerFixture): - """Test get_by_node_execution_id method.""" - session_obj, _ = session - # Set up mock - mock_select = mocker.patch("core.repositories.sqlalchemy_workflow_node_execution_repository.select") - mock_stmt = mocker.MagicMock() - mock_select.return_value = mock_stmt - mock_stmt.where.return_value = mock_stmt - - # Create a properly configured mock execution - mock_execution = mocker.MagicMock(spec=WorkflowNodeExecutionModel) - configure_mock_execution(mock_execution) - session_obj.scalar.return_value = mock_execution - - # Create a mock domain model to be returned by _to_domain_model - mock_domain_model = mocker.MagicMock() - # Mock the _to_domain_model method to return our mock domain model - repository._to_domain_model = mocker.MagicMock(return_value=mock_domain_model) - - # Call method - result = repository.get_by_node_execution_id("test-node-execution-id") - - # Assert select was called with correct parameters - mock_select.assert_called_once() - session_obj.scalar.assert_called_once_with(mock_stmt) - # Assert _to_domain_model was called with the mock execution - repository._to_domain_model.assert_called_once_with(mock_execution) - # Assert the result is our mock domain model - assert result is mock_domain_model - - def test_get_by_workflow_run(repository, session, mocker: MockerFixture): """Test get_by_workflow_run method.""" session_obj, _ = session @@ -202,88 +171,6 @@ def test_get_by_workflow_run(repository, session, mocker: MockerFixture): assert result[0] is mock_domain_model -def test_get_running_executions(repository, session, mocker: MockerFixture): - """Test get_running_executions method.""" - session_obj, _ = session - # Set up mock - mock_select = mocker.patch("core.repositories.sqlalchemy_workflow_node_execution_repository.select") - mock_stmt = mocker.MagicMock() - mock_select.return_value = mock_stmt - mock_stmt.where.return_value = mock_stmt - - # Create a properly configured mock execution - mock_execution = mocker.MagicMock(spec=WorkflowNodeExecutionModel) - configure_mock_execution(mock_execution) - session_obj.scalars.return_value.all.return_value = [mock_execution] - - # Create a mock domain model to be returned by _to_domain_model - mock_domain_model = mocker.MagicMock() - # Mock the _to_domain_model method to return our mock domain model - repository._to_domain_model = mocker.MagicMock(return_value=mock_domain_model) - - # Call method - result = repository.get_running_executions("test-workflow-run-id") - - # Assert select was called with correct parameters - mock_select.assert_called_once() - session_obj.scalars.assert_called_once_with(mock_stmt) - # Assert _to_domain_model was called with the mock execution - repository._to_domain_model.assert_called_once_with(mock_execution) - # Assert the result contains our mock domain model - assert len(result) == 1 - assert result[0] is mock_domain_model - - -def test_update_via_save(repository, session): - """Test updating an existing record via save method.""" - session_obj, _ = session - # Create a mock execution - execution = MagicMock(spec=WorkflowNodeExecutionModel) - execution.tenant_id = None - execution.app_id = None - execution.inputs = None - execution.process_data = None - execution.outputs = None - execution.metadata = None - - # Mock the to_db_model method to return the execution itself - # This simulates the behavior of setting tenant_id and app_id - repository.to_db_model = MagicMock(return_value=execution) - - # Call save method to update an existing record - repository.save(execution) - - # Assert to_db_model was called with the execution - repository.to_db_model.assert_called_once_with(execution) - - # Assert session.merge was called (for updates) - session_obj.merge.assert_called_once_with(execution) - - -def test_clear(repository, session, mocker: MockerFixture): - """Test clear method.""" - session_obj, _ = session - # Set up mock - mock_delete = mocker.patch("core.repositories.sqlalchemy_workflow_node_execution_repository.delete") - mock_stmt = mocker.MagicMock() - mock_delete.return_value = mock_stmt - mock_stmt.where.return_value = mock_stmt - - # Mock the execute result with rowcount - mock_result = mocker.MagicMock() - mock_result.rowcount = 5 # Simulate 5 records deleted - session_obj.execute.return_value = mock_result - - # Call method - repository.clear() - - # Assert delete was called with correct parameters - mock_delete.assert_called_once_with(WorkflowNodeExecutionModel) - 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