qlib/backend/tests/test_config.py

"""配置层测试：默认 MySQL URL 组装（config.yaml database.mysql）、.env 加载约定、
SQLite 兜底逻辑。conftest 已强制 DATABASE_URL=tmp sqlite，本文件内用
monkeypatch/cache_clear 单独验证「无 env 时走 mysql 段」的分支（只组装不连接）。"""

from __future__ import annotations

import os

import pytest
import yaml
from app.core.config import CONFIG_PATH, PROJECT_ROOT, _load_env_file, get_settings


def test_project_root_points_to_repo_root() -> None:
    # config.py 位于 <root>/backend/app/core/ → 项目根应含 config.yaml 与 backend/
    assert (PROJECT_ROOT / "config.yaml").exists()
    assert (PROJECT_ROOT / "backend").is_dir()


def test_test_runner_isolation_uses_tmp_sqlite() -> None:
    """conftest 强制每进程唯一 /tmp sqlite，测试绝不触碰 MySQL 开发库。"""
    settings = get_settings()
    assert settings.database_url.startswith("sqlite:////tmp/qlib-pytest-")


def test_mysql_default_from_config_yaml(monkeypatch) -> None:
    """未设 DATABASE_URL 时，config.yaml database.mysql 段组装 MySQL URL（不连接）。

    期望值直接取自 config.yaml，而不硬编码部署地址：切库（192.168.1.10 → 127.0.0.1）
    属于配置变更，不应让本用例失效。
    """
    mysql = yaml.safe_load(CONFIG_PATH.read_text(encoding="utf-8"))["database"]["mysql"]
    monkeypatch.delenv("DATABASE_URL", raising=False)
    get_settings.cache_clear()
    try:
        url = get_settings().database_url
    finally:
        get_settings.cache_clear()
    assert url.startswith(f"mysql+pymysql://{mysql['user']}:")
    expected = (
        f"{mysql['host']}:{mysql['port']}/{mysql['db']}?charset={mysql['charset']}"
    )
    # 断言只比较 @ 之后的部分：失败时 pytest 不会把含密码的 userinfo 打进日志
    assert expected in url.split("@")[-1]


def test_build_mysql_url(monkeypatch) -> None:
    from app.core.config import _build_mysql_url

    monkeypatch.setenv("MYSQL_PASSWORD", "p@ss:word")
    cfg = {
        "enabled": True,
        "host": "10.0.0.2",
        "port": 3307,
        "db": "q",
        "user": "u",
        "password_env": "MYSQL_PASSWORD",
    }
    url = _build_mysql_url(cfg)
    assert url == "mysql+pymysql://u:p%40ss%3Aword@10.0.0.2:3307/q?charset=utf8mb4"
    # 未启用 / 缺字段 → None（回退 sqlite 兜底）
    assert _build_mysql_url({**cfg, "enabled": False}) is None
    assert _build_mysql_url({**cfg, "host": None}) is None
    assert _build_mysql_url(None) is None


def test_mysql_url_password_optional(monkeypatch) -> None:
    """密码留空（未设 env）也可组装 URL —— 方便仅内网/免密场景。"""
    from app.core.config import _build_mysql_url

    monkeypatch.delenv("MYSQL_PASSWORD", raising=False)
    cfg = {"enabled": True, "host": "h", "db": "d", "user": "u"}
    assert _build_mysql_url(cfg) == "mysql+pymysql://u@h:3306/d?charset=utf8mb4"


def test_sqlite_url_normalization_and_fallback() -> None:
    """仅 sqlite 相对路径被解析为项目根绝对路径；其它 URL 原样透传。"""
    from app.core.config import _normalize_sqlite_url

    assert _normalize_sqlite_url("mysql+pymysql://u:p@h/d") == "mysql+pymysql://u:p@h/d"
    abs_url = _normalize_sqlite_url("sqlite:///./data/quant.db")
    assert abs_url.startswith("sqlite:///")
    # 绝对路径 sqlite 不再重复解析
    assert _normalize_sqlite_url("sqlite:////abs/x.db") == "sqlite:////abs/x.db"


def test_settings_loaded_from_config_yaml() -> None:
    settings = get_settings()
    assert settings.app_name == "qlib-platform"
    assert settings.api_prefix == "/api"
    assert settings.data_source_primary == "tushare"
    assert settings.data_source_fallback == "sina"


def test_env_file_loading(tmp_path, monkeypatch) -> None:
    monkeypatch.delenv("TUSHARE_TOKEN", raising=False)
    env_file = tmp_path / ".env"
    env_file.write_text(
        "# comment\nTUSHARE_TOKEN=demo-token\nDATABASE_URL=sqlite:////tmp/x.db\n",
        encoding="utf-8",
    )
    _load_env_file(env_file)
    assert os.environ.get("TUSHARE_TOKEN") == "demo-token"

    # 不覆盖已存在的环境变量（shell 环境优先）
    monkeypatch.setenv("TUSHARE_TOKEN", "kept")
    _load_env_file(env_file)
    assert os.environ["TUSHARE_TOKEN"] == "kept"


def test_llm_from_config_yaml() -> None:
    """URL 与模型名来自 config.yaml 明文，Key 仅来自环境（当前未设置）。"""
    settings = get_settings()
    assert settings.llm_model == "qwen-plus"
    assert settings.llm_base_url == "https://dashscope.aliyuncs.com/compatible-mode/v1"
    assert settings.llm_api_key == ""  # .env 未提供 → 空，Agent 层给出引导提示


def test_llm_env_overrides_yaml(monkeypatch) -> None:
    monkeypatch.setenv("LLM_MODEL", "env-model")
    monkeypatch.setenv("LLM_BASE_URL", "https://example.com/v1")
    monkeypatch.setenv("LLM_API_KEY", "sk-test")
    get_settings.cache_clear()  # Settings 有 lru_cache，刷新以读取新环境变量
    try:
        s = get_settings()
    finally:
        get_settings.cache_clear()
    assert s.llm_model == "env-model"
    assert s.llm_base_url == "https://example.com/v1"
    assert s.llm_api_key == "sk-test"


class TestForbiddenDbTarget:
    """用户约束：**禁止使用 192.168.1.10 作为 DB 目标**（只允许本机 MariaDB）。

    连错库是「最危险的静默错误」：不报错、界面正常，只是数据在另一台机器上读写。
    因此这里是硬失败（抛错），不是警告；本类把守卫生效与放行口子都钉住。
    """

    def _clear(self):
        get_settings.cache_clear()

    def test_forbidden_host_rejected_from_env_url(self, monkeypatch) -> None:
        monkeypatch.setenv(
            "DATABASE_URL", "mysql+pymysql://qlib:pw@192.168.1.10:3306/qlib"
        )
        monkeypatch.delenv("QLIB_FORBIDDEN_DB_HOSTS", raising=False)
        self._clear()
        try:
            with pytest.raises(RuntimeError, match="192.168.1.10"):
                get_settings()
        finally:
            self._clear()

    def test_forbidden_host_rejected_from_config_yaml(self, monkeypatch, tmp_path) -> None:
        """即使走 config.yaml 的 mysql 段（而非 env）也必须拦住。"""
        import app.core.config as config_mod

        cfg = {
            "database": {
                "mysql": {
                    "enabled": True,
                    "host": "192.168.1.10",
                    "port": 3306,
                    "db": "qlib",
                    "user": "qlib",
                    "charset": "utf8mb4",
                }
            }
        }
        path = tmp_path / "config.yaml"
        path.write_text(yaml.safe_dump(cfg, allow_unicode=True), encoding="utf-8")
        monkeypatch.delenv("DATABASE_URL", raising=False)
        monkeypatch.delenv("QLIB_FORBIDDEN_DB_HOSTS", raising=False)
        monkeypatch.setattr(config_mod, "CONFIG_PATH", path)
        self._clear()
        try:
            with pytest.raises(RuntimeError, match="已被禁止"):
                get_settings()
        finally:
            self._clear()

    def test_local_host_allowed(self, monkeypatch) -> None:
        monkeypatch.setenv(
            "DATABASE_URL", "mysql+pymysql://qlib:pw@127.0.0.1:3306/qlib"
        )
        self._clear()
        try:
            assert get_settings().database_url.startswith("mysql+pymysql://qlib")
        finally:
            self._clear()

    def test_forbidden_list_is_env_overridable(self, monkeypatch) -> None:
        """显式放开的口子必须存在（空串＝不限制），否则紧急情况下无法操作。"""
        monkeypatch.setenv(
            "DATABASE_URL", "mysql+pymysql://qlib:pw@192.168.1.10:3306/qlib"
        )
        monkeypatch.setenv("QLIB_FORBIDDEN_DB_HOSTS", "")
        self._clear()
        try:
            assert "192.168.1.10" in get_settings().database_url
        finally:
            self._clear()

    def test_sqlite_and_other_hosts_unaffected(self, monkeypatch) -> None:
        """SQLite（无 host）与其它主机不受影响 —— 守卫只拦精确匹配的禁用主机。"""
        from app.core.config import assert_db_target_allowed

        monkeypatch.delenv("QLIB_FORBIDDEN_DB_HOSTS", raising=False)
        assert_db_target_allowed("sqlite:////tmp/x.db")
        assert_db_target_allowed("mysql+pymysql://u:p@127.0.0.1:3306/qlib")
        assert_db_target_allowed("mysql+pymysql://u:p@db.internal:3306/qlib")
        assert_db_target_allowed("")
        with pytest.raises(RuntimeError):
            assert_db_target_allowed("mysql+pymysql://u:p@192.168.1.10:3306/qlib")

    def test_guard_only_matches_host_not_userinfo(self, monkeypatch) -> None:
        """只比 host：把禁用 IP 写进用户名/密码不应被误判（避免误报挡住正常连接）。"""
        from app.core.config import assert_db_target_allowed

        monkeypatch.delenv("QLIB_FORBIDDEN_DB_HOSTS", raising=False)
        assert_db_target_allowed("mysql+pymysql://192.168.1.10:pw@127.0.0.1:3306/qlib")