""" Unit-тесты для ProcurementService. Тестирует сервис для работы с данными о закупках. """ from apps.parsers.clients.zakupki.schemas import Procurement from apps.parsers.models import ProcurementRecord from apps.parsers.services import ProcurementService from apps.registers.models import Organization from django.test import TestCase from tests.apps.parsers.factories import ProcurementRecordFactory, fake def _digits(length: int) -> str: return "".join(str(fake.random_int(0, 9)) for _ in range(length)) def _region_code() -> str: return str(fake.random_int(min=1, max=99)).zfill(2) def _period() -> tuple[int, int]: return fake.random_int(min=2020, max=2025), fake.random_int(min=1, max=12) def _other_law(law_type: str) -> str: return "223-FZ" if law_type == "44-FZ" else "44-FZ" def _create_registry_organization(*, inn: str, ogrn: str) -> Organization: return Organization.objects.create( pn_name=fake.company(), mn_ogrn=int(ogrn), mn_inn=int(inn), in_kpp=int(_digits(9)), mn_okpo=_digits(8), ) def _build_procurement(**overrides) -> Procurement: data = { "purchase_number": _digits(19), "purchase_name": fake.sentence(nb_words=4), "customer_inn": _digits(10), "customer_kpp": _digits(9), "customer_ogrn": _digits(13), "customer_name": fake.company(), "max_price": str(fake.random_int(min=10_000, max=10_000_000)), "currency_code": fake.random_element(["RUB", "USD", "EUR"]), "placement_method": fake.word(), "publish_date": str(fake.date()), "end_date": str(fake.date()), "status": fake.word(), "law_type": fake.random_element(["44-FZ", "223-FZ"]), "purchase_object_info": fake.sentence(nb_words=6), "href": fake.url(), } data.update(overrides) return Procurement(**data) class ProcurementServiceSaveTestCase(TestCase): """Тесты метода save_procurements.""" def test_save_empty_list(self): """Сохранение пустого списка возвращает 0.""" saved = ProcurementService.save_procurements([], batch_id=1) self.assertEqual(saved, 0) def test_save_single_procurement(self): """Сохранение одной закупки.""" purchase_number = _digits(19) customer_inn = _digits(10) region_code = _region_code() year = fake.random_int(min=2020, max=2025) month = fake.random_int(min=1, max=12) procurement = _build_procurement( purchase_number=purchase_number, customer_inn=customer_inn, ) saved = ProcurementService.save_procurements( [procurement], batch_id=1, region_code=region_code, data_year=year, data_month=month, ) self.assertEqual(saved, 1) self.assertEqual(ProcurementRecord.objects.count(), 1) record = ProcurementRecord.objects.first() self.assertEqual(record.purchase_number, purchase_number) self.assertEqual(record.customer_inn, customer_inn) self.assertEqual(record.region_code, region_code) self.assertEqual(record.data_year, year) self.assertEqual(record.data_month, month) def test_save_multiple_procurements(self): """Сохранение нескольких закупок.""" procurements = [_build_procurement() for _ in range(5)] saved = ProcurementService.save_procurements(procurements, batch_id=1) self.assertEqual(saved, 5) self.assertEqual(ProcurementRecord.objects.count(), 5) def test_save_links_registry_organization_when_exists(self): """При совпадении ИНН/ОГРН должна ставиться связь с registers.Organization.""" inn = _digits(10) ogrn = _digits(13) organization = _create_registry_organization(inn=inn, ogrn=ogrn) purchase_number = _digits(19) procurement = _build_procurement( purchase_number=purchase_number, customer_inn=inn, customer_ogrn=ogrn, ) saved = ProcurementService.save_procurements([procurement], batch_id=1) self.assertEqual(saved, 1) record = ProcurementRecord.objects.get(purchase_number=purchase_number) self.assertEqual(record.registry_organization_id, organization.id) def test_save_updates_duplicates(self): """Повторная синхронизация обновляет существующую закупку.""" # Создаём существующую запись purchase_number = _digits(19) ProcurementRecordFactory(purchase_number=purchase_number) original = ProcurementRecord.objects.get(purchase_number=purchase_number) # Пытаемся сохранить с тем же номером procurement = _build_procurement( purchase_number=purchase_number, customer_inn=_digits(10), ) saved = ProcurementService.save_procurements([procurement], batch_id=2) # Существующая запись обновляется в пределах той же строки self.assertEqual(saved, 1) self.assertEqual(ProcurementRecord.objects.count(), 1) refreshed = ProcurementRecord.objects.get(purchase_number=purchase_number) self.assertEqual(refreshed.customer_inn, procurement.customer_inn) self.assertEqual(refreshed.load_batch, 2) self.assertEqual(refreshed.id, original.id) def test_save_with_chunking(self): """Сохранение большого количества записей чанками.""" procurements = [_build_procurement() for _ in range(100)] saved = ProcurementService.save_procurements( procurements, batch_id=1, chunk_size=25 ) self.assertEqual(saved, 100) self.assertEqual(ProcurementRecord.objects.count(), 100) class ProcurementServiceFindTestCase(TestCase): """Тесты методов поиска.""" def setUp(self): """Подготовка тестовых данных.""" self.inn_target = _digits(10) self.inn_other = _digits(10) self.region_a = _region_code() self.region_b = _region_code() while self.region_b == self.region_a: self.region_b = _region_code() self.name_key = fake.word() self.unique_token = fake.word() self.record1 = ProcurementRecordFactory( purchase_number=_digits(19), customer_inn=self.inn_target, customer_name=f"{self.unique_token} {self.name_key} {fake.company()}", region_code=self.region_a, load_batch=1, ) self.record2 = ProcurementRecordFactory( purchase_number=_digits(19), customer_inn=self.inn_other, customer_name=f"{self.name_key} {fake.company()}", region_code=self.region_a, load_batch=1, ) self.record3 = ProcurementRecordFactory( purchase_number=_digits(19), customer_inn=self.inn_target, # Тот же ИНН что и у первого customer_name=f"{self.name_key} {fake.company()}", region_code=self.region_b, load_batch=2, ) def test_find_by_inn(self): """Поиск по ИНН.""" results = ProcurementService.find_by_inn(self.inn_target) self.assertEqual(results.count(), 2) def test_find_by_inn_with_batch(self): """Поиск по ИНН с фильтром по batch.""" results = ProcurementService.find_by_inn(self.inn_target, batch_id=1) self.assertEqual(results.count(), 1) self.assertEqual(results.first().purchase_number, self.record1.purchase_number) def test_find_by_purchase_number(self): """Поиск по номеру закупки.""" results = ProcurementService.find_by_purchase_number( self.record2.purchase_number ) self.assertEqual(results.count(), 1) self.assertEqual(results.first().customer_inn, self.record2.customer_inn) def test_find_by_region(self): """Поиск по региону.""" results = ProcurementService.find_by_region(self.region_a) self.assertEqual(results.count(), 2) def test_find_by_region_with_batch(self): """Поиск по региону с фильтром по batch.""" results = ProcurementService.find_by_region(self.region_a, batch_id=1) self.assertEqual(results.count(), 2) def test_find_by_customer_name(self): """Поиск по названию заказчика (частичное совпадение).""" results = ProcurementService.find_by_customer_name(self.name_key) self.assertEqual(results.count(), 3) results = ProcurementService.find_by_customer_name(self.unique_token) self.assertEqual(results.count(), 1) class ProcurementServicePeriodTestCase(TestCase): """Тесты методов работы с периодами.""" def test_get_last_loaded_period_empty(self): """Если данных нет, возвращается (None, None).""" year, month = ProcurementService.get_last_loaded_period() self.assertIsNone(year) self.assertIsNone(month) def test_get_last_loaded_period(self): """Получение последнего загруженного периода.""" periods = [_period() for _ in range(3)] for year, month in periods: ProcurementRecordFactory(data_year=year, data_month=month) expected_year, expected_month = max(periods) year, month = ProcurementService.get_last_loaded_period() self.assertEqual(year, expected_year) self.assertEqual(month, expected_month) def test_get_last_loaded_period_by_region(self): """Получение последнего периода с фильтром по региону.""" region_a = _region_code() region_b = _region_code() while region_b == region_a: region_b = _region_code() period_a = _period() period_b = _period() ProcurementRecordFactory( data_year=period_a[0], data_month=period_a[1], region_code=region_a ) ProcurementRecordFactory( data_year=period_b[0], data_month=period_b[1], region_code=region_b ) year, month = ProcurementService.get_last_loaded_period(region_code=region_a) self.assertEqual(year, period_a[0]) self.assertEqual(month, period_a[1]) def test_get_last_loaded_period_by_law_type(self): """Получение последнего периода с фильтром по типу закона.""" law_type_a = fake.random_element(["44-FZ", "223-FZ"]) law_type_b = _other_law(law_type_a) period_a = _period() period_b = _period() ProcurementRecordFactory( data_year=period_a[0], data_month=period_a[1], law_type=law_type_a ) ProcurementRecordFactory( data_year=period_b[0], data_month=period_b[1], law_type=law_type_b ) year, month = ProcurementService.get_last_loaded_period(law_type=law_type_a) self.assertEqual(year, period_a[0]) self.assertEqual(month, period_a[1]) def test_has_data_for_period_true(self): """Проверка наличия данных за период - есть данные.""" year, month = _period() ProcurementRecordFactory(data_year=year, data_month=month) result = ProcurementService.has_data_for_period(year, month) self.assertTrue(result) def test_has_data_for_period_false(self): """Проверка наличия данных за период - нет данных.""" year, month = _period() ProcurementRecordFactory(data_year=year, data_month=month) other_month = month % 12 + 1 result = ProcurementService.has_data_for_period(year, other_month) self.assertFalse(result) def test_has_data_for_period_with_filters(self): """Проверка наличия данных с фильтрами.""" year, month = _period() region_code = _region_code() law_type = fake.random_element(["44-FZ", "223-FZ"]) ProcurementRecordFactory( data_year=year, data_month=month, region_code=region_code, law_type=law_type, ) # С правильными фильтрами - есть self.assertTrue( ProcurementService.has_data_for_period( year, month, region_code=region_code, law_type=law_type ) ) # С неправильным регионом - нет self.assertFalse( ProcurementService.has_data_for_period( year, month, region_code=_region_code() ) ) class ProcurementServiceBaseMethodsTestCase(TestCase): """Тесты базовых методов от BaseService.""" def test_count(self): """Подсчёт записей.""" ProcurementRecordFactory.create_batch(5) self.assertEqual(ProcurementService.count(), 5) def test_exists(self): """Проверка существования.""" record = ProcurementRecordFactory() self.assertTrue( ProcurementService.exists(purchase_number=record.purchase_number) ) self.assertFalse(ProcurementService.exists(purchase_number=_digits(19))) def test_get_by_id(self): """Получение по ID.""" record = ProcurementRecordFactory() found = ProcurementService.get_by_id(record.pk) self.assertEqual(found.pk, record.pk) def test_get_all(self): """Получение всех записей.""" ProcurementRecordFactory.create_batch(3) all_records = ProcurementService.get_all() self.assertEqual(all_records.count(), 3) def test_filter(self): """Фильтрация записей.""" law_type = fake.random_element(["44-FZ", "223-FZ"]) other_law = _other_law(law_type) ProcurementRecordFactory(law_type=law_type) ProcurementRecordFactory(law_type=law_type) ProcurementRecordFactory(law_type=other_law) filtered = ProcurementService.filter(law_type=law_type) self.assertEqual(filtered.count(), 2)