182 lines
7.1 KiB
Python
182 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
|
|
import pytest
|
|
|
|
|
|
def _airflow_available() -> bool:
|
|
try:
|
|
af = importlib.import_module("airflow")
|
|
except Exception:
|
|
return False
|
|
# Real Airflow exposes DAG at top-level
|
|
return hasattr(af, "DAG")
|
|
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not _airflow_available(), reason="Airflow is not installed for DAG smoke tests"
|
|
)
|
|
|
|
|
|
def _load_dag(module_name: str):
|
|
mod = importlib.import_module(module_name)
|
|
assert hasattr(mod, "dag"), f"{module_name} must expose variable 'dag'"
|
|
return getattr(mod, "dag")
|
|
|
|
|
|
def test_csv_to_greenplum_dag_structure():
|
|
dag = _load_dag("airflow.dags.csv_to_greenplum")
|
|
|
|
# tasks
|
|
expected_tasks = {
|
|
"create_orders_table",
|
|
"generate_csv",
|
|
"preview_csv",
|
|
"load_csv_to_greenplum",
|
|
}
|
|
assert expected_tasks.issubset(dag.task_dict.keys())
|
|
|
|
# linear dependencies
|
|
t1 = dag.get_task("create_orders_table")
|
|
t2 = dag.get_task("generate_csv")
|
|
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")
|
|
|
|
|
|
def test_csv_to_greenplum_dq_dag_structure():
|
|
dag = _load_dag("airflow.dags.csv_to_greenplum_dq")
|
|
|
|
expected_tasks = {
|
|
"check_orders_table_exists",
|
|
"check_orders_schema",
|
|
"check_orders_has_rows",
|
|
"check_order_duplicates",
|
|
"data_quality_summary",
|
|
}
|
|
assert expected_tasks.issubset(dag.task_dict.keys())
|
|
|
|
e = dag.get_task("check_orders_table_exists")
|
|
s = dag.get_task("check_orders_schema")
|
|
h = dag.get_task("check_orders_has_rows")
|
|
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")
|
|
|
|
|
|
def test_bookings_stg_ddl_dag_structure():
|
|
"""Проверка структуры DAG bookings_stg_ddl."""
|
|
dag = _load_dag("airflow.dags.bookings_stg_ddl")
|
|
|
|
expected_tasks = {
|
|
"apply_stg_bookings_ddl",
|
|
"apply_stg_tickets_ddl",
|
|
"apply_stg_airports_ddl",
|
|
"apply_stg_airplanes_ddl",
|
|
"apply_stg_routes_ddl",
|
|
"apply_stg_seats_ddl",
|
|
"apply_stg_flights_ddl",
|
|
"apply_stg_segments_ddl",
|
|
"apply_stg_boarding_passes_ddl",
|
|
}
|
|
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")
|
|
|
|
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")
|
|
|
|
|
|
def test_bookings_to_gp_stage_dag_structure():
|
|
"""Проверка структуры DAG bookings_to_gp_stage."""
|
|
dag = _load_dag("airflow.dags.bookings_to_gp_stage")
|
|
|
|
expected_tasks = {
|
|
"generate_bookings_day",
|
|
"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",
|
|
"finish_summary",
|
|
}
|
|
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")
|
|
|
|
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")
|