144 lines
5.4 KiB
Python
144 lines
5.4 KiB
Python
# -*- 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_identity_file_ascii_only():
|
||
"""④ 文件格式钉: 纯 ASCII（中文一律 \\u 转义）。
|
||
|
||
PS5.1 Get-Content 默认按 ANSI/GBK 解无 BOM 文件——文件含 UTF-8 中文
|
||
原始字节时 ConvertFrom-Json 直接 FormatException（2026-10-06 xt_eod
|
||
实证: BIGQMT_ACCOUNT_ID 注入失败→桥 shim import 期硬错 account_id
|
||
required→universe=0→rc=1）。纯 ASCII 双端等值: Python json.load 与
|
||
PS5.1 ConvertFrom-Json 都原生解 \\u 转义。phase-③ 生产 wrapper 读此
|
||
文件同样依赖本钉。
|
||
"""
|
||
raw = (_CFG / "qmt_identity.json").read_bytes()
|
||
raw.decode("ascii") # 非 ASCII 字节 = UnicodeDecodeError 直接红
|
||
|
||
|
||
# ---------------- ② 单源一致性 ----------------
|
||
|
||
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
|