From e723b348cf9cf412f0e9ffc1a20abdf56fe23bff Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Tue, 28 Jul 2026 14:54:17 +0900 Subject: [PATCH] test: use SQLite sessions in core app apps (#39103) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- .../test_generate_task_pipeline_core.py | 106 ++++++++---------- 1 file changed, 49 insertions(+), 57 deletions(-) diff --git a/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py b/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py index 04fe7a2ebed..9f83e657cfb 100644 --- a/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py +++ b/api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py @@ -1,11 +1,11 @@ from __future__ import annotations import logging -from contextlib import contextmanager from types import SimpleNamespace -from unittest.mock import MagicMock import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session from core.app.app_config.entities import AppAdditionalFeatures, WorkflowUIBasedAppConfig from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline @@ -54,6 +54,7 @@ from graphon.runtime import GraphRuntimeState, VariablePool from libs.datetime_utils import naive_utc_now from models.enums import CreatorUserRole from models.model import AppMode, EndUser +from models.workflow import WorkflowAppLog from tests.workflow_test_utils import build_test_variable_pool @@ -193,7 +194,7 @@ class TestWorkflowGenerateTaskPipeline: assert isinstance(responses[0], ValueError) - def test_handle_workflow_started_event_sets_run_id(self, monkeypatch: pytest.MonkeyPatch): + def test_handle_workflow_started_event_sets_run_id(self, monkeypatch: pytest.MonkeyPatch, sqlite_engine): pipeline = _make_pipeline() pipeline._graph_runtime_state = GraphRuntimeState( variable_pool=build_test_variable_pool(variables=build_system_variables(workflow_execution_id="run-id")), @@ -201,11 +202,10 @@ class TestWorkflowGenerateTaskPipeline: ) pipeline._workflow_response_converter.workflow_start_to_stream_response = lambda **kwargs: "started" - @contextmanager - def _fake_session(): - yield SimpleNamespace() - - monkeypatch.setattr(pipeline, "_database_session", _fake_session) + monkeypatch.setattr( + "core.app.apps.workflow.generate_task_pipeline.db", + SimpleNamespace(engine=sqlite_engine), + ) monkeypatch.setattr(pipeline, "_save_workflow_app_log", lambda **kwargs: None) responses = list(pipeline._handle_workflow_started_event(QueueWorkflowStartedEvent())) @@ -339,19 +339,18 @@ class TestWorkflowGenerateTaskPipeline: assert responses == ["finish"] - def test_save_workflow_app_log_created_from(self): + @pytest.mark.parametrize("sqlite_session", [(WorkflowAppLog,)], indirect=True) + def test_save_workflow_app_log_created_from(self, sqlite_session: Session): pipeline = _make_pipeline() pipeline._application_generate_entity.invoke_from = InvokeFrom.SERVICE_API pipeline._user_id = "user" - added: list[object] = [] + pipeline._save_workflow_app_log(session=sqlite_session, workflow_run_id="run-id") + sqlite_session.flush() - class _Session: - def add(self, item): - added.append(item) - - pipeline._save_workflow_app_log(session=_Session(), workflow_run_id="run-id") - - assert added + saved_log = sqlite_session.scalar(select(WorkflowAppLog)) + assert saved_log is not None + assert saved_log.workflow_run_id == "run-id" + assert saved_log.created_from == "service-api" def test_iteration_loop_and_human_input_handlers(self): pipeline = _make_pipeline() @@ -674,35 +673,29 @@ class TestWorkflowGenerateTaskPipeline: assert "Fails to get audio trunk, task_id: task" in caplog.messages assert any(isinstance(item, MessageAudioEndStreamResponse) for item in responses) - def test_database_session_rolls_back_on_error(self, monkeypatch: pytest.MonkeyPatch): + @pytest.mark.parametrize("sqlite_session", [(WorkflowAppLog,)], indirect=True) + def test_database_session_rolls_back_on_error( + self, monkeypatch: pytest.MonkeyPatch, sqlite_engine, sqlite_session: Session + ): pipeline = _make_pipeline() - calls = {"enter": 0, "exit_exc": None} + pipeline._application_generate_entity.invoke_from = InvokeFrom.SERVICE_API + pipeline._user_id = "user" + monkeypatch.setattr( + "core.app.apps.workflow.generate_task_pipeline.db", + SimpleNamespace(engine=sqlite_engine), + ) - class _BeginContext: - def __enter__(self): - calls["enter"] += 1 - return MagicMock() - - def __exit__(self, exc_type, exc, tb): - calls["exit_exc"] = exc_type - return False - - class _Sessionmaker: - def __init__(self, *args, **kwargs): - pass - - def begin(self): - return _BeginContext() - - monkeypatch.setattr("core.app.apps.workflow.generate_task_pipeline.sessionmaker", _Sessionmaker) - monkeypatch.setattr("core.app.apps.workflow.generate_task_pipeline.db", SimpleNamespace(engine=object())) - - with pytest.raises(RuntimeError, match="db error"): - with pipeline._database_session(): + def persist_then_fail() -> None: + with pipeline._database_session() as session: + pipeline._save_workflow_app_log(session=session, workflow_run_id="run-id") + session.flush() raise RuntimeError("db error") - assert calls["enter"] == 1 - assert calls["exit_exc"] is RuntimeError + with pytest.raises(RuntimeError, match="db error"): + persist_then_fail() + + sqlite_session.expire_all() + assert sqlite_session.scalar(select(WorkflowAppLog)) is None def test_node_retry_and_started_handlers_cover_none_and_value(self): pipeline = _make_pipeline() @@ -862,31 +855,30 @@ class TestWorkflowGenerateTaskPipeline: pipeline._handle_workflow_failed_and_stop_events = lambda event, **kwargs: iter(["stopped"]) assert list(pipeline._process_stream_response()) == ["stopped"] - def test_save_workflow_app_log_covers_invoke_from_variants(self): + @pytest.mark.parametrize("sqlite_session", [(WorkflowAppLog,)], indirect=True) + def test_save_workflow_app_log_covers_invoke_from_variants(self, sqlite_session: Session): pipeline = _make_pipeline() pipeline._user_id = "user-id" - added: list[object] = [] - - class _Session: - def add(self, item): - added.append(item) pipeline._application_generate_entity.invoke_from = InvokeFrom.EXPLORE - pipeline._save_workflow_app_log(session=_Session(), workflow_run_id="run-id") - assert added[-1].created_from == "installed-app" + pipeline._save_workflow_app_log(session=sqlite_session, workflow_run_id="run-id") pipeline._application_generate_entity.invoke_from = InvokeFrom.WEB_APP - pipeline._save_workflow_app_log(session=_Session(), workflow_run_id="run-id") - assert added[-1].created_from == "web-app" + pipeline._save_workflow_app_log(session=sqlite_session, workflow_run_id="run-id-2") + sqlite_session.flush() + saved_logs = sqlite_session.scalars(select(WorkflowAppLog).order_by(WorkflowAppLog.workflow_run_id)).all() + assert [log.created_from for log in saved_logs] == ["installed-app", "web-app"] - count_before = len(added) + count_before = len(saved_logs) pipeline._application_generate_entity.invoke_from = InvokeFrom.DEBUGGER - pipeline._save_workflow_app_log(session=_Session(), workflow_run_id="run-id") - assert len(added) == count_before + pipeline._save_workflow_app_log(session=sqlite_session, workflow_run_id="run-id-3") + sqlite_session.flush() + assert len(sqlite_session.scalars(select(WorkflowAppLog)).all()) == count_before pipeline._application_generate_entity.invoke_from = InvokeFrom.WEB_APP - pipeline._save_workflow_app_log(session=_Session(), workflow_run_id=None) - assert len(added) == count_before + pipeline._save_workflow_app_log(session=sqlite_session, workflow_run_id=None) + sqlite_session.flush() + assert len(sqlite_session.scalars(select(WorkflowAppLog)).all()) == count_before def test_save_output_for_event_writes_draft_variables(self): pipeline = _make_pipeline()