""" Unit-тесты для ProcurementService. Тестирует сервис для работы с данными о закупках. """ from apps.parsers.clients.zakupki.schemas import Procurement from apps.parsers.models import ParserLoadLog, ProcurementRecord from apps.parsers.services import ProcurementService from django.test import TestCase from organizations.models import Organization, OrganizationSourceRecord from tests.apps.parsers.factories import ProcurementRecordFactory, fake from tests.apps.parsers.organization_helpers import create_directory_organization 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 _other_region_code(region_code: str) -> str: next_region = int(region_code) % 99 + 1 return str(next_region).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 create_directory_organization( name=fake.company(), ogrn=ogrn, inn=inn, kpp=_digits(9), okpo=_digits(8), ) def _ensure_directory_organization_for_procurement( procurement: Procurement, ) -> Organization: return create_directory_organization( name=procurement.customer_name, ogrn=procurement.customer_ogrn, inn=procurement.customer_inn, kpp=procurement.customer_kpp, 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) def _procurement_records(): return OrganizationSourceRecord.objects.filter( source=ParserLoadLog.Source.PROCUREMENTS, ) def _save_procurement_source_record( *, batch_id: int = 1, region_code: str | None = None, data_year: int | None = None, data_month: int | None = None, **overrides, ) -> Procurement: procurement = _build_procurement(**overrides) _ensure_directory_organization_for_procurement(procurement) ProcurementService.save_procurements( [procurement], batch_id=batch_id, region_code=region_code, data_year=data_year, data_month=data_month, ) return procurement 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, ) _ensure_directory_organization_for_procurement(procurement) 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(), 0) self.assertEqual(_procurement_records().count(), 1) record = _procurement_records().first() self.assertEqual(record.external_id, purchase_number) self.assertEqual(record.extension.organization.inn, customer_inn) self.assertEqual(record.payload["region_code"], region_code) self.assertEqual(record.payload["data_year"], year) self.assertEqual(record.payload["data_month"], month) def test_save_multiple_procurements(self): """Сохранение нескольких закупок.""" procurements = [_build_procurement() for _ in range(5)] for procurement in procurements: _ensure_directory_organization_for_procurement(procurement) saved = ProcurementService.save_procurements(procurements, batch_id=1) self.assertEqual(saved, 5) self.assertEqual(ProcurementRecord.objects.count(), 0) self.assertEqual(_procurement_records().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 = _procurement_records().get(external_id=purchase_number) self.assertEqual(record.registry_organization_id, organization.id) def test_save_updates_duplicates(self): """Повторная синхронизация обновляет существующую закупку.""" purchase_number = _digits(19) _save_procurement_source_record(purchase_number=purchase_number, batch_id=1) original = _procurement_records().get(external_id=purchase_number) procurement = _build_procurement( purchase_number=purchase_number, customer_inn=_digits(10), ) _ensure_directory_organization_for_procurement(procurement) saved = ProcurementService.save_procurements([procurement], batch_id=2) self.assertEqual(saved, 1) self.assertEqual(ProcurementRecord.objects.count(), 0) self.assertEqual(_procurement_records().count(), 1) refreshed = _procurement_records().get(external_id=purchase_number) self.assertEqual(refreshed.extension.organization.inn, procurement.customer_inn) self.assertEqual(refreshed.load_batch, 2) self.assertEqual(refreshed.uid, original.uid) def test_save_with_chunking(self): """Сохранение большого количества записей чанками.""" procurements = [_build_procurement() for _ in range(100)] for procurement in procurements: _ensure_directory_organization_for_procurement(procurement) saved = ProcurementService.save_procurements( procurements, batch_id=1, chunk_size=25 ) self.assertEqual(saved, 100) self.assertEqual(ProcurementRecord.objects.count(), 0) self.assertEqual(_procurement_records().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 = _save_procurement_source_record( 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, batch_id=1, ) self.record2 = _save_procurement_source_record( purchase_number=_digits(19), customer_inn=self.inn_other, customer_name=f"{self.name_key} {fake.company()}", region_code=self.region_a, batch_id=1, ) self.record3 = _save_procurement_source_record( purchase_number=_digits(19), customer_inn=self.inn_target, # Тот же ИНН что и у первого customer_name=f"{self.name_key} {fake.company()}", region_code=self.region_b, batch_id=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().external_id, 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().extension.organization.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: _save_procurement_source_record(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() _save_procurement_source_record( data_year=period_a[0], data_month=period_a[1], region_code=region_a ) _save_procurement_source_record( 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() _save_procurement_source_record( data_year=period_a[0], data_month=period_a[1], law_type=law_type_a ) _save_procurement_source_record( 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() _save_procurement_source_record(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() _save_procurement_source_record(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"]) _save_procurement_source_record( 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=_other_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)