"""批次门禁：开跑条件、日切报表提前出现、同一输入重复调用。""" 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)