tradingSystem/scripts/test_batch16_units.py

# -*- coding: utf-8 -*-
"""
第十六批: 软归档 (archived_at) 的 SQL 与口径守卫 —— 不连库, 打桩捕获 SQL
=====================================================================
软归档是「把终态记录从在办/在途视图移除、但绝不删行」。这批用例不连库, 用打桩截住
pms_repo 的 execute / fetch_all, 把真实生成的 SQL 抓出来核对三件事:
  1. archive_* 生成的 UPDATE 只碰一张表 (过 assert_single_table 单表守卫), 只归档**终态**
     (status IN 终态集) 且只动 archived_at 仍为空的行;
  2. unarchive_* 只把 archived_at 清空, 不按状态过滤, 也过守卫;
  3. list_* 默认 include_archived=False 时 SQL 带 archived_at IS NULL, 置 True 时不带 ——
     保证已归档默认不出现在视图里, 而「显示已完成」能拉回来。
四类记录 (命令/策略/指令/提议) 的终态集与后端 archive_* 和前端可移除判据是同一份, 一并钉死。
运行: python scripts/test_batch16_units.py
"""
import os
import sys
import traceback

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.repo import pms_repo as R                                              # noqa: E402
from app.db.session import assert_single_table                                 # noqa: E402

RESULTS = []


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


def _cap_execute():
    cap = {}

    def fake(sql, params=None):
        cap["sql"] = sql
        cap["params"] = params or {}
        return 1
    return cap, fake


def _cap_fetch():
    cap = {}

    def fake(sql, params=None):
        cap["sql"] = sql
        cap["params"] = params or {}
        return []
    return cap, fake


def _run_capturing_execute(call):
    cap, fake = _cap_execute()
    orig = R.execute
    R.execute = fake
    try:
        call()
    finally:
        R.execute = orig
    return cap


@case("归档·archive_command 只碰单表 / 只归终态 / 带 archived_at IS NULL")
def _():
    cap, fake = _cap_execute()
    orig = R.execute
    R.execute = fake
    try:
        n = R.archive_command("CMD_X")
    finally:
        R.execute = orig
    assert n == 1, n
    assert_single_table(cap["sql"])                       # 守卫必须接受, 否则真机一执行就抛
    assert "UPDATE pms_command" in cap["sql"], cap["sql"]
    assert "archived_at IS NULL" in cap["sql"], cap["sql"]
    assert {"DONE", "CANCELLED", "SUPERSEDED"} <= set(cap["params"].values()), cap["params"]


@case("归档·四类记录的终态集与设计一致 (改一处必须同时改前端可移除判据)")
def _():
    assert R._ARCHIVE_TERMINAL["pms_command"] == ("DONE", "CANCELLED", "SUPERSEDED")
    assert R._ARCHIVE_TERMINAL["pms_strategy"] == ("CANCELLED", "DONE")
    assert R._ARCHIVE_TERMINAL["pms_instruction"] == ("CONFIRMED", "EXPIRED", "CANCELLED", "REJECTED")
    assert R._ARCHIVE_TERMINAL["pms_proposal"] == ("ACCEPTED", "DECLINED", "EXPIRED")


@case("归档·strategy/instruction/proposal 三个 archive_* 都单表 + 终态 + archived_at IS NULL")
def _():
    for fn, table, term in [
        (R.archive_strategy, "pms_strategy", {"CANCELLED", "DONE"}),
        (R.archive_instruction, "pms_instruction",
         {"CONFIRMED", "EXPIRED", "CANCELLED", "REJECTED"}),
        (R.archive_proposal, "pms_proposal", {"ACCEPTED", "DECLINED", "EXPIRED"}),
    ]:
        cap = _run_capturing_execute(lambda fn=fn: fn("ID_X"))
        assert_single_table(cap["sql"])
        assert f"UPDATE {table}" in cap["sql"], cap["sql"]
        assert "archived_at IS NULL" in cap["sql"], cap["sql"]
        assert term <= set(cap["params"].values()), (table, cap["params"])


@case("恢复·unarchive_* 只清 archived_at, 不按状态过滤, 过守卫")
def _():
    cap = _run_capturing_execute(lambda: R.unarchive_command("CMD_X"))
    assert_single_table(cap["sql"])
    assert "archived_at = NULL" in cap["sql"], cap["sql"]
    assert "status" not in cap["sql"], "恢复显示不该按状态过滤 (已撤下也能恢复): " + cap["sql"]


@case("列表·默认排除已归档 (SQL 带 archived_at IS NULL); include_archived=True 时不带")
def _():
    for lister in (R.list_commands, R.list_instructions, R.list_proposals, R.list_strategies):
        cap, fake = _cap_fetch()
        orig = R.fetch_all
        R.fetch_all = fake
        try:
            lister()                                   # 默认 include_archived=False
            sql_default = cap["sql"]
            lister(include_archived=True)
            sql_all = cap["sql"]
        finally:
            R.fetch_all = orig
        assert "archived_at IS NULL" in sql_default, (lister.__name__, sql_default)
        assert "archived_at IS NULL" not in sql_all, (lister.__name__, sql_all)
        assert_single_table(sql_default)
        assert_single_table(sql_all)


def main():
    passed, failed = 0, []
    for name, fn in RESULTS:
        try:
            fn()
            passed += 1
            print(f"  ✓ {name}")
        except Exception as e:
            failed.append((name, e))
            print(f"  ✗ {name}: {type(e).__name__}: {e}")
            traceback.print_exc()
    print()
    if failed:
        print(f"FAILED {len(failed)}/{len(RESULTS)}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()