from pathlib import Path import os import re import subprocess import textwrap import pytest REPO_ROOT = Path(__file__).resolve().parents[1] CHECK_SCRIPT = REPO_ROOT / "scripts" / "check_generated_analytics.sh" COMPOSE_FILE = REPO_ROOT / "docker-compose.yml" OPERATIONS_DOC = REPO_ROOT / "docs" / "OPERATIONS.md" CHAIN_CHECK_SCRIPT = REPO_ROOT / "scripts" / "check_generated_history_chain.sh" def _required_runtime_line(script, command): matches = [ line_number for line_number, line in enumerate(script.splitlines()) if line == command ] assert len(matches) == 1, f"{command} должна быть отдельной командой ровно один раз" return matches[0] def _fake_compose( tmp_path, *, manifest_profile="ci", live_rows="3", live_pairing="3\t0\t0\t0", seam_context="19\t19\t19", views_mode="ok", ): fake = tmp_path / "docker-compose" fake.write_text( textwrap.dedent( f"""\ #!/usr/bin/env bash set -euo pipefail args="$*" query="" previous="" for arg in "$@"; do if [[ "$previous" == "--query" ]]; then query="$arg" break fi previous="$arg" done if [[ "$args" == *"kafka-manifest-summary"* ]]; then printf '%s\\n' '16054\t2800\t930\t2026-01-01T00:00:00+00:00\t2026-01-01T05:59:00+00:00\t2026-01-01T00:00:00+00:00\t2026-01-01T06:00:00+00:00\t{manifest_profile}' exit 0 fi if [[ "$args" == *"clickhouse-client"* ]]; then if [[ "$query" == *"hex(sipHash128"* ]]; then printf '%s\\n' '16054\t2800\t930\t2026-01-01 00:00:00.000000\t2026-01-01 05:59:00.000000\t1\t1\t1\tdeadbeef' elif [[ "$query" == *"returning_users / users"* ]]; then printf '%s\\n' '930\t120\t0.129\t4' elif [[ "$query" == *"short_visit_share"* ]]; then printf '%s\\n' '2800\t0.2\t5\t5.7\t0\t120\t900\t18' elif [[ "$query" == *"minIf(event_ts"* ]]; then printf '%s\\n' '2800\t2100\t1500\t900\t650\t1\t0.23' elif [[ "$query" == *"has_home"* ]]; then printf '%s\\n' '2800\t2200\t1600\t950\t700\t1\t0.25' elif [[ "$query" == *"SELECT count()"* && "$query" == *"FROM dds.event"* ]]; then printf '%s\\n' '{live_rows}' elif [[ "$query" == *"uniqExact(event_id)"* ]]; then printf '%s\\n' '16200\t16200\t0' elif [[ "$query" == *"unpaired_location_rows"* ]]; then printf '%s\\n' '{live_pairing}' elif [[ "$query" == *"per_event_homogeneous_visits"* ]]; then printf '%s\\n' '{seam_context}' elif [[ "$query" == *"ods_device_rows"* ]]; then printf '%s\\n' '19\t19\t19\t0\t0' elif [[ "$query" == *"SELECT source, rows"* ]]; then if [[ "{views_mode}" == "fail" ]]; then exit 42 fi printf '%s\\n' \ 'dm.dq_summary\t1' \ 'dm.v_daily_traffic\t1' \ 'dm.v_events_enriched\t16054' \ 'dm.v_session_overview\t1' \ 'dm.v_top_pages_daily\t1' \ 'dm.v_utm_effectiveness\t1' else echo "unexpected ClickHouse query: $query" >&2 exit 91 fi exit 0 fi echo "unexpected compose call: $args" >&2 exit 92 """ ), encoding="utf-8", ) fake.chmod(0o755) return fake def _run_generated_history_check(tmp_path, *, fake_compose, **env_overrides): env = os.environ.copy() env.update( { "COMPOSE_BIN": str(fake_compose), "REQUIRE_SUPERSET": "0", "WAIT_LIVE_ROWS_SECONDS": "0", } ) env.update(env_overrides) return subprocess.run( ["bash", str(CHECK_SCRIPT)], cwd=REPO_ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30, check=False, ) def test_generated_history_check_uses_actual_manifest_boundary_and_profile(): """Проверка берёт стык и профиль из manifest фактического прогона.""" script = (REPO_ROOT / "scripts" / "check_generated_analytics.sh").read_text( encoding="utf-8" ) makefile = (REPO_ROOT / "Makefile").read_text(encoding="utf-8") assert "kafka-manifest-summary" in script assert "manifest_model_t_end" in script assert "CH_MODEL_T_END=\"$(clickhouse_datetime_literal \"${manifest_model_t_end}\")\"" in script assert "PROFILE ?= daily-wave" in makefile assert "GEN_LAUNCH_PROFILE" in makefile assert "PROFILE" in makefile def test_generated_history_check_does_not_skip_required_live_seam(): """Обычная проверка требует реальные live-строки после стыка.""" script = (REPO_ROOT / "scripts" / "check_generated_analytics.sh").read_text( encoding="utf-8" ) makefile = (REPO_ROOT / "Makefile").read_text(encoding="utf-8") assert 'CHECK_LIVE_SEAM="${CHECK_LIVE_SEAM:-1}"' in script assert 'if [[ "${should_check_live_seam}" == "1" ]] && (( live_rows == 0 )); then' in script assert "live-продолжение не записало строки" in script assert 'CHECK_LIVE_SEAM="$${CHECK_LIVE_SEAM:-1}"' in makefile def test_generated_history_check_rejects_manifest_profile_mismatch(tmp_path): """Проверка падает, если ожидаемый профиль не совпал с manifest.""" fake_compose = _fake_compose(tmp_path, manifest_profile="ci") result = _run_generated_history_check( tmp_path, fake_compose=fake_compose, PROFILE="daily-wave", CHECK_LIVE_SEAM="0", ) assert result.returncode != 0 assert "профиля daily-wave, но manifest от ci" in result.stderr def test_generated_history_check_requires_live_rows_when_seam_is_required(tmp_path): """CHECK_LIVE_SEAM=1 падает, если live-продолжение не записало строк.""" fake_compose = _fake_compose(tmp_path, live_rows="0") result = _run_generated_history_check( tmp_path, fake_compose=fake_compose, PROFILE="ci", CHECK_LIVE_SEAM="1", ) assert result.returncode != 0 assert "live-продолжение не записало строки" in result.stderr def test_generated_history_check_names_unpaired_live_rows(tmp_path): """Неполный live-batch получает отдельную ошибку до проверки фактуры.""" fake_compose = _fake_compose( tmp_path, live_pairing="3\t1\t0\t0", seam_context="19\t18\t19", ) result = _run_generated_history_check( tmp_path, fake_compose=fake_compose, PROFILE="ci", CHECK_LIVE_SEAM="1", ) assert result.returncode != 0 assert "непарные live-строки: location=1, device=0, geo=0 из browser=3" in result.stderr assert "per-event фактура меняется на стыке" not in result.stderr def test_live_pairing_precheck_counts_device_and_geo_messages_in_stg(): """Старая ODS-строка click_id не скрывает пропуск live device/geo.""" script = CHECK_SCRIPT.read_text(encoding="utf-8") live_pairing_query = script.split('live_pairing_query="', 1)[1].split( 'FORMAT TabSeparated"', 1, )[0] assert "FROM stg.browser_raw" in live_pairing_query assert "FROM stg.location_raw" in live_pairing_query assert "FROM stg.device_raw" in live_pairing_query assert "FROM stg.geo_raw" in live_pairing_query assert "browser_rows - ifNull(d.device_rows, 0)" in live_pairing_query assert "browser_rows - ifNull(g.geo_rows, 0)" in live_pairing_query assert "FROM ods.device_by_click" not in live_pairing_query assert "FROM ods.geo_by_click" not in live_pairing_query assert ( "parseDateTime64BestEffort('${manifest_model_t_end}', 6, 'UTC') AS t_end" in live_pairing_query ) assert live_pairing_query.count( "JSONExtractString(raw, 'event_timestamp'), 6, 'UTC'" ) == 2 def test_generated_history_check_still_rejects_real_per_event_change(tmp_path): """Полные пары не скрывают настоящую смену per-event фактуры.""" fake_compose = _fake_compose( tmp_path, live_pairing="3\t0\t0\t0", seam_context="19\t18\t19", ) result = _run_generated_history_check( tmp_path, fake_compose=fake_compose, PROFILE="ci", CHECK_LIVE_SEAM="1", ) assert result.returncode != 0 assert "per-event фактура меняется на стыке: 18/19" in result.stderr assert "непарные live-строки" not in result.stderr def test_generated_history_check_fails_when_dm_views_query_fails(tmp_path): """Ошибка запроса DM-витрин не превращается в пустой успешный цикл.""" fake_compose = _fake_compose(tmp_path, views_mode="fail") result = _run_generated_history_check( tmp_path, fake_compose=fake_compose, PROFILE="ci", CHECK_LIVE_SEAM="0", ) assert result.returncode != 0 assert "не удалось прочитать основные DM-витрины" in result.stderr def test_clean_generated_history_run_explicitly_skips_live_seam(): """Чистый backfill без live-продолжения отключает проверку стыка явно.""" script = (REPO_ROOT / "scripts" / "run_generated_history_analytics.sh").read_text( encoding="utf-8" ) assert "CHECK_LIVE_SEAM=0" in script assert script.index("CHECK_LIVE_SEAM=0") < script.index('bash "${SCRIPT_DIR}/check_generated_analytics.sh"') def test_runtime_gate_has_bounded_daily_wave_live_seam_path(): """Runtime gate issue 17 не использует полный daily-wave на 2 суток.""" makefile = (REPO_ROOT / "Makefile").read_text(encoding="utf-8") script = (REPO_ROOT / "scripts" / "run_generated_history_runtime_check.sh").read_text( encoding="utf-8" ) assert "generated-history-runtime-check" in makefile assert 'PROFILE="${PROFILE:-daily-wave}"' in script assert 'GEN_HISTORY_DURATION="${GEN_HISTORY_DURATION:-1h}"' in script assert 'LIVE_SECONDS="${LIVE_SECONDS:-25}"' in script assert "stg_crossing_visits()" in script assert "crossing_visits=\"$(stg_crossing_visits)\"" in script assert "crossing_visits > 0" in script assert "t_end - INTERVAL 30 MINUTE" in script assert "t_end + INTERVAL ${GEN_LIVE_CHECK_MINUTES} MINUTE" in script assert "HAVING first_ts < t_end AND last_ts >= t_end" in script assert ( "parseDateTime64BestEffort('${GEN_MODEL_T_END}', 6, 'UTC') AS t_end" in script ) assert "JSONExtractString(raw, 'event_timestamp'), 6, 'UTC'" in script assert "stg_rows_after_live > stg_rows_before_live" not in script assert "не создало переходящий визит" in script assert "GEN_RUN_MODE=live" in script assert "GEN_STATE_RESET=false" in script assert "up -d --build generator" in script stop_call_line = _required_runtime_line(script, "stop_live_generator") step_7_line = _required_runtime_line( script, 'echo "Шаг 7: второй batch после live"', ) assert stop_call_line < step_7_line assert "CHECK_LIVE_SEAM=1" in script assert "REQUIRE_SUPERSET=0" in script def test_runtime_gate_contract_rejects_removed_stop_call(): """Определение функции не скрывает удалённый вызов перед вторым batch.""" script = ( REPO_ROOT / "scripts" / "run_generated_history_runtime_check.sh" ).read_text(encoding="utf-8") script_without_call = script.replace("\nstop_live_generator\n", "\n", 1) assert "stop_live_generator" in script_without_call with pytest.raises(AssertionError, match="отдельной командой"): _required_runtime_line(script_without_call, "stop_live_generator") def test_generator_shutdown_grace_covers_slow_batch_and_is_documented(): """Compose даёт текущему batch минуту на завершение после SIGTERM.""" compose = COMPOSE_FILE.read_text(encoding="utf-8") operations = OPERATIONS_DOC.read_text(encoding="utf-8") generator_service = compose.split("\n generator:\n", 1)[1].split( "\n kafka-exporter:\n", 1, )[0] assert "stop_grace_period: 1m" in generator_service assert "stop_grace_period: 1m" in operations def test_generated_history_check_rejects_empty_ods_context(): """ODS-блок стыка падает, если в ODS нет строк для переходящих визитов.""" script = (REPO_ROOT / "scripts" / "check_generated_analytics.sh").read_text( encoding="utf-8" ) assert "ods_device_rows" in script assert "ods_geo_rows" in script assert "ODS device пустой на стыке" in script assert "ODS geo пустой на стыке" in script def test_boundary_chain_has_separate_batch_check_and_keeps_live_gate_unchanged(): """Цепочка boundaries проверяется отдельной целью, не режимом live-гейта.""" chain = CHAIN_CHECK_SCRIPT.read_text(encoding="utf-8") live = CHECK_SCRIPT.read_text(encoding="utf-8") makefile = (REPO_ROOT / "Makefile").read_text(encoding="utf-8") assert "generated-history-chain-check" in makefile assert "check_generated_history_chain.sh" in makefile assert "kafka-boundaries" in chain assert "boundaries" in chain assert "unpaired_location_rows" in chain assert "unpaired_device_rows" in chain assert "unpaired_geo_rows" in chain assert "browser_names" in chain assert "referer_urls" in chain assert "utm_sources" in chain assert "per_event_homogeneous_visits" in chain assert "toDateTime64('${boundary}', 6, 'UTC') AS boundary" in chain assert "index < boundary_count - 1" in chain assert "kafka-boundaries" not in live def _fake_chain_compose(tmp_path): fake = tmp_path / "chain-compose" fake.write_text( textwrap.dedent( """\ #!/usr/bin/env bash set -euo pipefail args="$*" query="" previous="" for arg in "$@"; do if [[ "$previous" == "--query" ]]; then query="$arg" break fi previous="$arg" done if [[ "$args" == *"kafka-boundaries"* ]]; then printf '%b' "${CHAIN_BOUNDARIES}" exit 0 fi if [[ "$args" == *"kafka-topic-rows"* ]]; then printf '%s\n' "${CHAIN_MANIFEST_TOPIC_ROWS:-10 10 10 10}" exit 0 fi if [[ "$args" == *"clickhouse-client"* ]]; then printf '%s\n' "$query" >> "${CHAIN_QUERY_LOG}" if [[ "$query" == *"data_tail_rows"* ]]; then printf '%s\n' "${CHAIN_TAIL_ROWS:-0}" elif [[ "$query" == *"actual_topic_rows"* ]]; then printf '%s\n' "${CHAIN_ACTUAL_TOPIC_ROWS:-10 10 10 10}" elif [[ "$query" == *"FROM dm.v_events_enriched"* ]]; then printf '%s\n' "${CHAIN_SEGMENT_ROWS:-1}" elif [[ "$query" == *"unpaired_location_rows"* ]]; then printf '%s\n' "${CHAIN_PAIRING:-1 0 0 0}" elif [[ "$query" == *"per_event_homogeneous_visits"* ]]; then printf '%s\n' "${CHAIN_CONTEXT:-1 1}" else printf '%s\n' '1' fi exit 0 fi echo "unexpected compose call: $args" >&2 exit 92 """ ), encoding="utf-8", ) fake.chmod(0o755) return fake def _run_chain_check(tmp_path, **overrides): query_log = tmp_path / "queries.log" env = os.environ.copy() env.update( { "COMPOSE_BIN": str(_fake_chain_compose(tmp_path)), "CHAIN_BOUNDARIES": ( "2026-01-01T00:00:00+00:00\\n" "2026-01-02T00:00:00+00:00\\n" ), "CHAIN_QUERY_LOG": str(query_log), } ) env.update(overrides) result = subprocess.run( ["bash", str(CHAIN_CHECK_SCRIPT)], cwd=REPO_ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30, check=False, ) return result, query_log.read_text(encoding="utf-8") def test_boundary_chain_rejects_data_tail_after_manifest_end(tmp_path): """Сообщения после последней границы означают незавершённый next-day.""" result, queries = _run_chain_check(tmp_path, CHAIN_TAIL_ROWS="1") assert result.returncode != 0 assert "хвост" in result.stderr assert "FROM stg.browser_raw" in queries assert ">= toDateTime64('2026-01-02 00:00:00.000000', 6, 'UTC')" in queries @pytest.mark.parametrize( ("actual_rows", "topic"), [ ("10\t11\t10\t10", "location_events"), ("10\t10\t11\t10", "device_events"), ("10\t10\t10\t11", "geo_events"), ], ) def test_boundary_chain_rejects_non_browser_topic_tail( tmp_path, actual_rows, topic, ): """Хвост отдельной Kafka-темы обнаруживается без browser-события.""" result, queries = _run_chain_check( tmp_path, CHAIN_ACTUAL_TOPIC_ROWS=actual_rows, ) assert result.returncode != 0 assert "хвост" in result.stderr assert topic in result.stderr assert "FROM stg.location_raw" in queries assert "FROM stg.device_raw" in queries assert "FROM stg.geo_raw" in queries def test_boundary_chain_reports_boundary_without_crossing_visits(tmp_path): """Естественный нулевой стык проходит с явным статусом без однородности.""" result, _ = _run_chain_check( tmp_path, CHAIN_BOUNDARIES=( "2026-01-01T00:00:00+00:00\\n" "2026-01-02T00:00:00+00:00\\n" "2026-01-03T00:00:00+00:00\\n" ), CHAIN_PAIRING="0\t0\t0\t0", CHAIN_CONTEXT="0\t0", ) assert result.returncode == 0, result.stderr assert ( "boundary 1/3: crossing_visits=0 — однородность неприменима" in result.stdout ) def test_boundary_chain_still_rejects_unpaired_rows_without_crossing_visits( tmp_path, ): """Нулевой стык не отключает обязательную проверку пар сообщений.""" result, _ = _run_chain_check( tmp_path, CHAIN_BOUNDARIES=( "2026-01-01T00:00:00+00:00\\n" "2026-01-02T00:00:00+00:00\\n" "2026-01-03T00:00:00+00:00\\n" ), CHAIN_PAIRING="0\t1\t0\t0", CHAIN_CONTEXT="0\t0", ) assert result.returncode != 0 assert "непарные строки" in result.stderr def test_boundary_chain_still_rejects_empty_segment(tmp_path): """Отсутствие переходящих визитов не разрешает пустую порцию дня.""" result, _ = _run_chain_check(tmp_path, CHAIN_SEGMENT_ROWS="0") assert result.returncode != 0 assert "порция" in result.stderr assert "пуста" in result.stderr def test_boundary_chain_normalizes_offsets_and_uses_explicit_utc(tmp_path): """Граница с +03:00 проверяет тот же UTC-интервал, а не местные часы.""" result, queries = _run_chain_check( tmp_path, CHAIN_BOUNDARIES=( "2026-01-01T03:00:00+03:00\\n" "2026-01-02T03:00:00+03:00\\n" "2026-01-03T03:00:00+03:00\\n" ), ) assert result.returncode == 0, result.stderr assert "toDateTime64('2026-01-01 00:00:00.000000', 6, 'UTC')" in queries assert "toDateTime64('2026-01-02 00:00:00.000000', 6, 'UTC')" in queries assert "toDateTime64('2026-01-03 00:00:00.000000', 6, 'UTC')" in queries literals = re.findall(r"toDateTime64\([^)]*\)", queries) assert len(literals) == 7 assert all(", 6, 'UTC')" in literal for literal in literals) def test_boundary_chain_preserves_fractional_boundary_seconds(tmp_path): """Приведение к UTC сохраняет микросекунды полуоткрытой границы.""" result, queries = _run_chain_check( tmp_path, CHAIN_BOUNDARIES=( "2026-01-01T00:00:00.500000+00:00\\n" "2026-01-02T00:00:00.500000+00:00\\n" ), ) assert result.returncode == 0, result.stderr assert "toDateTime64('2026-01-01 00:00:00.500000', 6, 'UTC')" in queries assert "toDateTime64('2026-01-02 00:00:00.500000', 6, 'UTC')" in queries