ddadminandClaude Fable 5
103ac021c8
- Зачем:
- world_next_day перечитывал всю историю топиков Kafka ради
накопительных счётчиков — время прогона росло с возрастом мира
(issue #5, находка F9).
- Что:
- счётчики засеваются при import из уже прочитанного артефакта и при
backfill из потока; next-day продвигает их только событиями нового
дня, полного чтения Kafka больше нет;
- катящаяся контрольная сумма — сумма SHA-256 событий по модулю 2^256
(инкремент равен полному пересчёту), старый формат артефакта
принимается без изменений;
- точные множества click_id/user_domain_id вынесены из manifest в
цепочку контент-адресуемых фрагментов (<=10 000 ID, SHA-256-цепочка,
отдельный топик counter_chunks) — потолок сообщения Kafka не грозит,
предел 900 000 байт проверяется явно с понятной ошибкой;
- порядок записи всюду: фрагменты -> manifest -> state; старое локальное
состояние отклоняется с подсказкой перезапустить import;
- документация manifest/state обновлена (ARCHITECTURE, OPERATIONS,
runbook startup-history).
- Проверка:
- make test (216+31) и make lint зелёные;
- живая приёмка на чистом стенде: import 235 с; три прогона
world_next_day — 716/718/716 с (плоское время, O(нового дня));
мир 3->6 дней, 561 942 события; make generated-history-chain-check —
все порции и стыки однородны;
- тест равенства инкремента и полного пересчёта:
test_incremental_counters_equal_full_recompute.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
895 lines
28 KiB
Python
895 lines
28 KiB
Python
"""
|
||
Тесты портативного артефакта стартовой истории.
|
||
"""
|
||
|
||
from dataclasses import replace
|
||
from datetime import datetime, timezone
|
||
import hashlib
|
||
import json
|
||
import lzma
|
||
|
||
import pytest
|
||
|
||
from generator import GeneratorState
|
||
|
||
|
||
def test_plain_json_roundtrip_keeps_existing_serialization(tmp_path):
|
||
"""Обычный JSON по-прежнему пишется без сжатия и читается обратно."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
load_startup_history_artifact,
|
||
write_startup_history_artifact,
|
||
)
|
||
|
||
artifact = {"text": "мир", "nested": {"value": 42}}
|
||
path = tmp_path / "artifact.json"
|
||
|
||
write_startup_history_artifact(path, artifact)
|
||
|
||
assert path.read_text(encoding="utf-8") == json.dumps(
|
||
artifact,
|
||
ensure_ascii=True,
|
||
indent=2,
|
||
)
|
||
assert load_startup_history_artifact(path) == artifact
|
||
|
||
|
||
def test_xz_artifact_roundtrip(tmp_path):
|
||
"""XZ-артефакт потоково пишется и читается без изменения данных."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
load_startup_history_artifact,
|
||
write_startup_history_artifact,
|
||
)
|
||
|
||
artifact = {"text": "мир", "nested": {"value": 42}}
|
||
path = tmp_path / "artifact.json.xz"
|
||
|
||
write_startup_history_artifact(path, artifact)
|
||
|
||
assert load_startup_history_artifact(path) == artifact
|
||
with lzma.open(path, "rt", encoding="utf-8") as input_file:
|
||
assert input_file.read() == json.dumps(
|
||
artifact,
|
||
ensure_ascii=True,
|
||
indent=2,
|
||
)
|
||
|
||
|
||
def _state() -> GeneratorState:
|
||
import random
|
||
|
||
now = datetime(2026, 1, 1, 1, 0, tzinfo=timezone.utc)
|
||
return GeneratorState(
|
||
tick=1,
|
||
rng_state=random.Random(42).getstate(),
|
||
last_batch_id="startup-history-abc",
|
||
last_timestamp=now,
|
||
model_timestamp=now,
|
||
wall_timestamp=now,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
model_t0=datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc),
|
||
gen_seed=42,
|
||
population=[
|
||
{
|
||
"user_domain_id": "user-1",
|
||
"seed_click_id": "seed-1",
|
||
"active_click_id": None,
|
||
"last_finished_at": None,
|
||
}
|
||
],
|
||
)
|
||
|
||
|
||
def _batch() -> dict[str, list[dict]]:
|
||
return {
|
||
"browser_events": [
|
||
{
|
||
"event_id": "event-1",
|
||
"click_id": "click-1",
|
||
"user_domain_id": "user-1",
|
||
"event_timestamp": "2026-01-01 00:00:00.000000",
|
||
}
|
||
],
|
||
"location_events": [
|
||
{
|
||
"event_id": "event-1",
|
||
"page_url_path": "/home",
|
||
}
|
||
],
|
||
"device_events": [
|
||
{
|
||
"click_id": "click-1",
|
||
"user_domain_id": "user-1",
|
||
}
|
||
],
|
||
"geo_events": [
|
||
{
|
||
"click_id": "click-1",
|
||
"country": "RU",
|
||
}
|
||
],
|
||
}
|
||
|
||
|
||
def test_incremental_counters_equal_full_recompute():
|
||
"""Накопление по частям даёт те же числа, что полный пересчёт мира."""
|
||
from clickstream_generator.startup_history_artifact import ManifestCounters
|
||
|
||
first_day = _batch()
|
||
second_day = {
|
||
"browser_events": [
|
||
{
|
||
"event_id": "event-2",
|
||
"click_id": "click-1",
|
||
"user_domain_id": "user-1",
|
||
"event_timestamp": "2026-01-02 00:00:00.000000",
|
||
},
|
||
{
|
||
"event_id": "event-3",
|
||
"click_id": "click-2",
|
||
"user_domain_id": "user-2",
|
||
"event_timestamp": "2026-01-02 00:01:00.000000",
|
||
},
|
||
],
|
||
"location_events": [
|
||
{"event_id": "event-2", "page_url_path": "/cart"},
|
||
{"event_id": "event-3", "page_url_path": "/home"},
|
||
],
|
||
"device_events": [
|
||
{"click_id": "click-1", "user_domain_id": "user-1"},
|
||
{"click_id": "click-2", "user_domain_id": "user-2"},
|
||
],
|
||
"geo_events": [
|
||
{"click_id": "click-1", "country": "RU"},
|
||
{"click_id": "click-2", "country": "KZ"},
|
||
],
|
||
}
|
||
|
||
incremental = ManifestCounters()
|
||
incremental.add_batch(first_day)
|
||
incremental = ManifestCounters.from_state(
|
||
incremental.to_state(),
|
||
known_click_ids={"click-1"},
|
||
known_user_ids={"user-1"},
|
||
)
|
||
incremental.add_batch(second_day)
|
||
|
||
full = ManifestCounters()
|
||
full.add_batch({
|
||
topic: first_day[topic] + second_day[topic]
|
||
for topic in first_day
|
||
})
|
||
|
||
assert incremental.to_manifest_topics() == full.to_manifest_topics()
|
||
assert incremental.to_manifest_totals() == full.to_manifest_totals()
|
||
|
||
|
||
def test_exact_id_sets_are_split_into_bounded_hash_chain(monkeypatch):
|
||
"""Manifest ссылается на цепочку малых точных фрагментов, а не растёт сам."""
|
||
from clickstream_generator import startup_history_artifact as module
|
||
|
||
monkeypatch.setattr(module, "ID_SET_CHUNK_SIZE", 2)
|
||
counters = module.ManifestCounters()
|
||
batch = _batch()
|
||
batch["browser_events"].extend([
|
||
{
|
||
"event_id": "event-2",
|
||
"click_id": "click-2",
|
||
"event_timestamp": "2026-01-01 00:01:00.000000",
|
||
},
|
||
{
|
||
"event_id": "event-3",
|
||
"click_id": "click-3",
|
||
"event_timestamp": "2026-01-01 00:02:00.000000",
|
||
},
|
||
])
|
||
counters.add_batch(batch)
|
||
|
||
state, chunks = counters.to_state_with_chunks()
|
||
|
||
assert len(chunks) == 2
|
||
assert state["totals"] == {"visits": 3, "users": 1}
|
||
assert state["id_sets"] == {
|
||
"storage": "manifest-topic-sha256-chain-v1",
|
||
"latest_sha256": chunks[-1][0],
|
||
"chunks": 2,
|
||
"click_ids": 3,
|
||
"user_ids": 1,
|
||
}
|
||
assert chunks[0][1]["previous_sha256"] is None
|
||
assert chunks[1][1]["previous_sha256"] == chunks[0][0]
|
||
assert "click-1" not in json.dumps(state)
|
||
|
||
restored = module.ManifestCounters.from_state(
|
||
state,
|
||
known_click_ids={"click-3"},
|
||
known_user_ids={"user-1"},
|
||
)
|
||
restored.add_batch({
|
||
"browser_events": [
|
||
{
|
||
"event_id": "event-4",
|
||
"click_id": "click-3",
|
||
"event_timestamp": "2026-01-02 00:00:00.000000",
|
||
},
|
||
{
|
||
"event_id": "event-5",
|
||
"click_id": "click-4",
|
||
"event_timestamp": "2026-01-02 00:01:00.000000",
|
||
},
|
||
],
|
||
"location_events": [],
|
||
"device_events": [{"click_id": "click-4", "user_domain_id": "user-2"}],
|
||
"geo_events": [],
|
||
})
|
||
next_state, next_chunks = restored.to_state_with_chunks()
|
||
|
||
assert next_state["totals"] == {"visits": 4, "users": 2}
|
||
assert len(next_chunks) == 1
|
||
assert next_chunks[0][1]["previous_sha256"] == chunks[-1][0]
|
||
assert next_state["id_sets"]["chunks"] == 3
|
||
|
||
|
||
def test_counter_chunk_has_explicit_kafka_size_and_hash_guards():
|
||
"""Большой или неверно адресованный фрагмент падает до отправки в Kafka."""
|
||
from clickstream_generator.kafka_io import (
|
||
MAX_COMPACT_MESSAGE_BYTES,
|
||
KafkaStartupHistoryManifest,
|
||
)
|
||
|
||
manager = object.__new__(KafkaStartupHistoryManifest)
|
||
with pytest.raises(ValueError, match="безопасный предел"):
|
||
manager._assert_message_size(
|
||
{"payload": "x" * MAX_COMPACT_MESSAGE_BYTES},
|
||
"фрагмент",
|
||
)
|
||
with pytest.raises(ValueError, match="хеш.*не совпадает"):
|
||
manager.save_counter_chunk("0" * 64, {"version": "1.0"})
|
||
|
||
assert manager.COUNTER_TOPIC != manager.MANIFEST_TOPIC
|
||
|
||
|
||
def test_artifact_roundtrip_keeps_events_state_and_manifest(base_config):
|
||
"""Артефакт хранит события, state и manifest одним проверяемым набором."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
load_startup_history_artifact,
|
||
validate_startup_history_artifact,
|
||
write_startup_history_artifact,
|
||
)
|
||
|
||
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,
|
||
)
|
||
|
||
artifact = builder.to_artifact(manifest=manifest, state=state)
|
||
(
|
||
loaded_state,
|
||
loaded_manifest,
|
||
loaded_topics,
|
||
loaded_raw_topics,
|
||
) = validate_startup_history_artifact(
|
||
artifact,
|
||
expected_config=replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
),
|
||
)
|
||
|
||
assert artifact["artifact_version"] == "1.0"
|
||
assert loaded_state.last_batch_id == state.last_batch_id
|
||
assert loaded_manifest["state"]["last_batch_id"] == state.last_batch_id
|
||
assert loaded_topics["browser_events"][0]["event_id"] == "event-1"
|
||
assert loaded_raw_topics["browser_events"][0]["value_json"].encode("utf-8") == (
|
||
b'{"event_id": "event-1", "click_id": "click-1", '
|
||
b'"user_domain_id": "user-1", '
|
||
b'"event_timestamp": "2026-01-01 00:00:00.000000"}'
|
||
)
|
||
|
||
path = base_config.data_dir.parent / "tmp-startup-history-artifact.json"
|
||
try:
|
||
write_startup_history_artifact(path, artifact)
|
||
loaded = load_startup_history_artifact(path)
|
||
assert loaded["manifest"]["totals"]["events"] == 1
|
||
finally:
|
||
path.unlink(missing_ok=True)
|
||
|
||
|
||
def test_legacy_artifact_checksum_remains_valid(base_config):
|
||
"""Эталонный артефакт со старой суммой не требует пересборки."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
validate_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
manifest = build_manifest(config, builder.counters, state)
|
||
for topic, events in builder.topics.items():
|
||
checksum = hashlib.sha256()
|
||
for event in events:
|
||
checksum.update(
|
||
json.dumps(event, sort_keys=True, ensure_ascii=True).encode("utf-8")
|
||
)
|
||
manifest["topics"][topic]["checksum_sha256"] = checksum.hexdigest()
|
||
|
||
artifact = builder.to_artifact(manifest=manifest, state=state)
|
||
|
||
_, loaded_manifest, _, _ = validate_startup_history_artifact(
|
||
artifact,
|
||
expected_config=config,
|
||
)
|
||
assert loaded_manifest["topics"] == artifact["manifest"]["topics"]
|
||
|
||
|
||
def test_manifest_and_artifact_show_launch_profile(base_config):
|
||
"""Manifest и файл артефакта показывают выбранный профиль запуска."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
load_startup_history_artifact,
|
||
write_startup_history_artifact,
|
||
)
|
||
|
||
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,
|
||
launch_profile="daily-wave",
|
||
),
|
||
counters=builder.counters,
|
||
state=state,
|
||
)
|
||
artifact = builder.to_artifact(manifest=manifest, state=state)
|
||
|
||
path = base_config.data_dir.parent / "tmp-startup-history-profile.json"
|
||
try:
|
||
write_startup_history_artifact(path, artifact)
|
||
loaded = load_startup_history_artifact(path)
|
||
assert manifest["launch_profile"] == "daily-wave"
|
||
assert loaded["manifest"]["launch_profile"] == "daily-wave"
|
||
finally:
|
||
path.unlink(missing_ok=True)
|
||
|
||
|
||
def test_manifest_records_initial_boundary_chain(base_config):
|
||
"""Новый manifest явно хранит начало и правую границу истории."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
manifest_boundaries,
|
||
)
|
||
|
||
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,
|
||
)
|
||
|
||
assert manifest["boundaries"] == [
|
||
"2026-01-01T00:00:00+00:00",
|
||
"2026-01-01T01:00:00+00:00",
|
||
]
|
||
assert manifest_boundaries(manifest) == manifest["boundaries"]
|
||
|
||
|
||
def test_legacy_manifest_without_boundaries_gets_endpoint_chain():
|
||
"""Старый manifest без boundaries читается как пара [T0, T_end]."""
|
||
from clickstream_generator.startup_history_artifact import manifest_boundaries
|
||
|
||
manifest = {
|
||
"model_t0": "2026-01-01T00:00:00+00:00",
|
||
"model_t_end": "2026-01-03T00:00:00+00:00",
|
||
}
|
||
|
||
assert manifest_boundaries(manifest) == [
|
||
"2026-01-01T00:00:00+00:00",
|
||
"2026-01-03T00:00:00+00:00",
|
||
]
|
||
|
||
|
||
def test_manifest_rejects_boundary_chain_with_wrong_endpoint():
|
||
"""Цепочка не может расходиться с текущим model_t_end."""
|
||
from clickstream_generator.startup_history_artifact import manifest_boundaries
|
||
|
||
manifest = {
|
||
"model_t0": "2026-01-01T00:00:00+00:00",
|
||
"model_t_end": "2026-01-03T00:00:00+00:00",
|
||
"boundaries": [
|
||
"2026-01-01T00:00:00+00:00",
|
||
"2026-01-02T00:00:00+00:00",
|
||
],
|
||
}
|
||
|
||
with pytest.raises(ValueError, match="model_t_end"):
|
||
manifest_boundaries(manifest)
|
||
|
||
|
||
def test_build_manifest_rejects_explicit_empty_boundaries(base_config):
|
||
"""Явно пустая цепочка не подменяется границами по умолчанию."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
)
|
||
|
||
state = _state()
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="T0 и T_end"):
|
||
build_manifest(config, builder.counters, state, boundaries=[])
|
||
|
||
|
||
def test_artifact_validation_rejects_mismatched_manifest(base_config):
|
||
"""Валидация отвергает артефакт, где manifest не совпадает с событиями."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
validate_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
manifest = build_manifest(config=config, counters=builder.counters, state=state)
|
||
artifact = builder.to_artifact(manifest=manifest, state=state)
|
||
artifact["manifest"]["gen_seed"] = 43
|
||
|
||
with pytest.raises(ValueError, match="GEN_SEED"):
|
||
validate_startup_history_artifact(artifact, expected_config=config)
|
||
|
||
|
||
def test_manifest_clickhouse_stats_detect_mismatch():
|
||
"""Сверка ClickHouse с manifest ловит расхождение контрольных чисел."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
compare_clickhouse_stats_to_manifest,
|
||
)
|
||
|
||
manifest = {
|
||
"totals": {
|
||
"events": 10,
|
||
"visits": 4,
|
||
"users": 3,
|
||
"min_event_timestamp": "2026-01-01 00:00:00.000000",
|
||
"max_event_timestamp": "2026-01-01 00:10:00.000000",
|
||
}
|
||
}
|
||
stats = {
|
||
"events": "9",
|
||
"visits": "4",
|
||
"users": "3",
|
||
"min_event_timestamp": "2026-01-01 00:00:00.000000",
|
||
"max_event_timestamp": "2026-01-01 00:10:00.000000",
|
||
}
|
||
|
||
assert compare_clickhouse_stats_to_manifest(manifest, stats) == ["events"]
|
||
|
||
|
||
def test_raw_topic_value_can_keep_original_json_bytes(base_config):
|
||
"""Raw value проверяется как JSON, но импортируется без пересборки строки."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
import_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
artifact = builder.to_artifact(
|
||
manifest=build_manifest(config=config, counters=builder.counters, state=state),
|
||
state=state,
|
||
)
|
||
assert "cumulative_manifest_counters" not in artifact["state"]
|
||
assert "cumulative_manifest_counters" not in artifact["manifest"]
|
||
raw_value = (
|
||
'{"event_timestamp":"2026-01-01 00:00:00.000000",'
|
||
'"user_domain_id":"user-1","click_id":"click-1","event_id":"event-1"}'
|
||
)
|
||
artifact["raw_topics"]["browser_events"][0]["value_json"] = raw_value
|
||
|
||
class Publisher:
|
||
def __init__(self):
|
||
self.records = None
|
||
|
||
def publish_records(self, topic, records):
|
||
if topic == "browser_events":
|
||
self.records = records
|
||
return len(records), 0
|
||
|
||
def flush(self):
|
||
return None
|
||
|
||
class CompactWriter:
|
||
def __init__(self):
|
||
self.chunks = []
|
||
|
||
def save(self, value):
|
||
self.saved = value
|
||
|
||
def save_counter_chunk(self, chunk_sha256, chunk):
|
||
self.chunks.append((chunk_sha256, chunk))
|
||
|
||
def flush(self):
|
||
return None
|
||
|
||
publisher = Publisher()
|
||
import_startup_history_artifact(
|
||
artifact,
|
||
publisher=publisher,
|
||
state_manager=CompactWriter(),
|
||
manifest_manager=CompactWriter(),
|
||
expected_config=config,
|
||
)
|
||
|
||
assert publisher.records[0]["value_json"] == raw_value
|
||
|
||
|
||
def test_artifact_v1_requires_raw_topics_for_import(base_config):
|
||
"""Artifact v1.0 без raw value не может обещать byte-for-byte импорт."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
import_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
artifact = builder.to_artifact(
|
||
manifest=build_manifest(config=config, counters=builder.counters, state=state),
|
||
state=state,
|
||
)
|
||
del artifact["raw_topics"]
|
||
|
||
class Publisher:
|
||
def publish_records(self, topic, records):
|
||
raise AssertionError("publish must not be called")
|
||
|
||
with pytest.raises(ValueError, match="raw_topics"):
|
||
import_startup_history_artifact(
|
||
artifact,
|
||
publisher=Publisher(),
|
||
state_manager=None,
|
||
manifest_manager=None,
|
||
expected_config=config,
|
||
)
|
||
|
||
|
||
def test_import_rejects_publisher_without_raw_publish_records(base_config):
|
||
"""Импорт не должен пересобирать dict, если publisher не умеет raw records."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
import_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
artifact = builder.to_artifact(
|
||
manifest=build_manifest(config=config, counters=builder.counters, state=state),
|
||
state=state,
|
||
)
|
||
artifact["raw_topics"]["browser_events"][0]["value_json"] = (
|
||
'{"event_timestamp":"2026-01-01 00:00:00.000000",'
|
||
'"user_domain_id":"user-1","click_id":"click-1","event_id":"event-1"}'
|
||
)
|
||
|
||
class DictPublisher:
|
||
def publish(self, topic, events):
|
||
raise AssertionError("dict publish must not be called")
|
||
|
||
def flush(self):
|
||
return None
|
||
|
||
with pytest.raises(TypeError, match="publish_records"):
|
||
import_startup_history_artifact(
|
||
artifact,
|
||
publisher=DictPublisher(),
|
||
state_manager=None,
|
||
manifest_manager=None,
|
||
expected_config=config,
|
||
)
|
||
|
||
|
||
def test_import_rejects_dirty_kafka_topics_before_publish(base_config):
|
||
"""Повторный импорт не должен дописывать дубли в непустые data-топики."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
import_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
artifact = builder.to_artifact(
|
||
manifest=build_manifest(config=config, counters=builder.counters, state=state),
|
||
state=state,
|
||
)
|
||
|
||
class Publisher:
|
||
def publish(self, topic, events):
|
||
raise AssertionError("publish must not be called")
|
||
|
||
class DirtyTopics:
|
||
def assert_data_topics_empty(self):
|
||
raise RuntimeError("Kafka data topics are not empty")
|
||
|
||
with pytest.raises(RuntimeError, match="not empty"):
|
||
import_startup_history_artifact(
|
||
artifact,
|
||
publisher=Publisher(),
|
||
state_manager=None,
|
||
manifest_manager=None,
|
||
expected_config=config,
|
||
topic_inspector=DirtyTopics(),
|
||
)
|
||
|
||
|
||
def test_import_rolls_back_kafka_topics_after_partial_publish(base_config):
|
||
"""При частичной публикации импорт откатывает Kafka-топики до повтора."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
import_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
artifact = builder.to_artifact(
|
||
manifest=build_manifest(config=config, counters=builder.counters, state=state),
|
||
state=state,
|
||
)
|
||
|
||
class PartialPublisher:
|
||
def __init__(self):
|
||
self.calls = []
|
||
|
||
def publish_records(self, topic, records):
|
||
self.calls.append(topic)
|
||
if topic == "location_events":
|
||
raise RuntimeError("location down")
|
||
return len(records), 0
|
||
|
||
class CompactWriter:
|
||
def save(self, value):
|
||
raise AssertionError("compact topics must not be written")
|
||
|
||
class Inspector:
|
||
def __init__(self):
|
||
self.snapshot = None
|
||
self.rollback = None
|
||
|
||
def assert_data_topics_empty(self):
|
||
return None
|
||
|
||
def snapshot_import_topics(self):
|
||
self.snapshot = {"topics": ["browser_events", "location_events"]}
|
||
return self.snapshot
|
||
|
||
def rollback_import_topics(self, snapshot):
|
||
self.rollback = snapshot
|
||
|
||
publisher = PartialPublisher()
|
||
inspector = Inspector()
|
||
|
||
with pytest.raises(RuntimeError, match="location down"):
|
||
import_startup_history_artifact(
|
||
artifact,
|
||
publisher=publisher,
|
||
state_manager=CompactWriter(),
|
||
manifest_manager=CompactWriter(),
|
||
expected_config=config,
|
||
topic_inspector=inspector,
|
||
)
|
||
|
||
assert publisher.calls == ["browser_events", "location_events"]
|
||
assert inspector.rollback == inspector.snapshot
|
||
|
||
|
||
def test_import_replays_events_and_compact_topics(base_config):
|
||
"""Импорт пишет события в Kafka и служебные compact-топики, не трогая ClickHouse."""
|
||
from clickstream_generator.startup_history_artifact import (
|
||
ManifestCounters,
|
||
StartupHistoryArtifactBuilder,
|
||
build_manifest,
|
||
cumulative_counter_reference,
|
||
import_startup_history_artifact,
|
||
)
|
||
|
||
state = _state()
|
||
config = replace(
|
||
base_config,
|
||
model_t0=state.model_t0,
|
||
model_t_end=state.model_timestamp,
|
||
model_time_speed=1,
|
||
model_timezone="UTC",
|
||
seed=42,
|
||
)
|
||
builder = StartupHistoryArtifactBuilder()
|
||
builder.add_batch(_batch())
|
||
artifact = builder.to_artifact(
|
||
manifest=build_manifest(config=config, counters=builder.counters, state=state),
|
||
state=state,
|
||
)
|
||
|
||
class Publisher:
|
||
def __init__(self):
|
||
self.calls = []
|
||
self.flushed = False
|
||
|
||
def publish_records(self, topic, records):
|
||
self.calls.append((topic, records))
|
||
return len(records), 0
|
||
|
||
def flush(self):
|
||
self.flushed = True
|
||
|
||
class CompactWriter:
|
||
def __init__(self):
|
||
self.saved = None
|
||
self.flushed = False
|
||
self.chunks = []
|
||
|
||
def save(self, value):
|
||
self.saved = value
|
||
|
||
def save_counter_chunk(self, chunk_sha256, chunk):
|
||
self.chunks.append((chunk_sha256, chunk))
|
||
|
||
def flush(self):
|
||
self.flushed = True
|
||
|
||
publisher = Publisher()
|
||
state_manager = CompactWriter()
|
||
manifest_manager = CompactWriter()
|
||
|
||
result = import_startup_history_artifact(
|
||
artifact,
|
||
publisher=publisher,
|
||
state_manager=state_manager,
|
||
manifest_manager=manifest_manager,
|
||
expected_config=config,
|
||
)
|
||
|
||
assert [topic for topic, _ in publisher.calls] == [
|
||
"browser_events",
|
||
"location_events",
|
||
"device_events",
|
||
"geo_events",
|
||
]
|
||
assert publisher.calls[0][1][0]["value_json"].encode("utf-8") == (
|
||
b'{"event_id": "event-1", "click_id": "click-1", '
|
||
b'"user_domain_id": "user-1", '
|
||
b'"event_timestamp": "2026-01-01 00:00:00.000000"}'
|
||
)
|
||
assert result["events"] == 4
|
||
assert state_manager.saved.last_batch_id == state.last_batch_id
|
||
assert state_manager.saved.cumulative_manifest_counters is not None
|
||
assert manifest_manager.saved["state"]["last_batch_id"] == state.last_batch_id
|
||
assert state_manager.saved.cumulative_manifest_counters == (
|
||
cumulative_counter_reference(
|
||
manifest_manager.saved["cumulative_manifest_counters"]
|
||
)
|
||
)
|
||
seeded = ManifestCounters.from_state(
|
||
manifest_manager.saved["cumulative_manifest_counters"],
|
||
known_click_ids={"click-1"},
|
||
known_user_ids={"user-1"},
|
||
)
|
||
assert seeded.click_ids == {"click-1"}
|
||
assert seeded.user_ids == {"user-1"}
|
||
assert len(manifest_manager.chunks) == 1
|
||
assert (
|
||
manifest_manager.saved["cumulative_manifest_counters"]["id_sets"][
|
||
"latest_sha256"
|
||
]
|
||
== manifest_manager.chunks[0][0]
|
||
)
|
||
assert publisher.flushed
|
||
assert state_manager.flushed
|
||
assert manifest_manager.flushed
|