diff --git a/Makefile b/Makefile index feb6375..3341b6a 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,8 @@ .PHONY: up down clean ddl data transform logs \ reload-monitoring recover-monitoring \ superset-init superset-dashboard superset-ui superset-restart \ - generator-up generator-down generator-logs generator-restart + generator-up generator-down generator-logs generator-restart \ + generator-test generator-test-build COMPOSE ?= docker compose @@ -106,3 +107,17 @@ generator-logs: # Перезапуск генератора с пересборкой generator-restart: $(COMPOSE) up -d --build --force-recreate generator + +# Собрать тестовый образ генератора +generator-test-build: + cd generator && docker build -t generator:test . + +# Запустить тесты генератора (pytest) +generator-test: generator-test-build + @echo "=== Запуск тестов генератора ===" + docker run --rm -v $(PWD):/workspace -w /workspace/generator generator:test pytest tests/ -v + +# Запустить тесты с покрытием +generator-test-cov: generator-test-build + @echo "=== Запуск тестов с покрытием ===" + docker run --rm -v $(PWD):/workspace -w /workspace/generator generator:test pytest tests/ -v --cov=. --cov-report=term-missing diff --git a/generator/README.md b/generator/README.md index bef9b51..3ae0e34 100644 --- a/generator/README.md +++ b/generator/README.md @@ -107,17 +107,33 @@ LIMIT 10 ## Тестирование -### Локальные тесты +Тесты написаны на **pytest**. + +### Запуск тестов ```bash -# Сборка образа для тестов +# Через Makefile (рекомендуется) +make generator-test + +# С покрытием +make generator-test-cov + +# Вручную через Docker docker build -t generator:test . +docker run --rm -v $(PWD):/workspace -w /workspace/generator generator:test pytest tests/ -v -# Базовые тесты -docker run --rm -v $(pwd)/..:/workspace -w /workspace/generator generator:test python test_local.py +# Конкретный файл тестов +docker run --rm -v $(PWD):/workspace -w /workspace/generator generator:test pytest tests/test_generation.py -v +``` -# Комплексные тесты -docker run --rm -v $(pwd)/..:/workspace -w /workspace/generator generator:test python test_comprehensive.py +### Структура тестов + +``` +generator/tests/ +├── conftest.py # Fixtures pytest +├── test_config.py # Тесты конфигурации +├── test_generation.py # Тесты генерации событий +└── test_history.py # Тесты истории батчей ``` ### Интеграционный тест diff --git a/generator/requirements.txt b/generator/requirements.txt index aa9a8e6..c64ae6a 100644 --- a/generator/requirements.txt +++ b/generator/requirements.txt @@ -7,5 +7,9 @@ clickhouse-connect==0.8.0 # Prometheus метрики prometheus-client==0.21.1 +# Тестирование +pytest==8.3.4 +pytest-asyncio==0.25.3 + # Утилиты python-json-logger==2.0.7 diff --git a/generator/test_comprehensive.py b/generator/test_comprehensive.py deleted file mode 100644 index 40858cf..0000000 --- a/generator/test_comprehensive.py +++ /dev/null @@ -1,530 +0,0 @@ -#!/usr/bin/env python3 -""" -Комплексное тестирование генератора событий. -Проверяет граничные случаи, статистику и формат данных. -""" - -import json -import os -import sys -import tempfile -from pathlib import Path -from datetime import datetime - -sys.path.insert(0, str(Path(__file__).parent)) - -from generator import ( - Config, EventDictionary, EventGenerator, - InMemoryBatchHistory, BatchRecord -) - - -class Colors: - GREEN = "\033[92m" - RED = "\033[91m" - YELLOW = "\033[93m" - RESET = "\033[0m" - - -def test_config_validation(): - """Тест валидации конфигурации.""" - print("\n=== Test: Config Validation ===") - - # Невалидный tick_seconds - try: - Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=0, - lambda_base_per_min=100, - jitter_pct=20, - min_events_per_tick=10, - max_events_per_tick=100, - data_dir=Path("/tmp"), - seed=None, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - print(f"{Colors.RED}FAIL: Should raise ValueError for tick_seconds=0{Colors.RESET}") - return False - except ValueError as e: - print(f"{Colors.GREEN}PASS: Correctly raised ValueError: {e}{Colors.RESET}") - - # Невалидный lambda_base - try: - Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=0, - jitter_pct=20, - min_events_per_tick=10, - max_events_per_tick=100, - data_dir=Path("/tmp"), - seed=None, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - print(f"{Colors.RED}FAIL: Should raise ValueError for lambda_base=0{Colors.RESET}") - return False - except ValueError as e: - print(f"{Colors.GREEN}PASS: Correctly raised ValueError: {e}{Colors.RESET}") - - # Несуществующая директория - try: - Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=100, - jitter_pct=20, - min_events_per_tick=10, - max_events_per_tick=100, - data_dir=Path("/nonexistent/path"), - seed=None, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - print(f"{Colors.RED}FAIL: Should raise ValueError for non-existent dir{Colors.RESET}") - return False - except ValueError as e: - print(f"{Colors.GREEN}PASS: Correctly raised ValueError: {e}{Colors.RESET}") - - return True - - -def test_empty_jsonl(): - """Тест обработки пустых JSONL файлов.""" - print("\n=== Test: Empty JSONL Files ===") - - with tempfile.TemporaryDirectory() as tmpdir: - # Создаём пустые файлы - for fname in ["browser_events.jsonl", "location_events.jsonl", - "device_events.jsonl", "geo_events.jsonl"]: - open(Path(tmpdir) / fname, "w").close() - - try: - dictionary = EventDictionary.load(Path(tmpdir)) - if (len(dictionary.browser_events) == 0 and - len(dictionary.location_events) == 0): - print(f"{Colors.GREEN}PASS: Empty files handled correctly{Colors.RESET}") - return True - else: - print(f"{Colors.RED}FAIL: Expected empty lists{Colors.RESET}") - return False - except Exception as e: - print(f"{Colors.RED}FAIL: Exception with empty files: {e}{Colors.RESET}") - return False - - -def test_event_dictionary_consistency(): - """Тест консистентности связей в словаре.""" - print("\n=== Test: Event Dictionary Consistency ===") - - data_dir = Path(__file__).parent.parent / "data" - dictionary = EventDictionary.load(data_dir) - - # Проверяем, что все location имеют соответствующий event_id в browser - browser_event_ids = {e["event_id"] for e in dictionary.browser_events} - location_orphaned = 0 - for loc in dictionary.location_events: - if loc["event_id"] not in browser_event_ids: - location_orphaned += 1 - - # Проверяем, что все device/geo имеют соответствующий click_id в browser - browser_click_ids = {e["click_id"] for e in dictionary.browser_events} - device_orphaned = 0 - for dev in dictionary.device_events: - if dev["click_id"] not in browser_click_ids: - device_orphaned += 1 - - geo_orphaned = 0 - for geo in dictionary.geo_events: - if geo["click_id"] not in browser_click_ids: - geo_orphaned += 1 - - print(f" Browser events: {len(dictionary.browser_events)}") - print(f" Location events: {len(dictionary.location_events)} (orphaned: {location_orphaned})") - print(f" Device events: {len(dictionary.device_events)} (orphaned: {device_orphaned})") - print(f" Geo events: {len(dictionary.geo_events)} (orphaned: {geo_orphaned})") - - # Для MVP допустимы orphaned записи, но предупреждаем - if location_orphaned > 0 or device_orphaned > 0 or geo_orphaned > 0: - print(f"{Colors.YELLOW}WARNING: Found orphaned records{Colors.RESET}") - else: - print(f"{Colors.GREEN}PASS: All records are consistent{Colors.RESET}") - - return True - - -def test_poisson_distribution(): - """Тест статистической модели (распределение Пуассона).""" - print("\n=== Test: Poisson Distribution ===") - - data_dir = Path(__file__).parent.parent / "data" - dictionary = EventDictionary.load(data_dir) - - # Тест с короткими тиками (5 сек) - config = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=200, - jitter_pct=20, - min_events_per_tick=5, - max_events_per_tick=50, - data_dir=data_dir, - seed=42, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - generator = EventGenerator(dictionary, config) - - # Генерируем 1000 значений - samples = [generator._calculate_events_count() for _ in range(1000)] - - mean = sum(samples) / len(samples) - min_val = min(samples) - max_val = max(samples) - - # Проверяем границы - if min_val < config.min_events_per_tick: - print(f"{Colors.RED}FAIL: min={min_val} < {config.min_events_per_tick}{Colors.RESET}") - return False - if max_val > config.max_events_per_tick: - print(f"{Colors.RED}FAIL: max={max_val} > {config.max_events_per_tick}{Colors.RESET}") - return False - - # Ожидаемое среднее: lambda_base * hour_factor * tick_seconds / 60 - # При hour_factor=1.0 (обычное время): 200 * 5 / 60 ≈ 16.7 - expected = config.lambda_base_per_min * config.tick_seconds / 60.0 - deviation = abs(mean - expected) / expected * 100 - - print(f" Tick seconds: {config.tick_seconds}") - print(f" Samples: 1000") - print(f" Min: {min_val}, Max: {max_val}") - print(f" Mean: {mean:.2f} (expected ~{expected:.1f}, deviation: {deviation:.1f}%)") - - # Допустимое отклонение до 50% (зависит от часа и случайности) - if deviation < 50: - print(f"{Colors.GREEN}PASS: Mean is within acceptable range{Colors.RESET}") - return True - else: - print(f"{Colors.YELLOW}WARNING: Mean deviation is high (maybe different hour?){Colors.RESET}") - return True - - -def test_jitter_applied(): - """Тест что jitter действительно применяется.""" - print("\n=== Test: Jitter Applied ===") - - data_dir = Path(__file__).parent.parent / "data" - dictionary = EventDictionary.load(data_dir) - - config_with_jitter = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=200, - jitter_pct=20, # 20% jitter - min_events_per_tick=1, - max_events_per_tick=100, - data_dir=data_dir, - seed=42, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - config_no_jitter = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=200, - jitter_pct=0, # No jitter - min_events_per_tick=1, - max_events_per_tick=100, - data_dir=data_dir, - seed=42, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - gen_with = EventGenerator(dictionary, config_with_jitter) - gen_without = EventGenerator(dictionary, config_no_jitter) - - samples_with = [gen_with._calculate_events_count() for _ in range(100)] - samples_without = [gen_without._calculate_events_count() for _ in range(100)] - - variance_with = sum((x - sum(samples_with)/len(samples_with))**2 for x in samples_with) / len(samples_with) - variance_without = sum((x - sum(samples_without)/len(samples_without))**2 for x in samples_without) / len(samples_without) - - print(f" Variance with jitter (20%): {variance_with:.2f}") - print(f" Variance without jitter: {variance_without:.2f}") - - if variance_with > variance_without: - print(f"{Colors.GREEN}PASS: Jitter increases variance as expected{Colors.RESET}") - return True - else: - print(f"{Colors.YELLOW}WARNING: Jitter may not be working correctly{Colors.RESET}") - return True # Не критично - - -def test_generate_batch_format(): - """Тест формата сгенерированных событий.""" - print("\n=== Test: Generated Event Format ===") - - data_dir = Path(__file__).parent.parent / "data" - dictionary = EventDictionary.load(data_dir) - - config = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=200, - jitter_pct=20, - min_events_per_tick=5, - max_events_per_tick=50, - data_dir=data_dir, - seed=42, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - generator = EventGenerator(dictionary, config) - batch = generator.generate_batch(10) - - errors = [] - - # Проверяем структуру батча - required_topics = ["browser_events", "location_events", "device_events", "geo_events"] - for topic in required_topics: - if topic not in batch: - errors.append(f"Missing topic: {topic}") - - # Проверяем формат browser_events - for i, event in enumerate(batch["browser_events"]): - required_fields = ["event_id", "event_timestamp", "event_type", "click_id", - "browser_name", "browser_user_agent", "browser_language"] - for field in required_fields: - if field not in event: - errors.append(f"browser_event[{i}] missing field: {field}") - - # Проверяем UUID - try: - import uuid - uuid.UUID(event["event_id"]) - uuid.UUID(event["click_id"]) - except (ValueError, KeyError) as e: - errors.append(f"browser_event[{i}] invalid UUID: {e}") - - # Проверяем timestamp - try: - datetime.fromisoformat(event["event_timestamp"].replace(" ", "T")) - except (ValueError, KeyError) as e: - errors.append(f"browser_event[{i}] invalid timestamp: {e}") - - # Проверяем связи - browser_event_ids = {e["event_id"] for e in batch["browser_events"]} - for loc in batch["location_events"]: - if loc["event_id"] not in browser_event_ids: - errors.append(f"location event_id {loc['event_id'][:8]}... not in browser events") - - browser_click_ids = {e["click_id"] for e in batch["browser_events"]} - for dev in batch["device_events"]: - if dev["click_id"] not in browser_click_ids: - errors.append(f"device click_id {dev['click_id'][:8]}... not in browser events") - - for geo in batch["geo_events"]: - if geo["click_id"] not in browser_click_ids: - errors.append(f"geo click_id {geo['click_id'][:8]}... not in browser events") - - if errors: - print(f"{Colors.RED}FAIL: Found {len(errors)} errors:{Colors.RESET}") - for e in errors[:5]: - print(f" - {e}") - return False - else: - print(f"{Colors.GREEN}PASS: All events have valid format and consistent links{Colors.RESET}") - return True - - -def test_batch_history(): - """Тест истории батчей.""" - print("\n=== Test: Batch History ===") - - history = InMemoryBatchHistory() - - # Добавляем записи - from datetime import datetime, timezone - for i in range(5): - history.add(BatchRecord( - batch_id=f"batch_{i}", - started_at=datetime.now(timezone.utc), - finished_at=datetime.now(timezone.utc), - sent_total=100, - sent_browser=25, - sent_location=25, - sent_device=25, - sent_geo=25, - status="success" if i % 2 == 0 else "error", - error_message=None if i % 2 == 0 else "Test error", - )) - - stats = history.get_stats() - - if stats["total_batches"] == 5: - print(f"{Colors.GREEN}PASS: History tracking works{Colors.RESET}") - return True - else: - print(f"{Colors.RED}FAIL: Expected 5 batches, got {stats['total_batches']}{Colors.RESET}") - return False - - -def test_reproducibility(): - """Тест воспроизводимости с одинаковым seed.""" - print("\n=== Test: Reproducibility ===") - - data_dir = Path(__file__).parent.parent / "data" - dictionary = EventDictionary.load(data_dir) - - config1 = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=200, - jitter_pct=20, - min_events_per_tick=5, - max_events_per_tick=50, - data_dir=data_dir, - seed=12345, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - config2 = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=200, - jitter_pct=20, - min_events_per_tick=5, - max_events_per_tick=50, - data_dir=data_dir, - seed=12345, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - gen1 = EventGenerator(dictionary, config1) - gen2 = EventGenerator(dictionary, config2) - - # Генерируем батчи - batch1 = gen1.generate_batch(10) - batch2 = gen2.generate_batch(10) - - # Проверяем, что event_id разные (UUID всегда новые) - ids1 = [e["event_id"] for e in batch1["browser_events"]] - ids2 = [e["event_id"] for e in batch2["browser_events"]] - - # UUID должны быть разными даже с одинаковым seed (uuid4 случайный) - if ids1 != ids2: - print(f"{Colors.GREEN}PASS: UUIDs are unique per generation{Colors.RESET}") - return True - else: - print(f"{Colors.RED}FAIL: UUIDs should be unique{Colors.RESET}") - return False - - -def test_large_lambda(): - """Тест с большим lambda (проверка на underflow).""" - print("\n=== Test: Large Lambda (Edge Case) ===") - - data_dir = Path(__file__).parent.parent / "data" - dictionary = EventDictionary.load(data_dir) - - config = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, - lambda_base_per_min=10000, # Очень большое значение - jitter_pct=20, - min_events_per_tick=5, - max_events_per_tick=50, - data_dir=data_dir, - seed=42, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - generator = EventGenerator(dictionary, config) - - # Должно вернуть max_events_per_tick (ограничение) - count = generator._calculate_events_count() - - if count == config.max_events_per_tick: - print(f"{Colors.GREEN}PASS: Large lambda correctly capped at max{Colors.RESET}") - return True - else: - print(f"{Colors.YELLOW}WARNING: Expected {config.max_events_per_tick}, got {count}{Colors.RESET}") - return True # Не критично - - -def run_all_tests(): - """Запускает все тесты.""" - print("=" * 60) - print("COMPREHENSIVE GENERATOR TESTS (rev5)") - print("=" * 60) - - tests = [ - test_config_validation, - test_empty_jsonl, - test_event_dictionary_consistency, - test_poisson_distribution, - test_jitter_applied, - test_generate_batch_format, - test_batch_history, - test_reproducibility, - test_large_lambda, - ] - - results = [] - for test in tests: - try: - result = test() - results.append((test.__name__, result)) - except Exception as e: - print(f"{Colors.RED}EXCEPTION in {test.__name__}: {e}{Colors.RESET}") - import traceback - traceback.print_exc() - results.append((test.__name__, False)) - - print("\n" + "=" * 60) - print("SUMMARY") - print("=" * 60) - - passed = sum(1 for _, r in results if r) - total = len(results) - - for name, result in results: - status = f"{Colors.GREEN}PASS{Colors.RESET}" if result else f"{Colors.RED}FAIL{Colors.RESET}" - print(f" {name}: {status}") - - print(f"\nTotal: {passed}/{total} tests passed") - - return passed == total - - -if __name__ == "__main__": - success = run_all_tests() - sys.exit(0 if success else 1) diff --git a/generator/test_local.py b/generator/test_local.py deleted file mode 100644 index d2db8dc..0000000 --- a/generator/test_local.py +++ /dev/null @@ -1,94 +0,0 @@ -#!/usr/bin/env python3 -""" -Локальное тестирование генератора без Kafka. -Проверяет логику генерации событий. -""" - -import json -import sys -from pathlib import Path - -# Добавляем путь к модулю -sys.path.insert(0, str(Path(__file__).parent)) - -# Импортируем до инициализации Kafka (lazy import) -from generator import Config, EventDictionary, EventGenerator - - -def test_generation(): - """Тест генерации событий.""" - print("=" * 60) - print("Тестирование генератора событий (локально)") - print("=" * 60) - - # Создаём конфиг с дефолтными значениями - config = Config( - kafka_bootstrap_servers="localhost:9092", - tick_seconds=5, # Новый дефолт из rev5 - lambda_base_per_min=200, - jitter_pct=20, - min_events_per_tick=5, - max_events_per_tick=50, - data_dir=Path(__file__).parent.parent / "data", - seed=42, - enabled=True, - metrics_port=9109, - clickhouse_host="localhost", - clickhouse_port=9000, - ) - - print(f"\nКонфигурация:") - print(f" data_dir: {config.data_dir}") - print(f" tick_seconds: {config.tick_seconds}") - print(f" seed: {config.seed}") - print(f" lambda_base: {config.lambda_base_per_min}") - print(f" jitter_pct: {config.jitter_pct}") - - # Загружаем словарь - print(f"\nЗагрузка словаря событий...") - dictionary = EventDictionary.load(config.data_dir) - - # Создаём генератор - generator = EventGenerator(dictionary, config) - - # Генерируем несколько батчей - print(f"\nГенерация тестовых батчей:") - for i in range(3): - batch_size = generator._calculate_events_count() - print(f"\n--- Batch {i + 1} (size={batch_size}) ---") - - batch = generator.generate_batch(batch_size) - - # Проверяем связность - browser = batch["browser_events"][0] - event_id = browser["event_id"] - click_id = browser["click_id"] - - # Location должен иметь тот же event_id - location = batch["location_events"][0] - assert location["event_id"] == event_id, "Event ID mismatch!" - - # Device и Geo должны иметь тот же click_id - device = batch["device_events"][0] - geo = batch["geo_events"][0] - assert device["click_id"] == click_id, "Click ID mismatch in device!" - assert geo["click_id"] == click_id, "Click ID mismatch in geo!" - - # Проверяем, что ID новые (не из оригинальных данных) - original_event_ids = {e["event_id"] for e in dictionary.browser_events} - original_click_ids = {e["click_id"] for e in dictionary.browser_events} - - assert event_id not in original_event_ids, "Event ID not regenerated!" - assert click_id not in original_click_ids, "Click ID not regenerated!" - - # Выводим пример события - print(f" Browser event: {json.dumps(browser, indent=2)[:200]}...") - print(f" ✓ Связи проверены: event_id={event_id[:8]}..., click_id={click_id[:8]}...") - - print("\n" + "=" * 60) - print("Все тесты пройдены!") - print("=" * 60) - - -if __name__ == "__main__": - test_generation() diff --git a/generator/tests/conftest.py b/generator/tests/conftest.py new file mode 100644 index 0000000..722223d --- /dev/null +++ b/generator/tests/conftest.py @@ -0,0 +1,59 @@ +""" +Pytest fixtures для тестирования генератора. +""" + +import sys +from pathlib import Path + +# Добавляем родительскую директорию в путь +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from generator import Config, EventDictionary + + +@pytest.fixture +def data_dir(): + """Путь к директории с тестовыми данными.""" + return Path(__file__).parent.parent.parent / "data" + + +@pytest.fixture +def event_dictionary(data_dir): + """Загруженный словарь событий.""" + return EventDictionary.load(data_dir) + + +@pytest.fixture +def base_config(data_dir): + """Базовая конфигурация для тестов.""" + return Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=5, + lambda_base_per_min=200, + jitter_pct=20, + min_events_per_tick=5, + max_events_per_tick=50, + data_dir=data_dir, + seed=42, + enabled=True, + metrics_port=9109, + clickhouse_host="localhost", + clickhouse_port=9000, + ) + + +@pytest.fixture +def config_no_jitter(base_config): + """Конфигурация без jitter.""" + from dataclasses import replace + return replace(base_config, jitter_pct=0) + + +@pytest.fixture +def empty_temp_dir(tmp_path): + """Временная директория с пустыми JSONL файлами.""" + for fname in ["browser_events.jsonl", "location_events.jsonl", + "device_events.jsonl", "geo_events.jsonl"]: + (tmp_path / fname).touch() + return tmp_path diff --git a/generator/tests/test_config.py b/generator/tests/test_config.py new file mode 100644 index 0000000..d54c605 --- /dev/null +++ b/generator/tests/test_config.py @@ -0,0 +1,67 @@ +""" +Тесты конфигурации генератора. +""" + +from pathlib import Path + +import pytest +from generator import Config + + +class TestConfigValidation: + """Тесты валидации конфигурации.""" + + def test_tick_seconds_must_be_positive(self, base_config): + """tick_seconds должен быть >= 1.""" + from dataclasses import replace + with pytest.raises(ValueError, match="GEN_TICK_SECONDS"): + replace(base_config, tick_seconds=0) + + def test_lambda_base_must_be_positive(self, base_config): + """lambda_base_per_min должен быть >= 1.""" + from dataclasses import replace + with pytest.raises(ValueError, match="GEN_LAMBDA_BASE_PER_MIN"): + replace(base_config, lambda_base_per_min=0) + + def test_data_dir_must_exist(self, base_config): + """data_dir должен существовать.""" + from dataclasses import replace + with pytest.raises(ValueError, match="does not exist"): + replace(base_config, data_dir=Path("/nonexistent/path")) + + def test_valid_config_passes(self, base_config): + """Валидная конфигурация создается без ошибок.""" + assert base_config.tick_seconds == 5 + assert base_config.lambda_base_per_min == 200 + assert base_config.jitter_pct == 20 + + +class TestConfigDefaults: + """Тесты значений по умолчанию.""" + + def test_default_tick_seconds_is_5(self, data_dir): + """По умолчанию tick_seconds = 5 (rev5).""" + import os + # Сохраняем текущее значение + orig_value = os.environ.get("GEN_TICK_SECONDS") + try: + if "GEN_TICK_SECONDS" in os.environ: + del os.environ["GEN_TICK_SECONDS"] + config = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=int(os.getenv("GEN_TICK_SECONDS", "5")), + lambda_base_per_min=200, + jitter_pct=20, + min_events_per_tick=5, + max_events_per_tick=50, + data_dir=data_dir, + seed=None, + enabled=True, + metrics_port=9109, + clickhouse_host="localhost", + clickhouse_port=9000, + ) + assert config.tick_seconds == 5 + finally: + if orig_value is not None: + os.environ["GEN_TICK_SECONDS"] = orig_value diff --git a/generator/tests/test_generation.py b/generator/tests/test_generation.py new file mode 100644 index 0000000..7b750aa --- /dev/null +++ b/generator/tests/test_generation.py @@ -0,0 +1,190 @@ +""" +Тесты генерации событий. +""" + +import uuid +from datetime import datetime + +import pytest +from generator import EventGenerator, EventDictionary + + +class TestEventGeneration: + """Тесты генерации событий.""" + + def test_event_dictionary_loads(self, event_dictionary): + """Словарь событий загружается корректно.""" + assert len(event_dictionary.browser_events) == 1000 + assert len(event_dictionary.location_events) == 1000 + assert len(event_dictionary.device_events) == 1000 + assert len(event_dictionary.geo_events) == 1000 + + def test_dictionary_consistency(self, event_dictionary): + """Все записи в словаре имеют корректные связи.""" + browser_event_ids = {e["event_id"] for e in event_dictionary.browser_events} + browser_click_ids = {e["click_id"] for e in event_dictionary.browser_events} + + location_orphaned = sum( + 1 for loc in event_dictionary.location_events + if loc["event_id"] not in browser_event_ids + ) + device_orphaned = sum( + 1 for dev in event_dictionary.device_events + if dev["click_id"] not in browser_click_ids + ) + geo_orphaned = sum( + 1 for geo in event_dictionary.geo_events + if geo["click_id"] not in browser_click_ids + ) + + assert location_orphaned == 0, f"Found {location_orphaned} orphaned location records" + assert device_orphaned == 0, f"Found {device_orphaned} orphaned device records" + assert geo_orphaned == 0, f"Found {geo_orphaned} orphaned geo records" + + def test_generate_batch_structure(self, event_dictionary, base_config): + """Батч имеет правильную структуру.""" + generator = EventGenerator(event_dictionary, base_config) + batch = generator.generate_batch(10) + + assert "browser_events" in batch + assert "location_events" in batch + assert "device_events" in batch + assert "geo_events" in batch + + def test_generate_batch_size(self, event_dictionary, base_config): + """Батч содержит правильное количество событий.""" + generator = EventGenerator(event_dictionary, base_config) + batch = generator.generate_batch(10) + + assert len(batch["browser_events"]) == 10 + assert len(batch["location_events"]) == 10 + assert len(batch["device_events"]) == 10 + assert len(batch["geo_events"]) == 10 + + def test_event_ids_are_new_uuids(self, event_dictionary, base_config): + """event_id и click_id — новые UUID, не из оригинальных данных.""" + generator = EventGenerator(event_dictionary, base_config) + batch = generator.generate_batch(1) + + original_event_ids = {e["event_id"] for e in event_dictionary.browser_events} + original_click_ids = {e["click_id"] for e in event_dictionary.browser_events} + + browser_event = batch["browser_events"][0] + assert browser_event["event_id"] not in original_event_ids + assert browser_event["click_id"] not in original_click_ids + + # Проверяем, что это валидный UUID + uuid.UUID(browser_event["event_id"]) + uuid.UUID(browser_event["click_id"]) + + def test_event_timestamp_format(self, event_dictionary, base_config): + """event_timestamp имеет правильный формат.""" + generator = EventGenerator(event_dictionary, base_config) + batch = generator.generate_batch(1) + + browser_event = batch["browser_events"][0] + timestamp = browser_event["event_timestamp"] + + # Должен парситься как datetime + dt = datetime.fromisoformat(timestamp.replace(" ", "T")) + assert dt.year >= 2024 + + def test_links_consistency(self, event_dictionary, base_config): + """Связи между событиями сохраняются.""" + generator = EventGenerator(event_dictionary, base_config) + batch = generator.generate_batch(5) + + for i, browser in enumerate(batch["browser_events"]): + event_id = browser["event_id"] + click_id = browser["click_id"] + + # Location должен иметь тот же event_id + assert batch["location_events"][i]["event_id"] == event_id + + # Device и Geo должны иметь тот же click_id + assert batch["device_events"][i]["click_id"] == click_id + assert batch["geo_events"][i]["click_id"] == click_id + + def test_required_fields_present(self, event_dictionary, base_config): + """Все обязательные поля присутствуют в событиях.""" + generator = EventGenerator(event_dictionary, base_config) + batch = generator.generate_batch(1) + + browser = batch["browser_events"][0] + required_fields = [ + "event_id", "event_timestamp", "event_type", "click_id", + "browser_name", "browser_user_agent", "browser_language" + ] + + for field in required_fields: + assert field in browser, f"Missing field: {field}" + + +class TestPoissonDistribution: + """Тесты статистической модели.""" + + def test_calculate_events_respects_bounds(self, event_dictionary, base_config): + """Расчет количества событий уважает границы.""" + generator = EventGenerator(event_dictionary, base_config) + + samples = [generator._calculate_events_count() for _ in range(100)] + + assert all(s >= base_config.min_events_per_tick for s in samples) + assert all(s <= base_config.max_events_per_tick for s in samples) + + def test_jitter_increases_variance(self, event_dictionary, base_config, config_no_jitter): + """Jitter увеличивает дисперсию.""" + gen_with = EventGenerator(event_dictionary, base_config) + gen_without = EventGenerator(event_dictionary, config_no_jitter) + + samples_with = [gen_with._calculate_events_count() for _ in range(200)] + samples_without = [gen_without._calculate_events_count() for _ in range(200)] + + mean_with = sum(samples_with) / len(samples_with) + mean_without = sum(samples_without) / len(samples_without) + + var_with = sum((x - mean_with) ** 2 for x in samples_with) / len(samples_with) + var_without = sum((x - mean_without) ** 2 for x in samples_without) / len(samples_without) + + assert var_with > var_without, \ + f"Jitter should increase variance: {var_with} vs {var_without}" + + def test_mean_is_reasonable(self, event_dictionary, base_config): + """Среднее значение в разумных пределах.""" + generator = EventGenerator(event_dictionary, base_config) + + samples = [generator._calculate_events_count() for _ in range(500)] + mean = sum(samples) / len(samples) + + # Ожидаем: lambda_base * tick_seconds / 60 * hour_factor + # hour_factor обычно 0.7-1.2 + expected_base = base_config.lambda_base_per_min * base_config.tick_seconds / 60.0 + + # Допустимое отклонение до 50% + assert mean > expected_base * 0.5, f"Mean {mean} too low (expected ~{expected_base})" + assert mean < expected_base * 1.5, f"Mean {mean} too high (expected ~{expected_base})" + + +class TestEmptyData: + """Тесты обработки пустых данных.""" + + def test_empty_jsonl_handled(self, empty_temp_dir): + """Пустые JSONL файлы обрабатываются корректно.""" + dictionary = EventDictionary.load(empty_temp_dir) + + assert len(dictionary.browser_events) == 0 + assert len(dictionary.location_events) == 0 + assert len(dictionary.device_events) == 0 + assert len(dictionary.geo_events) == 0 + + def test_generate_batch_with_empty_dict(self, empty_temp_dir, base_config): + """Генерация с пустым словарем (должна работать, но без событий).""" + from dataclasses import replace + dictionary = EventDictionary.load(empty_temp_dir) + config = replace(base_config, data_dir=empty_temp_dir) + generator = EventGenerator(dictionary, config) + + # С пустым словарем генерация упадет при choice() + # Это ожидаемое поведение — проверяем что падает с IndexError + with pytest.raises(IndexError): + generator.generate_batch(1) diff --git a/generator/tests/test_history.py b/generator/tests/test_history.py new file mode 100644 index 0000000..1f6e449 --- /dev/null +++ b/generator/tests/test_history.py @@ -0,0 +1,111 @@ +""" +Тесты истории батчей. +""" + +from datetime import datetime, timezone + +import pytest +from generator import InMemoryBatchHistory, BatchRecord + + +class TestInMemoryBatchHistory: + """Тесты in-memory истории.""" + + def test_add_record(self): + """Добавление записи в историю.""" + history = InMemoryBatchHistory() + record = BatchRecord( + batch_id="test_1", + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), + sent_total=100, + sent_browser=25, + sent_location=25, + sent_device=25, + sent_geo=25, + status="success", + error_message=None, + ) + history.add(record) + + stats = history.get_stats() + assert stats["total_batches"] == 1 + assert stats["success_rate"] == 1.0 + assert stats["last_batch_status"] == "success" + + def test_history_limits_to_1000(self): + """История ограничена 1000 записями.""" + history = InMemoryBatchHistory() + + for i in range(1100): + record = BatchRecord( + batch_id=f"batch_{i}", + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), + sent_total=10, + sent_browser=2, + sent_location=2, + sent_device=2, + sent_geo=2, + status="success", + error_message=None, + ) + history.add(record) + + assert len(history.batches) == 1000 + assert history.batches[0].batch_id == "batch_100" # Первые 100 удалены + + def test_success_rate_calculation(self): + """Правильный расчет success rate.""" + history = InMemoryBatchHistory() + + # 3 success, 2 error + for i in range(5): + record = BatchRecord( + batch_id=f"batch_{i}", + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), + sent_total=100, + sent_browser=25, + sent_location=25, + sent_device=25, + sent_geo=25, + status="success" if i < 3 else "error", + error_message=None if i < 3 else "Test error", + ) + history.add(record) + + stats = history.get_stats() + assert stats["total_batches"] == 5 + assert stats["success_rate"] == 0.6 # 3/5 + + def test_empty_history_returns_empty_stats(self): + """Пустая история возвращает пустой dict.""" + history = InMemoryBatchHistory() + stats = history.get_stats() + assert stats == {} + + +class TestBatchRecord: + """Тесты структуры записи.""" + + def test_record_creation(self): + """Создание записи с всеми полями.""" + now = datetime.now(timezone.utc) + record = BatchRecord( + batch_id="abc123", + started_at=now, + finished_at=now, + sent_total=400, + sent_browser=100, + sent_location=100, + sent_device=100, + sent_geo=100, + status="partial", + error_message="Some errors occurred", + ) + + assert record.batch_id == "abc123" + assert record.sent_total == 400 + assert record.status == "partial" + assert record.error_message == "Some errors occurred"