ddmitry/airflow-greenplum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Улучшение тестов

Browse Source
This commit is contained in:
ddadmin
2026-01-18 22:32:57 +03:00
parent 843500cf52
commit 9e08bc181c
2 changed files with 72 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/internal/bookings_stg_code_review.md
+2 -2
View File
@@ -62,7 +62,7 @@
### 2.3. Smoke-тесты DAG’ов (статус: исправлено) ### 2.3. Smoke-тесты DAG’ов (статус: исправлено)
Что сделано: Что сделано:
- Тесты усилены: теперь проверяются ключевые зависимости графа через `get_direct_relatives("downstream")`. - Тесты усилены: теперь проверяются ключевые зависимости графа через `get_direct_relatives(upstream=False)` и “барьеры” через `get_flat_relatives(upstream=False)`.
### 2.4. Документация по DAG (статус: синхронизировано) ### 2.4. Документация по DAG (статус: синхронизировано)
@@ -135,7 +135,7 @@ LEFT JOIN stg.airports AS a
```python ```python
tickets_dq = dag.get_task("check_tickets_dq") tickets_dq = dag.get_task("check_tickets_dq")
airports_load = dag.get_task("load_airports_to_stg") 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)
``` ```
--- ---
tests/test_dags_smoke.py
+70 -65
View File
@@ -25,6 +25,22 @@ def _load_dag(module_name: str):
return getattr(mod, "dag") 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(): def test_csv_to_greenplum_dag_structure():
dag = _load_dag("airflow.dags.csv_to_greenplum") 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") t3 = dag.get_task("preview_csv")
t4 = dag.get_task("load_csv_to_greenplum") t4 = dag.get_task("load_csv_to_greenplum")
assert t2 in t1.get_direct_relatives("downstream") assert t2 in t1.get_direct_relatives(upstream=False)
assert t3 in t2.get_direct_relatives("downstream") assert t3 in t2.get_direct_relatives(upstream=False)
assert t4 in t3.get_direct_relatives("downstream") assert t4 in t3.get_direct_relatives(upstream=False)
def test_csv_to_greenplum_dq_dag_structure(): 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") d = dag.get_task("check_order_duplicates")
q = dag.get_task("data_quality_summary") q = dag.get_task("data_quality_summary")
assert s in e.get_direct_relatives("downstream") assert s in e.get_direct_relatives(upstream=False)
assert h in s.get_direct_relatives("downstream") assert h in s.get_direct_relatives(upstream=False)
assert d in h.get_direct_relatives("downstream") assert d in h.get_direct_relatives(upstream=False)
assert q in d.get_direct_relatives("downstream") assert q in d.get_direct_relatives(upstream=False)
def test_bookings_stg_ddl_dag_structure(): 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()) assert expected_tasks.issubset(dag.task_dict.keys())
# Линейные зависимости: bookings/tickets → справочники → транзакции # Smoke-test графа: проверяем ключевые инварианты, не фиксируя линейный порядок.
t_bookings = dag.get_task("apply_stg_bookings_ddl") # Это позволяет в будущем распараллеливать независимые DDL-задачи.
t_tickets = dag.get_task("apply_stg_tickets_ddl") _assert_reachable(dag, "apply_stg_bookings_ddl", "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")
assert t_tickets in t_bookings.get_direct_relatives("downstream") for task_id in expected_tasks - {"apply_stg_bookings_ddl"}:
assert t_airports in t_tickets.get_direct_relatives("downstream") _assert_reachable(dag, "apply_stg_bookings_ddl", task_id)
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")
def test_bookings_to_gp_stage_dag_structure(): 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()) assert expected_tasks.issubset(dag.task_dict.keys())
# Линейные зависимости: bookings/tickets → справочники → транзакции → финальный лог # Smoke-test графа: проверяем инварианты, не фиксируя линейный порядок.
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")
assert t_bookings in t_generate.get_direct_relatives("downstream") # Базовая цепочка должна сохраниться: генерация → bookings → DQ → tickets → DQ.
assert t_bookings_dq in t_bookings.get_direct_relatives("downstream") _assert_reachable(dag, "generate_bookings_day", "load_bookings_to_stg")
assert t_tickets in t_bookings_dq.get_direct_relatives("downstream") _assert_reachable(dag, "load_bookings_to_stg", "check_row_counts")
assert t_tickets_dq in t_tickets.get_direct_relatives("downstream") _assert_reachable(dag, "check_row_counts", "load_tickets_to_stg")
assert t_airports in t_tickets_dq.get_direct_relatives("downstream") _assert_reachable(dag, "load_tickets_to_stg", "check_tickets_dq")
assert t_airports_dq in t_airports.get_direct_relatives("downstream")
assert t_airplanes in t_airports_dq.get_direct_relatives("downstream") # Инвариант "load → dq" для каждой таблицы.
assert t_airplanes_dq in t_airplanes.get_direct_relatives("downstream") load_to_dq = [
assert t_routes in t_airplanes_dq.get_direct_relatives("downstream") ("load_bookings_to_stg", "check_row_counts"),
assert t_routes_dq in t_routes.get_direct_relatives("downstream") ("load_tickets_to_stg", "check_tickets_dq"),
assert t_seats in t_routes_dq.get_direct_relatives("downstream") ("load_airports_to_stg", "check_airports_dq"),
assert t_seats_dq in t_seats.get_direct_relatives("downstream") ("load_airplanes_to_stg", "check_airplanes_dq"),
assert t_flights in t_seats_dq.get_direct_relatives("downstream") ("load_routes_to_stg", "check_routes_dq"),
assert t_flights_dq in t_flights.get_direct_relatives("downstream") ("load_seats_to_stg", "check_seats_dq"),
assert t_segments in t_flights_dq.get_direct_relatives("downstream") ("load_flights_to_stg", "check_flights_dq"),
assert t_segments_dq in t_segments.get_direct_relatives("downstream") ("load_segments_to_stg", "check_segments_dq"),
assert t_boarding in t_segments_dq.get_direct_relatives("downstream") ("load_boarding_passes_to_stg", "check_boarding_passes_dq"),
assert t_boarding_dq in t_boarding.get_direct_relatives("downstream") ]
assert t_finish in t_boarding_dq.get_direct_relatives("downstream") 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")