"""Phase-1 workflow definition store + validator + Coze compat tests.""" from __future__ import annotations import json import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy.dialects import mysql from sqlalchemy.schema import CreateTable from app.gateway.routers import workflow_resources, workflows, workflows_coze_compat from deerflow.persistence.workflows import ( MemoryWorkflowStore, WorkflowDraftConflictError, make_workflow_store, ) from deerflow.persistence.workflows.model import WorkflowDefinitionRow from deerflow.workflows.samples import load_sample_graph from deerflow.workflows.validator import validate_workflow_graph def test_workflow_definition_mysql_uses_longtext_without_a_server_default() -> None: """MySQL 5.7 rejects DEFAULT values on TEXT/LONGTEXT columns.""" ddl = str(CreateTable(WorkflowDefinitionRow.__table__).compile(dialect=mysql.dialect())).upper() assert "DRAFT_GRAPH_JSON LONGTEXT NOT NULL" in ddl assert "DRAFT_CANVAS_SCHEMA_JSON LONGTEXT NOT NULL" in ddl assert "DRAFT_CANVAS_SCHEMA_JSON LONGTEXT NOT NULL DEFAULT" not in ddl @pytest.mark.asyncio async def test_memory_store_create_draft_publish() -> None: store = make_workflow_store(None) assert isinstance(store, MemoryWorkflowStore) created = await store.create_definition( { "name": "专题报告", "owner_id": "user_a", "draft_graph": load_sample_graph("start_output").model_dump(by_alias=True), } ) assert created["draft_revision"] == 0 assert created["status"] == "draft" wf_id = created["id"] saved = await store.save_draft( wf_id, expected_revision=0, graph=load_sample_graph("start_output").model_dump(by_alias=True), ) assert saved["draft_revision"] == 1 with pytest.raises(WorkflowDraftConflictError) as exc: await store.save_draft( wf_id, expected_revision=0, graph=load_sample_graph("start_output").model_dump(by_alias=True), ) assert exc.value.current_revision == 1 graph = load_sample_graph("start_output") version = await store.publish_version( wf_id, graph=graph.model_dump(by_alias=True), graph_hash=graph.graph_hash(), published_by="user_a", change_note="v1", ) assert version["version_number"] == 1 assert version["graph_hash"] == graph.graph_hash() definition = await store.get_definition(wf_id) assert definition is not None assert definition["status"] == "active" versions = await store.list_versions(wf_id) assert len(versions) == 1 loaded = await store.get_version(version["id"]) assert loaded is not None assert loaded["graph"]["id"] == "wf_start_output" def test_validate_start_output_and_report_samples() -> None: start_output = load_sample_graph("start_output") assert validate_workflow_graph(start_output) == [] report = load_sample_graph("report_generation") issues = validate_workflow_graph(report) assert issues == [], [i.model_dump(by_alias=True) for i in issues] def test_validate_rejects_missing_output_and_cycle() -> None: issues = validate_workflow_graph( { "schemaVersion": "1.0", "id": "wf_bad", "nodes": [ {"id": "start_1", "type": "start", "config": {}}, {"id": "a", "type": "transform", "config": {"operations": []}}, {"id": "b", "type": "transform", "config": {"operations": []}}, ], "edges": [ {"id": "e1", "source": "start_1", "target": "a"}, {"id": "e2", "source": "a", "target": "b"}, {"id": "e3", "source": "b", "target": "a"}, ], } ) codes = {i.code for i in issues} assert "WORKFLOW_SCHEMA_INVALID" in codes # missing output assert "WORKFLOW_GRAPH_CYCLE" in codes def test_workflow_http_api_and_coze_compat(monkeypatch: pytest.MonkeyPatch) -> None: from deerflow.config.workflow_config import WorkflowConfig class _Cfg: workflows = WorkflowConfig() models = [] monkeypatch.setattr(workflows, "get_app_config", lambda: _Cfg()) monkeypatch.setattr(workflow_resources, "get_app_config", lambda: _Cfg()) monkeypatch.setattr(workflows_coze_compat, "get_app_config", lambda: _Cfg()) app = FastAPI() store = MemoryWorkflowStore() app.state.workflow_store = store app.include_router(workflow_resources.router) app.include_router(workflows.router) app.include_router(workflows_coze_compat.router) app.include_router(workflows_coze_compat.chrome_router) async def _fake_user(_request=None): return "user_a" monkeypatch.setattr(workflows, "get_current_user", _fake_user) monkeypatch.setattr(workflows, "get_optional_user_from_request", lambda _r: None) monkeypatch.setattr(workflow_resources, "get_current_user", _fake_user) monkeypatch.setattr(workflows_coze_compat, "get_current_user", _fake_user) client = TestClient(app) created = client.post("/api/workflows", json={"name": "demo"}).json() wf_id = created["id"] assert created["draft_revision"] == 0 graph = load_sample_graph("start_output").model_dump(by_alias=True) saved = client.put( f"/api/workflows/{wf_id}/draft", json={"expectedRevision": 0, "graph": graph}, ).json() assert saved["draft_revision"] == 1 conflict = client.put( f"/api/workflows/{wf_id}/draft", json={"expectedRevision": 0, "graph": graph}, ) assert conflict.status_code == 409 validated = client.post(f"/api/workflows/{wf_id}/validate", json={}).json() assert validated["ok"] is True published = client.post( f"/api/workflows/{wf_id}/publish", json={"changeNote": "first", "expectedRevision": 1}, ).json() assert published["version_number"] == 1 types = client.get("/api/workflows/node-types").json() # The built-in catalog may grow while preserving the minimum v1 nodes. assert len(types["nodeTypes"]) >= 13 canvas = client.post("/api/workflow_api/canvas", json={"workflowId": wf_id}).json() assert canvas["code"] == 0 workflow_brief = canvas["data"]["workflow"] assert workflow_brief["workflow_id"] == wf_id assert workflow_brief["is_published"] is True # Coze Playground uses this alongside VCS metadata to decide whether nodes # are draggable. An owned standalone workflow must not open as preview. assert workflow_brief["creator"]["self"] is True # Never opened in the Coze editor → built-in start→end canvas fallback. assert json.loads(workflow_brief["schema_json"])["nodes"] coze_canvas = json.dumps( { "nodes": [ {"id": "100001", "type": "1", "meta": {"position": {"x": 0, "y": 0}}}, {"id": "900001", "type": "2", "meta": {"position": {"x": 400, "y": 0}}}, ], "edges": [{"sourceNodeID": "100001", "targetNodeID": "900001"}], } ) saved = client.post( "/api/workflow_api/save", json={"workflow_id": wf_id, "schema": coze_canvas}, ) assert saved.status_code == 200, saved.text assert saved.json()["code"] == 0 reloaded = client.post("/api/workflow_api/canvas", json={"workflowId": wf_id}).json() # Storage canonicalises to compact JSON; compare as documents. assert json.loads(reloaded["data"]["workflow"]["schema_json"]) == json.loads(coze_canvas) # The Coze save must round-trip the canvas without touching the graph. definition = client.get(f"/api/workflows/{wf_id}").json() assert definition["draft_graph"]["nodes"], "execution graph must pass through unchanged" templates = client.post("/api/workflow_api/node_template_list").json() assert templates["code"] == 0 template_list = templates["data"]["template_list"] assert len(template_list) == 13 # Coze playground 形状：node_type 数字串 + 分类表 assert all("node_type" in item and "name" in item for item in template_list) assert templates["data"]["cate_list"] account = client.post("/api/passport/account/info/v2/").json() assert account["code"] == 0 assert account["data"]["user_id"] == "user_a" assert account["data"]["login_status"] == "logined" playground_spaces = client.post("/api/playground_api/space/list").json() assert playground_spaces["code"] == 0 assert playground_spaces["data"]["has_personal_space"] is True assert playground_spaces["data"]["bot_space_list"][0]["id"] == "1" listed = client.post("/api/space/list").json() assert listed["code"] == 0 assert listed["bot_space_list"][0]["id"] == "1" assert listed["data"]["bot_space_list"][0]["name"] == "默认空间" assert listed["data"]["bot_space_list"][0]["space_type"] == 1 saved_space = client.post("/api/playground_api/space/save").json() assert saved_space["code"] == 0 assert saved_space["data"]["id"] == "1" models = client.post("/api/bot/get_type_list").json() assert models["code"] == 0 assert models["data"]["model_list"] == [] copied = client.post(f"/api/workflows/{wf_id}/copy") assert copied.status_code == 200, copied.text assert copied.json()["id"] != wf_id assert copied.json()["name"].endswith("副本") def test_coze_studio_paths_are_csrf_exempt() -> None: from app.gateway.csrf_middleware import is_coze_studio_compat_path assert is_coze_studio_compat_path("/api/playground_api/space/list") assert is_coze_studio_compat_path("/api/passport/account/info/v2/") assert is_coze_studio_compat_path("/api/workflow_api/canvas") assert is_coze_studio_compat_path("/api/space/list") assert is_coze_studio_compat_path("/api/bot/get_type_list") assert not is_coze_studio_compat_path("/api/workflows") assert not is_coze_studio_compat_path("/api/v1/auth/me")