avm/mostovik-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8cfdbe5b49b1bf819ae0a5a447b424f2743bd12d
mostovik-backend/tests/apps/parsers/test_views.py
Aleksandr Meshchriakov 0d75d94139
All checks were successful
CI/CD Pipeline / Start Dev Containers in Dokploy (push) Has been skipped
Details
CI/CD Pipeline / Start Dev Containers in Dokploy (pull_request) Has been skipped
Details
CI/CD Pipeline / Quality Gate (pull_request) Successful in 2m29s
Details
CI/CD Pipeline / Build and Push Images (pull_request) Has been skipped
Details
CI/CD Pipeline / Internal Notify (pull_request) Successful in 1s
Details
CI/CD Pipeline / Quality Gate (push) Successful in 2m58s
Details
CI/CD Pipeline / Build and Push Images (push) Has been skipped
Details
CI/CD Pipeline / Internal Notify (push) Successful in 1s
Details
test: stabilize parser log search assertion
2026-04-28 10:27:09 +02:00

331 lines
13 KiB
Python
Raw Blame History

"""Integration tests for parsers API views (no mocks)."""
from __future__ import annotations
import io
import os
import tempfile
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_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")
Reference in New Issue View Git Blame Copy Permalink