# -*- coding: utf-8 -*-
"""第三十五批：单测隔离与观察读数内存仓储，全部离线。"""
import unit_test_guard  # noqa: F401
from datetime import datetime
import inspect
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import traceback

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from app.repo import consensus_stat_repo as csr
from app.services import consensus_stats as stats, proposal_service
from test_wiring import install_fakes, _target_fakes

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


@case("仓储四个入口均替换，签名与真实函数一致")
def _():
    originals = {name: getattr(csr, name) for name in
                 ("upsert_many", "delete_day_kind", "list_range", "count_by_kind")}
    fake = install_fakes()
    for name, original in originals.items():
        bound = getattr(csr, name)
        assert bound.__self__ is fake.consensus_stats
        def shape(fn):
            return [(p.name, p.kind, p.default) for p in inspect.signature(fn).parameters.values()]
        assert shape(original) == shape(bound), name


@case("重复读数累加轮次且保留首次时间，读取与日报都用内存")
def _():
    fake = install_fakes()
    out = {"self_decide_seen": [{"ts_code": "600000.SH", "decision": "accept",
                                 "why": "target_price", "action": "EXIT"}]}
    first, last = datetime(2026, 9, 18, 9, 31), datetime(2026, 9, 18, 9, 32)
    for now in (first, last):
        assert stats.record_round(out, {}, {}, now)["ok"]
    rows = csr.list_range(20260918, 20260918, ["auto_accept"])
    assert len(rows) == 1 and rows[0]["rounds"] == 2, rows
    assert rows[0]["first_at"] == first and rows[0]["last_at"] == last, rows
    assert stats.daily_summary(20260918)["auto_accept"] == {"rows": 1, "rounds": 2}
    assert len(fake.consensus_stats.rows) == 1


@case("整轮目标价扫描的自决读数落入内存，试算不写")
def _():
    fake, restore = _target_fakes(13.0, 12.5, params={"PMS_AUTONOMY": "full",
                                                   "PMS_SELF_DECIDE": "full"})
    try:
        proposal_service.scan_and_route(now=datetime(2026, 9, 18, 10, 30), dry_run=True)
        assert not fake.consensus_stats.rows
        result = proposal_service.scan_and_route(now=datetime(2026, 9, 18, 10, 30))
        rows = csr.list_range(20260918, 20260918, ["auto_accept"])
        assert result["ok"] and len(rows) == 1, (result, rows)
        assert rows[0]["ts_code"] == "600000.SH"
    finally:
        restore()


@case("重建覆盖先清当日无读数，保留其他日期与类别")
def _():
    install_fakes()
    for day in (17, 18):
        assert stats.record_map_cover(relevant_codes=["600000.SH", "000001.SZ"],
                                      state_codes=[], now=datetime(2026, 9, day, 6, 30))["ok"]
    assert stats.record_map_cover(relevant_codes=["600000.SH", "000001.SZ"],
                                  state_codes=["600000.SH"], now=datetime(2026, 9, 18, 8, 40))["ok"]
    today = csr.list_range(20260918, 20260918, ["tech_noread"])
    assert [r["ts_code"] for r in today] == ["000001.SZ"], today
    assert len(csr.list_range(20260917, 20260917, ["tech_noread"])) == 2
    assert stats.daily_summary(20260918)["map_cover"] == {"rows": 1, "rounds": 2}


@case("重新装桩清空读数，不把前一用例的数据带入下一例")
def _():
    first = install_fakes()
    csr.upsert_many([{"stat_date": 20260918, "ts_code": "600000.SH", "kind": "auto_accept"}])
    second = install_fakes()
    assert first.consensus_stats.rows and not second.consensus_stats.rows
    assert csr.list_range(20260918, 20260918) == []


def _child(source, env=None, dotenv=None):
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "probe.py"
        path.write_text("import sys\nsys.path.insert(0, %r)\n" % str(ROOT) + source)
        if dotenv:
            (Path(tmp) / ".env").write_text(dotenv)
        return subprocess.run([sys.executable, str(ROOT / "scripts/unit_test_guard.py"), str(path)],
                              cwd=tmp, env={**os.environ, **(env or {})}, text=True,
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20)


@case("联网异常被吞掉也必须失败，阻断发生在建立连接之前")
def _():
    r = _child("import socket\ntry:\n    socket.socket().connect(('127.0.0.1', 1))\n"
               "except Exception:\n    pass\nprint('swallowed')\n")
    assert "swallowed" in r.stdout and r.returncode == 1, r
    assert "TEST ISOLATION FAILED" in r.stderr and "socket.connect" in r.stderr, r.stderr


@case("测试入口不继承生产参数，也不读启动目录的 .env")
def _():
    r = _child("from config.settings import settings\n"
               "assert settings.PMS_TOTAL_SCALE == 2000000\nprint('clean')\n",
               env={"PMS_TOTAL_SCALE": "987654"}, dotenv="PMS_TOTAL_SCALE=123456\n")
    assert r.returncode == 0 and "clean" in r.stdout, (r.stdout, r.stderr)


@case("所有业务测试独立运行时也安装联网护栏")
def _():
    import ast
    import run_tests
    for name in run_tests.SUITES:
        tree = ast.parse((ROOT / "scripts" / name).read_text())
        imports = [n for n in tree.body if isinstance(n, (ast.Import, ast.ImportFrom))]
        assert isinstance(imports[0], ast.Import) and imports[0].names[0].name == "unit_test_guard", name


if __name__ == "__main__":
    ok = 0
    for name, fn in RESULTS:
        try:
            fn()
            ok += 1
            print("  ok  " + name)
        except Exception:
            print("  FAIL " + name)
            traceback.print_exc()
    print("ALL PASS (%d cases)" % ok if ok == len(RESULTS) else "FAILED %d/%d" % (len(RESULTS)-ok, len(RESULTS)))
    sys.exit(0 if ok == len(RESULTS) else 1)