test: use SQLite sessions in core repositories (#39067)

api/tests/unit_tests/core/repositories/test_human_input_repository.py

@@ -1,14 +1,13 @@
 from __future__ import annotations
 
-import dataclasses
 import json
-from collections.abc import Sequence
+from collections.abc import Iterator, Sequence
 from datetime import datetime, timedelta
-from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock
 
 import pytest
+from sqlalchemy import Engine, select
+from sqlalchemy.orm import Session, sessionmaker
 
 from core.repositories.human_input_repository import (
     FormCreateParams,
@@ -22,6 +21,7 @@ from core.repositories.human_input_repository import (
     _WorkspaceMemberInfo,
 )
 from core.workflow.human_input_adapter import (
+    DeliveryMethodType,
     EmailDeliveryConfig,
     EmailDeliveryMethod,
     EmailRecipients,
@@ -32,23 +32,35 @@ from core.workflow.human_input_adapter import (
 from core.workflow.nodes.human_input.entities import HumanInputNodeData, UserActionConfig
 from core.workflow.nodes.human_input.enums import HumanInputFormKind, HumanInputFormStatus
 from libs.datetime_utils import naive_utc_now
-from models.human_input import HumanInputFormRecipient, RecipientType
+from models.account import Account, TenantAccountJoin, TenantAccountRole
+from models.base import TypeBase
+from models.human_input import (
+    HumanInputDelivery,
+    HumanInputForm,
+    HumanInputFormRecipient,
+    RecipientType,
+)
 
 
-@pytest.fixture(autouse=True)
-def _stub_select(monkeypatch: pytest.MonkeyPatch) -> None:
-    class _FakeSelect:
-        def join(self, *_args: Any, **_kwargs: Any) -> _FakeSelect:
-            return self
+@pytest.fixture
+def repository_session(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[Session]:
+    """Bind repository-owned sessions to an isolated SQLite database."""
 
-        def where(self, *_args: Any, **_kwargs: Any) -> _FakeSelect:
-            return self
-
-        def options(self, *_args: Any, **_kwargs: Any) -> _FakeSelect:
-            return self
-
-    monkeypatch.setattr("core.repositories.human_input_repository.select", lambda *_args, **_kwargs: _FakeSelect())
-    monkeypatch.setattr("core.repositories.human_input_repository.selectinload", lambda *_args, **_kwargs: "_loader")
+    tables = [
+        HumanInputForm.__table__,
+        HumanInputDelivery.__table__,
+        HumanInputFormRecipient.__table__,
+        Account.__table__,
+        TenantAccountJoin.__table__,
+    ]
+    TypeBase.metadata.create_all(sqlite_engine, tables=tables)
+    repository_session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    monkeypatch.setattr(
+        "core.repositories.human_input_repository.session_factory.create_session",
+        repository_session_factory,
+    )
+    with repository_session_factory() as session:
+        yield session
 
 
 def _make_form_definition_json(*, include_expiration_time: bool) -> str:
@@ -63,196 +75,116 @@ def _make_form_definition_json(*, include_expiration_time: bool) -> str:
     return json.dumps(payload, default=str)
 
 
-@dataclasses.dataclass
-class _DummyForm:
-    id: str
-    workflow_run_id: str | None
-    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
+def _persist_form(
+    session: Session,
+    *,
+    form_id: str = "form-1",
+    tenant_id: str = "tenant",
+    workflow_run_id: str | None = "run",
+    node_id: str = "node",
+    status: HumanInputFormStatus = HumanInputFormStatus.WAITING,
+) -> HumanInputForm:
+    form = HumanInputForm(
+        id=form_id,
+        tenant_id=tenant_id,
+        app_id="app",
+        workflow_run_id=workflow_run_id,
+        conversation_id=None,
+        form_kind=HumanInputFormKind.RUNTIME,
+        node_id=node_id,
+        form_definition=_make_form_definition_json(include_expiration_time=True),
+        rendered_content="<p>x</p>",
+        expiration_time=naive_utc_now() + timedelta(hours=1),
+        status=status,
+    )
+    session.add(form)
+    session.commit()
+    return form
 
 
-@dataclasses.dataclass
-class _DummyRecipient:
-    id: str
-    form_id: str
-    recipient_type: RecipientType
-    access_token: str | None
-
-
-class _FakeScalarResult:
-    def __init__(self, obj: Any):
-        self._obj = obj
-
-    def first(self) -> Any:
-        if isinstance(self._obj, list):
-            return self._obj[0] if self._obj else None
-        return self._obj
-
-    def all(self) -> list[Any]:
-        if self._obj is None:
-            return []
-        if isinstance(self._obj, list):
-            return list(self._obj)
-        return [self._obj]
-
-
-class _FakeExecuteResult:
-    def __init__(self, rows: Sequence[tuple[Any, ...]]):
-        self._rows = list(rows)
-
-    def all(self) -> list[tuple[Any, ...]]:
-        return list(self._rows)
-
-
-class _FakeSession:
-    def __init__(
-        self,
-        *,
-        scalars_result: Any = None,
-        scalars_results: list[Any] | None = None,
-        forms: dict[str, _DummyForm] | None = None,
-        recipients: dict[str, _DummyRecipient] | None = None,
-        execute_rows: Sequence[tuple[Any, ...]] = (),
-    ):
-        if scalars_results is not None:
-            self._scalars_queue = list(scalars_results)
-        else:
-            self._scalars_queue = [scalars_result]
-        self._forms = forms or {}
-        self._recipients = recipients or {}
-        self._execute_rows = list(execute_rows)
-        self.added: list[Any] = []
-
-    def scalars(self, _query: Any) -> _FakeScalarResult:
-        if self._scalars_queue:
-            value = self._scalars_queue.pop(0)
-        else:
-            value = None
-        return _FakeScalarResult(value)
-
-    def execute(self, _stmt: Any) -> _FakeExecuteResult:
-        return _FakeExecuteResult(self._execute_rows)
-
-    def get(self, model_cls: Any, obj_id: str) -> Any:
-        name = getattr(model_cls, "__name__", "")
-        if name == "HumanInputForm":
-            return self._forms.get(obj_id)
-        if name == "HumanInputFormRecipient":
-            return self._recipients.get(obj_id)
-        return None
-
-    def add(self, obj: Any) -> None:
-        self.added.append(obj)
-
-    def add_all(self, objs: Sequence[Any]) -> None:
-        self.added.extend(list(objs))
-
-    def flush(self) -> None:
-        # Simulate DB default population for attributes referenced in entity wrappers.
-        for obj in self.added:
-            if hasattr(obj, "id") and obj.id in (None, ""):
-                obj.id = f"gen-{len(str(self.added))}"
-            if isinstance(obj, HumanInputFormRecipient) and obj.access_token is None:
-                if obj.recipient_type == RecipientType.CONSOLE:
-                    obj.access_token = "token-console"
-                elif obj.recipient_type == RecipientType.BACKSTAGE:
-                    obj.access_token = "token-backstage"
-                else:
-                    obj.access_token = "token-webapp"
-
-    def refresh(self, _obj: Any) -> None:
-        return None
-
-    def begin(self) -> _FakeSession:
-        return self
-
-    def __enter__(self) -> _FakeSession:
-        return self
-
-    def __exit__(self, exc_type, exc, tb) -> None:
-        return None
-
-
-class _SessionFactoryStub:
-    def __init__(self, session: _FakeSession):
-        self._session = session
-
-    def create_session(self) -> _FakeSession:
-        return self._session
-
-
-def _patch_session_factory(monkeypatch: pytest.MonkeyPatch, session: _FakeSession) -> None:
-    monkeypatch.setattr("core.repositories.human_input_repository.session_factory", _SessionFactoryStub(session))
+def _persist_recipient(
+    session: Session,
+    *,
+    form_id: str,
+    recipient_id: str = "recipient-1",
+    recipient_type: RecipientType = RecipientType.STANDALONE_WEB_APP,
+    access_token: str = "token-1",
+) -> HumanInputFormRecipient:
+    delivery = HumanInputDelivery(
+        id=f"delivery-{recipient_id}",
+        form_id=form_id,
+        delivery_method_type=DeliveryMethodType.WEBAPP,
+        delivery_config_id=None,
+        channel_payload="{}",
+    )
+    recipient = HumanInputFormRecipient(
+        id=recipient_id,
+        form_id=form_id,
+        delivery_id=delivery.id,
+        recipient_type=recipient_type,
+        recipient_payload="{}",
+        access_token=access_token,
+    )
+    session.add_all([delivery, recipient])
+    session.commit()
+    return recipient
 
 
 def test_recipient_entity_token_raises_when_missing() -> None:
-    recipient = SimpleNamespace(id="r1", access_token=None)
-    entity = _HumanInputFormRecipientEntityImpl(recipient)  # type: ignore[arg-type]
+    recipient = HumanInputFormRecipient(
+        id="r1",
+        form_id="f1",
+        delivery_id="d1",
+        recipient_type=RecipientType.CONSOLE,
+        recipient_payload="{}",
+        access_token=None,
+    )
+    entity = _HumanInputFormRecipientEntityImpl(recipient)
     with pytest.raises(AssertionError, match="access_token should not be None"):
         _ = entity.token
 
 
-def test_recipient_entity_id_and_token_success() -> None:
-    recipient = SimpleNamespace(id="r1", access_token="tok")
-    entity = _HumanInputFormRecipientEntityImpl(recipient)  # type: ignore[arg-type]
+def test_recipient_entity_id_and_token_success(repository_session: Session) -> None:
+    form = _persist_form(repository_session)
+    recipient = _persist_recipient(repository_session, form_id=form.id, recipient_id="r1", access_token="tok")
+    entity = _HumanInputFormRecipientEntityImpl(recipient)
     assert entity.id == "r1"
     assert entity.token == "tok"
 
 
-def test_form_entity_submission_token_prefers_console_then_webapp_then_none() -> None:
-    form = _DummyForm(
-        id="f1",
-        workflow_run_id="run",
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
+def test_form_entity_submission_token_prefers_console_then_webapp_then_none(repository_session: Session) -> None:
+    form = _persist_form(repository_session, form_id="f1")
+    console = _persist_recipient(
+        repository_session,
+        form_id=form.id,
+        recipient_id="c1",
+        recipient_type=RecipientType.CONSOLE,
+        access_token="ctok",
     )
-    console = _DummyRecipient(id="c1", form_id=form.id, recipient_type=RecipientType.CONSOLE, access_token="ctok")
-    webapp = _DummyRecipient(
-        id="w1", form_id=form.id, recipient_type=RecipientType.STANDALONE_WEB_APP, access_token="wtok"
+    webapp = _persist_recipient(
+        repository_session,
+        form_id=form.id,
+        recipient_id="w1",
+        recipient_type=RecipientType.STANDALONE_WEB_APP,
+        access_token="wtok",
     )
 
-    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[webapp, console])  # type: ignore[arg-type]
+    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[webapp, console])
     assert entity.submission_token == "ctok"
 
-    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[webapp])  # type: ignore[arg-type]
+    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[webapp])
     assert entity.submission_token == "wtok"
 
-    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[])  # type: ignore[arg-type]
+    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[])
     assert entity.submission_token is None
 
 
-def test_form_entity_submitted_data_parsed() -> None:
-    form = _DummyForm(
-        id="f1",
-        workflow_run_id="run",
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
-        submitted_data='{"a": 1}',
-        submitted_at=naive_utc_now(),
-    )
-    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[])  # type: ignore[arg-type]
+def test_form_entity_submitted_data_parsed(repository_session: Session) -> None:
+    form = _persist_form(repository_session, form_id="f1")
+    form.submitted_data = '{"a": 1}'
+    form.submitted_at = naive_utc_now()
+    repository_session.commit()
+    entity = _HumanInputFormEntityImpl(form_model=form, recipient_models=[])
     assert entity.submitted is True
     assert entity.submitted_data == {"a": 1}
     assert entity.rendered_content == "<p>x</p>"
@@ -260,42 +192,20 @@ def test_form_entity_submitted_data_parsed() -> None:
     assert entity.status == HumanInputFormStatus.WAITING
 
 
-def test_form_record_from_models_injects_expiration_time_when_missing() -> None:
+def test_form_record_from_models_injects_expiration_time_when_missing(repository_session: Session) -> None:
     expiration = naive_utc_now()
-    form = _DummyForm(
-        id="f1",
-        workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=False),
-        rendered_content="<p>x</p>",
-        expiration_time=expiration,
-        submitted_data='{"k": "v"}',
-    )
-    record = HumanInputFormRecord.from_models(form, None)  # type: ignore[arg-type]
+    form = _persist_form(repository_session, form_id="f1", workflow_run_id=None)
+    form.form_definition = _make_form_definition_json(include_expiration_time=False)
+    form.expiration_time = expiration
+    form.submitted_data = '{"k": "v"}'
+    repository_session.commit()
+    record = HumanInputFormRecord.from_models(form, None)
     assert record.definition.expiration_time == expiration
     assert record.submitted_data == {"k": "v"}
     assert record.submitted is False
 
 
-def test_create_email_recipients_from_resolved_dedupes_and_skips_blank(monkeypatch: pytest.MonkeyPatch) -> None:
-    created: list[SimpleNamespace] = []
-
-    def fake_new(cls, form_id: str, delivery_id: str, payload: Any):  # type: ignore[no-untyped-def]
-        recipient = SimpleNamespace(
-            id=f"{payload.TYPE}-{len(created)}",
-            form_id=form_id,
-            delivery_id=delivery_id,
-            recipient_type=payload.TYPE,
-            recipient_payload=payload.model_dump_json(),
-            access_token="tok",
-        )
-        created.append(recipient)
-        return recipient
-
-    monkeypatch.setattr("core.repositories.human_input_repository.HumanInputFormRecipient.new", classmethod(fake_new))
-
+def test_create_email_recipients_from_resolved_dedupes_and_skips_blank() -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
     recipients = repo._create_email_recipients_from_resolved(  # type: ignore[attr-defined]
         form_id="f",
@@ -310,26 +220,47 @@ def test_create_email_recipients_from_resolved_dedupes_and_skips_blank(monkeypat
     assert [r.recipient_type for r in recipients] == [RecipientType.EMAIL_MEMBER, RecipientType.EMAIL_EXTERNAL]
 
 
-def test_query_workspace_members_by_ids_empty_returns_empty() -> None:
+def test_query_workspace_members_by_ids_empty_returns_empty(repository_session: Session) -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
-    assert repo._query_workspace_members_by_ids(session=MagicMock(), restrict_to_user_ids=["", ""]) == []
+    assert repo._query_workspace_members_by_ids(session=repository_session, restrict_to_user_ids=["", ""]) == []
 
 
-def test_query_workspace_members_by_ids_maps_rows() -> None:
-    session = _FakeSession(execute_rows=[("u1", "a@example.com"), ("u2", "b@example.com")])
-    repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
-    rows = repo._query_workspace_members_by_ids(session=session, restrict_to_user_ids=["u1", "u2"])
-    assert rows == [
-        _WorkspaceMemberInfo(user_id="u1", email="a@example.com"),
-        _WorkspaceMemberInfo(user_id="u2", email="b@example.com"),
+def test_query_workspace_members_by_ids_maps_rows_and_scopes_tenant(repository_session: Session) -> None:
+    accounts = [
+        Account(name="One", email="a@example.com"),
+        Account(name="Two", email="b@example.com"),
+        Account(name="Other", email="other@example.com"),
     ]
-
-
-def test_query_all_workspace_members_maps_rows() -> None:
-    session = _FakeSession(execute_rows=[("u1", "a@example.com")])
+    repository_session.add_all(accounts)
+    repository_session.flush()
+    repository_session.add_all(
+        [
+            TenantAccountJoin(tenant_id="tenant", account_id=accounts[0].id, role=TenantAccountRole.NORMAL),
+            TenantAccountJoin(tenant_id="tenant", account_id=accounts[1].id, role=TenantAccountRole.NORMAL),
+            TenantAccountJoin(tenant_id="other-tenant", account_id=accounts[2].id, role=TenantAccountRole.NORMAL),
+        ]
+    )
+    repository_session.commit()
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
-    rows = repo._query_all_workspace_members(session=session)
-    assert rows == [_WorkspaceMemberInfo(user_id="u1", email="a@example.com")]
+    rows = repo._query_workspace_members_by_ids(
+        session=repository_session,
+        restrict_to_user_ids=[accounts[0].id, accounts[1].id, accounts[2].id],
+    )
+    assert set(rows) == {
+        _WorkspaceMemberInfo(user_id=accounts[0].id, email="a@example.com"),
+        _WorkspaceMemberInfo(user_id=accounts[1].id, email="b@example.com"),
+    }
+
+
+def test_query_all_workspace_members_maps_rows(repository_session: Session) -> None:
+    account = Account(name="One", email="a@example.com")
+    repository_session.add(account)
+    repository_session.flush()
+    repository_session.add(TenantAccountJoin(tenant_id="tenant", account_id=account.id, role=TenantAccountRole.NORMAL))
+    repository_session.commit()
+    repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
+    rows = repo._query_all_workspace_members(session=repository_session)
+    assert rows == [_WorkspaceMemberInfo(user_id=account.id, email="a@example.com")]
 
 
 def test_repository_init_sets_tenant_id() -> None:
@@ -337,11 +268,13 @@ def test_repository_init_sets_tenant_id() -> None:
     assert repo._tenant_id == "tenant"
 
 
-def test_delivery_method_to_model_webapp_creates_delivery_and_recipient(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_delivery_method_to_model_webapp_creates_delivery_and_recipient(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
     monkeypatch.setattr("core.repositories.human_input_repository.uuidv7", lambda: "del-1")
     result = repo._delivery_method_to_model(
-        session=MagicMock(), form_id="form-1", delivery_method=WebAppDeliveryMethod()
+        session=repository_session, form_id="form-1", delivery_method=WebAppDeliveryMethod()
     )
     assert result.delivery.id == "del-1"
     assert result.delivery.form_id == "form-1"
@@ -349,12 +282,14 @@ def test_delivery_method_to_model_webapp_creates_delivery_and_recipient(monkeypa
     assert result.recipients[0].recipient_type == RecipientType.STANDALONE_WEB_APP
 
 
-def test_delivery_method_to_model_email_uses_build_email_recipients(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_delivery_method_to_model_email_uses_build_email_recipients(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
     monkeypatch.setattr("core.repositories.human_input_repository.uuidv7", lambda: "del-1")
     called: dict[str, Any] = {}
 
-    def fake_build(*, session: Any, form_id: str, delivery_id: str, recipients_config: Any) -> list[Any]:
+    def fake_build(*, session: Session, form_id: str, delivery_id: str, recipients_config: Any) -> list[Any]:
         called.update(
             {"session": session, "form_id": form_id, "delivery_id": delivery_id, "recipients_config": recipients_config}
         )
@@ -372,12 +307,15 @@ def test_delivery_method_to_model_email_uses_build_email_recipients(monkeypatch:
             body="b",
         )
     )
-    result = repo._delivery_method_to_model(session="sess", form_id="form-1", delivery_method=method)
+    result = repo._delivery_method_to_model(session=repository_session, form_id="form-1", delivery_method=method)
     assert result.recipients == ["r"]
+    assert called["session"] is repository_session
     assert called["delivery_id"] == "del-1"
 
 
-def test_build_email_recipients_uses_all_members_when_whole_workspace(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_build_email_recipients_uses_all_members_when_whole_workspace(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
     monkeypatch.setattr(
         repo,
@@ -386,7 +324,7 @@ def test_build_email_recipients_uses_all_members_when_whole_workspace(monkeypatc
     )
     monkeypatch.setattr(repo, "_create_email_recipients_from_resolved", lambda **_: ["ok"])
     recipients = repo._build_email_recipients(
-        session=MagicMock(),
+        session=repository_session,
         form_id="f",
         delivery_id="d",
         recipients_config=EmailRecipients(include_bound_group=True, items=[ExternalRecipient(email="e@example.com")]),
@@ -394,7 +332,9 @@ def test_build_email_recipients_uses_all_members_when_whole_workspace(monkeypatc
     assert recipients == ["ok"]
 
 
-def test_build_email_recipients_uses_selected_members_when_not_whole_workspace(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_build_email_recipients_uses_selected_members_when_not_whole_workspace(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant")
 
     def fake_query(*, session: Any, restrict_to_user_ids: Sequence[str]) -> list[_WorkspaceMemberInfo]:
@@ -404,7 +344,7 @@ def test_build_email_recipients_uses_selected_members_when_not_whole_workspace(m
     monkeypatch.setattr(repo, "_query_workspace_members_by_ids", fake_query)
     monkeypatch.setattr(repo, "_create_email_recipients_from_resolved", lambda **_: ["ok"])
     recipients = repo._build_email_recipients(
-        session=MagicMock(),
+        session=repository_session,
         form_id="f",
         delivery_id="d",
         recipients_config=EmailRecipients(
@@ -415,29 +355,13 @@ def test_build_email_recipients_uses_selected_members_when_not_whole_workspace(m
     assert recipients == ["ok"]
 
 
-def test_get_form_returns_entity_and_none_when_missing(monkeypatch: pytest.MonkeyPatch) -> None:
-    _patch_session_factory(monkeypatch, _FakeSession(scalars_results=[None]))
+def test_get_form_returns_entity_and_none_when_missing(repository_session: Session) -> None:
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant", workflow_execution_id="run")
     assert repo.get_form("node") is None
 
-    form = _DummyForm(
-        id="f1",
-        workflow_run_id="run",
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
-    )
-    recipient = _DummyRecipient(
-        id="r1",
-        form_id=form.id,
-        recipient_type=RecipientType.STANDALONE_WEB_APP,
-        access_token="tok",
-    )
-    session = _FakeSession(scalars_results=[form, [recipient]])
-    _patch_session_factory(monkeypatch, session)
+    form = _persist_form(repository_session, form_id="f1")
+    _persist_form(repository_session, form_id="other-tenant-form", tenant_id="other-tenant")
+    recipient = _persist_recipient(repository_session, form_id=form.id, recipient_id="r1", access_token="tok")
     repo = HumanInputFormRepositoryImpl(tenant_id="tenant", workflow_execution_id="run")
     entity = repo.get_form("node")
     assert entity is not None
@@ -446,15 +370,15 @@ def test_get_form_returns_entity_and_none_when_missing(monkeypatch: pytest.Monke
     assert entity.recipients[0].token == "tok"
 
 
-def test_create_form_adds_console_and_backstage_recipients(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_create_form_adds_console_and_backstage_recipients(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     fixed_now = datetime(2024, 1, 1, 0, 0, 0)
     monkeypatch.setattr("core.repositories.human_input_repository.naive_utc_now", lambda: fixed_now)
 
     ids = iter(["form-id", "del-web", "del-console", "del-backstage"])
     monkeypatch.setattr("core.repositories.human_input_repository.uuidv7", lambda: next(ids))
 
-    session = _FakeSession()
-    _patch_session_factory(monkeypatch, session)
     repo = HumanInputFormRepositoryImpl(
         tenant_id="tenant",
         app_id="app",
@@ -485,19 +409,28 @@ def test_create_form_adds_console_and_backstage_recipients(monkeypatch: pytest.M
     assert entity.id == "form-id"
     assert entity.expiration_time == fixed_now + timedelta(hours=form_config.timeout)
     # Console token should take precedence when console recipient is present.
-    assert entity.submission_token == "token-console"
+    assert entity.submission_token is not None
     assert len(entity.recipients) == 3
+    repository_session.expire_all()
+    persisted_form = repository_session.get(HumanInputForm, "form-id")
+    assert persisted_form is not None
+    assert persisted_form.status == HumanInputFormStatus.WAITING
+    recipients = repository_session.scalars(
+        select(HumanInputFormRecipient).where(HumanInputFormRecipient.form_id == "form-id")
+    ).all()
+    assert {recipient.recipient_type for recipient in recipients} == {
+        RecipientType.STANDALONE_WEB_APP,
+        RecipientType.CONSOLE,
+        RecipientType.BACKSTAGE,
+    }
 
 
-def test_submission_get_by_token_returns_none_when_missing_or_form_missing(monkeypatch: pytest.MonkeyPatch) -> None:
-    _patch_session_factory(monkeypatch, _FakeSession(scalars_result=None))
+def test_submission_get_by_token_returns_none_when_missing_or_form_missing(repository_session: Session) -> None:
     repo = HumanInputFormSubmissionRepository()
     assert repo.get_by_token("tok") is None
 
-    recipient = SimpleNamespace(form=None)
-    _patch_session_factory(monkeypatch, _FakeSession(scalars_result=recipient))
-    repo = HumanInputFormSubmissionRepository()
-    assert repo.get_by_token("tok") is None
+    _persist_recipient(repository_session, form_id="missing-form", access_token="orphan-token")
+    assert repo.get_by_token("orphan-token") is None
 
 
 def test_submission_repository_init_no_args() -> None:
@@ -505,50 +438,36 @@ def test_submission_repository_init_no_args() -> None:
     assert isinstance(repo, HumanInputFormSubmissionRepository)
 
 
-def test_submission_get_by_token_and_get_by_form_id_success_paths(monkeypatch: pytest.MonkeyPatch) -> None:
-    form = _DummyForm(
-        id="f1",
-        workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
-    )
-    recipient = SimpleNamespace(
-        id="r1",
-        form_id=form.id,
-        recipient_type=RecipientType.STANDALONE_WEB_APP,
-        access_token="tok",
-        form=form,
-    )
-
-    _patch_session_factory(monkeypatch, _FakeSession(scalars_result=recipient))
+def test_submission_get_by_token_and_get_by_form_id_success_paths(repository_session: Session) -> None:
+    form = _persist_form(repository_session, form_id="f1", workflow_run_id=None)
+    recipient = _persist_recipient(repository_session, form_id=form.id, recipient_id="r1", access_token="tok")
     repo = HumanInputFormSubmissionRepository()
     record = repo.get_by_token("tok")
     assert record is not None
     assert record.access_token == "tok"
 
-    _patch_session_factory(monkeypatch, _FakeSession(scalars_result=recipient))
-    repo = HumanInputFormSubmissionRepository()
     record = repo.get_by_form_id_and_recipient_type(form_id=form.id, recipient_type=RecipientType.STANDALONE_WEB_APP)
     assert record is not None
     assert record.recipient_id == "r1"
 
+    record = repo.get_by_form_id(form.id)
+    assert record is not None
+    assert record.form_id == form.id
+    assert record.recipient_id is None
 
-def test_submission_get_by_form_id_returns_none_on_missing(monkeypatch: pytest.MonkeyPatch) -> None:
-    _patch_session_factory(monkeypatch, _FakeSession(scalars_result=None))
+
+def test_submission_get_by_form_id_returns_none_on_missing(repository_session: Session) -> None:
     repo = HumanInputFormSubmissionRepository()
     assert repo.get_by_form_id_and_recipient_type(form_id="f", recipient_type=RecipientType.CONSOLE) is None
+    assert repo.get_by_form_id("f") is None
 
 
-def test_mark_submitted_updates_and_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_mark_submitted_updates_and_raises_when_missing(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     fixed_now = datetime(2024, 1, 1, 0, 0, 0)
     monkeypatch.setattr("core.repositories.human_input_repository.naive_utc_now", lambda: fixed_now)
 
-    missing_session = _FakeSession(forms={})
-    _patch_session_factory(monkeypatch, missing_session)
     repo = HumanInputFormSubmissionRepository()
     with pytest.raises(FormNotFoundError, match="form not found"):
         repo.mark_submitted(
@@ -560,20 +479,14 @@ def test_mark_submitted_updates_and_raises_when_missing(monkeypatch: pytest.Monk
             submission_end_user_id=None,
         )
 
-    form = _DummyForm(
-        id="f",
-        workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=fixed_now,
+    form = _persist_form(repository_session, form_id="f", workflow_run_id=None)
+    recipient = _persist_recipient(
+        repository_session,
+        form_id=form.id,
+        recipient_id="r",
+        recipient_type=RecipientType.CONSOLE,
+        access_token="tok",
     )
-    recipient = _DummyRecipient(id="r", form_id=form.id, recipient_type=RecipientType.CONSOLE, access_token="tok")
-    session = _FakeSession(forms={form.id: form}, recipients={recipient.id: recipient})
-    _patch_session_factory(monkeypatch, session)
-    repo = HumanInputFormSubmissionRepository()
     record = repo.mark_submitted(
         form_id=form.id,
         recipient_id=recipient.id,
@@ -582,33 +495,29 @@ def test_mark_submitted_updates_and_raises_when_missing(monkeypatch: pytest.Monk
         submission_user_id="u",
         submission_end_user_id="eu",
     )
-    assert form.status == HumanInputFormStatus.SUBMITTED
-    assert form.submitted_at == fixed_now
+    repository_session.expire_all()
+    persisted = repository_session.get(HumanInputForm, form.id)
+    assert persisted is not None
+    assert persisted.status == HumanInputFormStatus.SUBMITTED
+    assert persisted.submitted_at == fixed_now
+    assert persisted.completed_by_recipient_id == recipient.id
     assert record.submitted_data == {"k": "v"}
 
 
-def test_mark_submitted_serializes_select_and_file_payloads(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_mark_submitted_serializes_select_and_file_payloads(
+    repository_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
     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="f-complex",
-        workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=fixed_now,
-    )
-    recipient = _DummyRecipient(
-        id="r-complex",
+    form = _persist_form(repository_session, form_id="f-complex", workflow_run_id=None)
+    recipient = _persist_recipient(
+        repository_session,
         form_id=form.id,
+        recipient_id="r-complex",
         recipient_type=RecipientType.CONSOLE,
         access_token="tok",
     )
-    session = _FakeSession(forms={form.id: form}, recipients={recipient.id: recipient})
-    _patch_session_factory(monkeypatch, session)
 
     payload = {
         "decision": "approve",
@@ -650,98 +559,71 @@ def test_mark_submitted_serializes_select_and_file_payloads(monkeypatch: pytest.
         submission_end_user_id="end-user-1",
     )
 
-    assert json.loads(form.submitted_data or "") == payload
+    repository_session.expire_all()
+    persisted = repository_session.get(HumanInputForm, form.id)
+    assert persisted is not None
+    assert json.loads(persisted.submitted_data or "") == payload
     assert record.submitted_data == payload
 
 
-def test_mark_timeout_invalid_status_raises(monkeypatch: pytest.MonkeyPatch) -> None:
-    form = _DummyForm(
-        id="f",
-        workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
-    )
-    session = _FakeSession(forms={form.id: form})
-    _patch_session_factory(monkeypatch, session)
+def test_mark_timeout_invalid_status_rolls_back(repository_session: Session) -> None:
+    form = _persist_form(repository_session, form_id="f", workflow_run_id=None)
     repo = HumanInputFormSubmissionRepository()
     with pytest.raises(_InvalidTimeoutStatusError, match="invalid timeout status"):
         repo.mark_timeout(form_id=form.id, timeout_status=HumanInputFormStatus.SUBMITTED)  # type: ignore[arg-type]
+    repository_session.expire_all()
+    persisted = repository_session.get(HumanInputForm, form.id)
+    assert persisted is not None
+    assert persisted.status == HumanInputFormStatus.WAITING
 
 
-def test_mark_timeout_already_timed_out_returns_record(monkeypatch: pytest.MonkeyPatch) -> None:
-    form = _DummyForm(
-        id="f",
+def test_mark_timeout_already_timed_out_returns_record(repository_session: Session) -> None:
+    form = _persist_form(
+        repository_session,
+        form_id="f",
         workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
         status=HumanInputFormStatus.TIMEOUT,
     )
-    session = _FakeSession(forms={form.id: form})
-    _patch_session_factory(monkeypatch, session)
     repo = HumanInputFormSubmissionRepository()
     record = repo.mark_timeout(form_id=form.id, timeout_status=HumanInputFormStatus.TIMEOUT, reason="r")
     assert record.status == HumanInputFormStatus.TIMEOUT
 
 
-def test_mark_timeout_submitted_raises_form_not_found(monkeypatch: pytest.MonkeyPatch) -> None:
-    form = _DummyForm(
-        id="f",
+def test_mark_timeout_submitted_raises_form_not_found(repository_session: Session) -> None:
+    form = _persist_form(
+        repository_session,
+        form_id="f",
         workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
         status=HumanInputFormStatus.SUBMITTED,
     )
-    session = _FakeSession(forms={form.id: form})
-    _patch_session_factory(monkeypatch, session)
     repo = HumanInputFormSubmissionRepository()
     with pytest.raises(FormNotFoundError, match="form already submitted"):
         repo.mark_timeout(form_id=form.id, timeout_status=HumanInputFormStatus.EXPIRED)
 
 
-def test_mark_timeout_updates_fields(monkeypatch: pytest.MonkeyPatch) -> None:
-    form = _DummyForm(
-        id="f",
-        workflow_run_id=None,
-        node_id="node",
-        tenant_id="tenant",
-        app_id="app",
-        form_definition=_make_form_definition_json(include_expiration_time=True),
-        rendered_content="<p>x</p>",
-        expiration_time=naive_utc_now(),
-        selected_action_id="a",
-        submitted_data="{}",
-        submission_user_id="u",
-        submission_end_user_id="eu",
-        completed_by_recipient_id="r",
-        status=HumanInputFormStatus.WAITING,
-    )
-    session = _FakeSession(forms={form.id: form})
-    _patch_session_factory(monkeypatch, session)
+def test_mark_timeout_updates_fields(repository_session: Session) -> None:
+    form = _persist_form(repository_session, form_id="f", workflow_run_id=None)
+    form.selected_action_id = "a"
+    form.submitted_data = "{}"
+    form.submission_user_id = "u"
+    form.submission_end_user_id = "eu"
+    form.completed_by_recipient_id = "r"
+    repository_session.commit()
     repo = HumanInputFormSubmissionRepository()
     record = repo.mark_timeout(form_id=form.id, timeout_status=HumanInputFormStatus.EXPIRED)
-    assert form.status == HumanInputFormStatus.EXPIRED
-    assert form.selected_action_id is None
-    assert form.submitted_data is None
-    assert form.submission_user_id is None
-    assert form.submission_end_user_id is None
-    assert form.completed_by_recipient_id is None
+    repository_session.expire_all()
+    persisted = repository_session.get(HumanInputForm, form.id)
+    assert persisted is not None
+    assert persisted.status == HumanInputFormStatus.EXPIRED
+    assert persisted.selected_action_id is None
+    assert persisted.submitted_data is None
+    assert persisted.submission_user_id is None
+    assert persisted.submission_end_user_id is None
+    assert persisted.completed_by_recipient_id is None
     assert record.status == HumanInputFormStatus.EXPIRED
 
 
-def test_mark_timeout_raises_when_form_missing(monkeypatch: pytest.MonkeyPatch) -> None:
-    _patch_session_factory(monkeypatch, _FakeSession(forms={}))
+def test_mark_timeout_raises_when_form_missing(repository_session: Session) -> None:
     repo = HumanInputFormSubmissionRepository()
     with pytest.raises(FormNotFoundError, match="form not found"):
         repo.mark_timeout(form_id="missing", timeout_status=HumanInputFormStatus.TIMEOUT)