Files
clickstream-ch-kafka-supers…/tests/test_generated_analytics_check_contract.py
ddadminandClaude Fable 5 b87dde79b7 feat(generator): дефолт профиля — daily-wave, ci стал служебным
- Зачем:
  - менти доставался плоский тестовый профиль ci; учебный профиль
    должен быть один — daily-wave с суточной волной (issue #6).
- Что:
  - дефолт daily-wave во всех точках: форма пульта generator_control,
    launch.py (API и CLI), Config, Makefile, четыре shell-скрипта.
  - автопроверки передают ci явно; контрактный тест пульта теперь
    проверяет сам дефолт Param профиля (через AST), а не подстроку.
  - доки в том же изменении: README, OPERATIONS, TEST_PLAN,
    generator/README, runbook startup-history.
- Проверка:
  - make test (206 + 31 passed) и make lint — зелёные.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-22 17:16:07 +03:00

576 lines
21 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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