""" E2E тесты для парсера zakupki.gov.ru. Тесты с реальной загрузкой данных. Для запуска E2E тестов с реальными данными: RUN_E2E_TESTS=1 uv run python manage.py test apps.parsers.tests.test_e2e Тесты пропускаются по умолчанию, чтобы не нагружать внешние API в обычных тестовых прогонах. """ import os import unittest from apps.parsers.clients.zakupki import ZakupkiClient from apps.parsers.models import ParserLoadLog, ProcurementRecord from apps.parsers.services import ParserLoadLogService, ProcurementService from django.conf import settings from django.test import TestCase, override_settings # Флаг для запуска E2E тестов RUN_E2E_TESTS = os.environ.get("RUN_E2E_TESTS", "").lower() in ("1", "true", "yes") # Токен из settings (или переменной окружения) ZAKUPKI_TOKEN = getattr(settings, "ZAKUPKI_TOKEN", "") or os.environ.get( "ZAKUPKI_TOKEN", "" ) @unittest.skipUnless(RUN_E2E_TESTS, "E2E tests disabled. Set RUN_E2E_TESTS=1 to enable") class ZakupkiClientE2ETestCase(TestCase): """ E2E тесты клиента ZakupkiClient. Выполняют реальные HTTP запросы к zakupki.gov.ru. """ def setUp(self): """Подготовка.""" self.client = ZakupkiClient(token=ZAKUPKI_TOKEN, timeout=60) def tearDown(self): """Очистка.""" self.client.close() def test_fetch_procurement_plans(self): """Получение списка файлов для загрузки.""" plans = self.client.fetch_procurement_plans(region_code="77", year=2025) self.assertIsInstance(plans, list) # Планы должны быть сгенерированы даже без реальных данных self.assertGreater(len(plans), 0) for plan in plans: self.assertEqual(plan.region_code, "77") self.assertEqual(plan.year, 2025) def test_fetch_procurements_with_invalid_region(self): """Загрузка с несуществующим регионом возвращает пустой список.""" # Используем несуществующий код региона procurements = self.client.fetch_procurements( region_code="99", year=2025, month=1, ) # Должен вернуть пустой список, а не упасть self.assertIsInstance(procurements, list) def test_fetch_with_progress_callback(self): """Тест callback для отслеживания прогресса.""" progress_updates = [] def progress_callback(percent: int, message: str) -> None: progress_updates.append({"percent": percent, "message": message}) self.client.fetch_procurements( region_code="77", year=2025, month=1, progress_callback=progress_callback, ) # Должен быть хотя бы один вызов callback self.assertGreater(len(progress_updates), 0) # Первый вызов должен быть с 0% self.assertEqual(progress_updates[0]["percent"], 0) @unittest.skipUnless(RUN_E2E_TESTS, "E2E tests disabled. Set RUN_E2E_TESTS=1 to enable") class ProcurementServiceE2ETestCase(TestCase): """ E2E тесты сервиса ProcurementService. Тестирует полный цикл: загрузка → парсинг → сохранение в БД. """ def test_full_load_cycle(self): """Полный цикл загрузки данных.""" # Подготовка source = ParserLoadLog.Source.PROCUREMENTS batch_id = ParserLoadLogService.get_next_batch_id(source) load_log = ParserLoadLogService.create_load_log( source=source, batch_id=batch_id, status="in_progress", ) # Загрузка данных with ZakupkiClient(token=ZAKUPKI_TOKEN, timeout=60) as client: procurements = client.fetch_procurements( region_code="77", year=2025, month=1, ) # Сохранение в БД if procurements: saved_count = ProcurementService.save_procurements( procurements, batch_id=batch_id, region_code="77", data_year=2025, data_month=1, ) ParserLoadLogService.update( load_log, status="success", records_count=saved_count, ) # Проверки self.assertGreater(saved_count, 0) self.assertEqual(ProcurementRecord.objects.count(), saved_count) # Проверяем что данные корректны record = ProcurementRecord.objects.first() self.assertIsNotNone(record.purchase_number) self.assertEqual(record.region_code, "77") self.assertEqual(record.data_year, 2025) self.assertEqual(record.data_month, 1) self.assertEqual(record.load_batch, batch_id) # Проверяем лог load_log.refresh_from_db() self.assertEqual(load_log.status, "success") self.assertEqual(load_log.records_count, saved_count) else: # Если данных нет - это тоже валидный результат ParserLoadLogService.update( load_log, status="success", records_count=0, ) @unittest.skipUnless(RUN_E2E_TESTS, "E2E tests disabled. Set RUN_E2E_TESTS=1 to enable") @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class ProcurementTasksE2ETestCase(TestCase): """ E2E тесты Celery задач. Запускает реальные задачи в синхронном режиме. """ def test_parse_procurements_task(self): """Тест задачи parse_procurements.""" import uuid from apps.parsers.tasks import parse_procurements # Запуск задачи синхронно через .apply() с явным task_id # (CELERY_TASK_ALWAYS_EAGER=True, но self.request.id = None без apply) task_id = str(uuid.uuid4()) async_result = parse_procurements.apply( kwargs={ "region_code": "77", "year": 2025, "month": 1, "law_type": "44", }, task_id=task_id, ) result = async_result.result # Проверки self.assertIn("batch_id", result) self.assertIn("status", result) self.assertIn("saved", result) # Статус должен быть success или failed (не упасть с исключением) self.assertIn(result["status"], ["success", "failed"]) if result["status"] == "success": # Если успех - проверяем что batch_id корректен self.assertGreater(result["batch_id"], 0) class ZakupkiClientOfflineTestCase(TestCase): """ Тесты клиента без реальных HTTP запросов. Эти тесты выполняются всегда и проверяют обработку ошибок и edge cases. """ def test_client_handles_connection_error(self): """Клиент корректно обрабатывает ошибки соединения.""" # Используем невалидный хост client = ZakupkiClient(host="invalid.host.local", timeout=5) # Должен вернуть пустой список, а не упасть procurements = client.fetch_procurements( region_code="77", year=2025, file_url="http://invalid.host.local/data.xml", ) self.assertEqual(procurements, []) client.close() def test_client_with_empty_response(self): """Клиент обрабатывает пустой ответ.""" client = ZakupkiClient() # Пустой XML procurements = client._parse_xml_content( b'', None, ) self.assertEqual(procurements, []) client.close() class ProxyIntegrationTestCase(TestCase): """ Тесты интеграции с прокси. Проверяет что клиент корректно использует прокси из БД. """ def test_client_accepts_proxy_list(self): """Клиент принимает список прокси.""" proxies = ["http://proxy1:8080", "http://proxy2:8080"] client = ZakupkiClient(proxies=proxies) self.assertEqual(client.proxies, proxies) client.close() def test_client_works_without_proxies(self): """Клиент работает без прокси.""" client = ZakupkiClient() self.assertIsNone(client.proxies) client.close() def test_proxy_service_integration(self): """Интеграция с ProxyService.""" from apps.parsers.services import ProxyService from apps.parsers.tests.factories import ProxyFactory # Создаём активные прокси ProxyFactory.create_batch(3, is_active=True) ProxyFactory.create_batch(2, is_active=False) # Получаем активные прокси proxies = ProxyService.get_active_proxies_or_none() self.assertIsNotNone(proxies) self.assertEqual(len(proxies), 3) # Создаём клиент с этими прокси client = ZakupkiClient(proxies=proxies) self.assertEqual(len(client.proxies), 3) client.close() def test_proxy_service_returns_none_when_empty(self): """ProxyService возвращает None если нет активных прокси.""" from apps.parsers.services import ProxyService proxies = ProxyService.get_active_proxies_or_none() self.assertIsNone(proxies)