avm/mostovik-backend
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: align media upload and gosedo records
All checks were successful
CI/CD Pipeline / Quality Gate (push) Successful in 1m13s
Details
CI/CD Pipeline / Build and Push Images (push) Successful in 20s
Details
CI/CD Pipeline / Internal Notify (push) Successful in 1s
Details
CI/CD Pipeline / Deploy Dev via Compose (push) Successful in 30s
Details

Browse Source
This commit is contained in:
Aleksandr Meshchriakov
2026-08-20 11:43:00 +02:00
parent d95a002d84
commit f2e36a8991
5 changed files with 207 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
tests/apps/organizations/test_api_v2_source_extensions.py
View File

@@ -389,6 +389,113 @@ class OrganizationSourceExtensionsApiV2Test(APITestCase):
expected,
)
def test_flat_gosedo_records_include_all_linked_types_and_record_sorting(self):
organization = create_frontend_organization(
name="Связанная организация",
inn="7707083828",
ogrn="1027700132028",
)
extension = ElectronicDocumentExchangeExtension.objects.create(
organization=organization,
title="Электронный документооборот",
)
for (
record_type,
external_id,
title,
full_name,
medo_address,
registration_number,
) in (
(
"participant",
"GOSEDO-PARTICIPANT",
"Участник",
"Полное имя участника",
"MEDO-3",
"3",
),
(
"operator",
"GOSEDO-OPERATOR",
"Оператор",
"Полное имя оператора",
"MEDO-1",
"1",
),
(
"organizer",
"GOSEDO-ORGANIZER",
"Организатор",
"Полное имя организатора",
"MEDO-2",
"2",
),
):
OrganizationSourceRecord.objects.create(
extension=extension,
record_type=record_type,
source="gosedo_address_directory",
external_id=external_id,
title=title,
payload={
"full_name": full_name,
"medo_address": medo_address,
"registration_number": registration_number,
},
)
endpoint = reverse("api_v2:organizations:organization-source-records-list")
params = {
"source_group": "electronic_document_exchange",
"source": "gosedo_address_directory",
"organization": str(organization.uid),
}
response = self.client.get(endpoint, params)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
{item["record_type"] for item in response.data["data"]},
{"participant", "operator", "organizer"},
)
expected_by_ordering = {
"record_type": [
"GOSEDO-OPERATOR",
"GOSEDO-ORGANIZER",
"GOSEDO-PARTICIPANT",
],
"external_id": [
"GOSEDO-OPERATOR",
"GOSEDO-ORGANIZER",
"GOSEDO-PARTICIPANT",
],
"payload__full_name": [
"GOSEDO-OPERATOR",
"GOSEDO-ORGANIZER",
"GOSEDO-PARTICIPANT",
],
"payload__medo_address": [
"GOSEDO-OPERATOR",
"GOSEDO-ORGANIZER",
"GOSEDO-PARTICIPANT",
],
"payload__registration_number": [
"GOSEDO-OPERATOR",
"GOSEDO-ORGANIZER",
"GOSEDO-PARTICIPANT",
],
}
for ordering, expected in expected_by_ordering.items():
with self.subTest(ordering=ordering):
response = self.client.get(endpoint, {**params, "ordering": ordering})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
[item["external_id"] for item in response.data["data"]],
expected,
)
def test_flat_media_records_support_frontend_sorting_fields(self):
organization = create_frontend_organization(
name="Media sorting organization",

53
tests/apps/parsers/test_gosedo_media_news.py
View File

@@ -18,6 +18,7 @@ from apps.parsers.gosedo import (
stable_gosedo_uid,
)
from apps.parsers.media_news import (
MEDIA_NEWS_MAX_BYTES,
import_media_news,
normalize_news_text,
stable_media_external_id,
@@ -619,6 +620,58 @@ class MediaNewsPermissionsTest(APITestCase):
task_id,
)
def test_upload_rejects_non_xlsx_before_saving_or_queueing(self):
self.client.force_authenticate(self.admin)
with patch("apps.parsers.views._save_uploaded_parser_file") as save_file, patch(
"apps.parsers.tasks.parse_media_news.apply_async"
) as apply_async:
response = self.client.post(
self.url,
{"file": SimpleUploadedFile("media.csv", b"headline")},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["errors"][0]["code"], "invalid_file_type")
save_file.assert_not_called()
apply_async.assert_not_called()
def test_upload_rejects_oversized_xlsx_before_saving_or_queueing(self):
self.client.force_authenticate(self.admin)
with patch("apps.parsers.views.MEDIA_NEWS_MAX_BYTES", 1), patch(
"apps.parsers.views._save_uploaded_parser_file"
) as save_file, patch(
"apps.parsers.tasks.parse_media_news.apply_async"
) as apply_async:
response = self.client.post(
self.url,
{"file": SimpleUploadedFile("media.xlsx", b"xx")},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["errors"][0]["code"], "file_too_large")
save_file.assert_not_called()
apply_async.assert_not_called()
def test_upload_accepts_xlsx_at_size_limit(self):
self.client.force_authenticate(self.admin)
with patch("apps.parsers.views.MEDIA_NEWS_MAX_BYTES", 1), patch(
"apps.parsers.views._save_uploaded_parser_file",
return_value="parser_uploads/media.xlsx",
), patch(
"apps.parsers.tasks.parse_media_news.apply_async",
side_effect=lambda **kwargs: SimpleNamespace(id=kwargs["task_id"]),
):
response = self.client.post(
self.url,
{"file": SimpleUploadedFile("media.xlsx", b"x")},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual(MEDIA_NEWS_MAX_BYTES, 25 * 1024 * 1024)
def test_gosedo_manual_run_is_admin_only_and_returns_task_ids(self):
url = reverse(
"api_v1:parsers:run-parser",