From edd6a6e77fd95133a641aca6c7842b4fc348e7b3 Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Thu, 30 Jul 2026 11:53:50 +0900 Subject: [PATCH] test: use SQLite sessions in core repositories (#39073) --- .../test_human_input_form_repository_impl.py | 512 ++++++------------ 1 file changed, 169 insertions(+), 343 deletions(-) diff --git a/api/tests/unit_tests/core/repositories/test_human_input_form_repository_impl.py b/api/tests/unit_tests/core/repositories/test_human_input_form_repository_impl.py index f8c6f6d81de..15925ccdc14 100644 --- a/api/tests/unit_tests/core/repositories/test_human_input_form_repository_impl.py +++ b/api/tests/unit_tests/core/repositories/test_human_input_form_repository_impl.py @@ -2,11 +2,12 @@ from __future__ import annotations -import dataclasses from datetime import datetime -from types import SimpleNamespace import pytest +from sqlalchemy import Engine, event +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session from core.repositories.human_input_repository import ( HumanInputFormRecord, @@ -27,9 +28,11 @@ from core.workflow.nodes.human_input.entities import ( ) from core.workflow.nodes.human_input.enums import HumanInputFormKind, HumanInputFormStatus from libs.datetime_utils import naive_utc_now +from models import Account, TenantAccountJoin from models.human_input import ( EmailExternalRecipientPayload, EmailMemberRecipientPayload, + HumanInputForm, HumanInputFormRecipient, RecipientType, StandaloneWebAppRecipientPayload, @@ -40,55 +43,40 @@ def _build_repository() -> HumanInputFormRepositoryImpl: return HumanInputFormRepositoryImpl(tenant_id="tenant-id") -def _patch_recipient_factory(monkeypatch: pytest.MonkeyPatch) -> list[SimpleNamespace]: - created: list[SimpleNamespace] = [] - - def fake_new(cls, form_id: str, delivery_id: str, payload): # type: ignore[no-untyped-def] - recipient = SimpleNamespace( - form_id=form_id, - delivery_id=delivery_id, - recipient_type=payload.TYPE, - recipient_payload=payload.model_dump_json(), - ) - created.append(recipient) - return recipient - - monkeypatch.setattr(HumanInputFormRecipient, "new", classmethod(fake_new)) - return created - - -@pytest.fixture(autouse=True) -def _stub_selectinload(monkeypatch: pytest.MonkeyPatch) -> None: - """Avoid SQLAlchemy mapper configuration in tests using fake sessions.""" - - class _FakeSelect: - def options(self, *_args, **_kwargs): # type: ignore[no-untyped-def] - return self - - def where(self, *_args, **_kwargs): # type: ignore[no-untyped-def] - return self - - monkeypatch.setattr( - "core.repositories.human_input_repository.selectinload", lambda *args, **kwargs: "_loader_option" - ) - monkeypatch.setattr("core.repositories.human_input_repository.select", lambda *args, **kwargs: _FakeSelect()) +def _add_workspace_member( + session: Session, + *, + user_id: str, + email: str, + tenant_id: str = "tenant-id", +) -> None: + account = Account(name=user_id, email=email) + account.id = user_id + session.add_all([account, TenantAccountJoin(tenant_id=tenant_id, account_id=user_id)]) + session.commit() class TestHumanInputFormRepositoryImplHelpers: - def test_build_email_recipients_with_member_and_external(self, monkeypatch: pytest.MonkeyPatch) -> None: + def test_build_email_recipients_with_member_and_external( + self, + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, + ) -> None: repo = _build_repository() - session_stub = object() - _patch_recipient_factory(monkeypatch) + _add_workspace_member(sqlite_session, user_id="member-1", email="member@example.com") + query_workspace_members_by_ids = repo._query_workspace_members_by_ids - def fake_query(self, session, restrict_to_user_ids): # type: ignore[no-untyped-def] - assert session is session_stub - assert restrict_to_user_ids == ["member-1"] - return [_WorkspaceMemberInfo(user_id="member-1", email="member@example.com")] + def query_members(session: Session, restrict_to_user_ids: list[str]) -> list[_WorkspaceMemberInfo]: + assert session is sqlite_session + return query_workspace_members_by_ids( + session=session, + restrict_to_user_ids=restrict_to_user_ids, + ) - monkeypatch.setattr(HumanInputFormRepositoryImpl, "_query_workspace_members_by_ids", fake_query) + monkeypatch.setattr(repo, "_query_workspace_members_by_ids", query_members) recipients = repo._build_email_recipients( - session=session_stub, + session=sqlite_session, form_id="form-id", delivery_id="delivery-id", recipients_config=EmailRecipients( @@ -111,20 +99,11 @@ class TestHumanInputFormRepositoryImplHelpers: external_payload = EmailExternalRecipientPayload.model_validate_json(external_recipient.recipient_payload) assert external_payload.email == "external@example.com" - def test_build_email_recipients_skips_unknown_members(self, monkeypatch: pytest.MonkeyPatch) -> None: + def test_build_email_recipients_skips_unknown_members(self, sqlite_session: Session) -> None: repo = _build_repository() - session_stub = object() - created = _patch_recipient_factory(monkeypatch) - - def fake_query(self, session, restrict_to_user_ids): # type: ignore[no-untyped-def] - assert session is session_stub - assert restrict_to_user_ids == ["missing-member"] - return [] - - monkeypatch.setattr(HumanInputFormRepositoryImpl, "_query_workspace_members_by_ids", fake_query) recipients = repo._build_email_recipients( - session=session_stub, + session=sqlite_session, form_id="form-id", delivery_id="delivery-id", recipients_config=EmailRecipients( @@ -138,24 +117,14 @@ class TestHumanInputFormRepositoryImplHelpers: assert len(recipients) == 1 assert recipients[0].recipient_type == RecipientType.EMAIL_EXTERNAL - assert len(created) == 1 # only external recipient created via factory - def test_build_email_recipients_whole_workspace_uses_all_members(self, monkeypatch: pytest.MonkeyPatch) -> None: + def test_build_email_recipients_whole_workspace_uses_all_members(self, sqlite_session: Session) -> None: repo = _build_repository() - session_stub = object() - _patch_recipient_factory(monkeypatch) - - def fake_query(self, session): # type: ignore[no-untyped-def] - assert session is session_stub - return [ - _WorkspaceMemberInfo(user_id="member-1", email="member1@example.com"), - _WorkspaceMemberInfo(user_id="member-2", email="member2@example.com"), - ] - - monkeypatch.setattr(HumanInputFormRepositoryImpl, "_query_all_workspace_members", fake_query) + _add_workspace_member(sqlite_session, user_id="member-1", email="member1@example.com") + _add_workspace_member(sqlite_session, user_id="member-2", email="member2@example.com") recipients = repo._build_email_recipients( - session=session_stub, + session=sqlite_session, form_id="form-id", delivery_id="delivery-id", recipients_config=EmailRecipients( @@ -168,20 +137,11 @@ class TestHumanInputFormRepositoryImplHelpers: emails = {EmailMemberRecipientPayload.model_validate_json(r.recipient_payload).email for r in recipients} assert emails == {"member1@example.com", "member2@example.com"} - def test_build_email_recipients_dedupes_external_by_email(self, monkeypatch: pytest.MonkeyPatch) -> None: + def test_build_email_recipients_dedupes_external_by_email(self, sqlite_session: Session) -> None: repo = _build_repository() - session_stub = object() - created = _patch_recipient_factory(monkeypatch) - - def fake_query(self, session, restrict_to_user_ids): # type: ignore[no-untyped-def] - assert session is session_stub - assert restrict_to_user_ids == [] - return [] - - monkeypatch.setattr(HumanInputFormRepositoryImpl, "_query_workspace_members_by_ids", fake_query) recipients = repo._build_email_recipients( - session=session_stub, + session=sqlite_session, form_id="form-id", delivery_id="delivery-id", recipients_config=EmailRecipients( @@ -194,24 +154,13 @@ class TestHumanInputFormRepositoryImplHelpers: ) assert len(recipients) == 1 - assert len(created) == 1 - def test_build_email_recipients_prefers_member_over_external_by_email( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: + def test_build_email_recipients_prefers_member_over_external_by_email(self, sqlite_session: Session) -> None: repo = _build_repository() - session_stub = object() - _patch_recipient_factory(monkeypatch) - - def fake_query(self, session, restrict_to_user_ids): # type: ignore[no-untyped-def] - assert session is session_stub - assert restrict_to_user_ids == ["member-1"] - return [_WorkspaceMemberInfo(user_id="member-1", email="shared@example.com")] - - monkeypatch.setattr(HumanInputFormRepositoryImpl, "_query_workspace_members_by_ids", fake_query) + _add_workspace_member(sqlite_session, user_id="member-1", email="shared@example.com") recipients = repo._build_email_recipients( - session=session_stub, + session=sqlite_session, form_id="form-id", delivery_id="delivery-id", recipients_config=EmailRecipients( @@ -228,20 +177,11 @@ class TestHumanInputFormRepositoryImplHelpers: def test_delivery_method_to_model_includes_external_recipients_with_whole_workspace( self, - monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, ) -> None: repo = _build_repository() - session_stub = object() - _patch_recipient_factory(monkeypatch) - - def fake_query(self, session): # type: ignore[no-untyped-def] - assert session is session_stub - return [ - _WorkspaceMemberInfo(user_id="member-1", email="member1@example.com"), - _WorkspaceMemberInfo(user_id="member-2", email="member2@example.com"), - ] - - monkeypatch.setattr(HumanInputFormRepositoryImpl, "_query_all_workspace_members", fake_query) + _add_workspace_member(sqlite_session, user_id="member-1", email="member1@example.com") + _add_workspace_member(sqlite_session, user_id="member-2", email="member2@example.com") method = EmailDeliveryMethod( config=EmailDeliveryConfig( @@ -254,7 +194,7 @@ class TestHumanInputFormRepositoryImplHelpers: ) ) - result = repo._delivery_method_to_model(session=session_stub, form_id="form-id", delivery_method=method) + result = repo._delivery_method_to_model(session=sqlite_session, form_id="form-id", delivery_method=method) assert len(result.recipients) == 3 member_emails = { @@ -279,155 +219,69 @@ def _make_form_definition() -> str: ).model_dump_json() -@dataclasses.dataclass -class _DummyForm: - id: str - workflow_run_id: str - node_id: str - tenant_id: str - app_id: str - form_definition: str - rendered_content: str - expiration_time: datetime - conversation_id: str | None = None - form_kind: HumanInputFormKind = HumanInputFormKind.RUNTIME - created_at: datetime = dataclasses.field(default_factory=naive_utc_now) - selected_action_id: str | None = None - submitted_data: str | None = None - submitted_at: datetime | None = None - submission_user_id: str | None = None - submission_end_user_id: str | None = None - completed_by_recipient_id: str | None = None - status: HumanInputFormStatus = HumanInputFormStatus.WAITING - - -@dataclasses.dataclass -class _DummyRecipient: - id: str - form_id: str - recipient_type: RecipientType - access_token: str - form: _DummyForm | None = None - recipient_payload: str = dataclasses.field( - default_factory=lambda: StandaloneWebAppRecipientPayload().model_dump_json() +def _make_form( + *, + form_id: str = "form-1", + workflow_run_id: str = "run-1", + node_id: str = "node-1", + tenant_id: str = "tenant-id", + selected_action_id: str | None = None, + submitted_data: str | None = None, + submitted_at: datetime | None = None, + expiration_time: datetime | None = None, +) -> HumanInputForm: + return HumanInputForm( + id=form_id, + workflow_run_id=workflow_run_id, + conversation_id=None, + node_id=node_id, + tenant_id=tenant_id, + app_id="app-id", + form_kind=HumanInputFormKind.RUNTIME, + form_definition=_make_form_definition(), + rendered_content="

hello

", + expiration_time=expiration_time or naive_utc_now(), + status=HumanInputFormStatus.WAITING, + selected_action_id=selected_action_id, + submitted_data=submitted_data, + submitted_at=submitted_at, ) -class _FakeScalarResult: - def __init__(self, obj): - self._obj = obj - - def first(self): - if isinstance(self._obj, list): - return self._obj[0] if self._obj else None - return self._obj - - def all(self): - if isinstance(self._obj, list): - return list(self._obj) - if self._obj is None: - return [] - return [self._obj] - - -class _FakeSession: - def __init__( - self, - *, - scalars_result=None, - scalars_results: list[object] | None = None, - forms: dict[str, _DummyForm] | None = None, - recipients: dict[str, _DummyRecipient] | None = None, - ): - if scalars_results is not None: - self._scalars_queue = list(scalars_results) - elif scalars_result is not None: - self._scalars_queue = [scalars_result] - else: - self._scalars_queue = [] - self.forms = forms or {} - self.recipients = recipients or {} - - def scalars(self, _query): - if self._scalars_queue: - result = self._scalars_queue.pop(0) - else: - result = None - return _FakeScalarResult(result) - - def get(self, model_cls, obj_id): # type: ignore[no-untyped-def] - if getattr(model_cls, "__name__", None) == "HumanInputForm": - return self.forms.get(obj_id) - if getattr(model_cls, "__name__", None) == "HumanInputFormRecipient": - return self.recipients.get(obj_id) - return None - - def add(self, _obj): - return None - - def flush(self): - return None - - def refresh(self, _obj): - return None - - def begin(self): - return self - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, tb): - return None - - -def _session_factory(session: _FakeSession): - class _SessionContext: - def __enter__(self): - return session - - def __exit__(self, exc_type, exc, tb): - return None - - def _factory(*_args, **_kwargs): - return _SessionContext() - - return _factory - - -def _patch_repo_session_factory(monkeypatch: pytest.MonkeyPatch, session: _FakeSession) -> None: - """Patch repository's global session factory to return our fake session. - - The repositories under test now use a global session factory; patch its - create_session method so unit tests don't hit a real database. - """ - monkeypatch.setattr( - "core.repositories.human_input_repository.session_factory.create_session", - _session_factory(session), - raising=True, +def _make_recipient( + form_id: str, + *, + recipient_id: str = "recipient-1", + recipient_type: RecipientType = RecipientType.STANDALONE_WEB_APP, + access_token: str = "token-123", +) -> HumanInputFormRecipient: + return HumanInputFormRecipient( + id=recipient_id, + form_id=form_id, + delivery_id="delivery-1", + recipient_type=recipient_type, + access_token=access_token, + recipient_payload=StandaloneWebAppRecipientPayload().model_dump_json(), ) +def _persist_form( + session: Session, + form: HumanInputForm, + recipients: list[HumanInputFormRecipient] | None = None, +) -> None: + session.add(form) + session.add_all(recipients or []) + session.commit() + + class TestHumanInputFormRepositoryImplPublicMethods: - def test_get_form_returns_entity_and_recipients(self, monkeypatch: pytest.MonkeyPatch): - form = _DummyForm( - id="form-1", - workflow_run_id="run-1", - node_id="node-1", - tenant_id="tenant-id", - app_id="app-id", - form_definition=_make_form_definition(), - rendered_content="

hello

", - expiration_time=naive_utc_now(), - ) - recipient = _DummyRecipient( - id="recipient-1", - form_id=form.id, - recipient_type=RecipientType.STANDALONE_WEB_APP, - access_token="token-123", - ) - session = _FakeSession(scalars_results=[form, [recipient]]) - _patch_repo_session_factory(monkeypatch, session) + def test_get_form_returns_entity_and_recipients(self, sqlite_session: Session): + form = _make_form() + recipient = _make_recipient(form.id) + other_tenant_form = _make_form(form_id="form-2", tenant_id="other-tenant") + sqlite_session.add(other_tenant_form) + _persist_form(sqlite_session, form, [recipient]) repo = HumanInputFormRepositoryImpl(tenant_id="tenant-id", workflow_execution_id=form.workflow_run_id) entity = repo.get_form(form.node_id) @@ -438,26 +292,14 @@ class TestHumanInputFormRepositoryImplPublicMethods: assert len(entity.recipients) == 1 assert entity.recipients[0].token == "token-123" - def test_get_form_returns_none_when_missing(self, monkeypatch: pytest.MonkeyPatch): - session = _FakeSession(scalars_results=[None]) - _patch_repo_session_factory(monkeypatch, session) + def test_get_form_returns_none_when_missing(self): repo = HumanInputFormRepositoryImpl(tenant_id="tenant-id", workflow_execution_id="run-1") assert repo.get_form("node-1") is None - def test_get_form_returns_unsubmitted_state(self, monkeypatch: pytest.MonkeyPatch): - form = _DummyForm( - id="form-1", - workflow_run_id="run-1", - node_id="node-1", - tenant_id="tenant-id", - app_id="app-id", - form_definition=_make_form_definition(), - rendered_content="

hello

", - expiration_time=naive_utc_now(), - ) - session = _FakeSession(scalars_results=[form, []]) - _patch_repo_session_factory(monkeypatch, session) + def test_get_form_returns_unsubmitted_state(self, sqlite_session: Session): + form = _make_form() + _persist_form(sqlite_session, form) repo = HumanInputFormRepositoryImpl(tenant_id="tenant-id", workflow_execution_id=form.workflow_run_id) entity = repo.get_form(form.node_id) @@ -467,22 +309,13 @@ class TestHumanInputFormRepositoryImplPublicMethods: assert entity.selected_action_id is None assert entity.submitted_data is None - def test_get_form_returns_submission_when_completed(self, monkeypatch: pytest.MonkeyPatch): - form = _DummyForm( - id="form-1", - workflow_run_id="run-1", - node_id="node-1", - tenant_id="tenant-id", - app_id="app-id", - form_definition=_make_form_definition(), - rendered_content="

hello

", - expiration_time=naive_utc_now(), + def test_get_form_returns_submission_when_completed(self, sqlite_session: Session): + form = _make_form( selected_action_id="approve", submitted_data='{"field": "value"}', submitted_at=naive_utc_now(), ) - session = _FakeSession(scalars_results=[form, []]) - _patch_repo_session_factory(monkeypatch, session) + _persist_form(sqlite_session, form) repo = HumanInputFormRepositoryImpl(tenant_id="tenant-id", workflow_execution_id=form.workflow_run_id) entity = repo.get_form(form.node_id) @@ -494,26 +327,10 @@ class TestHumanInputFormRepositoryImplPublicMethods: class TestHumanInputFormSubmissionRepository: - def test_get_by_token_returns_record(self, monkeypatch: pytest.MonkeyPatch): - form = _DummyForm( - id="form-1", - workflow_run_id="run-1", - node_id="node-1", - tenant_id="tenant-1", - app_id="app-1", - form_definition=_make_form_definition(), - rendered_content="

hello

", - expiration_time=naive_utc_now(), - ) - recipient = _DummyRecipient( - id="recipient-1", - form_id=form.id, - recipient_type=RecipientType.STANDALONE_WEB_APP, - access_token="token-123", - form=form, - ) - session = _FakeSession(scalars_result=recipient) - _patch_repo_session_factory(monkeypatch, session) + def test_get_by_token_returns_record(self, sqlite_session: Session): + form = _make_form(tenant_id="tenant-1") + recipient = _make_recipient(form.id) + _persist_form(sqlite_session, form, [recipient]) repo = HumanInputFormSubmissionRepository() record = repo.get_by_token("token-123") @@ -523,26 +340,10 @@ class TestHumanInputFormSubmissionRepository: assert record.recipient_type == RecipientType.STANDALONE_WEB_APP assert record.submitted is False - def test_get_by_form_id_and_recipient_type_uses_recipient(self, monkeypatch: pytest.MonkeyPatch): - form = _DummyForm( - id="form-1", - workflow_run_id="run-1", - node_id="node-1", - tenant_id="tenant-1", - app_id="app-1", - form_definition=_make_form_definition(), - rendered_content="

hello

", - expiration_time=naive_utc_now(), - ) - recipient = _DummyRecipient( - id="recipient-1", - form_id=form.id, - recipient_type=RecipientType.STANDALONE_WEB_APP, - access_token="token-123", - form=form, - ) - session = _FakeSession(scalars_result=recipient) - _patch_repo_session_factory(monkeypatch, session) + def test_get_by_form_id_and_recipient_type_uses_recipient(self, sqlite_session: Session): + form = _make_form(tenant_id="tenant-1") + recipient = _make_recipient(form.id) + _persist_form(sqlite_session, form, [recipient]) repo = HumanInputFormSubmissionRepository() record = repo.get_by_form_id_and_recipient_type( @@ -554,31 +355,17 @@ class TestHumanInputFormSubmissionRepository: assert record.recipient_id == recipient.id assert record.access_token == recipient.access_token - def test_mark_submitted_updates_fields(self, monkeypatch: pytest.MonkeyPatch): + def test_mark_submitted_updates_fields( + self, + monkeypatch: pytest.MonkeyPatch, + sqlite_session: Session, + ): fixed_now = datetime(2024, 1, 1, 0, 0, 0) monkeypatch.setattr("core.repositories.human_input_repository.naive_utc_now", lambda: fixed_now) - form = _DummyForm( - id="form-1", - workflow_run_id="run-1", - node_id="node-1", - tenant_id="tenant-1", - app_id="app-1", - form_definition=_make_form_definition(), - rendered_content="

hello

", - expiration_time=fixed_now, - ) - recipient = _DummyRecipient( - id="recipient-1", - form_id="form-1", - recipient_type=RecipientType.STANDALONE_WEB_APP, - access_token="token-123", - ) - session = _FakeSession( - forms={form.id: form}, - recipients={recipient.id: recipient}, - ) - _patch_repo_session_factory(monkeypatch, session) + form = _make_form(tenant_id="tenant-1", expiration_time=fixed_now) + recipient = _make_recipient(form.id) + _persist_form(sqlite_session, form, [recipient]) repo = HumanInputFormSubmissionRepository() record: HumanInputFormRecord = repo.mark_submitted( @@ -590,11 +377,50 @@ class TestHumanInputFormSubmissionRepository: submission_end_user_id="end-user-1", ) - assert form.selected_action_id == "approve" - assert form.completed_by_recipient_id == recipient.id - assert form.submission_user_id == "user-1" - assert form.submission_end_user_id == "end-user-1" - assert form.submitted_at == fixed_now + sqlite_session.expire_all() + persisted_form = sqlite_session.get(HumanInputForm, form.id) + assert persisted_form is not None + assert persisted_form.selected_action_id == "approve" + assert persisted_form.completed_by_recipient_id == recipient.id + assert persisted_form.submission_user_id == "user-1" + assert persisted_form.submission_end_user_id == "end-user-1" + assert persisted_form.submitted_at == fixed_now assert record.submitted is True assert record.selected_action_id == "approve" assert record.submitted_data == {"field": "value"} + + def test_mark_submitted_rolls_back_on_database_failure( + self, + sqlite_session: Session, + sqlite_engine: Engine, + ) -> None: + form = _make_form(tenant_id="tenant-1") + recipient = _make_recipient(form.id) + _persist_form(sqlite_session, form, [recipient]) + repo = HumanInputFormSubmissionRepository() + + def fail_form_update(_connection, _cursor, statement, _parameters, _context, _executemany) -> None: + if statement.lstrip().upper().startswith("UPDATE HUMAN_INPUT_FORMS"): + raise SQLAlchemyError("forced update failure") + + event.listen(sqlite_engine, "before_cursor_execute", fail_form_update) + try: + with pytest.raises(SQLAlchemyError, match="forced update failure"): + repo.mark_submitted( + form_id=form.id, + recipient_id=recipient.id, + selected_action_id="approve", + form_data={"field": "value"}, + submission_user_id="user-1", + submission_end_user_id="end-user-1", + ) + finally: + event.remove(sqlite_engine, "before_cursor_execute", fail_form_update) + + sqlite_session.expire_all() + persisted_form = sqlite_session.get(HumanInputForm, form.id) + assert persisted_form is not None + assert persisted_form.status == HumanInputFormStatus.WAITING + assert persisted_form.selected_action_id is None + assert persisted_form.submitted_data is None + assert persisted_form.submitted_at is None