"""Tests for core admin configurations.""" from datetime import date, timedelta from unittest.mock import patch from apps.core.admin import BackgroundJobAdmin from apps.core.admin_dashboard import build_admin_dashboard from apps.core.models import BackgroundJob from apps.parsers.models import FinancialReport, FinancialReportLine, ParserLoadLog from django.contrib.admin.sites import AdminSite from django.contrib.messages.storage.fallback import FallbackStorage from django.test import RequestFactory, TestCase from django.urls import reverse from django.utils import timezone from tests.apps.parsers.factories import ( IndustrialCertificateRecordFactory, InspectionRecordFactory, ManufacturerRecordFactory, ParserLoadLogFactory, ProcurementRecordFactory, ProxyFactory, ) from tests.apps.registers.factories import ( OrganizationFactory, RegisterFactory, RegisterUploadFactory, RegistryMembershipPeriodFactory, ) from tests.apps.user.factories import UserFactory from tests.utils.fixtures import fake class CoreAdminTest(TestCase): def setUp(self): self.site = AdminSite() self.admin = BackgroundJobAdmin(BackgroundJob, self.site) self.user = UserFactory.create_superuser() self.factory = RequestFactory() def _request(self): request = self.factory.get("/") request.user = self.user request.session = {} request._messages = FallbackStorage(request) return request def test_task_name_short(self): job = BackgroundJob.objects.create( task_id=fake.uuid4(), task_name="apps.parsers.tasks.parse_industrial_production", ) self.assertEqual(self.admin.task_name_short(job), "parse_industrial_production") job_short = BackgroundJob.objects.create( task_id=fake.uuid4(), task_name="short.task", ) self.assertEqual(self.admin.task_name_short(job_short), "short.task") def test_status_badge_and_progress(self): job = BackgroundJob.objects.create( task_id=fake.uuid4(), task_name="test.task", status="success", progress=100, ) badge = self.admin.status_badge(job) bar = self.admin.progress_bar(job) self.assertIn("span", str(badge)) self.assertIn("100%", str(bar)) def test_duration_display(self): job = BackgroundJob.objects.create( task_id=fake.uuid4(), task_name="test.task", ) self.assertEqual(self.admin.duration_display(job), "-") job.started_at = timezone.now() job.completed_at = job.started_at + timedelta(seconds=30) job.save(update_fields=["started_at", "completed_at"]) self.assertIn("сек", self.admin.duration_display(job)) job.completed_at = job.started_at + timedelta(seconds=90) job.save(update_fields=["started_at", "completed_at"]) self.assertIn("мин", self.admin.duration_display(job)) def test_permissions(self): request = self._request() self.assertFalse(self.admin.has_add_permission(request)) self.assertFalse(self.admin.has_change_permission(request)) def test_revoke_jobs_action_no_active(self): BackgroundJob.objects.create( task_id=fake.uuid4(), task_name="test.task", status="success", ) request = self._request() qs = BackgroundJob.objects.all() self.admin.revoke_jobs(request, qs) def test_revoke_jobs_action_pending(self): job = BackgroundJob.objects.create( task_id=fake.uuid4(), task_name="test.task", status="pending", ) request = self._request() qs = BackgroundJob.objects.all() with patch("celery.current_app.control.revoke") as revoke_mock: self.admin.revoke_jobs(request, qs) revoke_mock.assert_called_once_with(job.task_id, terminate=True) job.refresh_from_db() self.assertEqual(job.status, "revoked") class AdminDashboardTest(TestCase): def setUp(self): self.user = UserFactory.create_superuser() self.client.force_login(self.user) def test_admin_index_renders_custom_dashboard_with_live_metrics(self): primary_registry = RegisterFactory(name="Реестр Росатом ОПК тестовый") secondary_registry = RegisterFactory(name="Реестр Роскосмос ГОЗ тестовый") primary_upload = RegisterUploadFactory( registry=primary_registry, actual_date=date(2026, 3, 20), rows_count=24, ) secondary_upload = RegisterUploadFactory( registry=secondary_registry, actual_date=date(2026, 3, 21), rows_count=11, ) first_org = OrganizationFactory(pn_name='АО "Росатом Тест"') second_org = OrganizationFactory(pn_name='АО "Роскосмос Тест"') RegistryMembershipPeriodFactory( registry=primary_registry, organization=first_org, started_by_upload=primary_upload, ) RegistryMembershipPeriodFactory( registry=secondary_registry, organization=second_org, started_by_upload=secondary_upload, ) report = FinancialReport.objects.create( external_id="fns-dashboard-1", ogrn="1027700000001", file_name="fin_dashboard_1.xlsx", file_hash="fns-dashboard-file-hash-1", load_batch=101, status=FinancialReport.Status.SUCCESS, source=FinancialReport.SourceType.API, ) FinancialReportLine.objects.create( report=report, form_code="1", line_code="1100", line_name="Итого по разделу", year=2025, period_start=10, period_end=15, ) IndustrialCertificateRecordFactory(inn="7700000001") ManufacturerRecordFactory(inn="7700000002") InspectionRecordFactory(inn="7800000001") ProcurementRecordFactory( load_batch=303, region_code="77", customer_inn="7700000001", ) ProcurementRecordFactory( load_batch=303, region_code="77", customer_inn="7700000002", ) ProcurementRecordFactory( load_batch=303, region_code="78", customer_inn="7800000001", ) ProxyFactory(is_active=True) ParserLoadLogFactory( source=ParserLoadLog.Source.PROCUREMENTS, batch_id=303, status="success", records_count=3, ) ParserLoadLogFactory( source=ParserLoadLog.Source.INDUSTRIAL, status="failed", records_count=0, error_message="Источник временно недоступен", ) response = self.client.get(reverse("admin:index")) self.assertEqual(response.status_code, 200) self.assertNotContains(response, "Mostovik") self.assertContains(response, "Панель данных и загрузок") self.assertContains(response, "Обзор") self.assertContains(response, "Аналитика") self.assertContains(response, "Админка") self.assertContains(response, "Покрытие по внешним данным") self.assertContains(response, "Топ регионов по закупкам") self.assertContains(response, "Реестр Росатом ОПК тестовый") self.assertContains(response, "Реестр Роскосмос ГОЗ тестовый") self.assertContains(response, "Москва") self.assertContains(response, "Санкт-Петербург") self.assertContains(response, "Источник временно недоступен") self.assertContains( response, reverse("admin:registers_registerupload_upload_excel"), ) self.assertContains( response, reverse("admin:parsers_financialreport_upload_excel"), ) def test_dashboard_source_cards_use_latest_successful_source_slice(self): InspectionRecordFactory(load_batch=101, registration_number="old-1") InspectionRecordFactory(load_batch=101, registration_number="old-2") InspectionRecordFactory(load_batch=202, registration_number="new-1") ParserLoadLogFactory( source=ParserLoadLog.Source.INSPECTIONS, batch_id=101, status=ParserLoadLog.Status.SUCCESS, records_count=2, ) ParserLoadLogFactory( source=ParserLoadLog.Source.INSPECTIONS, batch_id=202, status=ParserLoadLog.Status.SUCCESS, records_count=1, ) dashboard = build_admin_dashboard() cards = {item["slug"]: item for item in dashboard["source_cards"]} self.assertEqual(cards["planned-inspections"]["records_count"], 1) self.assertEqual(cards["planned-inspections"]["organizations_count"], 1) self.assertEqual( cards["planned-inspections"]["metrics_scope_label"], "Последний успешный срез", ) def test_dashboard_regions_use_latest_successful_procurements_batch(self): ProcurementRecordFactory( load_batch=11, purchase_number="1111111111111111111", region_code="77", customer_inn="7700000001", ) ProcurementRecordFactory( load_batch=22, purchase_number="2222222222222222222", region_code="78", customer_inn="7800000001", ) ParserLoadLogFactory( source=ParserLoadLog.Source.PROCUREMENTS, batch_id=11, status=ParserLoadLog.Status.SUCCESS, records_count=1, ) ParserLoadLogFactory( source=ParserLoadLog.Source.PROCUREMENTS, batch_id=22, status=ParserLoadLog.Status.SUCCESS, records_count=1, ) dashboard = build_admin_dashboard() self.assertEqual(len(dashboard["region_rows"]), 1) self.assertEqual(dashboard["region_rows"][0]["code"], "78") self.assertEqual( dashboard["region_rows_note"], "Показываем только последний успешный срез ЕИС закупок.", )