from __future__ import annotations
from datetime import datetime, timedelta
from io import BytesIO
from types import SimpleNamespace
from unittest.mock import patch
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 (
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 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 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.tasks.parse_media_news.apply_async",
return_value=SimpleNamespace(id="media-task-1"),
):
response = self.client.post(
self.url,
{"file": SimpleUploadedFile("media.xlsx", b"queued")},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual(response.data["data"]["task_id"], "media-task-1")
self.assertEqual(response.data["data"]["task_ids"], ["media-task-1"])
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")