""" 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.test import TestCase, override_settings # Флаг для запуска E2E тестов RUN_E2E_TESTS = os.environ.get("RUN_E2E_TESTS", "").lower() in ("1", "true", "yes") @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(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(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)