post_clear_check/code/post_clear_check/tests/test_gate.py

"""批次门禁：开跑条件、日切报表提前出现、同一输入重复调用。"""

import asyncio
from datetime import date, datetime

import pytest

from post_clear_check.gate import (
    CARD_KIND_BATCH,
    FORMULA_EARLY_REPORT,
    FORMULA_ISSUE_BASELINE,
    FORMULA_ISSUE_PUBLISH,
    FORMULA_SALES_PUBLISH,
    FORMULA_SALES_UPSTREAM,
    FORMULA_TRS_DATE,
    FORMULA_TRS_VERSION,
    HEDGE_SQL,
    ISSUE_REPORT_TABLES,
    NO_HEDGE_BATCH_ID,
    NO_HEDGE_END_TIME,
    NO_HEDGE_REVISION,
    SALES_REPORT_TABLES,
    VERDICT_ERROR,
    Batch,
    BASIS_EARLY_REPORT,
    BatchRowError,
    GateScan,
    IssueRound,
    PublishedFacts,
    ReadyRun,
    batch_sql,
    evaluate_gate,
    evaluate_published,
    parse_batch_row,
    report_date_sql,
    scan_gate,
)
from post_clear_check.session import StatementRejected, assert_business_statement

TRADE_DAY = date(2026, 10, 17)
BUSINESS_DAY = date(2026, 10, 17)
ISSUE_END = datetime(2026, 10, 17, 16, 5, 0)
SALES_END = datetime(2026, 10, 17, 16, 8, 0)
TRS_END = datetime(2026, 10, 17, 16, 12, 0)
OLDER_END = datetime(2026, 10, 17, 15, 0, 0)


def _batch(
    *,
    batch_id: int,
    batch_no: str,
    status: str,
    end_time: datetime | None,
    revision: int | None,
    market: str = "a_share",
    trade_date: date = TRADE_DAY,
    business_date: date = BUSINESS_DAY,
) -> Batch:
    return Batch(
        id=batch_id,
        batch_no=batch_no,
        market=market,
        trade_date=trade_date,
        business_date=business_date,
        status=status,
        end_time=end_time,
        applied_revision=revision,
    )


def _issue_done(batch_id: int = 9, revision: int = 3) -> Batch:
    return _batch(
        batch_id=batch_id,
        batch_no=f"IB{batch_id}",
        status="completed",
        end_time=ISSUE_END,
        revision=revision,
    )


def _sales_done(batch_id: int = 4, revision: int = 2) -> Batch:
    return _batch(
        batch_id=batch_id,
        batch_no=f"SB{batch_id}",
        status="completed",
        end_time=SALES_END,
        revision=revision,
    )


def test_issue_completed_sales_open_does_not_start() -> None:
    """发行已完成、销售未完成：不开跑。已完成一侧的报表不单独构成错误。"""

    scan = evaluate_gate(
        issue_batches=(
            _issue_done(),
            _batch(
                batch_id=1,
                batch_no="IB1",
                status="completed",
                end_time=OLDER_END,
                revision=1,
            ),
        ),
        sales_batches=(
            _batch(
                batch_id=8,
                batch_no="SB-OPEN",
                status="running",
                end_time=None,
                revision=None,
            ),
        ),
        trs_batches=(),
        hedge_trade_dates=set(),
        issue_report_dates={BUSINESS_DAY},
        sales_report_dates=set(),
    )
    assert scan.ready == ()
    assert scan.lines == ()


def test_both_completed_without_hedge_starts_with_zero_trs() -> None:
    """两端都完成、无对冲：开跑，TRS 批次 id 为 0。同日取 id 最大的完成批次。"""

    scan = evaluate_gate(
        issue_batches=(
            _batch(
                batch_id=1,
                batch_no="IB1",
                status="completed",
                end_time=OLDER_END,
                revision=1,
            ),
            _issue_done(batch_id=9, revision=3),
            _batch(
                batch_id=12,
                batch_no="IB-NEWER",
                status="running",
                end_time=None,
                revision=None,
            ),
        ),
        sales_batches=(
            _batch(
                batch_id=2,
                batch_no="SB2",
                status="completed",
                end_time=OLDER_END,
                revision=1,
            ),
            _sales_done(batch_id=4, revision=2),
        ),
        trs_batches=(
            _batch(
                batch_id=7,
                batch_no="TB7",
                status="completed",
                end_time=TRS_END,
                revision=5,
            ),
        ),
        hedge_trade_dates=set(),
        issue_report_dates={BUSINESS_DAY},
        sales_report_dates={BUSINESS_DAY},
    )
    assert scan.lines == ()
    assert len(scan.ready) == 1
    ready = scan.ready[0]
    assert ready.market == "a_share"
    assert ready.trade_date == TRADE_DAY
    assert ready.business_date == BUSINESS_DAY
    assert ready.issue_batch_id == 9
    assert ready.sales_batch_id == 4
    assert ready.trs_batch_id == NO_HEDGE_BATCH_ID
    assert ready.issue_end_time == ISSUE_END
    assert ready.sales_end_time == SALES_END
    assert ready.trs_end_time == NO_HEDGE_END_TIME
    assert ready.issue_applied_revision == 3
    assert ready.sales_applied_revision == 2
    assert ready.trs_applied_revision == NO_HEDGE_REVISION


def test_hedge_without_trs_does_not_start() -> None:
    """有对冲且 TRS 未完成：不开跑。"""

    scan = evaluate_gate(
        issue_batches=(_issue_done(),),
        sales_batches=(_sales_done(),),
        trs_batches=(
            _batch(
                batch_id=3,
                batch_no="TB-OPEN",
                status="running",
                end_time=None,
                revision=None,
            ),
        ),
        hedge_trade_dates={TRADE_DAY},
        issue_report_dates=set(),
        sales_report_dates=set(),
    )
    assert scan.ready == ()
    assert scan.lines == ()
    assert len(scan.blocked) == 1
    assert scan.blocked[0].market == "a_share"
    assert scan.blocked[0].trade_date == TRADE_DAY


def test_hedge_with_completed_trs_uses_trs_batch() -> None:
    """有对冲且 TRS 已完成：开跑结构带上 TRS 的批次、结束时间和修订号。"""

    scan = evaluate_gate(
        issue_batches=(_issue_done(),),
        sales_batches=(_sales_done(),),
        trs_batches=(
            _batch(
                batch_id=3,
                batch_no="TB3",
                status="completed",
                end_time=TRS_END,
                revision=5,
            ),
        ),
        hedge_trade_dates={TRADE_DAY},
        issue_report_dates=set(),
        sales_report_dates=set(),
    )
    assert scan.lines == ()
    ready = scan.ready[0]
    assert ready.trs_batch_id == 3
    assert ready.trs_end_time == TRS_END
    assert ready.trs_applied_revision == 5


def test_report_without_completed_batch_is_error_and_does_not_start() -> None:
    """报表已有行且批次未完成：一条 error，没有开跑结构。"""

    scan = evaluate_gate(
        issue_batches=(_issue_done(),),
        sales_batches=(
            _batch(
                batch_id=6,
                batch_no="SB-OPEN",
                status="running",
                end_time=None,
                revision=None,
            ),
        ),
        trs_batches=(),
        hedge_trade_dates=set(),
        issue_report_dates=set(),
        sales_report_dates={BUSINESS_DAY},
    )
    assert scan.ready == ()
    assert len(scan.lines) == 1
    line = scan.lines[0]
    assert line.verdict == VERDICT_ERROR
    assert line.card_kind == CARD_KIND_BATCH
    assert line.biz_key == "SB-OPEN"
    assert line.formula_text == FORMULA_EARLY_REPORT
    assert line.calc_text == "营业日 2026-10-17 已有报表行，market = a_share"
    assert line.calc_value is None
    assert line.check_value is None
    assert line.check_text == "db_sales.clearing_batch.status = running，销售，batch_no = SB-OPEN"
    assert line.basis == BASIS_EARLY_REPORT
    assert line.market == "a_share"
    assert line.business_date == BUSINESS_DAY


def test_report_without_any_batch_names_the_gap() -> None:
    """还没有批次时，说明里写没有完成批次，业务键用营业日。"""

    scan = evaluate_gate(
        issue_batches=(),
        sales_batches=(),
        trs_batches=(),
        hedge_trade_dates=set(),
        issue_report_dates={BUSINESS_DAY},
        sales_report_dates=set(),
    )
    assert scan.ready == ()
    assert len(scan.lines) == 1
    line = scan.lines[0]
    assert line.verdict == VERDICT_ERROR
    assert line.biz_key == "2026-10-17"
    assert line.check_text == "db_issue.clearing_batch.status，发行，没有完成批次"
    assert "销售" not in line.check_text


def test_other_market_does_not_satisfy_the_same_day() -> None:
    """同一交易日的另一个市场完成，不能算本市场开跑。"""

    scan = evaluate_gate(
        issue_batches=(_issue_done(),),
        sales_batches=(_sales_done(batch_id=4, revision=2),),
        trs_batches=(),
        hedge_trade_dates=set(),
        issue_report_dates=set(),
        sales_report_dates=set(),
    )
    sales_other = _batch(
        batch_id=4,
        batch_no="SB4",
        status="completed",
        end_time=SALES_END,
        revision=2,
        market="hk",
    )
    blocked = evaluate_gate(
        issue_batches=(_issue_done(),),
        sales_batches=(sales_other,),
        trs_batches=(),
        hedge_trade_dates=set(),
        issue_report_dates=set(),
        sales_report_dates=set(),
    )
    assert len(scan.ready) == 1
    assert blocked.ready == ()
    assert blocked.lines == ()


def test_same_inputs_return_the_same_scan() -> None:
    """同一组 end_time 调用两次：返回内容相同。"""

    arguments = dict(
        issue_batches=(_issue_done(),),
        sales_batches=(_sales_done(),),
        trs_batches=(),
        hedge_trade_dates=set(),
        issue_report_dates=set(),
        sales_report_dates=set(),
    )
    first = evaluate_gate(**arguments)
    second = evaluate_gate(**arguments)
    assert first == second
    assert first.ready[0].issue_end_time == ISSUE_END
    assert first.ready[0].sales_end_time == SALES_END
    assert first.ready[0].trs_end_time == NO_HEDGE_END_TIME


def test_completed_batch_missing_clock_is_rejected() -> None:
    """完成批次没有结束时间时不能编出幂等键。"""

    with pytest.raises(BatchRowError, match="end_time"):
        evaluate_gate(
            issue_batches=(
                _batch(
                    batch_id=9,
                    batch_no="IB9",
                    status="completed",
                    end_time=None,
                    revision=3,
                ),
            ),
            sales_batches=(_sales_done(),),
            trs_batches=(),
            hedge_trade_dates=set(),
            issue_report_dates=set(),
            sales_report_dates=set(),
        )


def test_gate_statements_are_single_reads() -> None:
    """门禁语句都是单条 SELECT，表名来自清单。"""

    issue_sql = batch_sql("db_issue.clearing_batch")
    sales_sql = batch_sql("db_sales.clearing_batch")
    trs_sql = batch_sql("db_trs.clearing_batch")
    # 销售、TRS 没有 applied_revision 列。TRS 也没有 market 列。
    assert "applied_revision" in issue_sql
    assert "0 AS applied_revision" in sales_sql
    assert "market," in sales_sql
    assert "'a_share' AS market" in trs_sql
    assert "0 AS applied_revision" in trs_sql
    assert "WHERE business_date = %s" in issue_sql
    assert "WHERE business_date = %s" in sales_sql
    assert "WHERE business_date = %s" in trs_sql
    statements = [
        issue_sql,
        sales_sql,
        trs_sql,
        HEDGE_SQL,
        *(report_date_sql(table) for table in ISSUE_REPORT_TABLES),
        *(report_date_sql(table) for table in SALES_REPORT_TABLES),
    ]
    for sql in statements:
        assert_business_statement(sql)
    with pytest.raises(BatchRowError):
        batch_sql("db_issue.clearing_batch; SELECT 1")
    with pytest.raises(StatementRejected):
        assert_business_statement("INSERT INTO db_issue.clearing_batch (id) VALUES (1)")


def test_parse_batch_row_keeps_null_clock_on_open_batch() -> None:
    """未完成批次的结束时间可以空着，不能把整行丢弃。"""

    parsed = parse_batch_row((6, "SB-OPEN", "a_share", TRADE_DAY, BUSINESS_DAY, "running", None, None))
    assert parsed.id == 6
    assert parsed.end_time is None
    assert parsed.applied_revision is None


class _ScriptedGateway:
    """按语句返回预设行，并记下实际发出的语句。"""

    def __init__(self) -> None:
        self.statements: list[tuple[str, object]] = []

    async def execute(self, sql: str, args: object = None) -> list[tuple[object, ...]]:
        assert_business_statement(sql)
        self.statements.append((sql, args))
        if sql == HEDGE_SQL:
            return []
        if sql.startswith("SELECT DISTINCT business_date FROM "):
            return []
        if sql.startswith("SELECT id, batch_no"):
            if "db_issue.clearing_batch" in sql:
                return [(9, "IB9", "a_share", TRADE_DAY, BUSINESS_DAY, "completed", ISSUE_END, 3)]
            if "db_sales.clearing_batch" in sql:
                return [(4, "SB4", "a_share", TRADE_DAY, BUSINESS_DAY, "completed", SALES_END, 2)]
            return []
        # 已完成轮次上的是/否检查另行取数。这例只钉开跑条件，那些表按空表返回。
        if sql.lstrip().upper().startswith("SELECT"):
            return []
        raise AssertionError(sql)


def test_scan_gate_reads_batches_and_starts_without_hedge() -> None:
    """扫描只发只读语句。无对冲行时 TRS 批次 id 为 0。"""

    gateway = _ScriptedGateway()
    scan = asyncio.run(scan_gate(gateway, BUSINESS_DAY))
    assert scan.lines == ()
    assert scan.checks
    assert all(line.verdict == VERDICT_ERROR for line in scan.checks)
    assert all("completed" not in line.formula_text for line in scan.checks)
    assert scan.ready[0].trs_batch_id == NO_HEDGE_BATCH_ID
    assert scan.ready[0].trs_end_time == NO_HEDGE_END_TIME
    issued = [sql for sql, _args in gateway.statements]
    assert HEDGE_SQL in issued
    assert any(sql.startswith("SELECT DISTINCT business_date FROM db_issue.report_trade_order") for sql in issued)
    dated = [sql for sql in issued if sql.startswith(("SELECT id, batch_no", "SELECT DISTINCT business_date"))]
    assert dated
    assert all("WHERE business_date = %s" in sql for sql in dated)
    assert all(not sql.lstrip().upper().startswith("INSERT") for sql in issued)
    hedge_args = [args for sql, args in gateway.statements if sql == HEDGE_SQL]
    assert hedge_args == [(TRADE_DAY, "ORD%", 0)]
    scoped = [args for sql, args in gateway.statements if "WHERE business_date = %s" in sql]
    assert scoped
    assert scoped == [(BUSINESS_DAY,)] * len(scoped)


def _ready(*, trs_batch_id: int = NO_HEDGE_BATCH_ID) -> ReadyRun:
    return ReadyRun(
        market="a_share",
        trade_date=TRADE_DAY,
        business_date=BUSINESS_DAY,
        issue_batch_id=9,
        sales_batch_id=4,
        trs_batch_id=trs_batch_id,
        issue_end_time=ISSUE_END,
        sales_end_time=SALES_END,
        trs_end_time=NO_HEDGE_END_TIME if trs_batch_id == NO_HEDGE_BATCH_ID else TRS_END,
        issue_applied_revision=3,
        sales_applied_revision=0,
        trs_applied_revision=0,
    )


def _round(**overrides: object) -> IssueRound:
    payload: dict[str, object] = {
        "dataset_id": 7,
        "batch_id": 9,
        "revision": 3,
        "baseline_kind": "opening",
        "base_dataset_id": None,
        "base_trade_date": None,
        "base_row_trade_date": None,
        "base_missing": False,
        "upstream_fixed": 1,
        "upstream_batch_id": None,
        "upstream_dataset_id": None,
    }
    payload.update(overrides)
    return IssueRound(**payload)  # type: ignore[arg-type]


def _facts(**overrides: object) -> PublishedFacts:
    payload: dict[str, object] = {
        "issue_rounds": (_round(),),
        "issue_business_date": date(2026, 10, 20),
        "sales_versions": ("9:3:IB9",),
        "sales_dataset_id": 11,
        "sales_kind": "cleared",
        "sales_status": "active",
        "sales_dataset_batch_id": 4,
        "trs_dataset_id": None,
        "trs_kind": "",
        "trs_copy_no": None,
        "trs_status": "",
        "trs_dataset_batch_id": None,
        "trs_business_date": None,
    }
    payload.update(overrides)
    return PublishedFacts(**payload)  # type: ignore[arg-type]


def test_published_checks_stay_quiet_when_the_round_matches() -> None:
    """基线、唯一发布、清单版本和销售发布都对上时，不另出一张完成卡。"""

    lines = evaluate_published(_ready(), _facts())
    assert lines == ()


def test_published_checks_name_the_mismatch() -> None:
    """每一类不一致只出对应公式，公式里不出现批次 completed。"""

    missing = evaluate_published(_ready(), _facts(issue_rounds=()))
    assert {line.formula_text for line in missing} >= {FORMULA_ISSUE_PUBLISH, FORMULA_ISSUE_BASELINE}
    opening = evaluate_published(_ready(), _facts(issue_rounds=(_round(base_dataset_id=1),)))
    assert [line.formula_text for line in opening] == [FORMULA_ISSUE_BASELINE]
    stalled = evaluate_published(_ready(), _facts(issue_business_date=BUSINESS_DAY))
    assert FORMULA_ISSUE_PUBLISH in {line.formula_text for line in stalled}
    version = evaluate_published(_ready(), _facts(sales_versions=("8:3:IB9",)))
    assert [line.formula_text for line in version] == [FORMULA_SALES_UPSTREAM]
    draft = evaluate_published(_ready(), _facts(sales_status="backing_up"))
    assert [line.formula_text for line in draft] == [FORMULA_SALES_PUBLISH]
    hedged = _ready(trs_batch_id=8)
    late = _facts(
        issue_rounds=(_round(baseline_kind="published", base_dataset_id=2, base_trade_date=date(2026, 10, 16), base_row_trade_date=date(2026, 10, 16), upstream_batch_id=8, upstream_dataset_id=15),),
        trs_dataset_id=15,
        trs_kind="cleared",
        trs_copy_no=1,
        trs_status="backing_up",
        trs_dataset_batch_id=8,
        trs_business_date=BUSINESS_DAY,
    )
    named = {line.formula_text for line in evaluate_published(hedged, late)}
    assert named == {FORMULA_TRS_DATE, FORMULA_TRS_VERSION}
    quiet = evaluate_published(
        hedged,
        _facts(
            issue_rounds=late.issue_rounds,
            trs_dataset_id=15,
            trs_kind="cleared",
            trs_copy_no=1,
            trs_status="available",
            trs_dataset_batch_id=8,
            trs_business_date=date(2026, 10, 20),
        ),
    )
    assert quiet == ()
    for group in (missing, opening, stalled, version, draft, evaluate_published(hedged, late)):
        assert all("completed" not in line.formula_text for line in group)