207 lines
7.8 KiB
Python
207 lines
7.8 KiB
Python
"""Tests for core admin configurations."""
|
||
|
||
from datetime import date, timedelta
|
||
from unittest.mock import patch
|
||
|
||
from apps.core.admin import BackgroundJobAdmin
|
||
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(region_code="77", customer_inn="7700000001")
|
||
ProcurementRecordFactory(region_code="77", customer_inn="7700000002")
|
||
ProcurementRecordFactory(region_code="78", customer_inn="7800000001")
|
||
ProxyFactory(is_active=True)
|
||
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.PROCUREMENTS,
|
||
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.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"),
|
||
)
|