294 lines
10 KiB
Python
294 lines
10 KiB
Python
"""Tests for frontend source cards API."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
from apps.core.models import BackgroundJob, JobStatus
|
||
from apps.parsers.models import ParserLoadLog
|
||
from apps.parsers.source_cards import SourceCardService
|
||
from django.test import override_settings
|
||
from django.urls import NoReverseMatch, reverse
|
||
from organizations.models import Organization
|
||
from organizations.source_ingestion import (
|
||
OrganizationSourceIngestionService,
|
||
SourceRecordInput,
|
||
)
|
||
from rest_framework import status
|
||
from rest_framework.test import APITestCase
|
||
|
||
from tests.apps.parsers.factories import ParserLoadLogFactory
|
||
from tests.apps.parsers.organization_helpers import (
|
||
create_directory_organization,
|
||
get_or_create_directory_organization,
|
||
)
|
||
from tests.apps.user.factories import UserFactory
|
||
from tests.utils.fixtures import fake
|
||
|
||
|
||
def _digits(length: int) -> str:
|
||
return "".join(str(fake.random_int(0, 9)) for _ in range(length))
|
||
|
||
|
||
def _save_source_record(
|
||
*,
|
||
source: str,
|
||
external_id: str,
|
||
inn: str = "",
|
||
ogrn: str = "",
|
||
organization_name: str = "",
|
||
title: str = "",
|
||
) -> None:
|
||
payload: dict = {}
|
||
if inn or ogrn:
|
||
query = {}
|
||
if inn:
|
||
query["inn"] = inn
|
||
if ogrn:
|
||
query["ogrn"] = ogrn
|
||
if not Organization.objects.filter(**query).exists():
|
||
create_directory_organization(
|
||
name=organization_name or title or external_id,
|
||
inn=inn,
|
||
ogrn=ogrn,
|
||
)
|
||
else:
|
||
rn_key = organization_name or title or external_id
|
||
rn = str(int(hashlib.sha256(rn_key.encode()).hexdigest()[:12], 16) % 10**12)
|
||
payload["rn"] = rn
|
||
get_or_create_directory_organization(
|
||
rn=rn,
|
||
defaults={"name": organization_name or title or external_id},
|
||
)
|
||
OrganizationSourceIngestionService.save_records(
|
||
source=source,
|
||
load_batch=1,
|
||
records=[
|
||
SourceRecordInput(
|
||
external_id=external_id,
|
||
title=title,
|
||
organization_name=organization_name or title or external_id,
|
||
inn=inn,
|
||
ogrn=ogrn,
|
||
payload=payload,
|
||
)
|
||
],
|
||
)
|
||
|
||
|
||
class SourceCardsApiTestCase(APITestCase):
|
||
def setUp(self):
|
||
SourceCardService.clear_cache()
|
||
self.user = UserFactory.create_user()
|
||
self.admin = UserFactory.create_user(is_staff=True)
|
||
self.client.force_authenticate(self.user)
|
||
|
||
def test_source_cards_list_returns_aggregated_data(self):
|
||
report_ogrn = _digits(13)
|
||
_save_source_record(
|
||
source=ParserLoadLog.Source.FNS_REPORTS,
|
||
external_id=_digits(5),
|
||
ogrn=report_ogrn,
|
||
organization_name='ООО "Финансовая компания"',
|
||
title=f"fin_{_digits(5)}_{report_ogrn}.xlsx",
|
||
)
|
||
_save_source_record(
|
||
source=ParserLoadLog.Source.FNS_REPORTS,
|
||
external_id=_digits(5),
|
||
ogrn=report_ogrn,
|
||
organization_name='ООО "Финансовая компания"',
|
||
title=f"fin_{_digits(5)}_{report_ogrn}.xlsx",
|
||
)
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.FNS_REPORTS,
|
||
status="success",
|
||
records_count=2,
|
||
)
|
||
BackgroundJob.objects.create(
|
||
task_id="job-inspections-active",
|
||
task_name="apps.parsers.tasks.sync_inspections",
|
||
status=JobStatus.STARTED,
|
||
progress=63,
|
||
progress_message="sync",
|
||
user_id=self.user.id,
|
||
meta={"source": ParserLoadLog.Source.INSPECTIONS},
|
||
)
|
||
|
||
response = self.client.get(reverse("api_v1:sources:source-cards-list"))
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data["success"])
|
||
cards = {item["slug"]: item for item in response.data["data"]}
|
||
|
||
self.assertIn("financial-indicators", cards)
|
||
self.assertIn("planned-inspections", cards)
|
||
|
||
fns_card = cards["financial-indicators"]
|
||
self.assertEqual(fns_card["records_count"], 2)
|
||
self.assertEqual(fns_card["organizations_count"], 1)
|
||
self.assertEqual(fns_card["status"], "success")
|
||
self.assertFalse(fns_card["refresh_requires_params"])
|
||
|
||
inspections_card = cards["planned-inspections"]
|
||
self.assertEqual(inspections_card["status"], "in_progress")
|
||
self.assertEqual(inspections_card["progress"], 63)
|
||
|
||
def test_main_dashboard_route_is_removed_with_registers_api(self):
|
||
with self.assertRaises(NoReverseMatch):
|
||
reverse("api_v1:stat:main-dashboard")
|
||
|
||
def test_successful_parser_load_does_not_restore_removed_stat_route(self):
|
||
with self.captureOnCommitCallbacks(execute=True):
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.FNS_REPORTS,
|
||
status="success",
|
||
records_count=1,
|
||
)
|
||
|
||
with self.assertRaises(NoReverseMatch):
|
||
reverse("api_v1:stat:main-dashboard")
|
||
|
||
def test_source_card_detail_returns_combined_minprom_stats(self):
|
||
shared_inn = _digits(10)
|
||
_save_source_record(
|
||
source=ParserLoadLog.Source.INDUSTRIAL,
|
||
external_id="industrial-1",
|
||
inn=shared_inn,
|
||
organization_name='ООО "Производитель"',
|
||
title="Сертификат промышленной продукции",
|
||
)
|
||
_save_source_record(
|
||
source=ParserLoadLog.Source.INDUSTRIAL_PRODUCTS,
|
||
external_id="product-1",
|
||
inn=shared_inn,
|
||
organization_name='ООО "Производитель"',
|
||
title="Промышленная продукция",
|
||
)
|
||
_save_source_record(
|
||
source=ParserLoadLog.Source.MANUFACTURES,
|
||
external_id="manufacturer-1",
|
||
inn=shared_inn,
|
||
organization_name='ООО "Производитель"',
|
||
title="Производитель",
|
||
)
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.INDUSTRIAL,
|
||
status="success",
|
||
records_count=1,
|
||
)
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.INDUSTRIAL_PRODUCTS,
|
||
status="success",
|
||
records_count=1,
|
||
)
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.MANUFACTURES,
|
||
status="success",
|
||
records_count=1,
|
||
)
|
||
|
||
response = self.client.get(
|
||
reverse(
|
||
"api_v1:sources:source-cards-detail",
|
||
kwargs={"slug": "manufacturers-and-products"},
|
||
)
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data["success"])
|
||
card = response.data["data"]
|
||
|
||
self.assertEqual(card["records_count"], 3)
|
||
self.assertEqual(card["organizations_count"], 1)
|
||
self.assertEqual(card["status"], "success")
|
||
self.assertEqual(len(card["source_items"]), 3)
|
||
self.assertEqual(card["source_items"][0]["latest_load"]["status"], "success")
|
||
|
||
def test_source_task_statuses_returns_table_rows(self):
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.INDUSTRIAL,
|
||
status="success",
|
||
records_count=12,
|
||
)
|
||
ParserLoadLogFactory(
|
||
source=ParserLoadLog.Source.INSPECTIONS,
|
||
status="failed",
|
||
records_count=0,
|
||
)
|
||
|
||
response = self.client.get(reverse("api_v1:sources:source-cards-statuses"))
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data["success"])
|
||
rows = response.data["data"]
|
||
self.assertGreaterEqual(len(rows), 1)
|
||
|
||
row = rows[0]
|
||
self.assertIn("row_number", row)
|
||
self.assertIn("source", row)
|
||
self.assertIn("actualized_at", row)
|
||
self.assertIn("next_update_at", row)
|
||
self.assertIn("records_count", row)
|
||
self.assertIn("organizations_count", row)
|
||
self.assertIn("status", row)
|
||
self.assertIn("status_label", row)
|
||
self.assertIn("active_tasks", row)
|
||
|
||
@override_settings(CELERY_TASK_ALWAYS_EAGER=False)
|
||
def test_refresh_creates_background_job_and_returns_task(self):
|
||
self.client.force_authenticate(self.admin)
|
||
response = self.client.post(
|
||
reverse(
|
||
"api_v1:sources:source-cards-refresh",
|
||
kwargs={"slug": "financial-indicators"},
|
||
),
|
||
{},
|
||
format="json",
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
||
self.assertEqual(response.data["status"], "accepted")
|
||
self.assertEqual(set(response.data.keys()), {"task_id", "status"})
|
||
|
||
task_id = response.data["task_id"]
|
||
self.assertTrue(
|
||
BackgroundJob.objects.filter(
|
||
task_id=task_id,
|
||
task_name="apps.parsers.tasks.sync_fns_financial_reports",
|
||
user_id=self.admin.id,
|
||
).exists()
|
||
)
|
||
|
||
@override_settings(CELERY_TASK_ALWAYS_EAGER=False)
|
||
def test_refresh_procurements_does_not_require_region_code(self):
|
||
self.client.force_authenticate(self.admin)
|
||
with patch(
|
||
"apps.parsers.tasks.parse_registry_enrichment_sources.apply_async",
|
||
return_value=SimpleNamespace(id="task-procurements"),
|
||
):
|
||
response = self.client.post(
|
||
reverse(
|
||
"api_v1:sources:source-cards-refresh",
|
||
kwargs={"slug": "public-procurements"},
|
||
),
|
||
{},
|
||
format="json",
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
||
self.assertEqual(response.data["task_id"], "task-procurements")
|
||
|
||
def test_refresh_forbidden_for_regular_user(self):
|
||
response = self.client.post(
|
||
reverse(
|
||
"api_v1:sources:source-cards-refresh",
|
||
kwargs={"slug": "financial-indicators"},
|
||
),
|
||
{},
|
||
format="json",
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|