mirror of
https://github.com/langgenius/dify.git
synced 2026-08-01 09:50:50 +08:00
test: use SQLite sessions in core workflow (#39086)
This commit is contained in:
Asuka Minato
GitHub
parent
917207414b
commit
f6bbb94112
@ -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"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user