feat(forms): unify F2-F6 upload contracts and add shared serializers
This commit is contained in:
211
tests/apps/forms/test_upload_contracts_api.py
Normal file
211
tests/apps/forms/test_upload_contracts_api.py
Normal file
@@ -0,0 +1,211 @@
|
||||
"""Contract tests for F-2…F-6 upload endpoints."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.test import override_settings
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APITestCase
|
||||
|
||||
from tests.apps.user.factories import UserFactory
|
||||
|
||||
|
||||
@override_settings(ROOT_URLCONF="core.urls")
|
||||
class FormUploadContractsApiTest(APITestCase):
|
||||
"""Contract tests for multipart upload endpoints Ф-2 ... Ф-6."""
|
||||
|
||||
BACKGROUND_THRESHOLD_PLUS = 1024 * 1024 + 1
|
||||
RESULT_PAYLOAD = {
|
||||
"batch_id": 101,
|
||||
"loaded_count": 2,
|
||||
"skipped_count": 1,
|
||||
"errors": [],
|
||||
}
|
||||
CASES = {
|
||||
"f2": {
|
||||
"url": "/api/v1/forms/f2/upload/",
|
||||
"form": "f2",
|
||||
"payload": {"report_year": 2026, "report_quarter": 1},
|
||||
"period_field": "report_quarter",
|
||||
"period_value": 1,
|
||||
"report_period_display": "I квартал 2026",
|
||||
"parse_target": "apps.form_2.api.parse_form_f2_file",
|
||||
"task_target": "apps.form_2.api.process_form_f2_file",
|
||||
},
|
||||
"f3": {
|
||||
"url": "/api/v1/forms/f3/upload/",
|
||||
"form": "f3",
|
||||
"payload": {"report_year": 2026},
|
||||
"period_field": None,
|
||||
"period_value": None,
|
||||
"report_period_display": "2026 год",
|
||||
"parse_target": "apps.form_3.api.parse_form_f3_file",
|
||||
"task_target": "apps.form_3.api.process_form_f3_file",
|
||||
},
|
||||
"f4": {
|
||||
"url": "/api/v1/forms/f4/upload/",
|
||||
"form": "f4",
|
||||
"payload": {"report_year": 2026, "report_half_year": 1},
|
||||
"period_field": "report_half_year",
|
||||
"period_value": 1,
|
||||
"report_period_display": "I полугодие 2026",
|
||||
"parse_target": "apps.form_4.api.parse_form_f4_file",
|
||||
"task_target": "apps.form_4.api.process_form_f4_file",
|
||||
},
|
||||
"f5": {
|
||||
"url": "/api/v1/forms/f5/upload/",
|
||||
"form": "f5",
|
||||
"payload": {"report_year": 2026},
|
||||
"period_field": None,
|
||||
"period_value": None,
|
||||
"report_period_display": "2026 год",
|
||||
"parse_target": "apps.form_5.api.parse_form_f5_file",
|
||||
"task_target": "apps.form_5.api.process_form_f5_file",
|
||||
},
|
||||
"f6": {
|
||||
"url": "/api/v1/forms/f6/upload/",
|
||||
"form": "f6",
|
||||
"payload": {"report_year": 2026},
|
||||
"period_field": None,
|
||||
"period_value": None,
|
||||
"report_period_display": "2026 год",
|
||||
"parse_target": "apps.form_6.api.parse_form_f6_file",
|
||||
"task_target": "apps.form_6.api.process_form_f6_file",
|
||||
},
|
||||
}
|
||||
|
||||
def setUp(self):
|
||||
self.user = UserFactory.create_user()
|
||||
self.client.force_authenticate(self.user)
|
||||
|
||||
def _build_payload(self, base_payload: dict[str, int], file_size: int) -> dict:
|
||||
payload = dict(base_payload)
|
||||
payload["file"] = SimpleUploadedFile(
|
||||
name="report.xlsx",
|
||||
content=b"0" * file_size,
|
||||
content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
)
|
||||
return payload
|
||||
|
||||
def _assert_base_success_contract(
|
||||
self,
|
||||
response_data: dict,
|
||||
case: dict,
|
||||
status_code: str,
|
||||
):
|
||||
self.assertEqual(response_data["form"], case["form"])
|
||||
self.assertEqual(response_data["report_year"], 2026)
|
||||
self.assertEqual(response_data["status"], status_code)
|
||||
self.assertEqual(
|
||||
response_data["report_period_display"],
|
||||
case["report_period_display"],
|
||||
)
|
||||
datetime.fromisoformat(response_data["created_at"])
|
||||
self.assertIn("upload_id", response_data)
|
||||
if case["period_field"]:
|
||||
self.assertEqual(response_data[case["period_field"]], case["period_value"])
|
||||
else:
|
||||
self.assertNotIn("report_quarter", response_data)
|
||||
self.assertNotIn("report_half_year", response_data)
|
||||
|
||||
def test_upload_validation_error_contract(self):
|
||||
for _, case in self.CASES.items():
|
||||
with self.subTest(form=case["form"]):
|
||||
response = self.client.post(
|
||||
case["url"],
|
||||
case["payload"],
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertEqual(response.data["error_code"], "validation_error")
|
||||
self.assertEqual(response.data["error_message"], "Validation failed")
|
||||
self.assertIsInstance(response.data["details"], list)
|
||||
self.assertTrue(response.data["details"])
|
||||
self.assertTrue(
|
||||
any(item["field"] == "file" for item in response.data["details"])
|
||||
)
|
||||
|
||||
def test_upload_async_contract(self):
|
||||
for _, case in self.CASES.items():
|
||||
with self.subTest(form=case["form"]):
|
||||
task_mock = SimpleNamespace(
|
||||
delay=Mock(return_value=SimpleNamespace(id="background-task-id")),
|
||||
)
|
||||
with (
|
||||
patch(case["parse_target"]) as parse_mock,
|
||||
patch(case["task_target"], task_mock),
|
||||
):
|
||||
response = self.client.post(
|
||||
case["url"],
|
||||
self._build_payload(
|
||||
case["payload"],
|
||||
self.BACKGROUND_THRESHOLD_PLUS,
|
||||
),
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
||||
self.assertEqual(response.data["status"], "queued")
|
||||
self.assertIn("job_id", response.data)
|
||||
self.assertEqual(response.data["job_id"], "background-task-id")
|
||||
self.assertNotIn("result", response.data)
|
||||
self._assert_base_success_contract(
|
||||
response.data,
|
||||
case,
|
||||
status_code="queued",
|
||||
)
|
||||
parse_mock.assert_not_called()
|
||||
task_mock.delay.assert_called_once()
|
||||
|
||||
def test_upload_sync_contract(self):
|
||||
for _, case in self.CASES.items():
|
||||
with self.subTest(form=case["form"]):
|
||||
task_mock = SimpleNamespace(
|
||||
delay=Mock(return_value=SimpleNamespace(id="should-not-use")),
|
||||
)
|
||||
with patch(
|
||||
case["parse_target"],
|
||||
return_value=self.RESULT_PAYLOAD,
|
||||
) as parse_mock, patch(case["task_target"], task_mock):
|
||||
response = self.client.post(
|
||||
case["url"],
|
||||
self._build_payload(case["payload"], file_size=256),
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["status"], "done")
|
||||
self.assertIn("result", response.data)
|
||||
self.assertNotIn("job_id", response.data)
|
||||
self.assertEqual(response.data["result"], self.RESULT_PAYLOAD)
|
||||
self._assert_base_success_contract(
|
||||
response.data,
|
||||
case,
|
||||
status_code="done",
|
||||
)
|
||||
parse_mock.assert_called_once()
|
||||
task_mock.delay.assert_not_called()
|
||||
|
||||
def test_upload_processing_error_contract(self):
|
||||
for _, case in self.CASES.items():
|
||||
with self.subTest(form=case["form"]):
|
||||
with patch(
|
||||
case["parse_target"],
|
||||
side_effect=RuntimeError("parse failed"),
|
||||
) as parse_mock:
|
||||
response = self.client.post(
|
||||
case["url"],
|
||||
self._build_payload(case["payload"], file_size=256),
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertEqual(response.data["error_code"], "processing_error")
|
||||
self.assertEqual(response.data["error_message"], "parse failed")
|
||||
self.assertEqual(response.data["details"], [])
|
||||
parse_mock.assert_called_once()
|
||||
Reference in New Issue
Block a user