"""DB-free regression contract for the Yandex Market returns projection.""" from __future__ import annotations import re from contextlib import contextmanager import pytest from kernel.etl import build_fact_returns from kernel.etl.build_fact_returns import ( ACKNOWLEDGED_YM_LOSS_ENTRIES, ACKNOWLEDGED_YM_LOSS_ENTRY_COUNT, ACKNOWLEDGED_YM_LOSS_LEDGER_SHA256, YM_ATTEST_SQL, YM_CLEANUP_SQLS, YM_EXPECTED, YM_ITEMS, YM_ITEMS_ATTEST_SQL, YM_PREPARE_SQLS, YM_SQL, YM_STALE_CLEANUP_SQL, YM_TYPED_ATTEST_SQL, YMReturnsProjectionError, ) def test_ym_latest_snapshot_is_deterministic_per_marketplace_return() -> None: sql = "\n".join(YM_PREPARE_SQLS) assert "PARTITION BY r.mp_id, r.return_id" in sql assert "ORDER BY r.collected_at DESC, r.payload_id DESC, r.id DESC" in sql assert "WHERE latest.rn = 1" in sql assert "PRIMARY KEY (mp_id, return_id)" in sql def test_ym_items_prevent_order_model_fanout_and_preserve_count_two() -> None: prepare_sql = "\n".join(YM_PREPARE_SQLS) expected_sql = prepare_sql assert "'$.items[*]'" in prepare_sql # Коллация объявляется в самом JSON_TABLE с 10.08.2026: без неё значения # рождаются в коллации соединения и соединение с признанными строками по # shop_sku падало с 1267. Проверяем ПУТЬ и ТИП, не привязываясь к тому, # объявлена ли коллация — за неё отвечает test_vozvraty_ym_collation.py. assert re.search(r"shop_sku VARCHAR\(255\).*?PATH '\$\.shopSku'", prepare_sql, re.S) assert re.search(r"item_count_text VARCHAR\(64\).*?PATH '\$\.count'", prepare_sql, re.S) assert "REGEXP_LIKE(item_scope.item_count_text, '^[1-9][0-9]*$'" in prepare_sql assert "idn.mp_id = 6" in prepare_sql, "YM identifiers use canonical group mp_id=6" assert "idn.identifier_type = 'shopSku'" in prepare_sql assert "idn.is_current = 1" in prepare_sql assert "dp.artikul_upper = UPPER(TRIM(return_item.shop_sku))" in prepare_sql # A return item with count=2 must become quantity=2 and two units of the # matching order's per-unit revenue, not one return on every order model. assert "SUM(current_return.item_count) AS quantity" in expected_sql assert "current_return.order_revenue / current_return.order_quantity" in expected_sql assert "* current_return.item_count" in expected_sql assert f"FROM {YM_ITEMS} current_return" in expected_sql assert "idn.model_id <> dp.model_id THEN NULL" in prepare_sql def test_every_approved_typed_return_must_be_complete_before_projection() -> None: sql = " ".join(YM_TYPED_ATTEST_SQL.split()) assert "approved_ym_returns.payload_id = r.payload_id" in sql assert "r.mp_id NOT IN (6, 8)" in sql assert "TRIM(r.return_id) = ''" in sql assert "TRIM(r.order_id) = ''" in sql assert "r.return_date_mp IS NULL" in sql assert "JSON_VALID(r.return_json) = 0" in sql assert "JSON_LENGTH(JSON_EXTRACT(r.return_json, '$.items')) <= 0" in sql def test_ym_mp6_and_mp8_are_isolated_by_exact_order_marketplace() -> None: prepare_sql = "\n".join(YM_PREPARE_SQLS) assert "fact_order.mp_id = item_scope.mp_id" in prepare_sql assert "fact_order.order_id = item_scope.order_id" in prepare_sql assert "fact_order.model_id = item_scope.candidate_model_id" in prepare_sql assert "fact_order.mp_id IN (6, 8)" not in prepare_sql assert "current_return.mp_id NOT IN (6, 8)" in YM_ATTEST_SQL assert "expected_return.mp_id" in YM_SQL def test_ym_retracts_only_stale_s3_rows_from_authoritative_live_sets() -> None: prepare = "\n".join(YM_PREPARE_SQLS) cleanup = YM_STALE_CLEANUP_SQL # Empty/non-array item sets are not allowed to become an authoritative # retraction. Attestation aborts the whole ETL step before this DELETE. assert "JSON_TYPE(JSON_EXTRACT(latest.return_json, '$.items')) = 'ARRAY'" in prepare assert "THEN 1 ELSE 0" in prepare assert "JSON_LENGTH(JSON_EXTRACT(current_return.return_json, '$.items')) = 0" in YM_ATTEST_SQL assert "item_count_mismatches" in YM_ATTEST_SQL assert "parsed.identity_matches <> 1" in YM_ITEMS_ATTEST_SQL assert "parsed.order_matches <> 1" in YM_ITEMS_ATTEST_SQL assert "parsed.acknowledged_listed = 0" in YM_ITEMS_ATTEST_SQL assert "AS acknowledged_missing" in YM_ITEMS_ATTEST_SQL assert f"LEFT JOIN {YM_EXPECTED} expected_return" in cleanup assert "current_return.items_authoritative = 1" in cleanup assert "fact_return.source_payload_id IS NOT NULL" in cleanup assert "expected_return.model_id IS NULL" in cleanup assert "source_payload_id IS NULL" not in cleanup assert "payload_id IN" not in cleanup, "retraction is by live natural key, not old payload" def test_ym_upsert_and_reconciliation_are_idempotent() -> None: assert "ON DUPLICATE KEY UPDATE" in YM_SQL assert "legacy_return.source_payload_id IS NULL" in YM_SQL assert "WHERE legacy_return.id IS NULL" in YM_SQL assert "quantity=VALUES(quantity)" in YM_SQL assert "source_payload_id=VALUES(source_payload_id)" in YM_SQL assert YM_CLEANUP_SQLS == ( "DROP TEMPORARY TABLE IF EXISTS tmp_fact_returns_ym_expected", "DROP TEMPORARY TABLE IF EXISTS tmp_fact_returns_ym_items", "DROP TEMPORARY TABLE IF EXISTS tmp_fact_returns_ym_current", "DROP TEMPORARY TABLE IF EXISTS tmp_fact_returns_ym_acknowledged", ) def test_ym_acknowledged_loss_ledger_is_finite_and_exactly_materialized() -> None: prepare_sql = "\n".join(YM_PREPARE_SQLS) assert len(ACKNOWLEDGED_YM_LOSS_ENTRIES) == ACKNOWLEDGED_YM_LOSS_ENTRY_COUNT == 79 assert len(ACKNOWLEDGED_YM_LOSS_LEDGER_SHA256) == 64 assert "CREATE TEMPORARY TABLE tmp_fact_returns_ym_acknowledged" in prepare_sql assert "acknowledged.order_id = item_scope.order_id" in prepare_sql assert "acknowledged.item_count = CASE" in prepare_sql assert "acknowledged.model_id = item_scope.candidate_model_id" in prepare_sql assert "NULL AS amount" in prepare_sql class _AttestationCursor: def __init__(self, metrics: dict[str, int]) -> None: self.metrics = metrics self.executed: list[str] = [] self.rowcount = 0 def execute(self, sql: str) -> None: self.executed.append(sql) def fetchone(self) -> dict[str, int]: if self.executed[-1] == YM_TYPED_ATTEST_SQL: return {"source_rows": 1, "invalid_rows": 0} return self.metrics def _metrics(**overrides: int) -> dict[str, int]: result = { "selected_returns": 1, "invalid_return_shapes": 0, "expected_items": 1, "parsed_items": 1, "item_count_mismatches": 0, "invalid_items": 0, "identity_failures": 0, "order_failures": 0, "acknowledged_missing": 0, "projected_items": 1, } result.update(overrides) return result def _isolate_ym_attestation(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(build_fact_returns, "_require_lamoda_typed_input", lambda _c: None) monkeypatch.setattr(build_fact_returns, "_require_wb_projection", lambda _c: {}) monkeypatch.setattr(build_fact_returns, "_require_ozon_projection", lambda _c: {}) monkeypatch.setattr(build_fact_returns, "_require_lamoda_projection", lambda _c: {}) @pytest.mark.parametrize( ("case", "metrics"), ( ( "empty items array", _metrics( invalid_return_shapes=1, expected_items=0, parsed_items=0, projected_items=0, ), ), ("malformed item", _metrics(invalid_items=1, projected_items=0)), ("unresolved SKU", _metrics(identity_failures=1, projected_items=0)), ("missing exact order", _metrics(order_failures=1, projected_items=0)), ), ) def test_ym_semantic_gap_aborts_before_any_persistent_fact_dml( monkeypatch: pytest.MonkeyPatch, case: str, metrics: dict[str, int], ) -> None: cursor = _AttestationCursor(metrics) @contextmanager def cursor_context(): yield cursor monkeypatch.setattr(build_fact_returns, "get_cursor", cursor_context) monkeypatch.setattr( build_fact_returns, "require_build_input_projection", lambda _cursor: {}, ) _isolate_ym_attestation(monkeypatch) with pytest.raises(YMReturnsProjectionError, match="persistent DML is forbidden"): build_fact_returns.build() assert YM_ATTEST_SQL in cursor.executed, case assert not any( statement.lstrip().startswith("INSERT INTO gwptd_kernel.fact_returns") or statement.lstrip().startswith("DELETE fact_return") for statement in cursor.executed ), case def test_named_acknowledged_loss_is_the_only_nonfatal_missing_order() -> None: metrics = _metrics( parsed_items=1, projected_items=0, acknowledged_missing=1, order_failures=0, ) cursor = _AttestationCursor(metrics) assert build_fact_returns._require_ym_projection(cursor)["acknowledged_missing"] == 1 def test_ym_projection_never_reads_raw_payload() -> None: sql = "\n".join( ( YM_TYPED_ATTEST_SQL, *YM_PREPARE_SQLS, YM_ATTEST_SQL, YM_ITEMS_ATTEST_SQL, YM_STALE_CLEANUP_SQL, YM_SQL, ) ) assert "raw_payload" not in sql def test_invalid_typed_return_blocks_before_temp_or_persistent_dml( monkeypatch: pytest.MonkeyPatch, ) -> None: cursor = _AttestationCursor(_metrics()) def invalid_fetchone() -> dict[str, int]: return {"source_rows": 2, "invalid_rows": 1} cursor.fetchone = invalid_fetchone # type: ignore[method-assign] @contextmanager def cursor_context(): yield cursor monkeypatch.setattr(build_fact_returns, "get_cursor", cursor_context) monkeypatch.setattr( build_fact_returns, "require_build_input_projection", lambda _cursor: {}, ) _isolate_ym_attestation(monkeypatch) with pytest.raises(YMReturnsProjectionError, match="typed input"): build_fact_returns.build() assert cursor.executed == [YM_TYPED_ATTEST_SQL]