"""P1B-3 phase-one tests for the complete kernel input policy.""" from __future__ import annotations import re import shutil from pathlib import Path from typing import Callable import pytest import yaml from kernel.etl.build_input_manifest import ( BuildInputManifestError, load_build_input_manifest, validate_build_input_manifest, ) from kernel.etl.build_input_source_closure import validate_executable_source_closure REPO = Path(__file__).resolve().parents[2] MANIFEST = REPO / "specs" / "kernel" / "build-all-inputs.yaml" PRODUCTS = REPO / "specs" / "products" EXPECTED_STEPS = ( "dim_product", "resolve_identifiers", "wb_category", "fact_supplier_stock", "fact_stocks", "fact_prices", "fact_storage", "fact_finance", "fact_orders", "fact_returns", "fact_analytics", "dedupe_cross_mp_orders", "fact_turnover", "fact_rating", "validate_invariants", ) EXPECTED_ACTIVE_ENDPOINTS = { "1c.supplier_catalog", "lamoda.inventory", "lamoda.nomenclatures", "lamoda.orders", "lamoda.status_dates", "ozon.finance_transactions", "ozon.postings_fbo", "ozon.postings_fbs", "ozon.postings_fbs_changed", "ozon.prices", "ozon.product_info_list", "ozon.product_sku_map", "ozon.returns", "ozon.stock_on_warehouses", "ozon.stocks", "wb.analytics", "wb.cards_list", "wb.finance_reports", "wb.orders", "wb.orders_fbs", "wb.paid_storage", "wb.prices", "wb.returns", "wb.sales", "wb.stocks_fbs", "wb.warehouse_stocks", "ym.analytics", "ym.finance", "ym.offer_mappings", "ym.prices", "ym.returns", "ym.stats_orders", "ym.stats_orders_changed", "ym.stocks", "ym.storage", } def _modified_manifest( tmp_path: Path, mutate: Callable[[dict], None], ) -> Path: document = yaml.safe_load(MANIFEST.read_text(encoding="utf-8")) mutate(document) target = tmp_path / "build-all-inputs.yaml" target.write_text(yaml.safe_dump(document, sort_keys=False), encoding="utf-8") return target def _source(document: dict, source_id: str) -> dict: return next(source for source in document["sources"] if source["id"] == source_id) def _step(document: dict, name: str) -> dict: return next(step for step in document["steps"] if step["name"] == name) def test_manifest_matches_all_fifteen_executable_steps_and_exact_active_endpoints(): manifest = validate_build_input_manifest() assert manifest.step_order == EXPECTED_STEPS assert set(manifest.active_endpoints) == EXPECTED_ACTIVE_ENDPOINTS assert len(manifest.active_endpoints) == 35 assert re.fullmatch(r"[0-9a-f]{64}", manifest.policy_digest) def test_manifest_declares_exclusions_reference_and_formula_inputs(): manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) document = manifest.document wb_stocks = _source(document, "excluded:wb.stocks") assert wb_stocks["mode"] == "excluded_product_spec" assert wb_stocks["tables"] == ("wb_stocks",) stocks_step = _step(document, "fact_stocks") assert "excluded:wb.stocks" not in stocks_step["inputs"]["sources"] assert stocks_step["exclusions"] == ("excluded:wb.stocks",) excluded = _source(document, "excluded:lamoda.returns") assert excluded["mode"] == "excluded_product_spec" assert excluded["tables"] == ("lamoda_return_boxes", "lamoda_returns") assert _step(document, "fact_returns")["exclusions"] == ( "excluded:lamoda.returns", ) resolver_sources = set(_step(document, "resolve_identifiers")["inputs"]["sources"]) assert "reference:onec_nomenclature_ref" in resolver_sources rating_sources = set(_step(document, "fact_rating")["inputs"]["sources"]) assert rating_sources == { "external:cbr_usd_rub", "formula:business_rules", "metric:cost_price", "metric:rating", } ym_finance = _source(document, "product:ym.finance") assert ym_finance["id"] == "product:ym.finance" assert ym_finance["mode"] == "active_product_spec" assert ym_finance["inputMode"] == "rolling_journal" assert ym_finance["selection"] == "latest_attempt_exact_run_covers_replay_window_no_fallback" assert dict(ym_finance["contract"]) == { "freshnessHours": 27, "windowDays": 30, "endOffsetDays": 1, "requiredScopes": (), } assert ym_finance["endpoint"] == "ym.finance" assert ym_finance["tables"] == ("ym_finance",) assert _step(document, "fact_finance")["inputs"]["sources"] == ( "product:wb.finance_reports", "product:ozon.finance_transactions", "product:ym.finance", ) @pytest.mark.parametrize( ("marketplace", "collector_class", "parser_version", "marketplace_ids"), ( ("wb", "LegacyWbReturnsRecovery", "legacy-recovery-2026-08-04", (1, 2)), ("ym", "LegacyYmReturnsRecovery", "legacy-recovery-2026-08-03", (6, 8)), ), ) def test_manifest_declares_honest_legacy_recovery_without_freshness_credit( marketplace: str, collector_class: str, parser_version: str, marketplace_ids: tuple[int, ...], ): manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) recovery_endpoint = f"{marketplace}.returns.legacy_recovery" recovery = _source(manifest.document, f"recovery:{recovery_endpoint}") assert manifest.document["sourceModePolicies"]["legacy_recovery"] == { "countedActiveEndpoint": False, "buildEligible": True, "completeness": "recovered_historical_payload_not_live_collection", } assert recovery == { "id": f"recovery:{recovery_endpoint}", "mode": "legacy_recovery", "endpoint": recovery_endpoint, "tables": (f"{marketplace}_returns",), "producer": f"marketplace-collector-v3/scripts/recover_{marketplace}_returns_from_legacy.py", "collectorClass": collector_class, "parserVersion": parser_version, "sourceBusinessKeyPrefix": "legacy:lamoda_reports.mp_returns:", "sourceBusinessKeyAction": "return", # Маркетплейсы объявляет сам источник. Пока их держал резолвер # константой {6, 8}, восстановление WB отбивалось с чужой причиной. "marketplaceIds": marketplace_ids, "completenessGuard": "recovered_historical_payload_not_live_collection", } returns_sources = _step(manifest.document, "fact_returns")["inputs"]["sources"] assert returns_sources == ( "product:wb.returns", "recovery:wb.returns.legacy_recovery", "product:ozon.returns", "product:ym.returns", "recovery:ym.returns.legacy_recovery", "ledger:ym.returns.acknowledged_missing_orders", "product:lamoda.status_dates", ) assert recovery_endpoint not in manifest.active_endpoints assert len(manifest.active_endpoints) == 35 def test_manifest_binds_the_named_ym_order_recovery_and_loss_ledger(): manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) recovery = _source(manifest.document, "recovery:ym.stats_orders.legacy_recovery") ledger = _source(manifest.document, "ledger:ym.returns.acknowledged_missing_orders") assert recovery["tables"] == ("ym_stats_orders",) assert recovery["sourceBusinessKeyPrefix"] == "legacy:lamoda_reports.mp_orders_daily:" assert recovery["collectorClass"] == "LegacyYmOrdersRecovery" assert ledger["expectedEntryCount"] == 79 assert ledger["sourceEndpoint"] == "ym.returns.legacy_recovery" assert ledger["sha256"] == "484cda6ab4ee16dd8cb38845f18ced6907c425f92c98a683d62483b9a4f9491e" assert "recovery:ym.stats_orders.legacy_recovery" in _step( manifest.document, "fact_orders" )["inputs"]["sources"] assert "ledger:ym.returns.acknowledged_missing_orders" in _step( manifest.document, "fact_returns" )["inputs"]["sources"] @pytest.mark.parametrize( ("field", "value"), ( ("endpoint", "wb.returns.unreviewed_recovery"), ("collectorClass", "UnreviewedWbRecovery"), ("producer", "marketplace-collector-v3/scripts/unreviewed_recovery.py"), ), ) def test_manifest_rejects_unreviewed_wb_recovery_contract( tmp_path: Path, field: str, value: str ): def mutate(document: dict) -> None: recovery = _source(document, "recovery:wb.returns.legacy_recovery") recovery[field] = value path = _modified_manifest(tmp_path, mutate) with pytest.raises(BuildInputManifestError, match="reviewed legacy recovery|reviewed wb"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_manifest_binds_exact_snapshot_journal_and_scoped_input_semantics(): manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) active = [ source for source in manifest.document["sources"] if source["mode"] == "active_product_spec" ] snapshots = [source for source in active if source["inputMode"] == "full_snapshot"] journals = [source for source in active if source["inputMode"] == "rolling_journal"] assert len(snapshots) == 15 assert len(journals) == 20 assert {source["selection"] for source in snapshots} == { "latest_attempt_exact_run_no_fallback" } assert {source["selection"] for source in journals} == { "latest_attempt_exact_run_covers_replay_window_no_fallback" } expected_scopes = [{"key": "fbs", "mpId": 6}, {"key": "fby", "mpId": 8}] for endpoint in ( "ym.stocks", "ym.stats_orders", "ym.stats_orders_changed", "ym.returns", ): actual = _source(manifest.document, f"product:{endpoint}")["contract"][ "requiredScopes" ] assert [dict(scope) for scope in actual] == expected_scopes @pytest.mark.parametrize("mutation", ["freshness", "window", "scopes", "selection"]) def test_checker_rejects_product_input_semantic_drift(tmp_path: Path, mutation: str): def change(document: dict) -> None: source = _source(document, "product:ym.stats_orders") if mutation == "freshness": source["contract"]["freshnessHours"] += 1 elif mutation == "window": source["contract"]["windowDays"] += 1 elif mutation == "scopes": source["contract"]["requiredScopes"] = [{"key": "fbs", "mpId": 6}] else: source["selection"] = "latest_attempt_exact_run_no_fallback" path = _modified_manifest(tmp_path, change) with pytest.raises(BuildInputManifestError): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) @pytest.mark.parametrize( ("source_id", "endpoint", "table", "window_days", "end_offset_days", "scopes"), ( ( "product:ozon.postings_fbs_changed", "ozon.postings_fbs_changed", "ozon_postings_fbs", 30, -1, (), ), ( "product:ym.stats_orders_changed", "ym.stats_orders_changed", "ym_stats_orders", 30, 0, ({"key": "fbs", "mpId": 6}, {"key": "fby", "mpId": 8}), ), ), ) def test_changed_order_journals_are_exact_approved_fact_orders_inputs( source_id: str, endpoint: str, table: str, window_days: int, end_offset_days: int, scopes: tuple[dict[str, int | str], ...], ): """Changed-status feeds are complete sealed inputs, not optional table aliases.""" manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) source = dict(_source(manifest.document, source_id)) # intake_only проверяется отдельно: это перечень таблиц, в которые сбор # ПИШЕТ, но ядро не читает. У ozon.postings_fbs_changed он появился # 12.08.2026, когда добор привели к основному сбору один в один — до этого # он объявлял 16 полей из 58 и ни одного потомка, а в таблице было 5456 # отправлений с полными данными лишь у 21. Сам договор входа при этом не # менялся, поэтому сверяется он же, без этого поля. intake_only = tuple(source.pop("intake_only", ())) assert source == { "id": source_id, "mode": "active_product_spec", "inputMode": "rolling_journal", "selection": "latest_attempt_exact_run_covers_replay_window_no_fallback", "contract": { "freshnessHours": 27, "windowDays": window_days, "endOffsetDays": end_offset_days, "requiredScopes": scopes, }, "endpoint": endpoint, "tables": (table,), } for tablica in intake_only: assert tablica.startswith(table), ( f"{source_id}: {tablica} не потомок {table} — перечень intake_only " f"описывает таблицы, в которые пишет ЭТОТ эндпоинт" ) assert endpoint in manifest.active_endpoints assert source_id in _step(manifest.document, "fact_orders")["inputs"]["sources"] @pytest.mark.parametrize( "source_id", ("product:ozon.postings_fbs_changed", "product:ym.stats_orders_changed"), ) def test_changed_order_journal_cannot_be_removed_from_fact_orders_admission( tmp_path: Path, source_id: str, ): """The strict build manifest must fail before ETL if either changed feed is detached.""" def change(document: dict) -> None: fact_orders = _step(document, "fact_orders") fact_orders["inputs"]["sources"].remove(source_id) path = _modified_manifest(tmp_path, change) with pytest.raises(BuildInputManifestError, match="unassigned sources"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_executable_source_closure_is_exact_and_excludes_wb_stocks(): manifest = validate_build_input_manifest() closure = validate_executable_source_closure(manifest) stocks = closure.relations_by_step["fact_stocks"] assert "wb_warehouse_stocks" in stocks assert "wb_stocks_fbs" in stocks assert "wb_stocks" not in stocks assert closure.kernel_reads_by_step["validate_invariants"] == { "dim_warehouse", "fact_orders", "fact_stocks_daily", } assert closure.kernel_writes_by_step["fact_analytics"] == { "fact_analytics": {"insert"} } assert _step(manifest.document, "fact_analytics")["mutationMode"] == "upsert" @pytest.mark.parametrize( ("statement", "message"), [ ( 'cur.execute(f"DELETE FROM {K}.undeclared_kernel")', "undeclared_writes=.*undeclared_kernel", ), ( 'cur.execute(f"UPDATE {K}.undeclared_kernel SET id = id")', "undeclared_writes=.*undeclared_kernel", ), ( 'cur.execute(f"SELECT own.id FROM {K}.fact_prices_daily own ' 'JOIN {K}.undeclared_kernel hidden ON hidden.id = own.id")', "undeclared persistent kernel reads:.*undeclared_kernel", ), ], ids=["delete", "update", "join"], ) def test_kernel_closure_rejects_undeclared_persistent_access( tmp_path: Path, statement: str, message: str, ): copied = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied) target = copied / "build_fact_prices.py" target.write_text( target.read_text(encoding="utf-8") + "\ndef undeclared_kernel_probe(cur):\n" + f" {statement}\n", encoding="utf-8", ) manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) with pytest.raises(BuildInputManifestError, match=message): validate_executable_source_closure(manifest, module_root=copied) @pytest.mark.parametrize( "statement", [ 'cur.execute("DELETE FROM tmp_kernel_approved_payload")', 'cur.execute("DELETE FROM tmp_kernel_approved_payload__approved_ozon_row")', 'cur.execute("DELETE FROM tmp_kernel_approved_reference__approved_ref_exact")', ], ids=["base_payload", "alias_payload_copy", "alias_reference_copy"], ) def test_kernel_closure_rejects_business_mutation_of_approved_projection( tmp_path: Path, statement: str, ): copied = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied) target = copied / "build_fact_prices.py" target.write_text( target.read_text(encoding="utf-8") + "\ndef projection_mutation_probe(cur):\n" + f" {statement}\n", encoding="utf-8", ) manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) with pytest.raises(BuildInputManifestError, match="materializer-owned"): validate_executable_source_closure(manifest, module_root=copied) @pytest.mark.parametrize("drift", ["output", "mutation_mode"]) def test_kernel_closure_rejects_output_or_mutation_mode_drift( tmp_path: Path, drift: str, ): def change(document: dict) -> None: analytics = _step(document, "fact_analytics") if drift == "output": analytics["outputs"] = ["gwptd_kernel.fact_analytics_wrong"] else: analytics["mutationMode"] = "snapshot_reconcile" path = _modified_manifest(tmp_path, change) manifest = load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) message = "executable outputs differ" if drift == "output" else "destructive" with pytest.raises(BuildInputManifestError, match=message): validate_executable_source_closure(manifest) def test_static_kernel_reference_is_explicit_and_makes_no_digest_claim(): manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) source = _source(manifest.document, "kernel:dim_warehouse") assert source["table"] == "gwptd_kernel.dim_warehouse" assert source["consistency"] == "caller_transaction_visibility" assert source["completenessGuard"] == ( "declared_relation_only_no_content_digest" ) assert "kernel:dim_warehouse" in _step( manifest.document, "validate_invariants" )["inputs"]["sources"] def test_source_closure_rejects_new_undeclared_intake_relation(tmp_path: Path): copied = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied) target = copied / "build_fact_prices.py" target.write_text( target.read_text(encoding="utf-8") + '\ndef undeclared_probe(cur):\n' + ' cur.execute(f"SELECT * FROM {I}.secret_intake")\n', encoding="utf-8", ) manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) with pytest.raises(BuildInputManifestError, match="secret_intake"): validate_executable_source_closure(manifest, module_root=copied) def test_source_closure_rejects_unresolved_dynamic_relation(tmp_path: Path): copied = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied) target = copied / "build_fact_prices.py" target.write_text( target.read_text(encoding="utf-8") + '\ndef unresolved_probe(cur, table):\n' + ' cur.execute(f"SELECT * FROM {I}.{table}")\n', encoding="utf-8", ) manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) with pytest.raises(BuildInputManifestError, match="cannot close dynamic"): validate_executable_source_closure(manifest, module_root=copied) @pytest.mark.parametrize( "expression", [ '"SELECT * FROM {}.evil".format(I)', '"SELECT * FROM %s.evil" % I', '"".join(("SELECT * FROM ", I, ".evil"))', ], ids=["format", "percent", "join"], ) def test_source_closure_rejects_hidden_schema_construction( tmp_path: Path, expression: str, ): copied = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied) target = copied / "build_fact_prices.py" target.write_text( target.read_text(encoding="utf-8") + "\ndef hidden_relation_probe(cur):\n" + f" cur.execute({expression})\n", encoding="utf-8", ) manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) with pytest.raises(BuildInputManifestError, match="unrendered SQL schema"): validate_executable_source_closure(manifest, module_root=copied) def test_dead_sql_string_cannot_satisfy_a_removed_executable_read(tmp_path: Path): copied = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied) target = copied / "build_fact_prices.py" text = target.read_text(encoding="utf-8") text = text.replace("FROM {I}.wb_prices", "FROM {I}.ozon_prices") text = text.replace("JOIN {I}.wb_prices", "JOIN {I}.ozon_prices") text += '\nDEAD_DECOY = f"SELECT * FROM {I}.wb_prices"\n' target.write_text(text, encoding="utf-8") manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) with pytest.raises(BuildInputManifestError, match="declared_but_unread=.*wb_prices"): validate_executable_source_closure(manifest, module_root=copied) def test_policy_digest_is_deterministic_and_binds_product_spec_bytes(tmp_path: Path): first = load_build_input_manifest(expected_steps=EXPECTED_STEPS) second = load_build_input_manifest(expected_steps=EXPECTED_STEPS) assert second.policy_digest == first.policy_digest copied_products = tmp_path / "products" shutil.copytree(PRODUCTS, copied_products) wb_orders = copied_products / "wb.orders.yaml" wb_orders.write_text( wb_orders.read_text(encoding="utf-8") + "\n# digest-only contract change\n", encoding="utf-8", ) changed = load_build_input_manifest( product_specs_path=copied_products, expected_steps=EXPECTED_STEPS, ) assert changed.policy_digest != first.policy_digest def test_loaded_policy_snapshot_is_deeply_immutable(): manifest = load_build_input_manifest(expected_steps=EXPECTED_STEPS) digest = manifest.policy_digest wb_orders = _source(manifest.document, "product:wb.orders") with pytest.raises(TypeError): wb_orders["contract"]["freshnessHours"] = 999 with pytest.raises(TypeError): manifest.product_bindings["wb.orders"]["tables"][0] = "other" assert manifest.policy_digest == digest def test_checker_rejects_executable_steps_order_drift(): drifted = list(EXPECTED_STEPS) drifted[0], drifted[1] = drifted[1], drifted[0] with pytest.raises(BuildInputManifestError, match="STEPS order differs"): load_build_input_manifest(expected_steps=drifted) def test_checker_rejects_product_spec_table_mapping_drift(tmp_path: Path): path = _modified_manifest( tmp_path, lambda document: _source(document, "product:wb.orders").update( {"tables": ["wb_orders_wrong"]} ), ) with pytest.raises(BuildInputManifestError, match="ProductSpec/table mapping differs"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_checker_rejects_duplicate_source(tmp_path: Path): def duplicate(document: dict) -> None: document["sources"].append(dict(document["sources"][0])) path = _modified_manifest(tmp_path, duplicate) with pytest.raises(BuildInputManifestError, match="duplicate source id"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_checker_rejects_missing_declared_source(tmp_path: Path): def break_reference(document: dict) -> None: _step(document, "dim_product")["inputs"]["sources"] = ["product:missing"] path = _modified_manifest(tmp_path, break_reference) with pytest.raises(BuildInputManifestError, match="declared input missing source"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_checker_rejects_step_without_any_declared_input(tmp_path: Path): def remove_inputs(document: dict) -> None: inputs = _step(document, "dim_product")["inputs"] inputs["sources"] = [] inputs["derivedSteps"] = [] path = _modified_manifest(tmp_path, remove_inputs) with pytest.raises(BuildInputManifestError, match="has no declared input"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_checker_rejects_forward_or_missing_derived_dependency(tmp_path: Path): def break_dependency(document: dict) -> None: _step(document, "fact_prices")["inputs"]["derivedSteps"] = ["fact_rating"] path = _modified_manifest(tmp_path, break_dependency) with pytest.raises(BuildInputManifestError, match="missing or not earlier"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) def test_unique_yaml_loader_rejects_duplicate_mapping_keys(tmp_path: Path): path = tmp_path / "duplicate.yaml" path.write_text( MANIFEST.read_text(encoding="utf-8") + "\nsteps: []\n", encoding="utf-8", ) with pytest.raises(BuildInputManifestError, match="duplicate YAML key"): load_build_input_manifest(path, expected_steps=EXPECTED_STEPS) # --------------------------------------------------------------------------- # intake_only — таблицы приёма, не являющиеся входами ядра # --------------------------------------------------------------------------- def _make_product_spec_with_child( products_dir: Path, endpoint: str, child_table: str, ) -> None: """Добавить дочернюю таблицу в ProductSpec через intakeByKind. Нужно, чтобы _product_tables() вернула и основную, и дочернюю таблицу — тогда манифест может разнести их по tables / intake_only. Когда в outputs есть intakeByKind, компилятор заменяет основной intake целью из intakeByKind. Поэтому ОБЕ таблицы — и основная, и дочерняя — идут через intakeByKind.targets. Основная таблица делается defaultKind. """ import yaml as _yaml spec_path = products_dir / f"{endpoint}.yaml" document = _yaml.safe_load(spec_path.read_text(encoding="utf-8")) # Форм объявления вывода ДВЕ и обе действующие: одиночный `intake` и # `intakeByKind.targets.<вид>`. Разворот массивов переводит описание целиком # на вторую, и безусловное чтение "intake" роняло тест на живых описаниях. outputs = document["outputs"] if isinstance(outputs.get("intake"), dict): main_table = outputs["intake"]["table"] else: _first = next(iter((outputs["intakeByKind"]["targets"] or {}).values())) _targets = _first if isinstance(_first, list) else [_first] main_table = _targets[0]["table"] _main = outputs.get("intake") if isinstance(_main, dict): main_fields = _main.get("fields", {}) else: main_fields = _targets[0].get("fields", {}) # Переносим основной intake в intakeByKind, чтобы компилятор увидел обе таблицы. document["outputs"]["intakeByKind"] = { "field": "_kind", "defaultKind": "main", "targets": { "main": { "table": main_table, "upsert_keys": (_main or _targets[0]).get("upsert_keys", ["payload_id"]), "fields": main_fields, }, "child": { "table": child_table, "upsert_keys": ["payload_id"], "fields": { "payload_id": {"var": "payload_id"}, "mp_id": {"static": 1}, }, }, }, } # intakeByKind заменяет intake — удаляем отдельный intake, чтобы не было конфликта. # Описание могло УЖЕ быть только в форме intakeByKind: тогда удалять нечего. document["outputs"].pop("intake", None) spec_path.write_text( _yaml.safe_dump(document, sort_keys=False, allow_unicode=True), encoding="utf-8", ) def test_intake_only_table_not_read_by_step_passes_closure(tmp_path: Path): """Таблица в intake_only, шаг её не читает — НЕ ошибка. Это основное назначение intake_only: разрешить приёму опережать ядро, не ломая замыкание входов. """ child_table = "wb_prices_extra" copied_products = tmp_path / "products" shutil.copytree(PRODUCTS, copied_products) _make_product_spec_with_child(copied_products, "wb.prices", child_table) def mutate(document: dict) -> None: source = _source(document, "product:wb.prices") # Основная таблица остаётся в tables, дочерняя — в intake_only. source["tables"] = ["wb_prices"] source["intake_only"] = [child_table] manifest_path = _modified_manifest(tmp_path, mutate) manifest = load_build_input_manifest( manifest_path, product_specs_path=copied_products, expected_steps=EXPECTED_STEPS, ) # Замыкание входов: wb_prices читается, wb_prices_extra — нет. # Ошибки быть не должно. closure = validate_executable_source_closure(manifest) prices_relations = closure.relations_by_step["fact_prices"] assert "wb_prices" in prices_relations assert child_table not in prices_relations def test_intake_only_table_read_by_step_is_rejected(tmp_path: Path): """Таблица в intake_only, но шаг ядра её ЧИТАЕТ — ОШИБКА. Иначе intake_only становится лазейкой: вход перестают проверять, а ядро тем временем его читает. """ child_table = "wb_prices_extra" copied_products = tmp_path / "products" shutil.copytree(PRODUCTS, copied_products) _make_product_spec_with_child(copied_products, "wb.prices", child_table) def mutate(document: dict) -> None: source = _source(document, "product:wb.prices") source["tables"] = ["wb_prices"] source["intake_only"] = [child_table] manifest_path = _modified_manifest(tmp_path, mutate) manifest = load_build_input_manifest( manifest_path, product_specs_path=copied_products, expected_steps=EXPECTED_STEPS, ) # Подкладываем в fact_prices чтение intake_only-таблицы с правильной # проекцией (approved_payload_join), чтобы projection check прошёл, # а intake_only-проверка сработала. copied_etl = tmp_path / "etl" shutil.copytree(REPO / "marketplace-collector-v3/kernel/etl", copied_etl) target = copied_etl / "build_fact_prices.py" probe_code = ( "\n" + 'EXTRA_APPROVED = approved_payload_join("e", "wb.prices", "approved_wb_row")\n' + "def intake_only_probe(cur):\n" + f' cur.execute(f"SELECT e.* FROM {{I}}.{child_table} e ' + "JOIN {APPROVED_PAYLOAD_TABLE} p ON e.payload_id = p.payload_id " + """WHERE p.endpoint_code = 'wb.prices'")\n""" ) target.write_text(target.read_text(encoding="utf-8") + probe_code, encoding="utf-8") with pytest.raises( BuildInputManifestError, match=f"reads tables declared as intake_only.*{child_table}" ): validate_executable_source_closure(manifest, module_root=copied_etl) def test_intake_only_and_tables_overlap_rejected(tmp_path: Path): """Одна и та же таблица в tables и intake_only — ОШИБКА.""" def mutate(document: dict) -> None: source = _source(document, "product:wb.prices") source["intake_only"] = ["wb_prices"] # уже есть в tables manifest_path = _modified_manifest(tmp_path, mutate) with pytest.raises( BuildInputManifestError, match="tables and intake_only overlap" ): load_build_input_manifest(manifest_path, expected_steps=EXPECTED_STEPS) @pytest.mark.parametrize( ("intake_value", "match"), [ ([], "intake_only is invalid"), (["wb_prices", "wb_prices"], "intake_only is invalid"), (["UPPERCASE"], "intake_only is invalid"), ([""], "intake_only is invalid"), ], ids=["empty", "duplicate", "uppercase", "blank"], ) def test_intake_only_format_validation( tmp_path: Path, intake_value: list[str], match: str ): """intake_only обязан быть списком непустых строчных имён без повторов.""" def mutate(document: dict) -> None: source = _source(document, "product:wb.prices") source["intake_only"] = intake_value manifest_path = _modified_manifest(tmp_path, mutate) with pytest.raises(BuildInputManifestError, match=match): load_build_input_manifest(manifest_path, expected_steps=EXPECTED_STEPS) def test_product_tables_reads_both_output_forms(): """Описание объявляет вывод ДВУМЯ формами, и обе обязаны читаться. 09.08.2026 контракт читал `spec["intake"]` безусловно, и описание, переведённое целиком на `intakeByKind` (так делает разворот массивов в дочерние таблицы), ронял его с `KeyError: 'intake'`. Встала приёмка пакета с 21 описанием. """ from kernel.etl.build_input_contract import _product_tables, ProductInputContractError # только intake assert _product_tables({"intake": {"table": "wb_orders"}}) == ("wb_orders",) # только intakeByKind, БЕЗ intake — тот самый случай only_by_kind = { "intakeByKind": {"targets": {"posting": [ {"table": "ozon_postings_fbs"}, {"table": "ozon_postings_fbs_products"}, ]}} } assert _product_tables(only_by_kind) == ( "ozon_postings_fbs", "ozon_postings_fbs_products", ) # обе формы вместе — объединение без повторов both = {"intake": {"table": "ym_returns"}, "intakeByKind": {"targets": {"ret": [{"table": "ym_return_items"}, {"table": "ym_returns"}]}}} assert _product_tables(both) == ("ym_return_items", "ym_returns") # ни одной таблицы — отказ, а не пустой набор: пустой набор прошёл бы сравнение # с пустым манифестом и спрятал бы поломанное описание try: _product_tables({"quality": {}}) except ProductInputContractError as exc: assert "ни одной целевой таблицы" in str(exc) else: raise AssertionError("описание без таблиц обязано отвергаться")