Aleksandr Meshchriakov
3d298ce352
Some checks failed
CI/CD Pipeline / Run Tests (pull_request) Successful in 1m53s
CI/CD Pipeline / Telegram Notify Success (push) Has been cancelled
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 2m54s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
380 lines
14 KiB
Python
380 lines
14 KiB
Python
"""
|
||
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)
|