from __future__ import annotations
from datetime import datetime, timedelta
from io import BytesIO
from tempfile import TemporaryDirectory
from types import SimpleNamespace
from unittest.mock import patch
from apps.core.models import BackgroundJob, JobStatus
from apps.parsers.gosedo import (
GosedoNotModified,
GosedoParseResult,
GosedoRow,
GosedoValidationError,
download_gosedo,
parse_gosedo,
publish_gosedo_snapshot,
stable_gosedo_uid,
)
from apps.parsers.media_news import (
MEDIA_NEWS_MAX_BYTES,
import_media_news,
normalize_news_text,
stable_media_external_id,
)
from apps.parsers.models import ParserLoadLog, ParserSourceArtifact, ParserStagedRecord
from apps.parsers.source_artifacts import cleanup_parser_source_artifacts
from apps.parsers.tasks import parse_media_news
from django.core.files.base import ContentFile
from django.core.files.storage import default_storage
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase
from django.urls import reverse
from django.utils import timezone
from openpyxl import Workbook
from organizations.models import Organization, OrganizationSourceRecord
from organizations.source_ingestion import (
OrganizationSourceIngestionService,
SourceRecordInput,
)
from rest_framework import status
from rest_framework.test import APITestCase
from tests.apps.user.factories import UserFactory
def _gosedo_html() -> bytes:
rows = []
for row_class, external_id, label in (
("pt-on", "participant-on", "УЧАСТНИК МЭДО"),
("pt-off", "participant-off", "УЧАСТНИК МЭДО"),
("oper", "operator-non-rfc-id", "ОПЕРАТОР МЭДО"),
("org", "organizer-1", "ОРГАНИЗАТОР"),
):
rows.append(
f''
f"| {external_id} | Краткое имя | МЭДО-адрес |
"
''
f"- {label}
- Полное имя
"
"- Рег.номер
- ОГРН: 1027700132195
"
"- Статус участника
- АКТИВНЫЙ, не аттестован для ДСП
"
"- Новое поле
- Сохранить
"
" |
"
)
return (
"Версия: 633 от 10 августа 2026
"
'"
).encode("utf-8")
def _media_workbook(rows: list[list[object]]) -> BytesIO:
workbook = Workbook()
sheet = workbook.active
sheet.append(
[
"ОКПО",
"ИНН",
"Дата актуальности новости",
"Источник",
"URL",
"Текст",
"Оценка",
]
)
for row in rows:
sheet.append(row)
output = BytesIO()
workbook.save(output)
workbook.close()
output.seek(0)
return output
class _DownloadResponse:
def __init__(self, status_code=200, *, headers=None, body=b""):
self.status_code = status_code
self.headers = headers or {"Content-Type": "text/html; charset=utf-8"}
self.body = body
def iter_content(self, chunk_size: int):
del chunk_size
yield self.body
def close(self):
return None
class GosedoParserTest(TestCase):
def test_parses_all_record_types_unknown_fields_and_statuses(self):
parsed = parse_gosedo(BytesIO(_gosedo_html()))
self.assertEqual(parsed.version, "633")
self.assertEqual(parsed.published_at.isoformat(), "2026-08-10")
self.assertEqual(
[row.record_type for row in parsed.rows],
["participant", "participant", "operator", "organizer"],
)
self.assertEqual(parsed.rows[0].status, "active")
self.assertEqual(parsed.rows[1].status, "inactive")
self.assertEqual(
parsed.rows[0].normalized_data["attestation_status"],
"not_attested",
)
self.assertEqual(
parsed.rows[0].normalized_data["extra_fields"],
{"Новое поле": "Сохранить"},
)
self.assertEqual(parsed.unknown_fields, 4)
self.assertEqual(
stable_gosedo_uid("operator", "operator-non-rfc-id"),
stable_gosedo_uid("operator", "operator-non-rfc-id"),
)
self.assertNotEqual(
stable_gosedo_uid("operator", "same"),
stable_gosedo_uid("participant", "same"),
)
def test_missing_details_row_is_quarantinable(self):
html = (
'Версия: 1 от 1 января 2026
'
'| broken | Имя | |
'
'| next | Имя 2 | |
'
'- УЧАСТНИК МЭДО
'
"- Имя 2
- Рег.номер
- ОГРН: 1027700132195
"
"
|
"
).encode()
parsed = parse_gosedo(BytesIO(html))
self.assertEqual(parsed.rows[0].error, "invalid_row_pair")
self.assertEqual(parsed.rows[1].external_id, "next")
def test_conditional_get_uses_latest_headers_and_handles_304(self):
ParserSourceArtifact.objects.create(
source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
status=ParserSourceArtifact.Status.PUBLISHED,
etag='"etag-1"',
last_modified="Mon, 10 Aug 2026 10:00:00 GMT",
)
response = SimpleNamespace(status_code=304, headers={}, close=lambda: None)
session = SimpleNamespace()
session.get = lambda *args, **kwargs: (
setattr(session, "request_headers", kwargs["headers"]) or response
)
with self.assertRaises(GosedoNotModified):
download_gosedo(session=session)
self.assertEqual(session.request_headers["If-None-Match"], '"etag-1"')
self.assertIn("If-Modified-Since", session.request_headers)
def test_download_rejects_redirect_outside_allowlist(self):
response = _DownloadResponse(
302,
headers={"Location": "https://example.test/source.html"},
)
session = SimpleNamespace(get=lambda *args, **kwargs: response)
with self.assertRaisesRegex(GosedoValidationError, "unsafe_source_url"):
download_gosedo(session=session)
def test_download_rejects_mime_utf8_and_declared_size(self):
cases = (
(
_DownloadResponse(headers={"Content-Type": "application/json"}),
"invalid_content_type",
),
(_DownloadResponse(body=b"\xff"), "invalid_utf8"),
(
_DownloadResponse(
headers={
"Content-Type": "text/html",
"Content-Length": str(100 * 1024 * 1024 + 1),
}
),
"source_too_large",
),
)
for response, expected_error in cases:
with self.subTest(expected_error=expected_error):
session = SimpleNamespace(
get=lambda *args, _response=response, **kwargs: _response
)
with self.assertRaisesRegex(GosedoValidationError, expected_error):
download_gosedo(session=session)
def test_atomic_failure_keeps_previous_snapshot(self):
organization = Organization.objects.create(
name="АО Тест",
inn="7707083893",
ogrn="1027700132195",
okpo="12345678",
directory_imported_at=timezone.now(),
)
OrganizationSourceIngestionService.save_records(
source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
load_batch=1,
records=[
SourceRecordInput(
external_id="old",
record_type="participant",
title="Старый снимок",
organization_name=organization.name,
inn=organization.inn,
ogrn=organization.ogrn,
payload={"okpo": organization.okpo},
)
],
)
artifact = ParserSourceArtifact.objects.create(
source=ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY,
load_batch=2,
)
row = GosedoRow(
row_number=1,
external_id="new",
record_type="participant",
status="active",
normalized_data={
"registration_number": organization.ogrn,
"short_name": organization.name,
},
)
parsed = GosedoParseResult(
version="2",
published_at=timezone.localdate(),
rows=[row],
unknown_fields=0,
)
with patch.object(
OrganizationSourceIngestionService,
"save_records",
side_effect=RuntimeError("publish failed"),
), self.assertRaises(RuntimeError):
publish_gosedo_snapshot(artifact=artifact, parsed=parsed, load_batch=2)
self.assertTrue(
OrganizationSourceRecord.objects.filter(external_id="old").exists()
)
self.assertFalse(
OrganizationSourceRecord.objects.filter(external_id="new").exists()
)
class SourceArtifactRetentionTest(TestCase):
def test_cleanup_retains_latest_ten_versions_beyond_ninety_days(self):
created = []
base_time = timezone.now() - timedelta(days=120)
for index in range(12):
artifact = ParserSourceArtifact.objects.create(
source=ParserLoadLog.Source.MEDIA_NEWS,
version=str(index),
)
ParserSourceArtifact.objects.filter(pk=artifact.pk).update(
created_at=base_time + timedelta(hours=index)
)
created.append(artifact.uid)
deleted = cleanup_parser_source_artifacts()
self.assertEqual(deleted, 2)
remaining = set(ParserSourceArtifact.objects.values_list("uid", flat=True))
self.assertEqual(remaining, set(created[-10:]))
class MediaNewsImportTest(TestCase):
def setUp(self):
self.organization = Organization.objects.create(
name="АО СМИ",
inn="0012345678",
ogrn="1027700132195",
okpo="00123456",
directory_imported_at=timezone.now(),
)
def test_normalizes_text_and_stable_id(self):
full_text, excerpt = normalize_news_text(
"Источник_x000D_\r\n\r\nЗаголовок\nЛид\nСтрока 4\nСтрока 5"
)
first = stable_media_external_id(
inn="0012345678",
okpo="00123456",
published_at=timezone.localdate(),
news_source="СМИ",
url="",
full_text=full_text,
)
second = stable_media_external_id(
inn="0012345678",
okpo="00123456",
published_at=timezone.localdate(),
news_source="СМИ",
url="",
full_text=full_text,
)
self.assertEqual(excerpt, ["Источник", "Заголовок", "Лид", "Строка 4"])
self.assertEqual(first, second)
def test_import_upserts_and_quarantines_formula(self):
Organization.objects.create(
name="АО Другая",
inn="0099999999",
ogrn="1027700132196",
okpo="00999999",
directory_imported_at=timezone.now(),
)
workbook = _media_workbook(
[
[
"00123456",
"0012345678",
"2026-07-01",
"СМИ",
"",
"Источник\nЗаголовок\nЛид\nСтрока 4\nПолный текст",
"Положительная",
],
[
"00123456",
"=12345678",
"2026-07-02",
"СМИ",
"https://example.test/2",
"Источник\nЗаголовок\nЛид\nСтрока 4",
"Отрицательная",
],
[
"00999999",
"0012345678",
"2026-07-03",
"СМИ",
"https://example.test/3",
"Источник\nЗаголовок\nЛид\nСтрока 4",
"Отрицательная",
],
]
)
_, first = import_media_news(
handle=workbook,
original_name="media.xlsx",
load_batch=1,
uploaded_by_id=None,
)
workbook = _media_workbook(
[
[
"00123456",
"0012345678",
"2026-07-01",
"СМИ",
"",
"Источник\nЗаголовок\nЛид\nСтрока 4\nПолный текст",
"Положительная",
]
]
)
_, second = import_media_news(
handle=workbook,
original_name="media.xlsx",
load_batch=2,
uploaded_by_id=None,
)
self.assertEqual(first.published, 1)
self.assertEqual(first.quarantined, 2)
self.assertEqual(
first.reasons,
{"formula_not_allowed": 1, "inn_okpo_conflict": 1},
)
self.assertEqual(second.published, 1)
self.assertEqual(OrganizationSourceRecord.objects.count(), 1)
record = OrganizationSourceRecord.objects.get()
self.assertEqual(record.title, "Заголовок")
self.assertEqual(record.status, "positive")
self.assertEqual(
record.payload["excerpt_lines"],
["Источник", "Заголовок", "Лид", "Строка 4"],
)
self.assertEqual(
ParserStagedRecord.objects.filter(
disposition=ParserStagedRecord.Disposition.QUARANTINED
).count(),
2,
)
def test_import_accepts_native_date_typo_header_and_formatted_leading_zeroes(self):
workbook = Workbook()
sheet = workbook.active
sheet.append(
[
"ОКПО",
"ИНН",
"Дата акутальности новости",
"Источник",
"URL",
"Текст",
"Оценка",
]
)
sheet.append(
[
123456,
12345678,
datetime(2026, 7, 4),
"СМИ",
"https://example.test/native-date",
"Источник\nЗаголовок",
"Отрицательная",
]
)
sheet["A2"].number_format = "00000000"
sheet["B2"].number_format = "0000000000"
output = BytesIO()
workbook.save(output)
workbook.close()
output.seek(0)
_, result = import_media_news(
handle=output,
original_name="native.xlsx",
load_batch=3,
uploaded_by_id=None,
)
self.assertEqual(result.published, 1)
record = OrganizationSourceRecord.objects.get()
self.assertEqual(record.record_date, "2026-07-04")
self.assertEqual(record.payload["inn"], "0012345678")
self.assertEqual(record.payload["okpo"], "00123456")
self.assertEqual(record.status, "negative")
def test_same_article_for_two_organizations_stays_separate(self):
Organization.objects.create(
name="АО СМИ 2",
inn="0098765432",
ogrn="1027700132196",
okpo="00654321",
directory_imported_at=timezone.now(),
)
common = [
"2026-07-05",
"СМИ",
"https://example.test/shared",
"Источник\nЗаголовок\nТекст",
"Положительная",
]
workbook = _media_workbook(
[
["00123456", "0012345678", *common],
["00654321", "0098765432", *common],
]
)
_, result = import_media_news(
handle=workbook,
original_name="shared.xlsx",
load_batch=4,
uploaded_by_id=None,
)
self.assertEqual(result.published, 2)
self.assertEqual(OrganizationSourceRecord.objects.count(), 2)
self.assertEqual(
OrganizationSourceRecord.objects.values("external_id").distinct().count(),
2,
)
def test_invalid_date_and_sentiment_are_quarantined(self):
workbook = _media_workbook(
[
[
"00123456",
"0012345678",
"31.02.2026",
"СМИ",
"",
"Источник\nЗаголовок",
"Положительная",
],
[
"00123456",
"0012345678",
"2026-07-06",
"СМИ",
"",
"Источник\nЗаголовок",
"Нейтральная",
],
]
)
_, result = import_media_news(
handle=workbook,
original_name="invalid.xlsx",
load_batch=5,
uploaded_by_id=None,
)
self.assertEqual(result.published, 0)
self.assertEqual(
result.reasons,
{"invalid_date": 1, "invalid_sentiment": 1},
)
class MediaNewsTaskTest(TestCase):
def test_uploaded_workbook_task_completes_background_job(self):
Organization.objects.create(
name="АО Задача СМИ",
inn="0012345678",
ogrn="1027700132195",
okpo="00123456",
directory_imported_at=timezone.now(),
)
workbook = _media_workbook(
[
[
"00123456",
"0012345678",
"2026-07-01",
"СМИ",
"https://example.test/news",
"Источник\nЗаголовок\nЛид\nСтрока 4\nПолный текст",
"Положительная",
]
]
)
with TemporaryDirectory() as media_root, self.settings(MEDIA_ROOT=media_root):
file_path = default_storage.save(
"parser_uploads/media.xlsx",
ContentFile(workbook.getvalue()),
)
task_result = parse_media_news.apply(
kwargs={
"file_path": file_path,
"original_name": "media.xlsx",
},
task_id="media-news-task-test",
)
self.assertTrue(task_result.successful())
self.assertFalse(default_storage.exists(file_path))
job = BackgroundJob.objects.get(task_id="media-news-task-test")
self.assertEqual(job.status, JobStatus.SUCCESS)
self.assertEqual(job.progress, 100)
self.assertEqual(task_result.result["published"], 1)
self.assertEqual(
ParserLoadLog.objects.get(source=ParserLoadLog.Source.MEDIA_NEWS).status,
ParserLoadLog.Status.SUCCESS,
)
class MediaNewsPermissionsTest(APITestCase):
def setUp(self):
self.user = UserFactory.create_user()
self.admin = UserFactory.create_user(is_staff=True)
self.url = reverse("api_v1:parsers:upload-parser-data", args=["media_news"])
def test_upload_is_admin_only_and_returns_task_ids(self):
self.client.force_authenticate(self.user)
response = self.client.post(
self.url,
{"file": SimpleUploadedFile("media.xlsx", b"not-read-by-worker")},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.force_authenticate(self.admin)
with 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"queued")},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
task_id = response.data["data"]["task_id"]
self.assertEqual(response.data["data"]["task_ids"], [task_id])
card_response = self.client.get(
reverse(
"api_v1:sources:source-cards-detail",
kwargs={"slug": "media-mentions"},
)
)
self.assertEqual(card_response.status_code, status.HTTP_200_OK)
self.assertEqual(card_response.data["data"]["status"], "in_progress")
self.assertEqual(
card_response.data["data"]["active_tasks"][0]["task_id"],
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",
args=[ParserLoadLog.Source.GOSEDO_ADDRESS_DIRECTORY],
)
self.client.force_authenticate(self.user)
response = self.client.post(url, {}, format="json")
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.force_authenticate(self.admin)
with patch(
"apps.parsers.tasks.parse_gosedo_address_directory.apply_async",
return_value=SimpleNamespace(id="gosedo-task-1"),
):
response = self.client.post(url, {}, format="json")
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual(response.data["data"]["task_id"], "gosedo-task-1")
self.assertEqual(response.data["data"]["task_ids"], ["gosedo-task-1"])
def test_parser_log_exposes_artifact_quarantine_accounting(self):
organization = Organization.objects.create(name="АО Журнал")
log = ParserLoadLog.objects.create(
source=ParserLoadLog.Source.MEDIA_NEWS,
batch_id=44,
records_count=1,
status=ParserLoadLog.Status.SUCCESS,
)
artifact = ParserSourceArtifact.objects.create(
source=ParserLoadLog.Source.MEDIA_NEWS,
load_batch=44,
status=ParserSourceArtifact.Status.PUBLISHED,
parsed_count=2,
published_count=1,
quarantined_count=1,
rejection_reasons={"invalid_date": 1},
)
ParserStagedRecord.objects.create(
artifact=artifact,
row_number=1,
organization=organization,
disposition=ParserStagedRecord.Disposition.PUBLISHED,
)
ParserStagedRecord.objects.create(
artifact=artifact,
row_number=2,
disposition=ParserStagedRecord.Disposition.QUARANTINED,
reason="invalid_date",
)
self.client.force_authenticate(self.admin)
response = self.client.get(
reverse("api_v1:system:parser-logs-list"),
{"source": ParserLoadLog.Source.MEDIA_NEWS},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
row = next(item for item in response.data["results"] if item["id"] == log.id)
self.assertEqual(row["artifact_uid"], str(artifact.uid))
self.assertEqual(row["parsed_count"], 2)
self.assertEqual(row["published_count"], 1)
self.assertEqual(row["quarantined_count"], 1)
self.assertEqual(row["rejection_reasons"], {"invalid_date": 1})
self.assertEqual(row["organizations_count"], 1)
def test_media_list_omits_full_text_and_detail_includes_it(self):
organization = Organization.objects.create(
name="АО API СМИ",
inn="7707083801",
ogrn="1027700132101",
okpo="12345671",
opk_registry_membership=True,
directory_imported_at=timezone.now(),
)
OrganizationSourceIngestionService.save_records(
source=ParserLoadLog.Source.MEDIA_NEWS,
load_batch=1,
records=[
SourceRecordInput(
external_id="b" * 64,
record_type="media_mention",
title="Заголовок",
organization_name=organization.name,
inn=organization.inn,
ogrn=organization.ogrn,
record_date="2026-07-01",
status="positive",
payload={
"okpo": organization.okpo,
"news_source": "Тестовое СМИ",
"sentiment": "positive",
"excerpt_lines": ["1", "2", "3", "4"],
"full_text": "1\n2\n3\n4\n5",
},
)
],
)
record = OrganizationSourceRecord.objects.get()
self.client.force_authenticate(self.user)
list_response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{"source_group": "media_mentions", "status": "positive"},
)
detail_response = self.client.get(
reverse(
"api_v2:organizations:organization-source-records-detail",
args=[record.uid],
)
)
self.assertEqual(list_response.status_code, status.HTTP_200_OK)
self.assertNotIn("full_text", list_response.data["data"][0]["payload"])
self.assertEqual(
list_response.data["data"][0]["payload"]["excerpt_lines"],
["1", "2", "3", "4"],
)
self.assertEqual(detail_response.status_code, status.HTTP_200_OK)
self.assertEqual(detail_response.data["payload"]["full_text"], "1\n2\n3\n4\n5")