Aleksandr Meshchriakov
ee95628a0a
Some checks failed
CI/CD Pipeline / Run Tests (push) Failing after 37s
CI/CD Pipeline / Code Quality Checks (push) Failing after 43s
CI/CD Pipeline / Build & Push Images (push) Has been skipped
CI/CD Pipeline / Deploy (dev) (push) Has been skipped
CI/CD Pipeline / Deploy (prod) (push) Has been skipped
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 0s
CI/CD Pipeline / Run Tests (pull_request) Failing after 0s
CI/CD Pipeline / Build & Push Images (pull_request) Has been skipped
CI/CD Pipeline / Deploy (dev) (pull_request) Has been skipped
CI/CD Pipeline / Deploy (prod) (pull_request) Has been skipped
- Обновлены клиенты парсеров (checko, fns, minpromtorg, proverki, zakupki) - Добавлены новые миграции для моделей - Расширено покрытие тестами - Обновлены конфигурации и настройки проекта - Добавлены утилиты для тестирования Co-Authored-By: Warp <agent@warp.dev>
242 lines
9.8 KiB
Python
242 lines
9.8 KiB
Python
"""Tests for parsers admin configurations."""
|
|
|
|
from django.contrib.admin.sites import AdminSite
|
|
from django.contrib.messages.storage.fallback import FallbackStorage
|
|
from django.test import RequestFactory, TestCase
|
|
|
|
from apps.parsers.admin import (
|
|
FinancialReportAdmin,
|
|
HasCertificateNumberFilter,
|
|
IndustrialCertificateRecordAdmin,
|
|
InspectionRecordAdmin,
|
|
ManufacturerRecordAdmin,
|
|
ParserLoadLogAdmin,
|
|
ProcurementRecordAdmin,
|
|
ProxyAdmin,
|
|
)
|
|
from apps.parsers.models import (
|
|
FinancialReport,
|
|
FinancialReportLine,
|
|
IndustrialCertificateRecord,
|
|
InspectionRecord,
|
|
ManufacturerRecord,
|
|
ParserLoadLog,
|
|
ProcurementRecord,
|
|
Proxy,
|
|
)
|
|
from tests.apps.parsers.factories import (
|
|
IndustrialCertificateRecordFactory,
|
|
InspectionRecordFactory,
|
|
ManufacturerRecordFactory,
|
|
ParserLoadLogFactory,
|
|
ProxyFactory,
|
|
)
|
|
from tests.apps.user.factories import UserFactory
|
|
from tests.utils.fixtures import fake
|
|
|
|
|
|
_CYRILLIC_FINISHED = "\u0437\u0430\u0432\u0435\u0440\u0448\u0435\u043d"
|
|
_CYRILLIC_PUBLISHED = "\u043e\u043f\u0443\u0431\u043b\u0438\u043a"
|
|
|
|
|
|
def _digits(length: int) -> str:
|
|
return "".join(str(fake.random_int(0, 9)) for _ in range(length))
|
|
|
|
|
|
class ParsersAdminTest(TestCase):
|
|
def setUp(self):
|
|
self.site = AdminSite()
|
|
self.factory = RequestFactory()
|
|
self.user = UserFactory.create_superuser()
|
|
|
|
def _request(self):
|
|
request = self.factory.get("/")
|
|
request.user = self.user
|
|
request.session = {}
|
|
request._messages = FallbackStorage(request)
|
|
return request
|
|
|
|
def test_proxy_admin_actions(self):
|
|
admin = ProxyAdmin(Proxy, self.site)
|
|
proxy = ProxyFactory(is_active=False, fail_count=5)
|
|
request = self._request()
|
|
qs = Proxy.objects.filter(id=proxy.id)
|
|
|
|
admin.activate_proxies(request, qs)
|
|
proxy.refresh_from_db()
|
|
self.assertTrue(proxy.is_active)
|
|
|
|
admin.deactivate_proxies(request, qs)
|
|
proxy.refresh_from_db()
|
|
self.assertFalse(proxy.is_active)
|
|
|
|
admin.reset_fail_count(request, qs)
|
|
proxy.refresh_from_db()
|
|
self.assertEqual(proxy.fail_count, 0)
|
|
|
|
def test_proxy_active_badge(self):
|
|
admin = ProxyAdmin(Proxy, self.site)
|
|
active = ProxyFactory(is_active=True)
|
|
inactive = ProxyFactory(is_active=False)
|
|
self.assertIn("span", str(admin.is_active_badge(active)))
|
|
self.assertIn("span", str(admin.is_active_badge(inactive)))
|
|
|
|
def test_parser_load_log_admin_status_badge(self):
|
|
admin = ParserLoadLogAdmin(ParserLoadLog, self.site)
|
|
log = ParserLoadLogFactory(status="success")
|
|
badge = admin.status_badge(log)
|
|
self.assertIn("span", str(badge))
|
|
request = self._request()
|
|
self.assertFalse(admin.has_add_permission(request))
|
|
|
|
def test_certificate_admin_helpers(self):
|
|
admin = IndustrialCertificateRecordAdmin(IndustrialCertificateRecord, self.site)
|
|
record = IndustrialCertificateRecordFactory(organisation_name="X" * 80)
|
|
short_name = admin.organisation_name_short(record)
|
|
self.assertTrue(short_name.endswith("..."))
|
|
|
|
request = self._request()
|
|
self.assertFalse(admin.has_add_permission(request))
|
|
self.assertFalse(admin.has_change_permission(request))
|
|
|
|
def test_certificate_filter(self):
|
|
admin = IndustrialCertificateRecordAdmin(IndustrialCertificateRecord, self.site)
|
|
record_good = IndustrialCertificateRecordFactory(certificate_number="CN-1")
|
|
record_bad = IndustrialCertificateRecordFactory(certificate_number="-")
|
|
|
|
request = self._request()
|
|
filter_yes = HasCertificateNumberFilter(
|
|
request, {"has_cert_number": "yes"}, IndustrialCertificateRecord, admin
|
|
)
|
|
qs_yes = filter_yes.queryset(request, IndustrialCertificateRecord.objects.all())
|
|
self.assertIn(record_good, qs_yes)
|
|
self.assertNotIn(record_bad, qs_yes)
|
|
|
|
filter_no = HasCertificateNumberFilter(
|
|
request, {"has_cert_number": "no"}, IndustrialCertificateRecord, admin
|
|
)
|
|
qs_no = filter_no.queryset(request, IndustrialCertificateRecord.objects.all())
|
|
self.assertIn(record_bad, qs_no)
|
|
|
|
filter_none = HasCertificateNumberFilter(
|
|
request, {}, IndustrialCertificateRecord, admin
|
|
)
|
|
qs_none = filter_none.queryset(request, IndustrialCertificateRecord.objects.all())
|
|
self.assertIn(record_good, qs_none)
|
|
|
|
def test_manufacturer_admin_helpers(self):
|
|
admin = ManufacturerRecordAdmin(ManufacturerRecord, self.site)
|
|
record = ManufacturerRecordFactory(
|
|
full_legal_name="Y" * 80,
|
|
address="A" * 80,
|
|
)
|
|
self.assertTrue(admin.full_legal_name_short(record).endswith("..."))
|
|
self.assertTrue(admin.address_short(record).endswith("..."))
|
|
request = self._request()
|
|
self.assertFalse(admin.has_add_permission(request))
|
|
self.assertFalse(admin.has_change_permission(request))
|
|
|
|
def test_inspection_admin_helpers(self):
|
|
admin = InspectionRecordAdmin(InspectionRecord, self.site)
|
|
record = InspectionRecordFactory(
|
|
organisation_name="Org" * 30,
|
|
control_authority="Auth" * 20,
|
|
status=f"{_CYRILLIC_FINISHED}"
|
|
)
|
|
self.assertTrue(admin.organisation_name_short(record).endswith("..."))
|
|
self.assertTrue(admin.control_authority_short(record).endswith("..."))
|
|
self.assertIn("span", str(admin.status_badge(record)))
|
|
record_progress = InspectionRecordFactory(status="\u0432 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0435")
|
|
record_cancel = InspectionRecordFactory(status="\u043e\u0442\u043c\u0435\u043d\u0435\u043d\u0430")
|
|
record_other = InspectionRecordFactory(status="unknown")
|
|
self.assertIn("span", str(admin.status_badge(record_progress)))
|
|
self.assertIn("span", str(admin.status_badge(record_cancel)))
|
|
self.assertIn("span", str(admin.status_badge(record_other)))
|
|
request = self._request()
|
|
self.assertFalse(admin.has_add_permission(request))
|
|
self.assertFalse(admin.has_change_permission(request))
|
|
|
|
def test_procurement_admin_helpers(self):
|
|
admin = ProcurementRecordAdmin(ProcurementRecord, self.site)
|
|
base_number = _digits(18)
|
|
record = ProcurementRecord.objects.create(
|
|
load_batch=fake.random_int(min=1, max=1000),
|
|
purchase_number=f"{base_number}0",
|
|
purchase_name="Name" * 30,
|
|
customer_inn=_digits(10),
|
|
customer_kpp=_digits(9),
|
|
customer_ogrn=_digits(13),
|
|
customer_name="Customer" * 20,
|
|
max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)),
|
|
status=f"{_CYRILLIC_PUBLISHED}",
|
|
law_type="44-FZ",
|
|
)
|
|
self.assertTrue(admin.purchase_name_short(record).endswith("..."))
|
|
self.assertTrue(admin.customer_name_short(record).endswith("..."))
|
|
self.assertIn("span", str(admin.status_badge(record)))
|
|
record_finished = ProcurementRecord.objects.create(
|
|
load_batch=fake.random_int(min=1, max=1000),
|
|
purchase_number=f"{base_number}1",
|
|
purchase_name="Name",
|
|
customer_inn=_digits(10),
|
|
customer_kpp=_digits(9),
|
|
customer_ogrn=_digits(13),
|
|
customer_name="Customer",
|
|
max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)),
|
|
status="\u0437\u0430\u0432\u0435\u0440\u0448\u0435\u043d",
|
|
law_type="44-FZ",
|
|
)
|
|
record_cancel = ProcurementRecord.objects.create(
|
|
load_batch=fake.random_int(min=1, max=1000),
|
|
purchase_number=f"{base_number}2",
|
|
purchase_name="Name",
|
|
customer_inn=_digits(10),
|
|
customer_kpp=_digits(9),
|
|
customer_ogrn=_digits(13),
|
|
customer_name="Customer",
|
|
max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)),
|
|
status="\u043e\u0442\u043c\u0435\u043d\u0435\u043d\u0430",
|
|
law_type="44-FZ",
|
|
)
|
|
record_other = ProcurementRecord.objects.create(
|
|
load_batch=fake.random_int(min=1, max=1000),
|
|
purchase_number=f"{base_number}3",
|
|
purchase_name="Name",
|
|
customer_inn=_digits(10),
|
|
customer_kpp=_digits(9),
|
|
customer_ogrn=_digits(13),
|
|
customer_name="Customer",
|
|
max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)),
|
|
status="unknown",
|
|
law_type="44-FZ",
|
|
)
|
|
self.assertIn("span", str(admin.status_badge(record_finished)))
|
|
self.assertIn("span", str(admin.status_badge(record_cancel)))
|
|
self.assertIn("span", str(admin.status_badge(record_other)))
|
|
request = self._request()
|
|
self.assertFalse(admin.has_add_permission(request))
|
|
self.assertFalse(admin.has_change_permission(request))
|
|
|
|
def test_financial_report_admin(self):
|
|
admin = FinancialReportAdmin(FinancialReport, self.site)
|
|
report = FinancialReport.objects.create(
|
|
external_id=_digits(5),
|
|
ogrn=_digits(13),
|
|
file_name=f"fin_{_digits(5)}_{_digits(13)}.xlsx",
|
|
file_hash=fake.sha256(raw_output=False),
|
|
load_batch=fake.random_int(min=1, max=1000),
|
|
status=FinancialReport.Status.SUCCESS,
|
|
source=FinancialReport.SourceType.API,
|
|
)
|
|
FinancialReportLine.objects.create(
|
|
report=report,
|
|
form_code="1",
|
|
line_code=_digits(4),
|
|
line_name=fake.word(),
|
|
year=fake.random_int(min=2020, max=2025),
|
|
period_start=fake.random_int(min=1, max=999),
|
|
period_end=fake.random_int(min=1, max=999),
|
|
)
|
|
self.assertEqual(admin.lines_count(report), 1)
|
|
self.assertIn("span", str(admin.status_badge(report)))
|