330 lines
13 KiB
Python
330 lines
13 KiB
Python
"""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):
|
|
first_log = ParserLoadLogFactory(
|
|
source="manufactures",
|
|
batch_id=101,
|
|
status="success",
|
|
error_message="ok",
|
|
)
|
|
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": "101"},
|
|
)
|
|
|
|
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")
|