sanguo_vnpy_v2/tests/trader/test_qmt_identity.py

# -*- coding: utf-8 -*-
"""QMT 部署身份单一权威源契约（P2-18 根治, 2026-10-05）——config/qmt_identity.json。

三组钉（audit/20261005_qmt_identity_env_plan.md §四）:
① 装载链语义: env SANGUO_QMT_ACCOUNT / BIGQMT_ACCOUNT_ID(兼容别名)覆盖 > json;
   SANGUO_IDENTITY_JSON 显式指路不存在=RuntimeError(显式覆盖不静默回退)。
② 单源一致性(anti-drift, 休市表 CI 同步测先例): identity.default_account ≡
   live.yaml.account ≡ ∈ data_platform.yaml watch_accounts ≡ gate_common.ACCOUNT
   ≡ 各腿镜像解析值(env 清空)。
③ 字面量 grep-pin: "66639661" 只允许出现在 config 三件套; 生产码面
   (scripts/、sanguo_*)零字面量——任何新增扩散直接被本单测抓住。
"""
import json
import sys
from pathlib import Path

import pytest
import yaml

_REPO = Path(__file__).resolve().parents[2]
_CFG = _REPO / "config"
_GATE_DIR = _REPO / "scripts" / "qmt_relogin"

if str(_GATE_DIR) not in sys.path:
    sys.path.insert(0, str(_GATE_DIR))

import qmt_gate_common as gate  # noqa: E402

_ENV_NAMES = ("SANGUO_QMT_ACCOUNT", "BIGQMT_ACCOUNT_ID", "ACCOUNT_ID")


def _ident() -> dict:
    return json.loads(
        (_CFG / "qmt_identity.json").read_text(encoding="utf-8"))


# ---------------- ① 装载链语义 ----------------

def test_load_identity_env_overrides(monkeypatch):
    for n in _ENV_NAMES + ("SANGUO_IDENTITY_JSON",):
        monkeypatch.delenv(n, raising=False)
    base = gate.load_identity()
    assert base["default_account"] == _ident()["default_account"]
    monkeypatch.setenv("SANGUO_QMT_ACCOUNT", "123")
    assert gate.load_identity()["default_account"] == "123"
    monkeypatch.delenv("SANGUO_QMT_ACCOUNT")
    monkeypatch.setenv("BIGQMT_ACCOUNT_ID", "456")  # 兼容别名
    assert gate.load_identity()["default_account"] == "456"


def test_load_identity_explicit_path_missing_fails(monkeypatch):
    for n in _ENV_NAMES:
        monkeypatch.delenv(n, raising=False)
    monkeypatch.setenv("SANGUO_IDENTITY_JSON", "/nonexistent/x.json")
    with pytest.raises(RuntimeError, match="SANGUO_IDENTITY_JSON"):
        gate.load_identity()


def test_module_account_bound_to_identity():
    assert gate.ACCOUNT == _ident()["default_account"]


# ---------------- ② 单源一致性 ----------------

def test_single_source_agreement(monkeypatch):
    for n in _ENV_NAMES:
        monkeypatch.delenv(n, raising=False)
    monkeypatch.setenv("SANGUO_QMT_PATH", "/tmp")  # 防扫 C:\
    ident = _ident()

    live = yaml.safe_load((_CFG / "live.yaml").read_text(encoding="utf-8"))
    assert live["account"] == ident["default_account"]

    dp = yaml.safe_load(
        (_CFG / "data_platform.yaml").read_text(encoding="utf-8"))
    assert ident["default_account"] in dp["live_trading"]["watch_accounts"]

    assert gate.ACCOUNT == ident["default_account"]

    from sanguo_qmt_bridge import xt_gateway
    assert xt_gateway.ACCOUNT_ID == ident["default_account"]

    from sanguo_trader import qmt_gateway_client as qgc
    setting = qgc._QmtExec._setting()
    assert setting["交易账号"] == ident["default_account"]

    sys.path.insert(0, str(_REPO / "scripts"))
    sys.path.insert(0, str(_REPO / "scripts" / "pipeline"))
    try:
        import bridge_switch_ops as bso  # noqa: E402
        assert bso._qmt_account() == ident["default_account"]

        import is_price_source_vps as ips  # noqa: E402
        assert ips._identity_default_account() == ident["default_account"]
    finally:
        sys.path.remove(str(_REPO / "scripts"))
        sys.path.remove(str(_REPO / "scripts" / "pipeline"))


# ---------------- ③ 字面量 grep-pin ----------------

_CODE_ROOTS = ("scripts", "sanguo_live", "sanguo_trader", "sanguo_qmt_bridge",
               "sanguo_portfolio", "sanguo_data", "sanguo_factor")
_ALLOWED_EXACT = {
    "config/qmt_identity.json", "config/live.yaml", "config/data_platform.yaml",
}


def test_account_literal_confined_to_config():
    """生产码面零账号字面量: 只 config 三件套可含(P2-18 前码面曾有 17 处)。"""
    offenders = []
    for root in _CODE_ROOTS:
        base = _REPO / root
        if not base.is_dir():
            continue
        for p in base.rglob("*"):
            if (not p.is_file() or p.suffix not in
                    {".py", ".ps1", ".xml", ".md", ".yaml", ".json"}):
                continue
            rel = p.relative_to(_REPO).as_posix()
            if rel in _ALLOWED_EXACT:
                continue
            try:
                text = p.read_text(encoding="utf-8", errors="ignore")
            except OSError:
                continue
            if "66639661" in text:
                offenders.append(rel)
    assert offenders == [], "账号字面量泄漏进码面: %s" % offenders