"""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"))