tiennm99/DocsGPT
1
0 Fork 0
mirror of https://github.com/tiennm99/DocsGPT.git synced 2026-10-15 08:16:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2690 from arc53/workflow-export

Browse Source 
Workflow export
This commit is contained in:
Alex authored and GitHub committed 2026-08-23 11:46:26 +01:00
2 parents 52279d27b4 b47bfa8a37
commit 5c55d2610b
23 files changed
+2625 -3780
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api/test_agent_portability.py
+344
View File
@@ -286,10 +286,29 @@ def test_parse_rejects_bad_documents():
parse_agent_yaml("kind: Agent\napiVersion: other/v1\nspec:\n name: x\n")
with pytest.raises(AgentImportError):
parse_agent_yaml("kind: Agent\napiVersion: docsgpt.arc53.com/v1\nspec: {}\n")
# A workflow agent without a graph parses (drafts may have none)…
doc = parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n"
)
assert doc["spec"]["agent_type"] == "workflow"
# …but a present workflow block must be well-formed.
with pytest.raises(AgentImportError):
parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n workflow: nonsense\n"
)
with pytest.raises(AgentImportError):
parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n"
" workflow:\n nodes: {}\n edges: []\n"
)
with pytest.raises(AgentImportError):
parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n"
" workflow:\n nodes:\n - id: n1\n edges: []\n"
)
@@ -478,3 +497,328 @@ def test_tool_skipped_when_encryption_fails(pg_conn, monkeypatch):
agent = AgentsRepository(pg_conn).get(result["agent_id"], user)
assert agent["tools"] == []
assert any("encryption failed" in w for w in result["warnings"])
# ---------------------------------------------------------------------------
# Action customizations (per-parameter fixed values / filled_by_llm flags)
# ---------------------------------------------------------------------------
_SEND_MESSAGE_META = {
"name": "telegram_send_message",
"description": "Send a message",
"parameters": {
"type": "object",
"properties": {
"chat_id": {"type": "string", "description": "Chat id"},
"text": {"type": "string", "description": "Message text"},
},
"required": ["chat_id", "text"],
},
}
_GET_UPDATES_META = {
"name": "telegram_get_updates",
"description": "Poll updates",
"parameters": {"type": "object", "properties": {}, "required": []},
}
def _fake_telegram(monkeypatch):
import copy as _copy
fake_tool = Mock()
fake_tool.get_config_requirements.return_value = {
"token": {"secret": True, "required": True, "label": "Bot token"}
}
fake_tool.get_actions_metadata.side_effect = lambda: _copy.deepcopy(
[_SEND_MESSAGE_META, _GET_UPDATES_META]
)
monkeypatch.setattr(
"application.api.user.agents.portability._tool_instance",
lambda tool_type: fake_tool,
)
return fake_tool
def _stored_telegram_actions():
"""The shape ``update_tool_actions`` persists after the user fixes chat_id
and disables the updates action."""
return [
{
"name": "telegram_send_message",
"description": "Send a message",
"active": True,
"parameters": {
"type": "object",
"properties": {
"chat_id": {
"type": "string",
"description": "Chat id",
"filled_by_llm": False,
"value": "123456789",
},
"text": {
"type": "string",
"description": "Message text",
"filled_by_llm": True,
"value": "",
},
},
"required": ["chat_id", "text"],
},
},
{
"name": "telegram_get_updates",
"description": "Poll updates",
"active": False,
"parameters": {"type": "object", "properties": {}, "required": []},
},
]
def test_action_customizations_round_trip(pg_conn, monkeypatch):
user = "u_actions"
_fake_telegram(monkeypatch)
tool = UserToolsRepository(pg_conn).create(
user,
"telegram",
config={"encrypted_credentials": "BLOB"},
custom_name="My TG",
display_name="Telegram",
description="tg",
config_requirements={"token": {"secret": True, "required": True}},
actions=_stored_telegram_actions(),
)
agent = AgentsRepository(pg_conn).create(
user, "TG Bot", "published", description="d", tools=[str(tool["id"])]
)
export = serialize_agent(pg_conn, agent, user)
overrides = export["spec"]["tools"][0]["actions"]
# Only deviations from factory defaults travel: the fixed chat_id and the
# disabled action — the default-state ``text`` param does not.
assert overrides == [
{
"name": "telegram_send_message",
"parameters": {"chat_id": {"filled_by_llm": False, "value": "123456789"}},
},
{"name": "telegram_get_updates", "active": False},
]
# Import as a fresh user (tool re-created with the supplied secret).
doc = parse_agent_yaml(agent_to_yaml(export))
importer = "u_actions_2"
result = apply_import(
pg_conn,
importer,
doc,
resolution={"tools": {"tool-0": {"decision": "create", "secrets": {"token": "tok"}}}},
)
imported_agent = AgentsRepository(pg_conn).get(result["agent_id"], importer)
row = UserToolsRepository(pg_conn).get_any(imported_agent["tools"][0], importer)
by_name = {a["name"]: a for a in row["actions"]}
chat_id = by_name["telegram_send_message"]["parameters"]["properties"]["chat_id"]
text = by_name["telegram_send_message"]["parameters"]["properties"]["text"]
assert chat_id["filled_by_llm"] is False
assert chat_id["value"] == "123456789"
assert text["filled_by_llm"] is True and text["value"] == ""
assert by_name["telegram_get_updates"]["active"] is False
assert by_name["telegram_send_message"]["active"] is True
def test_action_header_values_never_exported(pg_conn):
user = "u_actions_hdr"
tool = UserToolsRepository(pg_conn).create(
user,
"custom_http",
config={},
custom_name="HTTP",
display_name="HTTP",
description="",
config_requirements={},
actions=[
{
"name": "call",
"active": True,
"headers": {
"properties": {
"Authorization": {
"type": "string",
"filled_by_llm": False,
"value": "Bearer sk-live-XYZ",
}
}
},
}
],
)
agent = AgentsRepository(pg_conn).create(
user, "Hdr Bot", "published", description="d", tools=[str(tool["id"])]
)
export = serialize_agent(pg_conn, agent, user)
text = agent_to_yaml(export)
assert "Bearer sk-live-XYZ" not in text
# The flag still travels so the imported action keeps its shape.
override = export["spec"]["tools"][0]["actions"][0]
assert override["headers"]["Authorization"] == {"filled_by_llm": False}
def test_action_override_for_removed_action_warns(pg_conn, monkeypatch):
user = "u_actions_gone"
_fake_telegram(monkeypatch)
doc = _doc(
name="Drifted Bot",
_slug="drifted",
tools=[
{
"type": "telegram",
"name": "My TG",
"config": {},
"requires_secrets": ["token"],
"actions": [
{"name": "telegram_removed_action", "active": False},
{
"name": "telegram_send_message",
"parameters": {
"gone_param": {"filled_by_llm": False, "value": "x"},
"chat_id": {"filled_by_llm": False, "value": "42"},
},
},
],
}
],
)
result = apply_import(
pg_conn,
user,
doc,
resolution={"tools": {"tool-0": {"secrets": {"token": "tok"}}}},
)
agent = AgentsRepository(pg_conn).get(result["agent_id"], user)
row = UserToolsRepository(pg_conn).get_any(agent["tools"][0], user)
by_name = {a["name"]: a for a in row["actions"]}
# The surviving customization applied; the drifted ones warned and skipped.
chat_id = by_name["telegram_send_message"]["parameters"]["properties"]["chat_id"]
assert chat_id["filled_by_llm"] is False and chat_id["value"] == "42"
assert any("telegram_removed_action" in w for w in result["warnings"])
assert any("gone_param" in w for w in result["warnings"])
def _fake_api_tool(monkeypatch):
"""api_tool's live surface: no declared config, no column actions."""
fake_tool = Mock()
fake_tool.get_config_requirements.return_value = {}
fake_tool.get_actions_metadata.return_value = []
monkeypatch.setattr(
"application.api.user.agents.portability._tool_instance",
lambda tool_type: fake_tool,
)
# The sanitizer SSRF-gates each action URL; DNS isn't available in tests.
monkeypatch.setattr(
"application.api.user.agents.portability.validate_url", lambda url: url
)
return fake_tool
def _stored_api_tool_config():
"""``config["actions"]`` as ToolConfig persists it (dict keyed by name)."""
return {
"actions": {
"get_pet": {
"name": "get_pet",
"url": "https://api.example.com/pets/{petId}?api_key=QSSECRET",
"method": "GET",
"active": True,
"headers": {
"properties": {
"Authorization": {
"type": "string",
"filled_by_llm": False,
"value": "Bearer sk-live-XYZ",
}
}
},
"query_params": {
"properties": {
"petId": {
"type": "string",
"filled_by_llm": True,
"value": "",
}
}
},
}
}
}
def test_api_tool_actions_round_trip_redacted(pg_conn, monkeypatch):
"""api_tool actions live in config, not the actions column — they must
travel (redacted) or the imported tool is dead (executor resolves calls
exclusively through ``config["actions"]``)."""
user = "u_api_tool"
_fake_api_tool(monkeypatch)
tool = UserToolsRepository(pg_conn).create(
user,
"api_tool",
config=_stored_api_tool_config(),
custom_name="Petstore",
display_name="Petstore",
description="pets",
config_requirements={},
actions=[],
)
agent = AgentsRepository(pg_conn).create(
user, "Pet Bot", "published", description="d", tools=[str(tool["id"])]
)
export = serialize_agent(pg_conn, agent, user)
text = agent_to_yaml(export)
assert "sk-live-XYZ" not in text and "QSSECRET" not in text
exported = export["spec"]["tools"][0]["config"]["actions"]["get_pet"]
# Shape travels; the URL loses its query string; the pinned header
# value is blanked while its flag and schema survive.
assert exported["url"] == "https://api.example.com/pets/{petId}"
auth = exported["headers"]["properties"]["Authorization"]
assert auth["filled_by_llm"] is False and auth["value"] == ""
assert exported["query_params"]["properties"]["petId"]["filled_by_llm"] is True
doc = parse_agent_yaml(text)
importer = "u_api_tool_2"
result = apply_import(
pg_conn,
importer,
doc,
resolution={"tools": {"tool-0": {"decision": "create"}}},
)
imported_agent = AgentsRepository(pg_conn).get(result["agent_id"], importer)
row = UserToolsRepository(pg_conn).get_any(imported_agent["tools"][0], importer)
action = row["config"]["actions"]["get_pet"]
assert action["url"] == "https://api.example.com/pets/{petId}"
assert action["method"] == "GET"
# The blanked pinned header is called out — the request would silently
# omit it until the importer sets a value in the tool's settings.
assert any("Authorization" in w for w in result["warnings"])
def test_api_tool_without_actions_imports_with_warning(pg_conn, monkeypatch):
"""Older exports carried no actions for api_tool; the tool is created but
the dead state is surfaced instead of silent."""
user = "u_api_tool_empty"
_fake_api_tool(monkeypatch)
doc = _doc(
name="Dead API Bot",
_slug="dead-api",
tools=[{"type": "api_tool", "name": "Legacy", "config": {}}],
)
result = apply_import(
pg_conn,
user,
doc,
resolution={"tools": {"tool-0": {"decision": "create"}}},
)
assert any("without any actions" in w for w in result["warnings"])
agent = AgentsRepository(pg_conn).get(result["agent_id"], user)
row = UserToolsRepository(pg_conn).get_any(agent["tools"][0], user)
assert row["config"]["actions"] == {}
tests/api/test_workflow_portability.py
+535
View File
@@ -0,0 +1,535 @@
"""Workflow-agent YAML export/import (application.api.user.agents.portability).
Real SQL against the ephemeral ``pg_conn`` fixture, like
``test_agent_portability.py``. Covers the graph block: reference rewriting
(tools → ``tool-N`` keys, sources → names, custom models → display names),
secret redaction inside node configs, graph validation at apply, and
idempotent re-import as a new graph version of the same workflow.
"""
from __future__ import annotations
import pytest
from application.agents.default_tools import default_tool_id
from application.api.user.agents.portability import (
AgentImportError,
agent_to_yaml,
apply_import,
parse_agent_yaml,
plan_import,
serialize_agent,
)
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.sources import SourcesRepository
from application.storage.db.repositories.user_custom_models import (
UserCustomModelsRepository,
)
from application.storage.db.repositories.user_tools import UserToolsRepository
from application.storage.db.repositories.workflow_edges import WorkflowEdgesRepository
from application.storage.db.repositories.workflow_nodes import WorkflowNodesRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
pytestmark = pytest.mark.integration
def _graph(conn, workflow_row):
version = int(workflow_row.get("current_graph_version") or 1)
nodes = WorkflowNodesRepository(conn).find_by_version(str(workflow_row["id"]), version)
edges = WorkflowEdgesRepository(conn).find_by_version(str(workflow_row["id"]), version)
return nodes, edges
def _seed_workflow_agent(conn, user, *, tool_id=None, source_id=None, model_id=None):
"""Workflow agent with a start → agent → end graph referencing real resources."""
agent_cfg = {
"agent_type": "classic",
"system_prompt": "Answer briefly.",
"tools": [t for t in [tool_id, default_tool_id("read_document")] if t],
"sources": [s for s in [source_id] if s],
"model_id": model_id,
}
wf = WorkflowsRepository(conn).create(user, "Pipeline", description="wd")
wf_id = str(wf["id"])
nodes = WorkflowNodesRepository(conn).bulk_create(
wf_id,
1,
[
{"node_id": "start-1", "node_type": "start", "title": "Start"},
{"node_id": "agent-1", "node_type": "agent", "title": "Answer", "config": agent_cfg},
{"node_id": "end-1", "node_type": "end", "title": "End"},
],
)
by_str = {n["node_id"]: n["id"] for n in nodes}
WorkflowEdgesRepository(conn).bulk_create(
wf_id,
1,
[
{"edge_id": "e1", "from_node_id": by_str["start-1"], "to_node_id": by_str["agent-1"]},
{"edge_id": "e2", "from_node_id": by_str["agent-1"], "to_node_id": by_str["end-1"]},
],
)
agent = AgentsRepository(conn).create(
user,
"Flow Bot",
"published",
description="wf agent",
agent_type="workflow",
workflow_id=wf_id,
)
return agent, wf
def _seed_tool(conn, user):
return UserToolsRepository(conn).create(
user,
"brave",
config={"encrypted_credentials": "SECRETBLOB", "region": "US"},
custom_name="My Brave",
display_name="Brave Search",
description="search",
config_requirements={"token": {"secret": True, "required": True}, "region": {}},
actions=[],
)
def test_export_rewrites_node_refs_and_redacts_secrets(pg_conn):
user = "u_wf_export"
tool = _seed_tool(pg_conn, user)
src = SourcesRepository(pg_conn).create("KB", user_id=user, type="file")
model = UserCustomModelsRepository(pg_conn).create(
user, "llama-3", "My Llama", "https://models.example.com/v1", "sk-plain"
)
agent, wf = _seed_workflow_agent(
pg_conn,
user,
tool_id=str(tool["id"]),
source_id=str(src["id"]),
model_id=str(model["id"]),
)
export = serialize_agent(pg_conn, agent, user)
text = agent_to_yaml(export)
spec = export["spec"]
assert "SECRETBLOB" not in text and "sk-plain" not in text
# Raw ids never appear inside the graph block (the sections may carry
# ``ref`` hints, same as classic export).
workflow_text = agent_to_yaml(spec["workflow"])
assert str(tool["id"]) not in workflow_text
assert str(src["id"]) not in workflow_text
assert str(model["id"]) not in workflow_text
assert spec["workflow"]["name"] == "Pipeline"
assert [n["id"] for n in spec["workflow"]["nodes"]] == ["agent-1", "end-1", "start-1"]
agent_node = next(n for n in spec["workflow"]["nodes"] if n["id"] == "agent-1")
assert agent_node["config"]["tools"] == ["tool-0", "tool-1"]
assert agent_node["config"]["sources"] == ["KB"]
assert agent_node["config"]["model_id"] == "My Llama"
# The collected sections use the classic serializers' portable identities.
assert spec["tools"][0]["type"] == "brave"
assert spec["tools"][0]["requires_secrets"] == ["token"]
assert spec["tools"][1] == {
"type": "read_document",
"builtin": True,
"ref": default_tool_id("read_document"),
}
assert spec["sources"][0]["name"] == "KB"
assert spec["model"]["available"][0]["display_name"] == "My Llama"
assert spec["prompt"] == "default"
edges = spec["workflow"]["edges"]
assert {(e["source"], e["target"]) for e in edges} == {
("start-1", "agent-1"),
("agent-1", "end-1"),
}
def test_export_without_workflow_emits_null_block(pg_conn):
user = "u_wf_export2"
agent = AgentsRepository(pg_conn).create(
user, "Draft Flow", "draft", agent_type="workflow"
)
export = serialize_agent(pg_conn, agent, user)
assert export["spec"]["workflow"] is None
assert export["spec"]["tools"] == []
assert export["spec"]["sources"] == []
def test_round_trip_same_user_reuses_everything(pg_conn):
user = "u_wf_round"
tool = _seed_tool(pg_conn, user)
src = SourcesRepository(pg_conn).create("KB", user_id=user, type="file")
model = UserCustomModelsRepository(pg_conn).create(
user, "llama-3", "My Llama", "https://models.example.com/v1", "sk-plain"
)
agent, wf = _seed_workflow_agent(
pg_conn,
user,
tool_id=str(tool["id"]),
source_id=str(src["id"]),
model_id=str(model["id"]),
)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
result = apply_import(pg_conn, user, doc)
assert result["action"] == "updated"
assert result["agent_id"] == str(agent["id"])
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
# Same workflow row, bumped to a new graph version.
assert str(updated["workflow_id"]) == str(wf["id"])
wf_row = WorkflowsRepository(pg_conn).get(str(wf["id"]), user)
assert int(wf_row["current_graph_version"]) == 2
nodes, edges = _graph(pg_conn, wf_row)
assert len(nodes) == 3 and len(edges) == 2
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
# References resolved back to the same owned resources — no duplicates.
assert set(agent_node["config"]["tools"]) == {
str(tool["id"]), default_tool_id("read_document"),
}
assert agent_node["config"]["sources"] == [str(src["id"])]
assert agent_node["config"]["model_id"] == str(model["id"])
assert len(UserToolsRepository(pg_conn).list_for_user(user)) == 1
# Old graph versions are pruned.
assert WorkflowNodesRepository(pg_conn).find_by_version(str(wf["id"]), 1) == []
def test_import_into_fresh_user_creates_workflow(pg_conn):
owner = "u_wf_owner"
tool = _seed_tool(pg_conn, owner)
src = SourcesRepository(pg_conn).create("KB", user_id=owner, type="file")
agent, _ = _seed_workflow_agent(
pg_conn, owner, tool_id=str(tool["id"]), source_id=str(src["id"])
)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, owner)))
importer = "u_wf_importer"
result = apply_import(
pg_conn,
importer,
doc,
{"tools": {"tool-0": {"decision": "create", "secrets": {"token": "tok"}}}},
)
assert result["action"] == "created"
created = AgentsRepository(pg_conn).get(result["agent_id"], importer)
assert created["status"] == "draft"
assert created["workflow_id"] is not None
wf_row = WorkflowsRepository(pg_conn).get(str(created["workflow_id"]), importer)
assert wf_row is not None and wf_row["name"] == "Pipeline"
nodes, edges = _graph(pg_conn, wf_row)
assert len(nodes) == 3 and len(edges) == 2
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
# The tool was created for the importer; the missing source was dropped.
my_tool = UserToolsRepository(pg_conn).list_for_user(importer)[0]
assert set(agent_node["config"]["tools"]) == {
str(my_tool["id"]), default_tool_id("read_document"),
}
assert agent_node["config"]["sources"] == []
assert any("Source 'KB' not found" in w for w in result["warnings"])
assert any("not resolved" in w for w in result["warnings"])
def test_plan_reports_workflow_block(pg_conn):
user = "u_wf_plan"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
plan = plan_import(pg_conn, user, doc)
assert plan["workflow"] == {"nodes": 3, "edges": 2, "action": "update"}
other_plan = plan_import(pg_conn, "someone_else", doc)
assert other_plan["workflow"]["action"] == "create"
def test_import_rejects_invalid_graph(pg_conn):
user = "u_wf_invalid"
doc = {
"apiVersion": "docsgpt.arc53.com/v1",
"kind": "Agent",
"metadata": {"slug": "broken-flow"},
"spec": {
"name": "Broken Flow",
"agent_type": "workflow",
"workflow": {
"name": "Broken",
# No start/end nodes — the API's validation gate must reject it.
"nodes": [{"id": "a", "type": "agent", "config": {}}],
"edges": [],
},
},
}
with pytest.raises(AgentImportError, match="Workflow validation failed"):
apply_import(pg_conn, user, doc)
# Nothing half-imported.
assert AgentsRepository(pg_conn).list_for_user(user) == []
assert WorkflowsRepository(pg_conn).list_for_user(user) == []
def test_import_drops_foreign_raw_ids_from_node_config(pg_conn):
"""A hand-edited file carrying someone else's raw ids must not link them."""
victim = "u_wf_victim"
victim_tool = _seed_tool(pg_conn, victim)
victim_src = SourcesRepository(pg_conn).create("Secret KB", user_id=victim, type="file")
doc = {
"apiVersion": "docsgpt.arc53.com/v1",
"kind": "Agent",
"metadata": {"slug": "sneaky-flow"},
"spec": {
"name": "Sneaky Flow",
"agent_type": "workflow",
"workflow": {
"name": "Sneaky",
"nodes": [
{"id": "s", "type": "start"},
{
"id": "a",
"type": "agent",
"config": {
"tools": [str(victim_tool["id"])],
"sources": [str(victim_src["id"])],
},
},
{"id": "e", "type": "end"},
],
"edges": [
{"id": "e1", "source": "s", "target": "a"},
{"id": "e2", "source": "a", "target": "e"},
],
},
},
}
result = apply_import(pg_conn, "u_wf_attacker", doc)
created = AgentsRepository(pg_conn).get(result["agent_id"], "u_wf_attacker")
wf_row = WorkflowsRepository(pg_conn).get(str(created["workflow_id"]), "u_wf_attacker")
nodes, _ = _graph(pg_conn, wf_row)
agent_node = next(n for n in nodes if n["node_id"] == "a")
assert agent_node["config"]["tools"] == []
assert agent_node["config"]["sources"] == []
assert len(result["warnings"]) == 2
def test_import_route_rejects_invalid_graph_as_client_error(pg_conn):
"""Apply-time graph rejection is a 400 carrying the validation detail.
Regression: an ``AgentImportError`` raised while applying (the file parses,
but the graph fails the workflow API's validation gate) used to fall into
the route's generic handler and surface as a 500 with no explanation.
"""
from contextlib import contextmanager
from unittest.mock import patch
from flask import Flask, request
from application.api.user.agents.portability import ImportAgent
@contextmanager
def _conn():
yield pg_conn
yaml_text = (
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n"
" name: Broken Flow\n"
" agent_type: workflow\n"
" workflow:\n"
" name: Broken\n"
" nodes:\n"
" - id: a\n"
" type: agent\n"
" edges: []\n"
)
app = Flask(__name__)
with patch(
"application.api.user.agents.portability.db_session", _conn
), app.test_request_context(
"/api/import_agent", method="POST", json={"yaml": yaml_text}
):
request.decoded_token = {"sub": "u_wf_route"}
response = ImportAgent().post()
assert response.status_code == 400
body = response.get_json()
assert body["success"] is False
assert "Workflow validation failed" in body["message"]
assert AgentsRepository(pg_conn).list_for_user("u_wf_route") == []
def test_import_without_workflow_keeps_published_agents_graph(pg_conn):
user = "u_wf_keep"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
del doc["spec"]["workflow"]
result = apply_import(pg_conn, user, doc)
assert result["action"] == "updated"
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert str(updated["workflow_id"]) == str(wf["id"])
def test_import_with_null_workflow_keeps_published_but_clears_draft(pg_conn):
user = "u_wf_null"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
doc["spec"]["workflow"] = None
# Published agent: explicit null must not break it — kept with a warning.
result = apply_import(pg_conn, user, doc)
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert str(updated["workflow_id"]) == str(wf["id"])
assert any("kept its existing" in w for w in result["warnings"])
# Draft agent: explicit null clears the link AND reaps the now-orphaned
# workflow row — there is no workflow-list API independent of agents, so
# an unlinked row would be unreachable forever.
AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"})
apply_import(pg_conn, user, doc)
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert updated["workflow_id"] is None
assert WorkflowsRepository(pg_conn).get(str(wf["id"]), user) is None
def test_null_workflow_keeps_row_still_referenced_by_another_agent(pg_conn):
user = "u_wf_null_shared"
agent, wf = _seed_workflow_agent(pg_conn, user)
# A second draft agent points at the same workflow.
AgentsRepository(pg_conn).create(
user, "Sibling", "draft", agent_type="workflow", workflow_id=str(wf["id"])
)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
doc["spec"]["workflow"] = None
AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"})
apply_import(pg_conn, user, doc)
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert updated["workflow_id"] is None
# Still referenced — must not be deleted.
assert WorkflowsRepository(pg_conn).get(str(wf["id"]), user) is not None
def test_publish_guard_rejects_workflow_agent_with_no_graph(pg_conn):
"""A file that makes a published agent a graph-less workflow agent is refused.
The create and update routes both reject that state; import must not
install it through the back door.
"""
user = "u_wf_publish_guard"
agent = AgentsRepository(pg_conn).create(
user, "Probe Bot", "published", agent_type="classic", slug="probe-bot"
)
doc = {
"apiVersion": "docsgpt.arc53.com/v1",
"kind": "Agent",
"metadata": {"id": str(agent["id"]), "slug": "probe-bot"},
"spec": {"name": "Probe Bot", "agent_type": "workflow"},
}
# Key omitted entirely.
with pytest.raises(AgentImportError, match="published workflow agent needs a workflow"):
apply_import(pg_conn, user, doc)
# Explicit null, with no prior workflow to fall back on.
doc["spec"]["workflow"] = None
with pytest.raises(AgentImportError, match="published workflow agent needs a workflow"):
apply_import(pg_conn, user, doc)
unchanged = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert unchanged["agent_type"] == "classic"
assert unchanged["status"] == "published"
def test_plan_distinguishes_absent_workflow_from_explicit_null(pg_conn):
"""Omitting the key is a no-op; an explicit null destroys the graph.
The two used to produce an identical plan, so the modal showed nothing
either way while the outcomes differed completely.
"""
user = "u_wf_plan_null"
agent, wf = _seed_workflow_agent(pg_conn, user)
AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"})
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
absent = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
del absent["spec"]["workflow"]
assert plan_import(pg_conn, user, absent)["workflow"] is None
doc["spec"]["workflow"] = None
assert plan_import(pg_conn, user, doc)["workflow"] == {
"action": "delete",
"nodes": 3,
"edges": 2,
}
def test_plan_reports_no_removal_when_workflow_survives(pg_conn):
"""No delete block when the row lives on — published, or still referenced."""
user = "u_wf_plan_keep"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
doc["spec"]["workflow"] = None
# Published: _apply_workflow keeps the existing graph.
assert plan_import(pg_conn, user, doc)["workflow"] is None
# Draft, but a sibling agent still references the row.
AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"})
AgentsRepository(pg_conn).create(
user, "Sibling", "draft", agent_type="workflow", workflow_id=str(wf["id"])
)
assert plan_import(pg_conn, user, doc)["workflow"] is None
def test_null_workflow_deletion_is_warned_about(pg_conn):
"""The destructive reap must not be silent."""
user = "u_wf_null_warn"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
doc["spec"]["workflow"] = None
AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"})
result = apply_import(pg_conn, user, doc)
assert WorkflowsRepository(pg_conn).get(str(wf["id"]), user) is None
assert any("was deleted" in w and "run history" in w for w in result["warnings"])
def test_duplicate_source_names_warn_and_dedupe(pg_conn):
"""Two sources sharing a name collapse onto the oldest — say so."""
user = "u_wf_dup_src"
repo = SourcesRepository(pg_conn)
older = repo.create("KB", user_id=user, type="file")
newer = repo.create("KB", user_id=user, type="file")
agent, _wf = _seed_workflow_agent(pg_conn, user, source_id=str(older["id"]))
# Point the node at both same-named sources.
version = 1
nodes_repo = WorkflowNodesRepository(pg_conn)
node = [n for n in nodes_repo.find_by_version(str(agent["workflow_id"]), version)
if n["node_type"] == "agent"][0]
config = dict(node["config"])
config["sources"] = [str(older["id"]), str(newer["id"])]
nodes_repo.bulk_create(
str(agent["workflow_id"]),
version,
[{
"node_id": node["node_id"],
"node_type": node["node_type"],
"title": node.get("title"),
"position": node.get("position"),
"config": config,
}],
)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
assert [s["name"] for s in doc["spec"]["sources"]] == ["KB", "KB"]
result = apply_import(pg_conn, user, doc)
assert any("share the name 'KB'" in w for w in result["warnings"])
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
graph_nodes, _ = _graph(pg_conn, WorkflowsRepository(pg_conn).get(str(updated["workflow_id"]), user))
agent_node = [n for n in graph_nodes if n["node_type"] == "agent"][0]
# Collapsed onto the oldest match, and not stored twice.
assert agent_node["config"]["sources"] == [str(older["id"])]
tests/api/user/agents/test_workflow_agents.py
+461
View File
@@ -0,0 +1,461 @@
"""Workflow-agent behavior of the agents routes, against real Postgres.
Rebuilds the workflow-related coverage of the retired Mongo-era
``tests/api/user/test_agents_routes.py`` on the ``pg_conn`` fixture:
adopting a workflow template (the graph must be deep-copied, never
shared by id), deleting a workflow agent (only the owner's graph may
go), and creating/updating agents with a workflow reference.
"""
from contextlib import contextmanager
from unittest.mock import patch
import pytest
from flask import Flask
@pytest.fixture
def app():
return Flask(__name__)
@contextmanager
def _patch_db(conn):
@contextmanager
def _yield():
yield conn
with patch(
"application.api.user.agents.routes.db_session", _yield
), patch(
"application.api.user.agents.routes.db_readonly", _yield
):
yield
def _seed_workflow(pg_conn, user, *, name="WF"):
"""Create a start → agent → end workflow; returns (workflow, nodes, edges)."""
from application.storage.db.repositories.workflow_edges import (
WorkflowEdgesRepository,
)
from application.storage.db.repositories.workflow_nodes import (
WorkflowNodesRepository,
)
from application.storage.db.repositories.workflows import WorkflowsRepository
wf = WorkflowsRepository(pg_conn).create(user, name, description="d")
wf_id = str(wf["id"])
nodes = WorkflowNodesRepository(pg_conn).bulk_create(
wf_id,
1,
[
{"node_id": "start-1", "node_type": "start", "title": "Start"},
{
"node_id": "agent-1",
"node_type": "agent",
"title": "Agent",
"config": {"system_prompt": "hi", "tools": [], "sources": []},
},
{"node_id": "end-1", "node_type": "end", "title": "End"},
],
)
by_str = {n["node_id"]: n["id"] for n in nodes}
edges = WorkflowEdgesRepository(pg_conn).bulk_create(
wf_id,
1,
[
{
"edge_id": "e1",
"from_node_id": by_str["start-1"],
"to_node_id": by_str["agent-1"],
},
{
"edge_id": "e2",
"from_node_id": by_str["agent-1"],
"to_node_id": by_str["end-1"],
},
],
)
return wf, nodes, edges
def _graph(pg_conn, workflow_id, version=1):
from application.storage.db.repositories.workflow_edges import (
WorkflowEdgesRepository,
)
from application.storage.db.repositories.workflow_nodes import (
WorkflowNodesRepository,
)
nodes = WorkflowNodesRepository(pg_conn).find_by_version(str(workflow_id), version)
edges = WorkflowEdgesRepository(pg_conn).find_by_version(str(workflow_id), version)
return nodes, edges
class TestCloneToUser:
def test_clones_graph_for_new_owner(self, pg_conn):
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "owner-a", name="Source WF")
clone = WorkflowsRepository(pg_conn).clone_to_user(
str(wf["id"]), "owner-b", from_owner="owner-a"
)
assert clone is not None
assert str(clone["id"]) != str(wf["id"])
assert clone["user_id"] == "owner-b"
assert clone["name"] == "Source WF"
nodes, edges = _graph(pg_conn, clone["id"])
assert {n["node_id"] for n in nodes} == {"start-1", "agent-1", "end-1"}
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
assert agent_node["config"] == {
"system_prompt": "hi", "tools": [], "sources": [],
}
# Edges are re-keyed onto the clone's node rows, not the source's.
clone_node_uuids = {n["id"] for n in nodes}
assert len(edges) == 2
for e in edges:
assert e["from_node_id"] in clone_node_uuids
assert e["to_node_id"] in clone_node_uuids
# The source graph is untouched.
src_nodes, src_edges = _graph(pg_conn, wf["id"])
assert len(src_nodes) == 3 and len(src_edges) == 2
def test_wrong_owner_returns_none(self, pg_conn):
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "owner-a")
assert (
WorkflowsRepository(pg_conn).clone_to_user(
str(wf["id"]), "thief", from_owner="not-the-owner"
)
is None
)
def test_missing_workflow_returns_none(self, pg_conn):
from application.storage.db.repositories.workflows import WorkflowsRepository
assert (
WorkflowsRepository(pg_conn).clone_to_user(
"00000000-0000-0000-0000-000000000000", "u"
)
is None
)
class TestAdoptWorkflowAgent:
def test_adopt_clones_the_graph(self, app, pg_conn):
from application.api.user.agents.routes import AdoptAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "__system__", name="Template WF")
repo = AgentsRepository(pg_conn)
template = repo.create(
"__system__",
"WF Template",
"template",
agent_type="workflow",
workflow_id=str(wf["id"]),
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/adopt_agent?id={template['id']}", method="POST"
):
from flask import request
request.decoded_token = {"sub": "u-wf-adopter"}
response = AdoptAgent().post()
assert response.status_code == 200
adopted = next(
a for a in repo.list_for_user("u-wf-adopter") if a["name"] == "WF Template"
)
assert adopted["workflow_id"] is not None
assert str(adopted["workflow_id"]) != str(wf["id"])
cloned = WorkflowsRepository(pg_conn).get(
str(adopted["workflow_id"]), "u-wf-adopter"
)
assert cloned is not None
nodes, edges = _graph(pg_conn, cloned["id"])
assert {n["node_id"] for n in nodes} == {"start-1", "agent-1", "end-1"}
assert len(edges) == 2
def test_adopt_refuses_unresolvable_workflow(self, app, pg_conn):
"""A template pointing at a graph its owner doesn't own cannot adopt.
Publishing the agent anyway would hand the adopter a workflow agent
with no graph — permanently unrunnable — so the adopt fails instead.
"""
from application.api.user.agents.routes import AdoptAgent
from application.storage.db.repositories.agents import AgentsRepository
wf, _, _ = _seed_workflow(pg_conn, "someone-else")
repo = AgentsRepository(pg_conn)
template = repo.create(
"__system__",
"Broken WF Template",
"template",
agent_type="workflow",
workflow_id=str(wf["id"]),
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/adopt_agent?id={template['id']}", method="POST"
):
from flask import request
request.decoded_token = {"sub": "u-wf-adopter2"}
response = AdoptAgent().post()
assert response.status_code == 500
assert not [
a
for a in repo.list_for_user("u-wf-adopter2")
if a["name"] == "Broken WF Template"
]
# The unowned graph was not cloned or mutated.
nodes, edges = _graph(pg_conn, wf["id"])
assert len(nodes) == 3 and len(edges) == 2
def test_adopt_strips_foreign_node_refs(self, app, pg_conn):
"""Cloned agent nodes keep builtin tool ids but shed the owner's refs."""
from application.agents.default_tools import default_tool_id
from application.api.user.agents.routes import AdoptAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.user_tools import (
UserToolsRepository,
)
from application.storage.db.repositories.workflow_nodes import (
WorkflowNodesRepository,
)
from application.storage.db.repositories.workflows import WorkflowsRepository
owner_tool = UserToolsRepository(pg_conn).create(
"__system__", "brave", config={}, display_name="Brave"
)
builtin_id = default_tool_id("artifact_generator")
wf = WorkflowsRepository(pg_conn).create("__system__", "Reffy WF")
WorkflowNodesRepository(pg_conn).bulk_create(
str(wf["id"]),
1,
[
{"node_id": "start-1", "node_type": "start", "title": "Start"},
{
"node_id": "agent-1",
"node_type": "agent",
"title": "Agent",
"config": {
"tools": [builtin_id, str(owner_tool["id"])],
"sources": ["00000000-0000-0000-0000-00000000dead"],
"model_id": "00000000-0000-0000-0000-00000000beef",
},
},
],
)
repo = AgentsRepository(pg_conn)
template = repo.create(
"__system__",
"Reffy Template",
"template",
agent_type="workflow",
workflow_id=str(wf["id"]),
tools=[builtin_id, str(owner_tool["id"])],
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/adopt_agent?id={template['id']}", method="POST"
):
from flask import request
request.decoded_token = {"sub": "u-wf-adopter3"}
response = AdoptAgent().post()
assert response.status_code == 200
adopted = next(
a
for a in repo.list_for_user("u-wf-adopter3")
if a["name"] == "Reffy Template"
)
# Classic-column tools: the builtin survives, the foreign row doesn't.
assert adopted["tools"] == [builtin_id]
nodes, _ = _graph(pg_conn, adopted["workflow_id"])
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
assert agent_node["config"]["tools"] == [builtin_id]
assert agent_node["config"]["sources"] == []
assert agent_node["config"]["model_id"] is None
# The template's own graph is untouched.
src_nodes, _ = _graph(pg_conn, wf["id"])
src_agent = next(n for n in src_nodes if n["node_id"] == "agent-1")
assert str(owner_tool["id"]) in src_agent["config"]["tools"]
class TestDeleteWorkflowAgent:
def test_delete_removes_owned_workflow(self, app, pg_conn):
from application.api.user.agents.routes import DeleteAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
user = "u-wf-del"
wf, _, _ = _seed_workflow(pg_conn, user)
repo = AgentsRepository(pg_conn)
agent = repo.create(
user, "WF Agent", "draft", agent_type="workflow", workflow_id=str(wf["id"])
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/delete_agent?id={agent['id']}", method="DELETE"
):
from flask import request
request.decoded_token = {"sub": user}
response = DeleteAgent().delete()
assert response.status_code == 200
assert repo.get_any(str(agent["id"]), user) is None
assert WorkflowsRepository(pg_conn).get_by_id(str(wf["id"])) is None
nodes, edges = _graph(pg_conn, wf["id"])
assert nodes == [] and edges == []
def test_delete_leaves_unowned_workflow_intact(self, app, pg_conn):
"""Regression: deleting an agent whose ``workflow_id`` points at another
user's workflow (the pre-clone adopted shape) must not touch that graph.
The old explicit node/edge cleanup was not owner-scoped and gutted it."""
from application.api.user.agents.routes import DeleteAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "__system__", name="Shared Template WF")
repo = AgentsRepository(pg_conn)
agent = repo.create(
"u-wf-del2",
"Adopted WF Agent",
"draft",
agent_type="workflow",
workflow_id=str(wf["id"]),
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/delete_agent?id={agent['id']}", method="DELETE"
):
from flask import request
request.decoded_token = {"sub": "u-wf-del2"}
response = DeleteAgent().delete()
assert response.status_code == 200
assert repo.get_any(str(agent["id"]), "u-wf-del2") is None
# The system's workflow and its whole graph survive.
assert WorkflowsRepository(pg_conn).get_by_id(str(wf["id"])) is not None
nodes, edges = _graph(pg_conn, wf["id"])
assert len(nodes) == 3 and len(edges) == 2
class TestCreateWorkflowAgent:
def test_create_published_with_owned_workflow(self, app, pg_conn):
from application.api.user.agents.routes import CreateAgent
from application.storage.db.repositories.agents import AgentsRepository
user = "u-wf-create"
wf, _, _ = _seed_workflow(pg_conn, user)
with _patch_db(pg_conn), app.test_request_context(
"/api/create_agent",
method="POST",
json={
"name": "My WF Agent",
"description": "d",
"agent_type": "workflow",
"status": "published",
"workflow": str(wf["id"]),
},
):
from flask import request
request.decoded_token = {"sub": user}
response = CreateAgent().post()
assert response.status_code == 201
agent = next(
a
for a in AgentsRepository(pg_conn).list_for_user(user)
if a["name"] == "My WF Agent"
)
assert str(agent["workflow_id"]) == str(wf["id"])
assert agent["status"] == "published"
def test_create_published_with_unowned_workflow_returns_404(self, app, pg_conn):
from application.api.user.agents.routes import CreateAgent
wf, _, _ = _seed_workflow(pg_conn, "someone-else")
with _patch_db(pg_conn), app.test_request_context(
"/api/create_agent",
method="POST",
json={
"name": "Stolen WF Agent",
"description": "d",
"agent_type": "workflow",
"status": "published",
"workflow": str(wf["id"]),
},
):
from flask import request
request.decoded_token = {"sub": "u-wf-thief"}
response = CreateAgent().post()
assert response.status_code == 404
class TestUpdateWorkflowAgent:
def test_update_sets_and_clears_workflow(self, app, pg_conn):
from application.api.user.agents.routes import UpdateAgent
from application.storage.db.repositories.agents import AgentsRepository
user = "u-wf-update"
wf, _, _ = _seed_workflow(pg_conn, user)
repo = AgentsRepository(pg_conn)
agent = repo.create(user, "WF Agent", "draft", agent_type="workflow")
agent_id = str(agent["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/update_agent/{agent_id}",
method="PUT",
json={"workflow": str(wf["id"])},
):
from flask import request
request.decoded_token = {"sub": user}
response = UpdateAgent().put(agent_id)
assert response.status_code == 200
assert str(repo.get(agent_id, user)["workflow_id"]) == str(wf["id"])
# Clearing is allowed while the agent stays a draft.
with _patch_db(pg_conn), app.test_request_context(
f"/api/update_agent/{agent_id}",
method="PUT",
json={"workflow": None},
):
from flask import request
request.decoded_token = {"sub": user}
response = UpdateAgent().put(agent_id)
assert response.status_code == 200
assert repo.get(agent_id, user)["workflow_id"] is None
def test_publish_without_workflow_is_rejected(self, app, pg_conn):
from application.api.user.agents.routes import UpdateAgent
from application.storage.db.repositories.agents import AgentsRepository
user = "u-wf-update2"
agent = AgentsRepository(pg_conn).create(
user, "WF Agent", "draft", agent_type="workflow"
)
agent_id = str(agent["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/update_agent/{agent_id}",
method="PUT",
json={"status": "published"},
):
from flask import request
request.decoded_token = {"sub": user}
response = UpdateAgent().put(agent_id)
assert response.status_code == 400
tests/api/user/test_agents_routes.py
-3645
View File
File diff suppressed because it is too large. Load diff
tests/api/user/test_workflows.py
-12
View File
@@ -1,12 +0,0 @@
"""Tests for application/api/user/workflows/routes.py.
Previously asserted on bson.ObjectId serialization. Workflow persistence is
now Postgres-backed; coverage will be rebuilt via pg_conn + WorkflowsRepository.
"""
import pytest
@pytest.mark.skip(reason="needs PG fixture rewrite - tracked separately")
def test_user_workflows_routes_pending_pg_rewrite():
pass