416 lines
14 KiB
Python
416 lines
14 KiB
Python
"""Regression coverage for openGauss/Dolphin through the MySQL protocol."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy import Column, Integer, MetaData, String, Table
|
|
|
|
from deerflow.persistence.gauss_mysql_compat import (
|
|
GAUSS_MYSQL_SQLALCHEMY_VERSION,
|
|
_gauss_mysql_default_schema_name,
|
|
_gauss_mysql_has_table,
|
|
_gauss_mysql_isolation_level,
|
|
_gauss_mysql_server_version_info,
|
|
decode_gauss_mysql_json,
|
|
install_gauss_mysql_dialect_compatibility,
|
|
is_gauss_mysql_product,
|
|
)
|
|
from deerflow.persistence.gauss_schema import ensure_gauss_schema_sync
|
|
from deerflow.persistence.types import PortableJSON
|
|
|
|
|
|
def _mysql_escape(value: str) -> str:
|
|
table = {
|
|
"\0": "\\0",
|
|
"\\": "\\\\",
|
|
"\n": "\\n",
|
|
"\r": "\\r",
|
|
"\x1a": "\\Z",
|
|
'"': '\\"',
|
|
"'": "\\'",
|
|
}
|
|
return "".join(table.get(character, character) for character in value)
|
|
|
|
|
|
def test_product_switch_is_explicit_and_case_insensitive(monkeypatch) -> None:
|
|
monkeypatch.delenv("DEERFLOW_DATABASE_PRODUCT", raising=False)
|
|
assert not is_gauss_mysql_product()
|
|
monkeypatch.setenv("DEERFLOW_DATABASE_PRODUCT", "mysql")
|
|
assert not is_gauss_mysql_product()
|
|
monkeypatch.setenv("DEERFLOW_DATABASE_PRODUCT", " GAUSS_MYSQL ")
|
|
assert is_gauss_mysql_product()
|
|
|
|
|
|
def test_json_decoder_accepts_strict_and_bounded_legacy_formats() -> None:
|
|
strict = {"name": "中文", "enabled": True, "items": [1, None]}
|
|
assert decode_gauss_mysql_json(json.dumps(strict, ensure_ascii=False)) == strict
|
|
assert decode_gauss_mysql_json("{'name': 'legacy', 'enabled': True}") == {
|
|
"name": "legacy",
|
|
"enabled": True,
|
|
}
|
|
escaped = _mysql_escape(json.dumps(["web", "file:read", "bash"]))
|
|
assert decode_gauss_mysql_json(escaped) == ["web", "file:read", "bash"]
|
|
assert decode_gauss_mysql_json(memoryview(escaped.encode())) == ["web", "file:read", "bash"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
[
|
|
r"[\"web\",",
|
|
r"[\"bad\q\"]",
|
|
r'[\"a\", "b"]',
|
|
"__import__('os').system('echo unsafe')",
|
|
"{1, 2}",
|
|
"(1, 2)",
|
|
"{'nested': {1, 2}}",
|
|
"{'nested': (1, 2)}",
|
|
"{1: 'non-string key'}",
|
|
"{'nested': b'bytes'}",
|
|
],
|
|
)
|
|
def test_json_decoder_rejects_malformed_or_unsafe_values(value: str) -> None:
|
|
with pytest.raises(ValueError):
|
|
decode_gauss_mysql_json(value)
|
|
|
|
|
|
def test_portable_json_fallback_is_dolphin_only() -> None:
|
|
ordinary_mysql = SimpleNamespace(name="mysql")
|
|
dolphin = SimpleNamespace(name="mysql", _deerflow_gauss_mysql_compat_installed=True)
|
|
column_type = PortableJSON()
|
|
|
|
with pytest.raises(json.JSONDecodeError):
|
|
column_type.process_result_value("{'enabled': True}", ordinary_mysql)
|
|
assert column_type.process_result_value("{'enabled': True}", dolphin) == {"enabled": True}
|
|
|
|
|
|
def test_dialect_helpers_use_current_schema_version_and_isolation() -> None:
|
|
class Cursor:
|
|
def __init__(self, row):
|
|
self.row = row
|
|
self.statement = None
|
|
self.closed = False
|
|
|
|
def execute(self, statement):
|
|
self.statement = statement
|
|
|
|
def fetchone(self):
|
|
return self.row
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
version_cursor = Cursor(("openGauss 6.0.5",))
|
|
dialect = SimpleNamespace(
|
|
_set_mariadb=lambda *_args: None,
|
|
server_version_info=None,
|
|
)
|
|
connection = SimpleNamespace(connection=SimpleNamespace(cursor=lambda: version_cursor))
|
|
assert _gauss_mysql_server_version_info(dialect, connection) == GAUSS_MYSQL_SQLALCHEMY_VERSION
|
|
assert dialect.server_version_info == GAUSS_MYSQL_SQLALCHEMY_VERSION
|
|
assert version_cursor.statement == "SELECT VERSION()"
|
|
assert version_cursor.closed
|
|
|
|
isolation_cursor = Cursor(("read-committed",))
|
|
dbapi = SimpleNamespace(cursor=lambda: isolation_cursor)
|
|
assert _gauss_mysql_isolation_level(dialect, dbapi) == "READ COMMITTED"
|
|
assert isolation_cursor.statement == "SELECT @@transaction_isolation"
|
|
|
|
class ScalarResult:
|
|
def scalar(self):
|
|
return b"deerflow"
|
|
|
|
class SchemaConnection:
|
|
def exec_driver_sql(self, statement):
|
|
assert statement == "SELECT current_schema()"
|
|
return ScalarResult()
|
|
|
|
assert _gauss_mysql_default_schema_name(dialect, SchemaConnection()) == "deerflow"
|
|
|
|
|
|
def test_has_table_uses_native_catalog_instead_of_describe() -> None:
|
|
observed = {}
|
|
|
|
class Result:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def fetchone(self):
|
|
return (1,)
|
|
|
|
class Connection:
|
|
def exec_driver_sql(self, statement, parameters, execution_options):
|
|
observed.update(
|
|
statement=statement,
|
|
parameters=parameters,
|
|
execution_options=execution_options,
|
|
)
|
|
return Result()
|
|
|
|
dialect = SimpleNamespace(
|
|
default_schema_name="deerflow",
|
|
_ensure_has_table_connection=lambda _connection: None,
|
|
)
|
|
assert _gauss_mysql_has_table(dialect, Connection(), "users")
|
|
assert "pg_catalog.pg_class" in observed["statement"]
|
|
assert "DESCRIBE" not in observed["statement"].upper()
|
|
assert observed["parameters"] == ("deerflow", "users")
|
|
|
|
|
|
def test_installer_is_driver_scoped_and_idempotent(monkeypatch) -> None:
|
|
from sqlalchemy.dialects.mysql import asyncmy as asyncmy_module
|
|
|
|
class FakeDialect:
|
|
calls = 0
|
|
|
|
def create_connect_args(self, _url):
|
|
type(self).calls += 1
|
|
return [], {"charset": "utf8mb4"}
|
|
|
|
monkeypatch.setattr(asyncmy_module, "MySQLDialect_asyncmy", FakeDialect)
|
|
install_gauss_mysql_dialect_compatibility("asyncmy")
|
|
first_original = FakeDialect._deerflow_gauss_mysql_original_create_connect_args
|
|
install_gauss_mysql_dialect_compatibility("asyncmy")
|
|
|
|
_args, kwargs = FakeDialect().create_connect_args(object())
|
|
assert kwargs["charset"] == "utf8mb4"
|
|
assert kwargs["init_command"] == "SET standard_conforming_strings TO off"
|
|
assert FakeDialect.calls == 1
|
|
assert FakeDialect._deerflow_gauss_mysql_original_create_connect_args is first_original
|
|
assert FakeDialect._deerflow_gauss_mysql_compat_installed is True
|
|
|
|
|
|
def test_installer_rejects_conflicting_init_command(monkeypatch) -> None:
|
|
from sqlalchemy.dialects.mysql import asyncmy as asyncmy_module
|
|
|
|
class FakeDialect:
|
|
def create_connect_args(self, _url):
|
|
return [], {"init_command": "SET time_zone = '+00:00'"}
|
|
|
|
monkeypatch.setattr(asyncmy_module, "MySQLDialect_asyncmy", FakeDialect)
|
|
install_gauss_mysql_dialect_compatibility("asyncmy")
|
|
|
|
with pytest.raises(ValueError, match="standard_conforming_strings"):
|
|
FakeDialect().create_connect_args(object())
|
|
|
|
|
|
class _RowsResult:
|
|
def __init__(self, *, scalar=None, rows=()):
|
|
self._scalar = scalar
|
|
self._rows = list(rows)
|
|
|
|
def scalar_one_or_none(self):
|
|
return self._scalar
|
|
|
|
def fetchall(self):
|
|
return list(self._rows)
|
|
|
|
|
|
def test_schema_integrity_preflights_all_columns_before_ddl() -> None:
|
|
metadata = MetaData()
|
|
Table(
|
|
"example",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("safe_note", String(20), nullable=True),
|
|
Column("unsafe_required", String(20), nullable=False),
|
|
)
|
|
|
|
class Connection:
|
|
engine = SimpleNamespace(url=SimpleNamespace(database="deerflow"))
|
|
|
|
def __init__(self):
|
|
self.statements = []
|
|
|
|
def exec_driver_sql(self, statement, parameters=None):
|
|
self.statements.append((statement, parameters))
|
|
if statement == "SELECT current_schema()":
|
|
return _RowsResult(scalar="deerflow")
|
|
if "pg_catalog.pg_attribute" in statement:
|
|
return _RowsResult(rows=[("example", "r", "id")])
|
|
raise AssertionError(f"DDL must not run after unsafe preflight: {statement}")
|
|
|
|
connection = Connection()
|
|
with pytest.raises(RuntimeError, match="NOT NULL column has no server_default"):
|
|
ensure_gauss_schema_sync(connection, metadata)
|
|
assert not any(statement.lstrip().startswith("ALTER TABLE") for statement, _ in connection.statements)
|
|
|
|
|
|
def test_active_feature_models_are_registered_without_enterprise_research() -> None:
|
|
import deerflow.persistence.models # noqa: F401
|
|
from deerflow.persistence.base import Base
|
|
|
|
expected = {
|
|
"ai_writing_sessions",
|
|
"ai_writing_article_types",
|
|
"scheduled_html_pages",
|
|
"html_page_favorites",
|
|
"task_analysis_tasks",
|
|
"task_analysis_settings",
|
|
}
|
|
assert expected <= set(Base.metadata.tables)
|
|
assert "enterprise_research_tasks" not in Base.metadata.tables
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("product", "expects_gauss"),
|
|
[("mysql", False), ("gauss_mysql", True)],
|
|
)
|
|
async def test_engine_selects_dolphin_hooks_without_changing_mysql(
|
|
monkeypatch,
|
|
product: str,
|
|
expects_gauss: bool,
|
|
) -> None:
|
|
import deerflow.persistence.engine as engine_module
|
|
import deerflow.persistence.gauss_mysql_compat as compat_module
|
|
|
|
calls: list[str] = []
|
|
|
|
class Connection:
|
|
async def run_sync(self, function, *_args):
|
|
calls.append(function.__name__)
|
|
|
|
class Begin:
|
|
async def __aenter__(self):
|
|
return Connection()
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
class Engine:
|
|
dialect = SimpleNamespace(name="mysql")
|
|
|
|
def begin(self):
|
|
return Begin()
|
|
|
|
async def dispose(self):
|
|
calls.append("dispose")
|
|
|
|
fake_engine = Engine()
|
|
monkeypatch.setenv("DEERFLOW_DATABASE_PRODUCT", product)
|
|
monkeypatch.setattr(engine_module, "find_spec", lambda _name: object())
|
|
monkeypatch.setattr(engine_module, "_add_runtime_vendor_path", lambda: None)
|
|
monkeypatch.setattr(engine_module, "create_async_engine", lambda *_args, **_kwargs: fake_engine)
|
|
monkeypatch.setattr(engine_module, "_acquire_gauss_mysql_schema_bootstrap_lock", lambda: None)
|
|
monkeypatch.setattr(engine_module, "_release_gauss_mysql_schema_bootstrap_lock", lambda _handle: None)
|
|
monkeypatch.setattr(
|
|
compat_module,
|
|
"install_gauss_mysql_dialect_compatibility",
|
|
lambda driver: calls.append(f"install:{driver}"),
|
|
)
|
|
|
|
await engine_module.close_engine()
|
|
try:
|
|
await engine_module.init_engine(
|
|
"mysql",
|
|
url="mysql+asyncmy://user:pass@db.example/deerflow",
|
|
)
|
|
finally:
|
|
await engine_module.close_engine()
|
|
|
|
assert ("install:asyncmy" in calls) is expects_gauss
|
|
assert ("ensure_gauss_schema_sync" in calls) is expects_gauss
|
|
assert ("_ensure_orm_columns_sync" in calls) is (not expects_gauss)
|
|
assert ("_backfill_thread_meta_task_projections_sync" in calls) is (not expects_gauss)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dolphin_unknown_schema_never_runs_mysql_create_database(monkeypatch) -> None:
|
|
import deerflow.persistence.engine as engine_module
|
|
import deerflow.persistence.gauss_mysql_compat as compat_module
|
|
|
|
calls: list[str] = []
|
|
|
|
class Connection:
|
|
async def run_sync(self, function, *_args):
|
|
if function.__name__ == "create_all":
|
|
raise RuntimeError("1049 unknown database")
|
|
|
|
class Begin:
|
|
async def __aenter__(self):
|
|
return Connection()
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
class Engine:
|
|
dialect = SimpleNamespace(name="mysql")
|
|
|
|
def begin(self):
|
|
return Begin()
|
|
|
|
async def dispose(self):
|
|
return None
|
|
|
|
monkeypatch.setenv("DEERFLOW_DATABASE_PRODUCT", "gauss_mysql")
|
|
monkeypatch.setattr(engine_module, "find_spec", lambda _name: object())
|
|
monkeypatch.setattr(engine_module, "_add_runtime_vendor_path", lambda: None)
|
|
monkeypatch.setattr(engine_module, "create_async_engine", lambda *_args, **_kwargs: Engine())
|
|
monkeypatch.setattr(engine_module, "_acquire_gauss_mysql_schema_bootstrap_lock", lambda: None)
|
|
monkeypatch.setattr(engine_module, "_release_gauss_mysql_schema_bootstrap_lock", lambda _handle: None)
|
|
monkeypatch.setattr(
|
|
engine_module,
|
|
"_auto_create_mysql_db",
|
|
lambda _url: calls.append("mysql-create-database"),
|
|
)
|
|
monkeypatch.setattr(compat_module, "install_gauss_mysql_dialect_compatibility", lambda _driver: None)
|
|
|
|
await engine_module.close_engine()
|
|
try:
|
|
# Startup remains best-effort by repository invariant, but the MySQL
|
|
# CREATE DATABASE retry must not be attempted against Dolphin.
|
|
await engine_module.init_engine(
|
|
"mysql",
|
|
url="mysql+asyncmy://user:pass@db.example/deerflow",
|
|
)
|
|
finally:
|
|
await engine_module.close_engine()
|
|
|
|
assert "mysql-create-database" not in calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dolphin_analytics_skips_mysql_session_variables(monkeypatch) -> None:
|
|
from app.gateway.routers.admin_users import _set_analytics_read_timeout
|
|
|
|
class Session:
|
|
async def connection(self):
|
|
raise AssertionError("Dolphin must return before touching MySQL session variables")
|
|
|
|
monkeypatch.setenv("DEERFLOW_DATABASE_PRODUCT", "gauss_mysql")
|
|
await _set_analytics_read_timeout(Session())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_display_table_accepts_only_duplicate_create_race(monkeypatch) -> None:
|
|
import app.gateway.routers.skills as skills_router
|
|
|
|
class DuplicateBegin:
|
|
async def __aenter__(self):
|
|
raise RuntimeError('relation "skill_display_adapters" already exists')
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
class DuplicateEngine:
|
|
def begin(self):
|
|
return DuplicateBegin()
|
|
|
|
duplicate_engine = DuplicateEngine()
|
|
monkeypatch.setattr(skills_router, "_skill_display_table_engine", None)
|
|
monkeypatch.setattr(skills_router, "_SKILL_DISPLAY_TABLE_LOCK", asyncio.Lock())
|
|
monkeypatch.setattr(skills_router, "get_engine", lambda: duplicate_engine)
|
|
monkeypatch.setattr(skills_router, "get_session_factory", lambda: object())
|
|
|
|
assert await skills_router._ensure_skill_display_table()
|
|
assert skills_router._skill_display_table_engine is duplicate_engine
|
|
assert skills_router._skill_display_table_already_exists(RuntimeError("error 1050"))
|
|
assert not skills_router._skill_display_table_already_exists(RuntimeError("permission denied"))
|