ddadminandClaude Fable 5
10ee1a1cb6
- Зачем:
- менти собирал мир генерацией (минуты и десятки минут CPU); теперь
готовый трёхдневный мир загружается импортом за ~2 минуты, и числа
у всех менти совпадают число-в-число (issue #3).
- Что:
- артефакт data/startup_history/reference-world.json.xz в git:
3 модельных дня daily-wave, ~850 МБ JSON → 32 МБ xz;
- чтение и запись артефакта понимают .xz потоково (lzma); пустое поле
artifact_path в пульте и make startup-history-import читают эталон;
- длительность профиля daily-wave стала 3d — в тон эталонному миру;
- предпроверка чистого стенда ставит зависимости генератора через uv;
экспорт и импорт разведены отдельными переменными Makefile;
- доки и runbook обновлены; новые контрактные тесты: xz round-trip
и дефолтные пути артефакта.
- Проверка:
- make test (210 + 31) и make lint зелёные; импорт на чистом стенде
за 2м03с, manifest совпал (280437 событий), Superset-проверка
зелёная; независимое ревью — APPROVED.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
306 lines
11 KiB
Python
306 lines
11 KiB
Python
"""
|
||
Тесты чистой логики пульта Airflow для генератора.
|
||
"""
|
||
|
||
from dataclasses import replace
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
|
||
def test_default_artifact_paths_keep_import_and_backfill_separate():
|
||
"""Import читает эталонный мир, а backfill пишет в прежний файл."""
|
||
from clickstream_generator.airflow_control import default_artifact_path
|
||
|
||
assert default_artifact_path("import") == (
|
||
"/opt/airflow/data/startup_history/reference-world.json.xz"
|
||
)
|
||
assert default_artifact_path("backfill") == (
|
||
"/opt/airflow/data/startup-history.json"
|
||
)
|
||
|
||
|
||
def test_build_control_env_uses_profile_duration_and_airflow_data_dir():
|
||
"""Пульт строит env для backfill из профиля и каталога Airflow."""
|
||
from clickstream_generator.airflow_control import build_control_env
|
||
|
||
env = build_control_env(
|
||
"backfill",
|
||
profile_name="ci",
|
||
duration="",
|
||
artifact_path="/opt/airflow/data/startup-history.json",
|
||
overrides={"GEN_LAMBDA_BASE_PER_MIN": "120", "GEN_SEED": "99"},
|
||
)
|
||
|
||
assert env["GEN_RUN_MODE"] == "backfill"
|
||
assert env["GEN_HISTORY_DURATION"] == "6h"
|
||
assert env["GEN_MODEL_T_END"] == "2026-01-01T06:00:00+00:00"
|
||
assert env["GEN_DATA_DIR"] == "/opt/airflow/data"
|
||
assert env["GEN_METRICS_ENABLED"] == "false"
|
||
assert env["GEN_STARTUP_HISTORY_ARTIFACT"] == "/opt/airflow/data/startup-history.json"
|
||
assert env["GEN_LAMBDA_BASE_PER_MIN"] == "120"
|
||
assert env["GEN_SEED"] == "99"
|
||
|
||
|
||
def test_import_env_requires_artifact_path_and_uses_backfill_contract():
|
||
"""Import валидируется как тот же мир, что backfill."""
|
||
from clickstream_generator.airflow_control import build_control_env
|
||
|
||
with pytest.raises(ValueError, match="artifact_path"):
|
||
build_control_env("import", profile_name="ci", artifact_path="")
|
||
|
||
env = build_control_env(
|
||
"import",
|
||
profile_name="daily-wave",
|
||
artifact_path="/opt/airflow/data/history.json",
|
||
)
|
||
|
||
assert env["GEN_RUN_MODE"] == "backfill"
|
||
assert env["GEN_STATE_RESET"] == "true"
|
||
assert env["GEN_HISTORY_DURATION"] == "3d"
|
||
|
||
|
||
def test_next_day_env_uses_world_settings_and_current_manifest_boundary():
|
||
"""Next-day восстанавливает настройки мира из manifest, а не из формы DAG."""
|
||
from clickstream_generator.airflow_control import build_next_day_env
|
||
|
||
manifest = {
|
||
"gen_seed": 42,
|
||
"model_t0": "2026-01-01T00:00:00+00:00",
|
||
"model_t_end": "2026-01-03T00:00:00+00:00",
|
||
"model_timezone": "UTC",
|
||
"launch_profile": "daily-wave",
|
||
"generation_settings": {
|
||
"tick_seconds": 1,
|
||
"lambda_base_per_min": 60,
|
||
"jitter_pct": 0,
|
||
"min_events_per_tick": 1,
|
||
"max_events_per_tick": 1000,
|
||
"max_session_events": 30,
|
||
"max_active_sessions": 200,
|
||
"population_max": 300,
|
||
"p_new_user": 0.15,
|
||
"min_return_minutes": 30,
|
||
"model_time_speed": 60,
|
||
},
|
||
}
|
||
|
||
env = build_next_day_env(manifest)
|
||
|
||
assert env["GEN_RUN_MODE"] == "next-day"
|
||
assert env["GEN_STATE_RESET"] == "false"
|
||
assert env["GEN_MODEL_T_END"] == "2026-01-03T00:00:00+00:00"
|
||
assert env["GEN_MODEL_TIME_SPEED"] == "60"
|
||
assert env["GEN_MAX_SESSION_EVENTS"] == "30"
|
||
assert env["GEN_DATA_DIR"] == "/opt/airflow/data"
|
||
|
||
|
||
def test_next_day_env_map_covers_every_generation_setting(base_config):
|
||
"""Новая настройка генерации не может потеряться между manifest и env."""
|
||
from clickstream_generator.airflow_control import NEXT_DAY_SETTING_ENV_KEYS
|
||
from clickstream_generator.startup_history_artifact import (
|
||
generation_settings_from_config,
|
||
)
|
||
|
||
assert set(NEXT_DAY_SETTING_ENV_KEYS) == set(
|
||
generation_settings_from_config(base_config)
|
||
)
|
||
|
||
|
||
def test_next_day_precheck_rejects_stale_expected_boundary_with_both_values():
|
||
"""Повторный запуск с прежней границей падает до записи данных."""
|
||
from clickstream_generator.airflow_control import assert_expected_t_end
|
||
|
||
with pytest.raises(RuntimeError) as exc_info:
|
||
assert_expected_t_end(
|
||
"2026-01-03T00:00:00+00:00",
|
||
"2026-01-04T00:00:00+00:00",
|
||
)
|
||
|
||
message = str(exc_info.value)
|
||
assert "2026-01-03T00:00:00+00:00" in message
|
||
assert "2026-01-04T00:00:00+00:00" in message
|
||
|
||
|
||
def test_next_day_precheck_requires_manifest_and_matching_state():
|
||
"""Next-day громко отвергает пустой стенд и state не на T_end."""
|
||
from clickstream_generator.airflow_control import assert_next_day_snapshot
|
||
from test_startup_history_artifact import _state
|
||
|
||
state = _state()
|
||
with pytest.raises(RuntimeError, match="Manifest"):
|
||
assert_next_day_snapshot(None, state)
|
||
|
||
manifest = {
|
||
"run_mode": "backfill",
|
||
"model_t0": state.model_t0.isoformat(),
|
||
"model_t_end": "2026-01-01T02:00:00+00:00",
|
||
"gen_seed": state.gen_seed,
|
||
"model_timezone": state.model_timezone,
|
||
"state_version": state.version,
|
||
"state": {
|
||
"last_batch_id": state.last_batch_id,
|
||
"model_timestamp": "2026-01-01T02:00:00+00:00",
|
||
},
|
||
"generation_settings": {"model_time_speed": state.model_time_speed},
|
||
}
|
||
with pytest.raises(RuntimeError, match="state.*T_end"):
|
||
assert_next_day_snapshot(manifest, state)
|
||
|
||
|
||
def test_target_dag_trigger_error_covers_missing_paused_and_ready_states():
|
||
"""Проверка зависимого DAG различает три реальные ветки."""
|
||
from clickstream_generator.airflow_control import target_dag_trigger_error
|
||
|
||
assert "ещё не найден" in target_dag_trigger_error("etl_pipeline", None)
|
||
assert "стоит на паузе" in target_dag_trigger_error(
|
||
"etl_pipeline",
|
||
SimpleNamespace(is_paused=True),
|
||
)
|
||
assert target_dag_trigger_error(
|
||
"etl_pipeline",
|
||
SimpleNamespace(is_paused=False),
|
||
) is None
|
||
|
||
|
||
def test_assert_stand_clean_rejects_non_empty_stg_before_writes():
|
||
"""Backfill/import не стартуют на непустом STG."""
|
||
from clickstream_generator.airflow_control import assert_stg_tables_empty
|
||
|
||
class Hook:
|
||
def execute(self, sql):
|
||
self.sql = sql
|
||
return [(0, 2, 0, 0)]
|
||
|
||
hook = Hook()
|
||
|
||
with pytest.raises(RuntimeError, match="make clean"):
|
||
assert_stg_tables_empty(hook)
|
||
assert "stg.browser_raw" in hook.sql
|
||
assert "stg.location_raw" in hook.sql
|
||
|
||
|
||
def test_assert_stand_clean_rejects_non_empty_kafka_with_make_clean_hint(monkeypatch):
|
||
"""Непустые Kafka-топики дают ту же подсказку про make clean."""
|
||
from clickstream_generator import airflow_control, stand_clean
|
||
|
||
class Inspector:
|
||
def __init__(self, bootstrap_servers):
|
||
self.bootstrap_servers = bootstrap_servers
|
||
|
||
def assert_data_topics_empty(self):
|
||
raise RuntimeError("Kafka data topics are not empty: browser_events")
|
||
|
||
class Hook:
|
||
def execute(self, sql):
|
||
return [(0, 0, 0, 0)]
|
||
|
||
monkeypatch.setattr(stand_clean, "KafkaTopicInspector", Inspector)
|
||
|
||
with pytest.raises(RuntimeError, match="make clean"):
|
||
airflow_control.assert_stand_clean("kafka:29092", Hook())
|
||
|
||
|
||
def test_shared_stand_clean_rejects_host_stg_counts():
|
||
"""Хостовая предпроверка использует ту же STG-границу, что пульт."""
|
||
from clickstream_generator.stand_clean import assert_stg_counts_empty
|
||
|
||
with pytest.raises(RuntimeError, match="stg.location_raw=3.*make clean"):
|
||
assert_stg_counts_empty((0, 3, 0, 0))
|
||
|
||
|
||
def test_continue_rejects_dirty_stand_without_state(base_config, monkeypatch):
|
||
"""Continue без state не стартует новый мир поверх грязного STG."""
|
||
from clickstream_generator import stand_clean
|
||
|
||
class StateManager:
|
||
def __init__(self, bootstrap_servers):
|
||
self.bootstrap_servers = bootstrap_servers
|
||
|
||
def load(self):
|
||
return None
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
class Inspector:
|
||
def __init__(self, bootstrap_servers):
|
||
self.bootstrap_servers = bootstrap_servers
|
||
|
||
def assert_data_topics_empty(self):
|
||
pass
|
||
|
||
monkeypatch.setattr(stand_clean, "KafkaStateManager", StateManager)
|
||
monkeypatch.setattr(stand_clean, "KafkaTopicInspector", Inspector)
|
||
|
||
with pytest.raises(RuntimeError, match="stg.browser_raw=1.*make clean"):
|
||
stand_clean.assert_continue_has_state_or_clean_stand(
|
||
"localhost:9092",
|
||
(1, 0, 0, 0),
|
||
base_config,
|
||
)
|
||
|
||
|
||
def test_check_manifest_compares_clickhouse_stats(base_config):
|
||
"""Check падает, когда контрольные числа ClickHouse расходятся с manifest."""
|
||
from clickstream_generator.airflow_control import assert_clickhouse_matches_manifest
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
)
|
||
from test_startup_history_artifact import _batch, _state
|
||
|
||
state = _state()
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
manifest = build_manifest(
|
||
config=replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
),
|
||
counters=builder.counters,
|
||
state=state,
|
||
)
|
||
|
||
class Hook:
|
||
def execute(self, sql):
|
||
self.sql = sql
|
||
return [(
|
||
0,
|
||
1,
|
||
1,
|
||
"2026-01-01 00:00:00.000000",
|
||
"2026-01-01 00:00:00.000000",
|
||
)]
|
||
|
||
with pytest.raises(RuntimeError, match="events"):
|
||
assert_clickhouse_matches_manifest(manifest, Hook())
|
||
|
||
|
||
def test_generator_metrics_server_can_be_disabled(base_config, monkeypatch):
|
||
"""Airflow-задача может запускать генератор без HTTP-сервера метрик."""
|
||
from clickstream_generator.service import GeneratorService
|
||
|
||
config = replace(base_config, enabled=False, metrics_enabled=False)
|
||
service = GeneratorService(config)
|
||
called = False
|
||
|
||
def start_http_server(_port):
|
||
nonlocal called
|
||
called = True
|
||
|
||
monkeypatch.setattr(
|
||
"clickstream_generator.service.start_http_server",
|
||
start_http_server,
|
||
)
|
||
|
||
service.start()
|
||
|
||
assert called is False
|