clickstream-ch-kafka-superset-demo/generator/tests/test_airflow_control.py

"""
Тесты чистой логики пульта Airflow для генератора.
"""

from dataclasses import replace
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace

import pytest


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"] == "2d"


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