董宏凯
ac92f29db1
提交内容：为数据库生命周期增加按白名单清单重建 Workflow Studio 表的受限流程，限定仅 MySQL 且检查当前连接数据库名称一致，避免误删非目标 schema。 启动流程中新增 DEER_FLOW_REBUILD_WORKFLOW_SCHEMA 环境变量入口与持久层重建 API，并在 lifespan 里按需执行；导出到 persistence 包，便于后续调用与测试复用。 迁移与模型修复：将 workflow 相关大文本字段切到 MySQL LONGTEXT，不再给 TEXT/LONGTEXT 设置 server_default；新增 draft_canvas_schema_json 的迁移回填与非空约束变更，规避 MySQL 5.7 对默认值限制。 测试覆盖：补充 workflow 定义 ddl 长文本断言与 schema 重建保护性测试（仅允许清单表、数据库名不匹配直接拒绝），并把新测试文件加入版本追踪。
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""Safety tests for the one-time MySQL Workflow Studio schema rebuild."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
|
|
import deerflow.persistence.engine as persistence_engine
|
|
from deerflow.persistence.base import Base
|
|
|
|
|
|
class _ScalarResult:
|
|
def __init__(self, value: str | None) -> None:
|
|
self._value = value
|
|
|
|
def scalar_one_or_none(self) -> str | None:
|
|
return self._value
|
|
|
|
|
|
class _SyncConnection:
|
|
def __init__(self, database: str | None) -> None:
|
|
self._database = database
|
|
|
|
def exec_driver_sql(self, statement: str) -> _ScalarResult:
|
|
assert statement == "SELECT DATABASE()"
|
|
return _ScalarResult(self._database)
|
|
|
|
|
|
class _AsyncConnection:
|
|
def __init__(self, sync_connection: _SyncConnection) -> None:
|
|
self._sync_connection = sync_connection
|
|
|
|
async def run_sync(self, callback):
|
|
return callback(self._sync_connection)
|
|
|
|
|
|
class _AsyncEngine:
|
|
def __init__(self, database: str, connected_database: str | None) -> None:
|
|
self.dialect = SimpleNamespace(name="mysql")
|
|
self.url = SimpleNamespace(database=database)
|
|
self._connection = _AsyncConnection(_SyncConnection(connected_database))
|
|
|
|
@asynccontextmanager
|
|
async def begin(self):
|
|
yield self._connection
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rebuild_workflow_schema_only_targets_known_tables(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
engine = _AsyncEngine("deerflow", "deerflow")
|
|
monkeypatch.setattr(persistence_engine, "get_engine", lambda: engine)
|
|
monkeypatch.setattr(
|
|
sa,
|
|
"inspect",
|
|
lambda _connection: SimpleNamespace(
|
|
get_table_names=lambda: {"workflow_definitions", "unrelated_workflow_cache"}
|
|
),
|
|
)
|
|
|
|
calls: dict[str, list[str]] = {}
|
|
monkeypatch.setattr(
|
|
Base.metadata,
|
|
"drop_all",
|
|
lambda _connection, *, tables, checkfirst: calls.setdefault("drop", [table.name for table in tables]),
|
|
)
|
|
monkeypatch.setattr(
|
|
Base.metadata,
|
|
"create_all",
|
|
lambda _connection, *, tables, checkfirst: calls.setdefault("create", [table.name for table in tables]),
|
|
)
|
|
|
|
rebuilt = await persistence_engine.rebuild_workflow_schema_for_mysql()
|
|
|
|
expected = list(persistence_engine._WORKFLOW_SCHEMA_TABLE_NAMES)
|
|
assert rebuilt == ["workflow_definitions"]
|
|
assert calls == {"drop": expected, "create": expected}
|
|
assert "unrelated_workflow_cache" not in calls["drop"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rebuild_workflow_schema_refuses_unexpected_database(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
engine = _AsyncEngine("deerflow", "another_database")
|
|
monkeypatch.setattr(persistence_engine, "get_engine", lambda: engine)
|
|
monkeypatch.setattr(sa, "inspect", lambda _connection: pytest.fail("must not inspect tables"))
|
|
|
|
with pytest.raises(RuntimeError, match="does not match configured database"):
|
|
await persistence_engine.rebuild_workflow_schema_for_mysql()
|