Aleksandr Meshchriakov
c72343a375
All checks were successful
CI/CD Pipeline / Manual Action Help (push) Has been skipped
CI/CD Pipeline / Start Dev Containers in Dokploy (push) Has been skipped
CI/CD Pipeline / Drop and Recreate Dev Database (push) Has been skipped
CI/CD Pipeline / Quality Gate (push) Successful in 1m53s
CI/CD Pipeline / Build and Push Images (push) Successful in 2m42s
CI/CD Pipeline / Internal Notify (push) Successful in 1s
CI/CD Pipeline / Deploy Dev in Dokploy (push) Successful in 1s
370 lines
14 KiB
Python
370 lines
14 KiB
Python
"""Integration tests for parsers API views (no mocks)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import Mock, patch
|
|
|
|
from apps.parsers.models import FinancialReport, FinancialReportLine, ProcurementRecord
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.urls import reverse
|
|
from openpyxl import Workbook
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
|
|
from tests.apps.parsers.factories import (
|
|
IndustrialCertificateRecordFactory,
|
|
IndustrialProductRecordFactory,
|
|
InspectionRecordFactory,
|
|
ManufacturerRecordFactory,
|
|
ParserLoadLogFactory,
|
|
)
|
|
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 _build_fns_excel_bytes() -> bytes:
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
year = fake.random_int(min=2020, max=2025)
|
|
ws.append(["Form", None, year, None])
|
|
ws.append([None, "Code", "Start", "End"])
|
|
ws.append(
|
|
[fake.word(), _digits(4), fake.random_int(10, 999), fake.random_int(10, 999)]
|
|
)
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
wb.close()
|
|
return buf.getvalue()
|
|
|
|
|
|
def _create_procurement_record() -> ProcurementRecord:
|
|
return ProcurementRecord.objects.create(
|
|
load_batch=fake.random_int(min=1, max=1000),
|
|
purchase_number=_digits(19),
|
|
purchase_name=fake.sentence(nb_words=6),
|
|
customer_inn=_digits(10),
|
|
customer_kpp=_digits(9),
|
|
customer_ogrn=_digits(13),
|
|
customer_name=fake.company(),
|
|
max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)),
|
|
status=fake.word(),
|
|
law_type="44-FZ",
|
|
href=fake.url(),
|
|
region_code=f"{fake.random_int(min=1, max=99):02d}",
|
|
)
|
|
|
|
|
|
class ParsersViewSetTest(APITestCase):
|
|
def setUp(self):
|
|
self.user = UserFactory.create_user()
|
|
self.admin = UserFactory.create_superuser()
|
|
|
|
def test_certificates_list_and_retrieve(self):
|
|
record = IndustrialCertificateRecordFactory()
|
|
self.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:minpromtorg:certificates-list")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail = self.client.get(
|
|
reverse("api_v1:minpromtorg:certificates-detail", args=[record.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
|
|
def test_manufacturers_list_and_retrieve(self):
|
|
record = ManufacturerRecordFactory()
|
|
self.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:minpromtorg:manufacturers-list")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail = self.client.get(
|
|
reverse("api_v1:minpromtorg:manufacturers-detail", args=[record.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
|
|
def test_products_list_and_retrieve(self):
|
|
record = IndustrialProductRecordFactory()
|
|
self.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:minpromtorg:industrial-products-list")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail = self.client.get(
|
|
reverse("api_v1:minpromtorg:industrial-products-detail", args=[record.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
|
|
def test_inspections_list_and_retrieve(self):
|
|
record = InspectionRecordFactory()
|
|
self.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:proverki:inspections-list")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail = self.client.get(
|
|
reverse("api_v1:proverki:inspections-detail", args=[record.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
|
|
def test_procurements_list_and_retrieve(self):
|
|
record = _create_procurement_record()
|
|
self.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:zakupki:procurements-list")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail = self.client.get(
|
|
reverse("api_v1:zakupki:procurements-detail", args=[record.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
|
|
def test_dashboard_data_exposes_source_groups_for_page(self):
|
|
self.client.force_authenticate(self.user)
|
|
|
|
response = self.client.get("/api/v1/parsers/dashboard/")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
payload = response.data["data"]
|
|
self.assertIn("groups", payload)
|
|
self.assertIn("api", payload["groups"])
|
|
self.assertIn("uploads", payload["groups"])
|
|
self.assertEqual(payload["api_sources"], payload["groups"]["api"])
|
|
self.assertEqual(payload["file_sources"], payload["groups"]["uploads"])
|
|
|
|
def test_financial_reports_list_and_retrieve(self):
|
|
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.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:fns:fns-reports-list")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["data"][0]["lines_count"], 1)
|
|
detail = self.client.get(
|
|
reverse("api_v1:fns:fns-reports-detail", args=[report.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
self.assertIn("lines", detail.data)
|
|
|
|
def test_system_logs_admin_only(self):
|
|
log = ParserLoadLogFactory()
|
|
url_logs = reverse("api_v1:system:parser-logs-list")
|
|
|
|
response = self.client.get(url_logs)
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
self.client.force_authenticate(self.user)
|
|
response = self.client.get(url_logs)
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
self.client.force_authenticate(self.admin)
|
|
response = self.client.get(url_logs)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
detail = self.client.get(
|
|
reverse("api_v1:system:parser-logs-detail", args=[log.id])
|
|
)
|
|
self.assertEqual(detail.status_code, status.HTTP_200_OK)
|
|
|
|
def test_system_logs_support_search_and_organizations_count(self):
|
|
search_marker = "manufactures-unique-search-marker"
|
|
first_log = ParserLoadLogFactory(
|
|
source="manufactures",
|
|
batch_id=101,
|
|
status="success",
|
|
error_message=search_marker,
|
|
)
|
|
ParserLoadLogFactory(
|
|
source="inspections",
|
|
batch_id=202,
|
|
status="failed",
|
|
error_message="timeout",
|
|
)
|
|
ManufacturerRecordFactory(load_batch=101, inn="7701000001")
|
|
ManufacturerRecordFactory(load_batch=101, inn="7701000002")
|
|
|
|
self.client.force_authenticate(self.admin)
|
|
response = self.client.get(
|
|
reverse("api_v1:system:parser-logs-list"),
|
|
{"search": search_marker},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
rows = response.data["results"]
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0]["id"], first_log.id)
|
|
self.assertEqual(rows[0]["organizations_count"], 2)
|
|
|
|
def test_system_logs_export_returns_csv(self):
|
|
ParserLoadLogFactory(
|
|
source="manufactures",
|
|
batch_id=333,
|
|
status="success",
|
|
records_count=4,
|
|
)
|
|
ManufacturerRecordFactory(load_batch=333, inn="7701000001")
|
|
|
|
self.client.force_authenticate(self.admin)
|
|
response = self.client.get(
|
|
reverse("api_v1:system:parser-logs-export"),
|
|
{"search": "333"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response["Content-Type"], "text/csv; charset=utf-8")
|
|
content = response.content.decode("utf-8")
|
|
self.assertIn("organizations_count", content)
|
|
self.assertIn("333", content)
|
|
|
|
def test_system_logs_support_search_by_source_label(self):
|
|
ParserLoadLogFactory(
|
|
source="fns_reports",
|
|
batch_id=909,
|
|
status="success",
|
|
)
|
|
|
|
self.client.force_authenticate(self.admin)
|
|
response = self.client.get(
|
|
reverse("api_v1:system:parser-logs-list"),
|
|
{"search": "финансово"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(len(response.data["results"]), 1)
|
|
self.assertEqual(response.data["results"][0]["source"], "financial-indicators")
|
|
|
|
def test_fns_upload_invalid_filename(self):
|
|
self.client.force_authenticate(self.admin)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
watch_dir = os.path.join(tmpdir, "watch")
|
|
processed_dir = os.path.join(tmpdir, "processed")
|
|
failed_dir = os.path.join(tmpdir, "failed")
|
|
content = _build_fns_excel_bytes()
|
|
upload = SimpleUploadedFile(
|
|
f"bad_{fake.random_int()}.xlsx",
|
|
content,
|
|
content_type=(
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
),
|
|
)
|
|
with self.settings(
|
|
FNS_WATCH_DIRECTORY=watch_dir,
|
|
FNS_PROCESSED_DIRECTORY=processed_dir,
|
|
FNS_FAILED_DIRECTORY=failed_dir,
|
|
):
|
|
url = reverse("api_v1:fns:fns-upload")
|
|
response = self.client.post(
|
|
url, {"files": [upload]}, format="multipart"
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_fns_upload_forbidden_for_regular_user(self):
|
|
self.client.force_authenticate(self.user)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
watch_dir = os.path.join(tmpdir, "watch")
|
|
processed_dir = os.path.join(tmpdir, "processed")
|
|
failed_dir = os.path.join(tmpdir, "failed")
|
|
content = _build_fns_excel_bytes()
|
|
upload = SimpleUploadedFile(
|
|
f"fin_{_digits(5)}_{_digits(13)}.xlsx",
|
|
content,
|
|
content_type=(
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
),
|
|
)
|
|
with self.settings(
|
|
FNS_WATCH_DIRECTORY=watch_dir,
|
|
FNS_PROCESSED_DIRECTORY=processed_dir,
|
|
FNS_FAILED_DIRECTORY=failed_dir,
|
|
):
|
|
url = reverse("api_v1:fns:fns-upload")
|
|
response = self.client.post(
|
|
url, {"files": [upload]}, format="multipart"
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
def test_fns_upload_accepts_single_file_payload(self):
|
|
self.client.force_authenticate(self.admin)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
watch_dir = os.path.join(tmpdir, "watch")
|
|
processed_dir = os.path.join(tmpdir, "processed")
|
|
failed_dir = os.path.join(tmpdir, "failed")
|
|
content = _build_fns_excel_bytes()
|
|
upload = SimpleUploadedFile(
|
|
f"fin_{_digits(5)}_{_digits(13)}.xlsx",
|
|
content,
|
|
content_type=(
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
),
|
|
)
|
|
with self.settings(
|
|
FNS_WATCH_DIRECTORY=watch_dir,
|
|
FNS_PROCESSED_DIRECTORY=processed_dir,
|
|
FNS_FAILED_DIRECTORY=failed_dir,
|
|
):
|
|
url = reverse("api_v1:fns:fns-upload")
|
|
response = self.client.post(url, {"file": upload}, format="multipart")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["success"], True)
|
|
self.assertIn("message", response.data)
|
|
|
|
def test_parsing_settings_get_and_patch(self):
|
|
self.client.force_authenticate(self.admin)
|
|
url = reverse("api_v1:parsing:parsing-settings")
|
|
|
|
initial = self.client.get(url)
|
|
self.assertEqual(initial.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(initial.data["planned_inspections"], "monthly")
|
|
|
|
updated = self.client.patch(
|
|
url,
|
|
{"planned_inspections": "weekly"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(updated.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(updated.data["planned_inspections"], "weekly")
|
|
|
|
def test_run_sync_inspections_accepts_limited_sync_params(self):
|
|
self.client.force_authenticate(self.user)
|
|
url = reverse("api_v1:parsers:run-parser", args=["sync_inspections"])
|
|
payload = {
|
|
"max_months_per_law": 1,
|
|
"start_year": 2026,
|
|
"start_month": 4,
|
|
"include_fz294": True,
|
|
"include_fz248": False,
|
|
"current_year": 2026,
|
|
"current_month": 4,
|
|
}
|
|
|
|
with patch(
|
|
"apps.parsers.views.tasks.sync_inspections.apply_async",
|
|
return_value=Mock(id="task-123"),
|
|
) as apply_async_mock:
|
|
response = self.client.post(url, payload, format="json")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
|
task_kwargs = apply_async_mock.call_args.kwargs["kwargs"]
|
|
for key, value in payload.items():
|
|
self.assertEqual(task_kwargs[key], value)
|
|
self.assertEqual(task_kwargs["requested_by_id"], self.user.id)
|