From f6bbb94112a783e5de94f50f6a0f63d1bb107484 Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Fri, 31 Jul 2026 21:54:30 +0900 Subject: [PATCH] test: use SQLite sessions in core workflow (#39086) --- .../nodes/agent_v2/test_binding_resolver.py | 462 +++++++++++------- 1 file changed, 295 insertions(+), 167 deletions(-) diff --git a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py index d52abb46128..c7774e6836f 100644 --- a/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py +++ b/api/tests/unit_tests/core/workflow/nodes/agent_v2/test_binding_resolver.py @@ -1,10 +1,13 @@ -import pytest -from sqlalchemy.orm import Session +from uuid import uuid4 -from core.workflow.nodes.agent_v2.binding_resolver import ( - WorkflowAgentBindingError, - WorkflowAgentBindingResolver, -) +import pytest +from sqlalchemy import event, inspect +from sqlalchemy.engine import Engine +from sqlalchemy.orm import ORMExecuteState, Session, sessionmaker +from sqlalchemy.sql import Executable + +import core.workflow.nodes.agent_v2.binding_resolver as resolver_module +from core.workflow.nodes.agent_v2.binding_resolver import WorkflowAgentBindingError, WorkflowAgentBindingResolver from models.agent import ( Agent, AgentConfigRevision, @@ -18,53 +21,58 @@ from models.agent import ( ) from models.agent_config_entities import AgentSoulConfig, AgentSoulModelConfig, WorkflowNodeJobConfig - -class FakeSession: - def __init__(self, scalar_results): - self._scalar_results = list(scalar_results) - self.expunge_calls = [] - self.scalar_statements = [] - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, tb): - return False - - def scalar(self, stmt): - self.scalar_statements.append(stmt) - if not self._scalar_results: - return None - return self._scalar_results.pop(0) - - def expunge(self, value): - self.expunge_calls.append(value) +RESOLVER_MODELS = (WorkflowAgentNodeBinding, Agent, AgentConfigSnapshot, AgentConfigRevision) -def _binding() -> WorkflowAgentNodeBinding: - return WorkflowAgentNodeBinding( - id="binding-1", - tenant_id="tenant-1", - app_id="app-1", - workflow_id="workflow-1", - node_id="agent-node", - agent_id="agent-1", - current_snapshot_id="snapshot-1", - node_job_config=WorkflowNodeJobConfig(), +def _resolve_ids() -> dict[str, str]: + return { + "tenant_id": str(uuid4()), + "app_id": str(uuid4()), + "workflow_id": str(uuid4()), + "node_id": "agent-node", + } + + +def _agent( + *, + tenant_id: str, + status: AgentStatus = AgentStatus.ACTIVE, + scope: AgentScope = AgentScope.WORKFLOW_ONLY, + source: AgentSource = AgentSource.WORKFLOW, + app_id: str | None = None, + active_config_snapshot_id: str | None = None, + active_config_is_published: bool = True, +) -> Agent: + return Agent( + tenant_id=tenant_id, + name=f"Agent {uuid4()}", + description="", + role="", + icon_type=None, + icon=None, + icon_background=None, + scope=scope, + source=source, + app_id=app_id, + backing_app_id=None, + workflow_id=None, + workflow_node_id=None, + active_config_snapshot_id=active_config_snapshot_id, + active_config_has_model=True, + active_config_is_published=active_config_is_published, + status=status, + created_by=None, + updated_by=None, + archived_by=None, + archived_at=None, ) -def _agent(*, status: AgentStatus = AgentStatus.ACTIVE) -> Agent: - return Agent(id="agent-1", tenant_id="tenant-1", name="Agent", status=status) - - -def _snapshot() -> AgentConfigSnapshot: +def _snapshot(*, tenant_id: str, agent_id: str) -> AgentConfigSnapshot: return AgentConfigSnapshot( - id="snapshot-1", - tenant_id="tenant-1", - agent_id="agent-1", + tenant_id=tenant_id, + agent_id=agent_id, version=1, - home_snapshot_id="home-1", config_snapshot=AgentSoulConfig( model=AgentSoulModelConfig( plugin_id="langgenius/openai", @@ -72,99 +80,186 @@ def _snapshot() -> AgentConfigSnapshot: model="gpt-test", ) ), + summary=None, + version_note=None, + created_by=None, ) -def _resolve() -> dict[str, str]: - return { - "tenant_id": "tenant-1", - "app_id": "app-1", - "workflow_id": "workflow-1", - "node_id": "agent-node", - } - - -def test_binding_resolver_returns_detached_binding_bundle(monkeypatch: pytest.MonkeyPatch): - fake_session = FakeSession([_binding(), _agent(), _snapshot()]) - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: fake_session, +def _binding( + *, ids: dict[str, str], agent_id: str, snapshot_id: str, binding_type: WorkflowAgentBindingType +) -> WorkflowAgentNodeBinding: + return WorkflowAgentNodeBinding( + tenant_id=ids["tenant_id"], + app_id=ids["app_id"], + workflow_id=ids["workflow_id"], + workflow_version="draft", + node_id=ids["node_id"], + binding_type=binding_type, + agent_id=agent_id, + current_snapshot_id=snapshot_id, + node_job_config=WorkflowNodeJobConfig(), + created_by=None, + updated_by=None, ) - bundle = WorkflowAgentBindingResolver().resolve(**_resolve()) - assert bundle.binding.id == "binding-1" - assert bundle.agent.id == "agent-1" - assert bundle.snapshot.id == "snapshot-1" - assert fake_session.expunge_calls == [bundle.binding, bundle.agent, bundle.snapshot] +def _bind_factory(monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> list[Executable]: + scalar_statements: list[Executable] = [] + + class RecordingSession(Session): + pass + + def record_statement(execute_state: ORMExecuteState) -> None: + scalar_statements.append(execute_state.statement) + + event.listen(RecordingSession, "do_orm_execute", record_statement) + factory = sessionmaker(bind=sqlite_engine, class_=RecordingSession, expire_on_commit=False) + monkeypatch.setattr(resolver_module.session_factory, "create_session", factory) + return scalar_statements -def test_binding_resolver_uses_active_snapshot_for_roster_agent(monkeypatch: pytest.MonkeyPatch): - binding = _binding() - binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT - binding.current_snapshot_id = "old-snapshot" - agent = _agent() - agent.active_config_snapshot_id = "active-snapshot" - snapshot = _snapshot() - snapshot.id = "active-snapshot" - fake_session = FakeSession([binding, agent, snapshot]) - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: fake_session, +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) +def test_binding_resolver_returns_detached_binding_bundle( + monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session +) -> None: + ids = _resolve_ids() + agent = _agent(tenant_id=ids["tenant_id"]) + sqlite_session.add(agent) + sqlite_session.flush() + snapshot = _snapshot(tenant_id=ids["tenant_id"], agent_id=agent.id) + sqlite_session.add(snapshot) + sqlite_session.flush() + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=snapshot.id, + binding_type=WorkflowAgentBindingType.INLINE_AGENT, ) + sqlite_session.add(binding) + sqlite_session.commit() + _bind_factory(monkeypatch, sqlite_engine) - bundle = WorkflowAgentBindingResolver().resolve(**_resolve()) + bundle = WorkflowAgentBindingResolver().resolve(**ids) - assert bundle.snapshot.id == "active-snapshot" + assert bundle.binding.id == binding.id + assert bundle.agent.id == agent.id + assert bundle.snapshot.id == snapshot.id + assert inspect(bundle.binding).detached + assert inspect(bundle.agent).detached + assert inspect(bundle.snapshot).detached + + +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) +def test_binding_resolver_uses_active_snapshot_for_roster_agent( + monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session +) -> None: + ids = _resolve_ids() + agent = _agent( + tenant_id=ids["tenant_id"], + scope=AgentScope.ROSTER, + source=AgentSource.ROSTER, + ) + sqlite_session.add(agent) + sqlite_session.flush() + active_snapshot = _snapshot(tenant_id=ids["tenant_id"], agent_id=agent.id) + sqlite_session.add(active_snapshot) + sqlite_session.flush() + agent.active_config_snapshot_id = active_snapshot.id + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=str(uuid4()), + binding_type=WorkflowAgentBindingType.ROSTER_AGENT, + ) + sqlite_session.add(binding) + sqlite_session.commit() + _bind_factory(monkeypatch, sqlite_engine) + + bundle = WorkflowAgentBindingResolver().resolve(**ids) + + assert bundle.snapshot.id == active_snapshot.id @pytest.mark.parametrize( - "binding_type", - [WorkflowAgentBindingType.ROSTER_AGENT, WorkflowAgentBindingType.INLINE_AGENT], + ("binding_type", "scope", "source"), + [ + (WorkflowAgentBindingType.ROSTER_AGENT, AgentScope.ROSTER, AgentSource.ROSTER), + (WorkflowAgentBindingType.INLINE_AGENT, AgentScope.WORKFLOW_ONLY, AgentSource.WORKFLOW), + ], ) +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) def test_binding_resolver_uses_pinned_snapshot_for_existing_node_execution( monkeypatch: pytest.MonkeyPatch, + sqlite_engine: Engine, + sqlite_session: Session, binding_type: WorkflowAgentBindingType, -): - binding = _binding() - binding.binding_type = binding_type - agent = _agent() - agent.active_config_snapshot_id = "active-snapshot" - snapshot = _snapshot() - snapshot.id = "pinned-snapshot" - fake_session = FakeSession([binding, agent, snapshot]) - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: fake_session, + scope: AgentScope, + source: AgentSource, +) -> None: + ids = _resolve_ids() + agent = _agent( + tenant_id=ids["tenant_id"], + scope=scope, + source=source, + active_config_snapshot_id=str(uuid4()), ) + sqlite_session.add(agent) + sqlite_session.flush() + pinned_snapshot = _snapshot(tenant_id=ids["tenant_id"], agent_id=agent.id) + sqlite_session.add(pinned_snapshot) + sqlite_session.flush() + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=str(uuid4()), + binding_type=binding_type, + ) + sqlite_session.add(binding) + sqlite_session.commit() + scalar_statements = _bind_factory(monkeypatch, sqlite_engine) bundle = WorkflowAgentBindingResolver().resolve( - **_resolve(), - binding_id="binding-1", - snapshot_id="pinned-snapshot", + **ids, + binding_id=binding.id, + snapshot_id=pinned_snapshot.id, ) - assert bundle.snapshot.id == "pinned-snapshot" - assert "binding-1" in fake_session.scalar_statements[0].compile().params.values() - assert "pinned-snapshot" in fake_session.scalar_statements[-1].compile().params.values() + assert bundle.snapshot.id == pinned_snapshot.id + assert binding.id in scalar_statements[0].compile().params.values() + assert pinned_snapshot.id in scalar_statements[-1].compile().params.values() -def test_binding_resolver_does_not_fallback_from_an_explicit_empty_snapshot(monkeypatch: pytest.MonkeyPatch): - binding = _binding() - binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT - agent = _agent() - agent.active_config_snapshot_id = "active-snapshot" - fake_session = FakeSession([binding, agent, None]) - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: fake_session, +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) +def test_binding_resolver_does_not_fallback_from_an_explicit_empty_snapshot( + monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session +) -> None: + ids = _resolve_ids() + agent = _agent( + tenant_id=ids["tenant_id"], + scope=AgentScope.ROSTER, + source=AgentSource.ROSTER, ) + sqlite_session.add(agent) + sqlite_session.flush() + active_snapshot = _snapshot(tenant_id=ids["tenant_id"], agent_id=agent.id) + sqlite_session.add(active_snapshot) + sqlite_session.flush() + agent.active_config_snapshot_id = active_snapshot.id + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=str(uuid4()), + binding_type=WorkflowAgentBindingType.ROSTER_AGENT, + ) + sqlite_session.add(binding) + sqlite_session.commit() + _bind_factory(monkeypatch, sqlite_engine) with pytest.raises(WorkflowAgentBindingError) as exc_info: - WorkflowAgentBindingResolver().resolve(**_resolve(), binding_id="binding-1", snapshot_id="") + WorkflowAgentBindingResolver().resolve(**ids, binding_id=binding.id, snapshot_id="") assert exc_info.value.error_code == "agent_config_snapshot_not_found" - assert "" in fake_session.scalar_statements[-1].compile().params.values() @pytest.mark.parametrize( @@ -177,7 +272,7 @@ def test_binding_resolver_rejects_half_pinned_generation( ) -> None: with pytest.raises(WorkflowAgentBindingError) as exc_info: WorkflowAgentBindingResolver().resolve( - **_resolve(), + **_resolve_ids(), binding_id=binding_id, snapshot_id=snapshot_id, ) @@ -185,120 +280,153 @@ def test_binding_resolver_rejects_half_pinned_generation( assert exc_info.value.error_code == "agent_binding_generation_invalid" -def test_binding_resolver_rejects_unpublished_roster_agent(monkeypatch: pytest.MonkeyPatch): - binding = _binding() - binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: FakeSession([binding, None]), +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) +def test_binding_resolver_rejects_unpublished_roster_agent( + monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session +) -> None: + ids = _resolve_ids() + snapshot_id = str(uuid4()) + agent = _agent( + tenant_id=ids["tenant_id"], + scope=AgentScope.ROSTER, + source=AgentSource.IMPORTED, + app_id=str(uuid4()), + active_config_snapshot_id=snapshot_id, + active_config_is_published=False, ) + sqlite_session.add(agent) + sqlite_session.flush() + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=snapshot_id, + binding_type=WorkflowAgentBindingType.ROSTER_AGENT, + ) + sqlite_session.add(binding) + sqlite_session.commit() + _bind_factory(monkeypatch, sqlite_engine) with pytest.raises(WorkflowAgentBindingError) as exc_info: - WorkflowAgentBindingResolver().resolve(**_resolve()) + WorkflowAgentBindingResolver().resolve(**ids) assert exc_info.value.error_code == "agent_not_available" assert "not been published" in str(exc_info.value) -@pytest.mark.parametrize( - "sqlite_session", - [(Agent, AgentConfigSnapshot, AgentConfigRevision, WorkflowAgentNodeBinding)], - indirect=True, -) +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) def test_binding_resolver_requires_publish_provenance_for_active_roster_snapshot( monkeypatch: pytest.MonkeyPatch, + sqlite_engine: Engine, sqlite_session: Session, ) -> None: - binding = _binding() - binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT - binding.workflow_version = "draft" - agent = Agent( - id="agent-1", - tenant_id="tenant-1", - name="Imported Agent", + ids = _resolve_ids() + agent = _agent( + tenant_id=ids["tenant_id"], scope=AgentScope.ROSTER, source=AgentSource.IMPORTED, - app_id="agent-app-1", - status=AgentStatus.ACTIVE, - active_config_snapshot_id="snapshot-1", - active_config_has_model=True, - # Dirty draft state must not hide a snapshot after it has publish provenance. + app_id=str(uuid4()), active_config_is_published=False, ) + sqlite_session.add(agent) + sqlite_session.flush() + snapshot = _snapshot(tenant_id=ids["tenant_id"], agent_id=agent.id) + sqlite_session.add(snapshot) + sqlite_session.flush() + agent.active_config_snapshot_id = snapshot.id + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=snapshot.id, + binding_type=WorkflowAgentBindingType.ROSTER_AGENT, + ) sqlite_session.add_all( [ binding, - agent, - _snapshot(), AgentConfigRevision( - id="revision-import", - tenant_id="tenant-1", - agent_id="agent-1", - current_snapshot_id="snapshot-1", + tenant_id=ids["tenant_id"], + agent_id=agent.id, + current_snapshot_id=snapshot.id, revision=1, operation=AgentConfigRevisionOperation.IMPORT_PACKAGE, ), ] ) sqlite_session.commit() - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: sqlite_session, - ) + _bind_factory(monkeypatch, sqlite_engine) with pytest.raises(WorkflowAgentBindingError) as exc_info: - WorkflowAgentBindingResolver().resolve(**_resolve()) + WorkflowAgentBindingResolver().resolve(**ids) assert exc_info.value.error_code == "agent_not_available" sqlite_session.add( AgentConfigRevision( - id="revision-publish", - tenant_id="tenant-1", - agent_id="agent-1", - current_snapshot_id="snapshot-1", + tenant_id=ids["tenant_id"], + agent_id=agent.id, + current_snapshot_id=snapshot.id, revision=2, operation=AgentConfigRevisionOperation.PUBLISH_DRAFT, ) ) sqlite_session.commit() - bundle = WorkflowAgentBindingResolver().resolve(**_resolve()) + bundle = WorkflowAgentBindingResolver().resolve(**ids) assert bundle.agent.id == agent.id - assert bundle.snapshot.id == "snapshot-1" + assert bundle.snapshot.id == snapshot.id -def test_binding_resolver_raises_when_binding_missing(monkeypatch: pytest.MonkeyPatch): - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: FakeSession([None]), - ) +def test_binding_resolver_raises_when_binding_missing(monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None: + _bind_factory(monkeypatch, sqlite_engine) with pytest.raises(WorkflowAgentBindingError) as exc_info: - WorkflowAgentBindingResolver().resolve(**_resolve()) + WorkflowAgentBindingResolver().resolve(**_resolve_ids()) assert exc_info.value.error_code == "agent_binding_not_found" -def test_binding_resolver_raises_when_agent_archived(monkeypatch: pytest.MonkeyPatch): - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: FakeSession([_binding(), _agent(status=AgentStatus.ARCHIVED)]), +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) +def test_binding_resolver_raises_when_agent_archived( + monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session +) -> None: + ids = _resolve_ids() + agent = _agent(tenant_id=ids["tenant_id"], status=AgentStatus.ARCHIVED) + sqlite_session.add(agent) + sqlite_session.flush() + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=str(uuid4()), + binding_type=WorkflowAgentBindingType.INLINE_AGENT, ) + sqlite_session.add(binding) + sqlite_session.commit() + _bind_factory(monkeypatch, sqlite_engine) with pytest.raises(WorkflowAgentBindingError) as exc_info: - WorkflowAgentBindingResolver().resolve(**_resolve()) + WorkflowAgentBindingResolver().resolve(**ids) assert exc_info.value.error_code == "agent_not_available" -def test_binding_resolver_raises_when_snapshot_missing(monkeypatch: pytest.MonkeyPatch): - monkeypatch.setattr( - "core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session", - lambda: FakeSession([_binding(), _agent(), None]), +@pytest.mark.parametrize("sqlite_session", [RESOLVER_MODELS], indirect=True) +def test_binding_resolver_raises_when_snapshot_missing( + monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session +) -> None: + ids = _resolve_ids() + agent = _agent(tenant_id=ids["tenant_id"]) + sqlite_session.add(agent) + sqlite_session.flush() + binding = _binding( + ids=ids, + agent_id=agent.id, + snapshot_id=str(uuid4()), + binding_type=WorkflowAgentBindingType.INLINE_AGENT, ) + sqlite_session.add(binding) + sqlite_session.commit() + _bind_factory(monkeypatch, sqlite_engine) with pytest.raises(WorkflowAgentBindingError) as exc_info: - WorkflowAgentBindingResolver().resolve(**_resolve()) + WorkflowAgentBindingResolver().resolve(**ids) assert exc_info.value.error_code == "agent_config_snapshot_not_found"