deer-flow/backend/tests/test_config_version.py

"""Tests for config version check and upgrade logic."""

from __future__ import annotations

import logging
import os
import tempfile
from pathlib import Path

import pytest
import yaml
from support.shell import find_script_bash

from deerflow.config.app_config import AppConfig
from deerflow.config.knowledge_base_config import KnowledgeBaseConfig
from deerflow.tools.tools import get_available_tools


def test_knowledge_base_config_is_provider_agnostic() -> None:
    assert set(KnowledgeBaseConfig.model_fields) == {"enabled", "scope_selection_enabled"}
    config = KnowledgeBaseConfig.model_validate(
        {
            "enabled": True,
            "scope_selection_enabled": True,
            "base_url": "http://legacy-ragflow.test",
            "api_key": "legacy-secret",
        }
    )
    assert config.model_dump() == {"enabled": True, "scope_selection_enabled": True}


# Only the upgrade-script test shells out; it needs Git Bash on Windows (the
# WSL launcher and Store alias stubs cannot run the repo scripts).
SCRIPT_BASH = find_script_bash()


def _make_config_files(tmpdir: Path, user_config: dict, example_config: dict) -> Path:
    """Write user config.yaml and config.example.yaml to a temp dir, return config path."""
    config_path = tmpdir / "config.yaml"
    example_path = tmpdir / "config.example.yaml"

    # Minimal valid config needs sandbox
    defaults = {
        "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
    }
    for cfg in (user_config, example_config):
        for k, v in defaults.items():
            cfg.setdefault(k, v)

    with open(config_path, "w", encoding="utf-8") as f:
        yaml.dump(user_config, f)
    with open(example_path, "w", encoding="utf-8") as f:
        yaml.dump(example_config, f)

    return config_path


def test_missing_version_treated_as_zero(caplog):
    """Config without config_version should be treated as version 0."""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={},  # no config_version
            example_config={"config_version": 1},
        )
        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version(
                {"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}},
                config_path,
            )
        assert "outdated" in caplog.text
        assert "version 0" in caplog.text
        assert "version is 1" in caplog.text


def test_matching_version_no_warning(caplog):
    """Config with matching version should not emit a warning."""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={"config_version": 1},
            example_config={"config_version": 1},
        )
        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version(
                {"config_version": 1},
                config_path,
            )
        assert "outdated" not in caplog.text


def test_outdated_version_emits_warning(caplog):
    """Config with lower version should emit a warning."""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={"config_version": 1},
            example_config={"config_version": 2},
        )
        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version(
                {"config_version": 1},
                config_path,
            )
        assert "outdated" in caplog.text
        assert "version 1" in caplog.text
        assert "version is 2" in caplog.text


def test_no_example_file_no_warning(caplog):
    """If config.example.yaml doesn't exist, no warning should be emitted."""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = Path(tmpdir) / "config.yaml"
        with open(config_path, "w", encoding="utf-8") as f:
            yaml.dump({"sandbox": {"use": "test"}}, f)
        # No config.example.yaml created

        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version({}, config_path)
        assert "outdated" not in caplog.text


def test_string_config_version_does_not_raise_type_error(caplog):
    """config_version stored as a YAML string should not raise TypeError on comparison."""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={"config_version": "1"},  # string, as YAML can produce
            example_config={"config_version": 2},
        )
        # Must not raise TypeError: '<' not supported between instances of 'str' and 'int'
        AppConfig._check_config_version({"config_version": "1"}, config_path)


def test_newer_user_version_no_warning(caplog):
    """If user has a newer version than example (edge case), no warning."""
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={"config_version": 3},
            example_config={"config_version": 2},
        )
        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version(
                {"config_version": 3},
                config_path,
            )
        assert "outdated" not in caplog.text


@pytest.mark.skipif(SCRIPT_BASH is None, reason="repo shell-script tests need Git Bash on Windows")
def test_version_26_config_upgrades_to_checkpoint_channel_mode(tmp_path, caplog):
    """A v26 user config must be flagged outdated and merge the new persisted field.

    `database.checkpoint_channel_mode` shipped with config_version 27; the
    upgrade path must add it with the safe default (``full``) without touching
    the user's existing database backend settings. Uses the repository's real
    config.example.yaml and the real config-upgrade script.
    """
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    example_data = yaml.safe_load(example_src.read_text(encoding="utf-8"))
    expected_version = example_data["config_version"]
    assert expected_version > 26, "config.example.yaml must be bumped past 26 for checkpoint_channel_mode"

    config_path = tmp_path / "config.yaml"
    (tmp_path / "config.example.yaml").write_text(example_src.read_text(encoding="utf-8"), encoding="utf-8")
    user_config = {
        "config_version": 26,
        "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
        "database": {"backend": "sqlite", "sqlite_dir": "custom-data"},
    }
    config_path.write_text(yaml.dump(user_config), encoding="utf-8")

    with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
        AppConfig._check_config_version(dict(user_config), config_path)
    assert "outdated" in caplog.text
    assert "(version 26)" in caplog.text

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version
    assert upgraded["database"]["checkpoint_channel_mode"] == "full"
    assert upgraded["database"]["backend"] == "sqlite"
    assert upgraded["database"]["sqlite_dir"] == "custom-data"
    assert upgraded["verification"]["receipts_enabled"] is True
    assert upgraded["verification"]["receipts_render_mode"] == "delegation_only"
    assert upgraded["verification"]["judge_enabled"] is False
    assert upgraded["verification"]["judge_model_name"] is None


@pytest.mark.skipif(SCRIPT_BASH is None, reason="repo shell-script tests need Git Bash on Windows")
def test_version_41_config_moves_legacy_ragflow_settings_to_tool(tmp_path):
    """The v46 migration keeps provider settings on the RAGFlow tool entry."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    expected_version = yaml.safe_load(example_src.read_text(encoding="utf-8"))["config_version"]
    assert expected_version >= 46

    config_path = tmp_path / "config.yaml"
    legacy = {
        "config_version": 41,
        "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
        "knowledge_base": {
            "enabled": True,
            "scope_selection_enabled": True,
            "base_url": "http://legacy-ragflow:9380",
            "api_key": "$LEGACY_RAGFLOW_API_KEY",
            "page_size": 12,
        },
        "tools": [
            {
                "name": "knowledge_search",
                "group": "knowledge",
                "use": "deerflow.community.ragflow.tools:knowledge_search_tool",
                # Explicit tool values win over the legacy global value.
                "api_key": "$CURRENT_RAGFLOW_API_KEY",
            }
        ],
    }
    config_path.write_text(yaml.dump(legacy), encoding="utf-8")

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version, result.stdout + result.stderr
    assert upgraded["knowledge_base"] == {
        "enabled": True,
        "scope_selection_enabled": True,
    }
    tool = upgraded["tools"][0]
    assert tool["base_url"] == "http://legacy-ragflow:9380"
    assert tool["page_size"] == 12
    assert tool["api_key"] == "$CURRENT_RAGFLOW_API_KEY"


@pytest.mark.skipif(SCRIPT_BASH is None, reason="repo shell-script tests need Git Bash on Windows")
def test_version_41_tools_only_ragflow_config_enables_knowledge_capability(tmp_path):
    """Tools-only legacy configs must not be disabled by the new capability gate."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    expected_version = yaml.safe_load(example_src.read_text(encoding="utf-8"))["config_version"]
    assert expected_version >= 46

    config_path = tmp_path / "config.yaml"
    legacy = {
        "config_version": 41,
        "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
        # This was the documented enablement path before knowledge_base existed.
        "tools": [
            {
                "name": "knowledge_search",
                "group": "knowledge",
                "use": "deerflow.community.ragflow.tools:knowledge_search_tool",
                "base_url": "http://legacy-ragflow:9380",
                "api_key": "$RAGFLOW_API_KEY",
            }
        ],
    }
    config_path.write_text(yaml.dump(legacy), encoding="utf-8")

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr
    assert "knowledge_base.enabled set to true" in result.stdout

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version
    assert upgraded["knowledge_base"] == {
        "enabled": True,
        "scope_selection_enabled": False,
    }
    assert upgraded["tools"][0]["base_url"] == "http://legacy-ragflow:9380"


def test_version_45_tools_only_ragflow_config_runs_knowledge_migration(tmp_path):
    """The knowledge migration must run for configs at the former base version."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    expected_version = yaml.safe_load(example_src.read_text(encoding="utf-8"))["config_version"]
    assert expected_version > 45

    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        yaml.dump(
            {
                "config_version": 45,
                "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
                "tools": [
                    {
                        "name": "knowledge_search",
                        "group": "knowledge",
                        "use": "deerflow.community.ragflow.tools:knowledge_search_tool",
                        "base_url": "http://legacy-ragflow:9380",
                        "api_key": "$RAGFLOW_API_KEY",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr
    assert "knowledge_base.enabled set to true" in result.stdout

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version
    assert upgraded["knowledge_base"] == {
        "enabled": True,
        "scope_selection_enabled": False,
    }
    assert upgraded["tools"][0]["base_url"] == "http://legacy-ragflow:9380"


def test_version_45_tools_only_lightrag_config_keeps_knowledge_tool_available(tmp_path):
    """Upgrading a configured LightRAG provider must enable the new knowledge gate."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    expected_version = yaml.safe_load(example_src.read_text(encoding="utf-8"))["config_version"]
    assert expected_version > 45

    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        yaml.dump(
            {
                "config_version": 45,
                "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
                "tools": [
                    {
                        "name": "knowledge_search",
                        "group": "knowledge",
                        "use": "deerflow.community.lightrag.tools:knowledge_search_tool",
                        "base_url": "http://legacy-lightrag:9621",
                        "api_key": "$LIGHTRAG_API_KEY",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr
    assert "knowledge_base.enabled set to true" in result.stdout

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version
    assert upgraded["knowledge_base"] == {
        "enabled": True,
        "scope_selection_enabled": False,
    }

    app_config = AppConfig.model_validate(upgraded)
    tools = get_available_tools(
        groups=["knowledge"],
        include_mcp=False,
        include_upload_tool=False,
        app_config=app_config,
    )
    assert "knowledge_search" in {tool.name for tool in tools}


def test_version_45_lightrag_config_preserves_explicit_disabled_gate(tmp_path):
    """Migration must not override an operator's explicit knowledge gate value."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        yaml.dump(
            {
                "config_version": 45,
                "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
                "knowledge_base": {"enabled": False},
                "tools": [
                    {
                        "name": "knowledge_search",
                        "group": "knowledge",
                        "use": "deerflow.community.lightrag.tools:knowledge_search_tool",
                        "base_url": "http://legacy-lightrag:9621",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["knowledge_base"]["enabled"] is False

    app_config = AppConfig.model_validate(upgraded)
    tools = get_available_tools(
        groups=["knowledge"],
        include_mcp=False,
        include_upload_tool=False,
        app_config=app_config,
    )
    assert "knowledge_search" not in {tool.name for tool in tools}


def _load_repo_example() -> dict:
    """Load the real repo config.example.yaml (first-run template)."""
    example_path = Path(__file__).resolve().parents[2] / "config.example.yaml"
    with open(example_path, encoding="utf-8") as f:
        return yaml.safe_load(f) or {}


def _merge_missing(target: dict, source: dict) -> None:
    """Add-missing-keys-only recursive merge mirroring scripts/config-upgrade.sh."""
    for key, value in source.items():
        if key not in target:
            import copy

            target[key] = copy.deepcopy(value)
        elif isinstance(value, dict) and isinstance(target[key], dict):
            _merge_missing(target[key], value)


def test_security_fail_closed_bumped_config_version():
    """The example must ship security_fail_closed under a version > 26 so v26 configs upgrade."""
    example = _load_repo_example()
    assert example.get("config_version", 0) >= 27
    assert example["skill_evolution"]["security_fail_closed"] is True


def test_version_26_config_reported_outdated_against_example(caplog):
    """A version-26 user config is flagged outdated against the real example version."""
    example = _load_repo_example()
    example_version = example["config_version"]
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={"config_version": 26},
            example_config=example,
        )
        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version({"config_version": 26}, config_path)
        assert "outdated" in caplog.text
        assert "version 26" in caplog.text
        assert f"version is {example_version}" in caplog.text


def test_config_upgrade_adds_security_fail_closed_preserving_user_values():
    """config-upgrade merges security_fail_closed: true without touching existing skill_evolution values."""
    example = _load_repo_example()
    # A version-26 user who customized skill_evolution but predates the new field.
    user = {
        "config_version": 26,
        "skill_evolution": {
            "enabled": True,
            "moderation_model_name": "custom-moderation-model",
        },
    }

    _merge_missing(user, example)
    user["config_version"] = example["config_version"]

    # New persisted field is merged in with the example's fail-closed default.
    assert user["skill_evolution"]["security_fail_closed"] is True
    # The user's existing skill_evolution values are preserved unchanged.
    assert user["skill_evolution"]["enabled"] is True
    assert user["skill_evolution"]["moderation_model_name"] == "custom-moderation-model"
    assert user["config_version"] == example["config_version"]


def test_version_46_pii_enabled_config_reported_outdated_against_example(caplog):
    """token_secret became mandatory for enabled redaction in v47; a v46
    deployment with redaction on must be flagged outdated so the upgrade
    warning fires instead of a raw startup validation error with no guidance."""
    example = _load_repo_example()
    example_version = example["config_version"]
    assert example_version >= 47, "config.example.yaml must be bumped past 46 for the mandatory token_secret"
    with tempfile.TemporaryDirectory() as tmpdir:
        config_path = _make_config_files(
            Path(tmpdir),
            user_config={"config_version": 46, "pii_redaction": {"enabled": True}},
            example_config=example,
        )
        with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
            AppConfig._check_config_version({"config_version": 46}, config_path)
        assert "outdated" in caplog.text
        assert "(version 46)" in caplog.text
        assert f"version is {example_version}" in caplog.text


@pytest.mark.skipif(SCRIPT_BASH is None, reason="repo shell-script tests need Git Bash on Windows")
def test_version_46_pii_enabled_config_upgrade_generates_token_secret(tmp_path):
    """Upgrading a v46 config with redaction enabled persists a generated
    token_secret, leaving the deployment startable under the v47 mandatory
    validation instead of failing startup after the version bump."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    expected_version = yaml.safe_load(example_src.read_text(encoding="utf-8"))["config_version"]
    assert expected_version >= 47

    config_path = tmp_path / "config.yaml"
    (tmp_path / "config.example.yaml").write_text(example_src.read_text(encoding="utf-8"), encoding="utf-8")
    config_path.write_text(
        yaml.dump(
            {
                "config_version": 46,
                "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
                "pii_redaction": {"enabled": True},
            }
        ),
        encoding="utf-8",
    )

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr
    assert "token_secret generated" in result.stdout

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version
    assert upgraded["pii_redaction"]["enabled"] is True
    secret = upgraded["pii_redaction"]["token_secret"]
    assert isinstance(secret, str) and len(secret) >= 16

    # The upgraded deployment passes the now-mandatory validation.
    app_config = AppConfig.model_validate(upgraded)
    assert app_config.pii_redaction.token_secret == secret


@pytest.mark.skipif(SCRIPT_BASH is None, reason="repo shell-script tests need Git Bash on Windows")
def test_version_46_pii_disabled_config_upgrade_skips_token_secret(tmp_path):
    """The migration must not invent a secret when redaction is off."""
    import subprocess

    repo_root = Path(__file__).resolve().parents[2]
    example_src = repo_root / "config.example.yaml"
    expected_version = yaml.safe_load(example_src.read_text(encoding="utf-8"))["config_version"]
    assert expected_version >= 47

    config_path = tmp_path / "config.yaml"
    (tmp_path / "config.example.yaml").write_text(example_src.read_text(encoding="utf-8"), encoding="utf-8")
    config_path.write_text(
        yaml.dump(
            {
                "config_version": 46,
                "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
                "pii_redaction": {"enabled": False},
            }
        ),
        encoding="utf-8",
    )

    env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
    result = subprocess.run(
        [SCRIPT_BASH, str(repo_root / "scripts" / "config-upgrade.sh")],
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr

    upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    assert upgraded["config_version"] == expected_version
    assert "token_secret" not in upgraded["pii_redaction"]
    AppConfig.model_validate(upgraded)