From 159ee99a3258c7734f3a7dc041a66cc39342c2e7 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 18 Jan 2026 22:32:57 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A3=D0=BB=D1=83=D1=87=D1=88=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D0=B5=20=D1=82=D0=B5=D1=81=D1=82=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/internal/bookings_stg_code_review.md | 4 +- tests/test_dags_smoke.py | 135 +++++++++++----------- 2 files changed, 72 insertions(+), 67 deletions(-) diff --git a/docs/internal/bookings_stg_code_review.md b/docs/internal/bookings_stg_code_review.md index 3e33530..8a909b2 100644 --- a/docs/internal/bookings_stg_code_review.md +++ b/docs/internal/bookings_stg_code_review.md @@ -62,7 +62,7 @@ ### 2.3. Smoke-тесты DAG’ов (статус: исправлено) Что сделано: -- Тесты усилены: теперь проверяются ключевые зависимости графа через `get_direct_relatives("downstream")`. +- Тесты усилены: теперь проверяются ключевые зависимости графа через `get_direct_relatives(upstream=False)` и “барьеры” через `get_flat_relatives(upstream=False)`. ### 2.4. Документация по DAG (статус: синхронизировано) @@ -135,7 +135,7 @@ LEFT JOIN stg.airports AS a ```python tickets_dq = dag.get_task("check_tickets_dq") airports_load = dag.get_task("load_airports_to_stg") -assert airports_load in tickets_dq.get_direct_relatives("downstream") +assert airports_load in tickets_dq.get_direct_relatives(upstream=False) ``` --- diff --git a/tests/test_dags_smoke.py b/tests/test_dags_smoke.py index 7e76ab8..100b722 100644 --- a/tests/test_dags_smoke.py +++ b/tests/test_dags_smoke.py @@ -25,6 +25,22 @@ def _load_dag(module_name: str): return getattr(mod, "dag") +def _assert_direct_edge(dag, upstream_task_id: str, downstream_task_id: str) -> None: + upstream = dag.get_task(upstream_task_id) + downstream = dag.get_task(downstream_task_id) + assert downstream in upstream.get_direct_relatives(upstream=False), ( + f"Expected direct edge {upstream_task_id} -> {downstream_task_id}" + ) + + +def _assert_reachable(dag, upstream_task_id: str, downstream_task_id: str) -> None: + upstream = dag.get_task(upstream_task_id) + downstream = dag.get_task(downstream_task_id) + assert downstream in upstream.get_flat_relatives(upstream=False), ( + f"Expected {downstream_task_id} to be downstream of {upstream_task_id}" + ) + + def test_csv_to_greenplum_dag_structure(): dag = _load_dag("airflow.dags.csv_to_greenplum") @@ -43,9 +59,9 @@ def test_csv_to_greenplum_dag_structure(): t3 = dag.get_task("preview_csv") t4 = dag.get_task("load_csv_to_greenplum") - assert t2 in t1.get_direct_relatives("downstream") - assert t3 in t2.get_direct_relatives("downstream") - assert t4 in t3.get_direct_relatives("downstream") + assert t2 in t1.get_direct_relatives(upstream=False) + assert t3 in t2.get_direct_relatives(upstream=False) + assert t4 in t3.get_direct_relatives(upstream=False) def test_csv_to_greenplum_dq_dag_structure(): @@ -66,10 +82,10 @@ def test_csv_to_greenplum_dq_dag_structure(): d = dag.get_task("check_order_duplicates") q = dag.get_task("data_quality_summary") - assert s in e.get_direct_relatives("downstream") - assert h in s.get_direct_relatives("downstream") - assert d in h.get_direct_relatives("downstream") - assert q in d.get_direct_relatives("downstream") + assert s in e.get_direct_relatives(upstream=False) + assert h in s.get_direct_relatives(upstream=False) + assert d in h.get_direct_relatives(upstream=False) + assert q in d.get_direct_relatives(upstream=False) def test_bookings_stg_ddl_dag_structure(): @@ -89,25 +105,12 @@ def test_bookings_stg_ddl_dag_structure(): } assert expected_tasks.issubset(dag.task_dict.keys()) - # Линейные зависимости: bookings/tickets → справочники → транзакции - t_bookings = dag.get_task("apply_stg_bookings_ddl") - t_tickets = dag.get_task("apply_stg_tickets_ddl") - t_airports = dag.get_task("apply_stg_airports_ddl") - t_airplanes = dag.get_task("apply_stg_airplanes_ddl") - t_routes = dag.get_task("apply_stg_routes_ddl") - t_seats = dag.get_task("apply_stg_seats_ddl") - t_flights = dag.get_task("apply_stg_flights_ddl") - t_segments = dag.get_task("apply_stg_segments_ddl") - t_boarding = dag.get_task("apply_stg_boarding_passes_ddl") + # Smoke-test графа: проверяем ключевые инварианты, не фиксируя линейный порядок. + # Это позволяет в будущем распараллеливать независимые DDL-задачи. + _assert_reachable(dag, "apply_stg_bookings_ddl", "apply_stg_tickets_ddl") - assert t_tickets in t_bookings.get_direct_relatives("downstream") - assert t_airports in t_tickets.get_direct_relatives("downstream") - assert t_airplanes in t_airports.get_direct_relatives("downstream") - assert t_routes in t_airplanes.get_direct_relatives("downstream") - assert t_seats in t_routes.get_direct_relatives("downstream") - assert t_flights in t_seats.get_direct_relatives("downstream") - assert t_segments in t_flights.get_direct_relatives("downstream") - assert t_boarding in t_segments.get_direct_relatives("downstream") + for task_id in expected_tasks - {"apply_stg_bookings_ddl"}: + _assert_reachable(dag, "apply_stg_bookings_ddl", task_id) def test_bookings_to_gp_stage_dag_structure(): @@ -138,44 +141,46 @@ def test_bookings_to_gp_stage_dag_structure(): } assert expected_tasks.issubset(dag.task_dict.keys()) - # Линейные зависимости: bookings/tickets → справочники → транзакции → финальный лог - t_generate = dag.get_task("generate_bookings_day") - t_bookings = dag.get_task("load_bookings_to_stg") - t_bookings_dq = dag.get_task("check_row_counts") - t_tickets = dag.get_task("load_tickets_to_stg") - t_tickets_dq = dag.get_task("check_tickets_dq") - t_airports = dag.get_task("load_airports_to_stg") - t_airports_dq = dag.get_task("check_airports_dq") - t_airplanes = dag.get_task("load_airplanes_to_stg") - t_airplanes_dq = dag.get_task("check_airplanes_dq") - t_routes = dag.get_task("load_routes_to_stg") - t_routes_dq = dag.get_task("check_routes_dq") - t_seats = dag.get_task("load_seats_to_stg") - t_seats_dq = dag.get_task("check_seats_dq") - t_flights = dag.get_task("load_flights_to_stg") - t_flights_dq = dag.get_task("check_flights_dq") - t_segments = dag.get_task("load_segments_to_stg") - t_segments_dq = dag.get_task("check_segments_dq") - t_boarding = dag.get_task("load_boarding_passes_to_stg") - t_boarding_dq = dag.get_task("check_boarding_passes_dq") - t_finish = dag.get_task("finish_summary") + # Smoke-test графа: проверяем инварианты, не фиксируя линейный порядок. + # Это позволяет в будущем распараллеливать независимые загрузки справочников/транзакций. - assert t_bookings in t_generate.get_direct_relatives("downstream") - assert t_bookings_dq in t_bookings.get_direct_relatives("downstream") - assert t_tickets in t_bookings_dq.get_direct_relatives("downstream") - assert t_tickets_dq in t_tickets.get_direct_relatives("downstream") - assert t_airports in t_tickets_dq.get_direct_relatives("downstream") - assert t_airports_dq in t_airports.get_direct_relatives("downstream") - assert t_airplanes in t_airports_dq.get_direct_relatives("downstream") - assert t_airplanes_dq in t_airplanes.get_direct_relatives("downstream") - assert t_routes in t_airplanes_dq.get_direct_relatives("downstream") - assert t_routes_dq in t_routes.get_direct_relatives("downstream") - assert t_seats in t_routes_dq.get_direct_relatives("downstream") - assert t_seats_dq in t_seats.get_direct_relatives("downstream") - assert t_flights in t_seats_dq.get_direct_relatives("downstream") - assert t_flights_dq in t_flights.get_direct_relatives("downstream") - assert t_segments in t_flights_dq.get_direct_relatives("downstream") - assert t_segments_dq in t_segments.get_direct_relatives("downstream") - assert t_boarding in t_segments_dq.get_direct_relatives("downstream") - assert t_boarding_dq in t_boarding.get_direct_relatives("downstream") - assert t_finish in t_boarding_dq.get_direct_relatives("downstream") + # Базовая цепочка должна сохраниться: генерация → bookings → DQ → tickets → DQ. + _assert_reachable(dag, "generate_bookings_day", "load_bookings_to_stg") + _assert_reachable(dag, "load_bookings_to_stg", "check_row_counts") + _assert_reachable(dag, "check_row_counts", "load_tickets_to_stg") + _assert_reachable(dag, "load_tickets_to_stg", "check_tickets_dq") + + # Инвариант "load → dq" для каждой таблицы. + load_to_dq = [ + ("load_bookings_to_stg", "check_row_counts"), + ("load_tickets_to_stg", "check_tickets_dq"), + ("load_airports_to_stg", "check_airports_dq"), + ("load_airplanes_to_stg", "check_airplanes_dq"), + ("load_routes_to_stg", "check_routes_dq"), + ("load_seats_to_stg", "check_seats_dq"), + ("load_flights_to_stg", "check_flights_dq"), + ("load_segments_to_stg", "check_segments_dq"), + ("load_boarding_passes_to_stg", "check_boarding_passes_dq"), + ] + for load_task_id, dq_task_id in load_to_dq: + _assert_direct_edge(dag, load_task_id, dq_task_id) + + # Барьеры по данным (не обязательно прямые рёбра). + # routes_dq использует airports/airplanes текущего batch_id. + _assert_reachable(dag, "check_airports_dq", "check_routes_dq") + _assert_reachable(dag, "check_airplanes_dq", "check_routes_dq") + + # seats_dq использует airplanes текущего batch_id. + _assert_reachable(dag, "check_airplanes_dq", "check_seats_dq") + + # flights_dq использует routes текущего batch_id. + _assert_reachable(dag, "check_routes_dq", "check_flights_dq") + + # segments_dq проверяет наличие flights (STG-история); для первой загрузки flights должны быть до segments. + _assert_reachable(dag, "check_flights_dq", "check_segments_dq") + + # boarding_passes_dq проверяет наличие segments/tickets (STG-история); для первой загрузки segments должны быть до DQ. + _assert_reachable(dag, "check_segments_dq", "check_boarding_passes_dq") + + # Финальная сводка должна быть в конце графа. + _assert_reachable(dag, "check_boarding_passes_dq", "finish_summary")