"""Regression coverage for official registry ingestion without upstream IO."""

from __future__ import annotations

import zipfile
from collections import Counter
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.budget_registry import BUDGET_SOURCE, refresh_budget_registry
from apps.parsers.models import ParserLoadLog, ParserSourceArtifact, ParserStagedRecord
from apps.parsers.registry_snapshots import SnapshotValidationError, publish_snapshot
from apps.parsers.services import ParserLoadLogService
from apps.parsers.sme_support import (
    SME_SOURCE,
    discover_sme_archive,
    import_sme_support,
    iter_sme_measures,
)
from apps.parsers.tasks_registry_snapshots import _run_snapshot
from django.test import TestCase, override_settings
from django.utils import timezone
from lxml import etree
from organizations.models import Organization, OrganizationSourceRecord
from organizations.serializers import (
    OrganizationSourceRecordListSerializer,
    OrganizationSourceRecordSerializer,
)
from organizations.source_record_export import _source_group_queryset
from rest_framework.test import APIClient

from tests.apps.user.factories import UserFactory


def _xml(
    *,
    numbers=("old-support", "new-support"),
    inn="1234567890",
    documents=1,
    units=("1", "3"),
):
    root = etree.Element(
        "Файл",
        ИдФайл="fixture",
        ВерсФорм="4.04",
        ТипИнф="РЕЕСТРМСП-ПП",
        КолДок=str(documents),
    )
    doc = etree.SubElement(
        root, "Документ", ИдДок="recipient-doc", ДатаСост="15.08.2026"
    )
    etree.SubElement(
        doc, "СвЮЛ", НаимОрг="АО Проверка", ИННЮЛ=inn, ОГРН="1027700132195"
    )
    for number in numbers:
        measure = etree.SubElement(
            doc,
            "СвПредПод",
            НомерПод=number,
            ДатаСвед="04.10.2024",
            ДатаОбнов="01.07.2026",
            ВидПП="1",
            НаимОрг="Поставщик",
            ИННЮЛ="9876543210",
            КатСуб="1",
            СрокПод="01.01.2027",
            ДатаПрин="01.10.2024",
            ИнфНаруш="1",
        )
        etree.SubElement(measure, "ФормПод", КодФорм="0001", НаимФорм="Форма")
        etree.SubElement(measure, "ВидПод", КодВид="0002", НаимВид="Вид")
        for unit in units:
            etree.SubElement(measure, "РазмПод", ЕдПод=unit, РазмПод="2.50")
        etree.SubElement(measure, "Нарушения", ВидНаруш="1", ДатаНаруш="01.02.2025")
        etree.SubElement(measure, "РегДок", НомерРД="A", НаимРД="Первый документ")
        etree.SubElement(measure, "РегДок", НомерРД="B", НаимРД="Второй документ")
    return etree.tostring(root, encoding="UTF-8", xml_declaration=True)


def _archive(xml: bytes) -> BytesIO:
    handle = BytesIO()
    with zipfile.ZipFile(handle, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("support.xml", xml)
    handle.seek(0)
    return handle


def _budget_record(identifier="1", *, inn="1234567890", status="2"):
    return {
        "id": identifier,
        "info": {
            "inn": inn,
            "ogrn": "1027700132195",
            "fullName": "АО Проверка",
            "okpoCode": "12345678",
            "statusCode": status,
            "dateUpdate": "2024-10-01T15:00:00",
            "code": "own-code",
            "recordNum": "00000001",
        },
        "activities": [{"name": "activity"}],
        "unknown_new_block": [{"raw": "preserved"}],
    }


class RegistrySnapshotTest(TestCase):
    def setUp(self):
        self.storage = TemporaryDirectory()
        self.addCleanup(self.storage.cleanup)
        self.settings = override_settings(MEDIA_ROOT=self.storage.name)
        self.settings.enable()
        self.addCleanup(self.settings.disable)
        self.organization = Organization.objects.create(
            name="АО Проверка",
            inn="1234567890",
            ogrn="1027700132195",
            okpo="12345678",
            directory_imported_at=timezone.now(),
            opk_registry_membership=True,
        )

    def import_sme(self, xml=None, batch=1):
        return import_sme_support(
            handle=_archive(xml or _xml()),
            original_name="snapshot.zip",
            snapshot_date="2026-08-15",
            load_batch=batch,
        )

    def test_sme_multiple_measures_nested_data_and_stable_ids(self):
        artifact, result = self.import_sme()
        self.assertEqual((result.parsed, result.published), (2, 2))
        records = list(OrganizationSourceRecord.objects.order_by("external_id"))
        self.assertEqual(records[0].record_date, "2024-10-01")
        self.assertEqual(str(records[0].amount), "2.50")
        self.assertEqual(len(records[0].payload["support_sizes"]), 2)
        self.assertEqual(len(records[0].payload["regulatory_documents"]), 2)
        self.assertEqual(records[0].payload["source_updated_date"], "2026-07-01")
        self.assertEqual(len(records[0].payload["violations"]), 1)
        before = {record.external_id: record.uid for record in records}
        self.import_sme(batch=2)
        self.assertEqual(
            dict(OrganizationSourceRecord.objects.values_list("external_id", "uid")),
            before,
        )
        self.assertEqual(Organization.objects.count(), 1)
        self.assertEqual(artifact.status, ParserSourceArtifact.Status.PUBLISHED)

    def test_sme_non_rub_amount_is_null_and_foreign_subject_not_created(self):
        self.import_sme(_xml(units=("3", "4")))
        self.assertIsNone(OrganizationSourceRecord.objects.first().amount)
        _, result = self.import_sme(_xml(inn="5678901234"), batch=2)
        self.assertEqual(result.published, 0)
        self.assertEqual(result.skipped, 2)
        self.assertEqual(Organization.objects.count(), 1)
        self.assertEqual(OrganizationSourceRecord.objects.count(), 0)

    def test_sme_invalid_truncated_or_conflicting_snapshot_keeps_old_records(self):
        self.import_sme()
        before = set(OrganizationSourceRecord.objects.values_list("uid", flat=True))
        for xml in (_xml(numbers=("same", "same")), _xml()[:-15]):
            with self.assertRaises((SnapshotValidationError, etree.XMLSyntaxError)):
                self.import_sme(xml, batch=2)
            self.assertEqual(
                set(OrganizationSourceRecord.objects.values_list("uid", flat=True)),
                before,
            )

    def test_official_document_count_mismatch_is_retained_as_diagnostic(self):
        root = etree.fromstring(_xml())  # noqa: S320 - locally generated fixture
        other = etree.fromstring(_xml(numbers=("third-measure",)))  # noqa: S320
        root.append(other.find("Документ"))
        artifact, result = self.import_sme(etree.tostring(root, encoding="UTF-8"))
        self.assertEqual(result.published, 3)
        self.assertEqual(artifact.metadata["documents_count"], 2)
        self.assertEqual(artifact.metadata["declared_documents_count"], 1)
        self.assertEqual(artifact.metadata["document_count_mismatch_files"], 1)

    def test_sme_publish_failure_rolls_back_updates_and_deletes(self):
        self.import_sme()
        before = set(OrganizationSourceRecord.objects.values_list("uid", flat=True))
        with self.assertRaises(RuntimeError), patch(
            "apps.parsers.registry_snapshots.OrganizationSourceIngestionService.save_records",
            side_effect=RuntimeError("stop"),
        ):
            self.import_sme(_xml(numbers=("replacement",)), batch=2)
        self.assertEqual(
            set(OrganizationSourceRecord.objects.values_list("uid", flat=True)), before
        )

    def test_repeated_publication_keeps_previous_records_and_counts(self):
        artifact, original = self.import_sme()
        before = set(OrganizationSourceRecord.objects.values_list("uid", flat=True))
        self.assertEqual(publish_snapshot(artifact), original)
        self.assertEqual(
            set(OrganizationSourceRecord.objects.values_list("uid", flat=True)), before
        )
        artifact.refresh_from_db()
        self.assertEqual(artifact.published_count, 2)

    def test_cache_failure_rolls_back_publication(self):
        self.import_sme()
        before = set(OrganizationSourceRecord.objects.values_list("uid", flat=True))
        with self.assertRaises(RuntimeError), patch(
            "apps.parsers.registry_snapshots.invalidate_source_data_cache",
            side_effect=RuntimeError("cache"),
        ):
            self.import_sme(_xml(numbers=("replacement",)), batch=2)
        self.assertEqual(
            set(OrganizationSourceRecord.objects.values_list("uid", flat=True)), before
        )

    def test_cache_is_invalidated_again_only_after_commit(self):
        with patch(
            "apps.parsers.registry_snapshots.invalidate_source_data_cache"
        ) as invalidate, self.captureOnCommitCallbacks(execute=True) as callbacks:
            artifact, _ = self.import_sme()
            self.assertEqual(invalidate.call_count, 1)
            artifact.refresh_from_db()
            self.assertEqual(artifact.status, ParserSourceArtifact.Status.PUBLISHED)
        self.assertEqual(len(callbacks), 1)
        self.assertEqual(invalidate.call_count, 2)

    def test_after_commit_cache_failure_does_not_reject_published_data(self):
        with patch(
            "apps.parsers.registry_snapshots.invalidate_source_data_cache",
            side_effect=[None, RuntimeError("cache")],
        ), self.assertLogs(
            "apps.parsers.registry_snapshots", level="ERROR"
        ), self.captureOnCommitCallbacks(execute=True):
            artifact, result = self.import_sme()
        artifact.refresh_from_db()
        self.assertEqual(artifact.status, ParserSourceArtifact.Status.PUBLISHED)
        self.assertEqual(OrganizationSourceRecord.objects.count(), result.published)

    def test_job_finalization_failure_rolls_back_publication(self):
        self.import_sme()
        before = set(OrganizationSourceRecord.objects.values_list("uid", flat=True))
        task = SimpleNamespace(
            name="task", request=SimpleNamespace(id="finalize-failure")
        )

        def refresh(**kwargs):
            return import_sme_support(
                handle=_archive(_xml(numbers=("replacement",))),
                original_name="next.zip",
                snapshot_date="2026-08-15",
                **kwargs,
            )

        with self.assertRaises(RuntimeError), patch.object(
            BackgroundJob, "complete", side_effect=RuntimeError("finalization")
        ):
            _run_snapshot(
                task, source=SME_SOURCE, refresh=refresh, requested_by_id=None
            )
        self.assertEqual(
            set(OrganizationSourceRecord.objects.values_list("uid", flat=True)), before
        )
        self.assertEqual(
            BackgroundJob.objects.get(task_id="finalize-failure").status,
            JobStatus.FAILURE,
        )
        self.assertEqual(
            ParserLoadLog.objects.get().status, ParserLoadLog.Status.FAILED
        )

    def test_revoked_job_cannot_replace_previous_published_snapshot(self):
        self.import_sme()
        before = set(OrganizationSourceRecord.objects.values_list("uid", flat=True))
        task = SimpleNamespace(
            name="task", request=SimpleNamespace(id="revoked-import")
        )

        def refresh(**kwargs):
            BackgroundJob.objects.filter(task_id="revoked-import").update(
                status=JobStatus.REVOKED
            )
            return import_sme_support(
                handle=_archive(_xml(numbers=("replacement",))),
                original_name="next.zip",
                snapshot_date="2026-08-15",
                **kwargs,
            )

        with self.assertRaisesMessage(
            SnapshotValidationError, "snapshot_job_no_longer_active"
        ):
            _run_snapshot(
                task, source=SME_SOURCE, refresh=refresh, requested_by_id=None
            )
        self.assertEqual(
            set(OrganizationSourceRecord.objects.values_list("uid", flat=True)), before
        )
        self.assertEqual(
            BackgroundJob.objects.get(task_id="revoked-import").status,
            JobStatus.REVOKED,
        )
        self.assertEqual(
            ParserLoadLog.objects.get().status, ParserLoadLog.Status.FAILED
        )

    def test_card_list_dashboard_and_export_count_all_published_organizations(self):
        Organization.objects.create(
            name="Справочная организация",
            inn="5678901234",
            ogrn="1027700132195",
            okpo="87654321",
            directory_imported_at=timezone.now(),
            opk_registry_membership=False,
        )
        root = etree.fromstring(_xml())  # noqa: S320 - locally generated fixture
        other = etree.fromstring(_xml(numbers=("non-opk",), inn="5678901234"))  # noqa: S320
        root.append(other.find("Документ"))
        root.set("КолДок", "2")
        _, result = self.import_sme(etree.tostring(root, encoding="UTF-8"))
        self.assertEqual(result.published, 3)
        client = APIClient()
        client.force_authenticate(UserFactory.create_user())
        card = client.get("/api/v1/sources/sme-support-recipients-registry/")
        records = client.get(
            "/api/v2/organization-source-records/", {"source": SME_SOURCE}
        )
        dashboard = client.get("/api/v1/parsers/dashboard/")
        self.assertEqual(
            (card.status_code, records.status_code, dashboard.status_code),
            (200, 200, 200),
        )
        self.assertEqual(card.data["data"]["records_count"], 3)
        self.assertEqual(records.data["meta"]["pagination"]["total_count"], 3)
        self.assertEqual(dashboard.data["data"]["source_counts"][SME_SOURCE], 3)
        self.assertEqual(
            _source_group_queryset("government_support", export_year=2026).count(), 3
        )

    def test_list_omits_nested_detail_but_detail_and_export_keep_all_years(self):
        self.import_sme()
        record = OrganizationSourceRecord.objects.first()
        self.assertNotIn(
            "violations", OrganizationSourceRecordListSerializer(record).data["payload"]
        )
        self.assertEqual(
            len(
                OrganizationSourceRecordSerializer(record).data["payload"]["violations"]
            ),
            1,
        )
        self.assertEqual(
            _source_group_queryset("government_support", export_year=2026).count(), 2
        )

    def test_budget_scans_foreign_without_detail_or_raw_retention(self):
        own, foreign = _budget_record(), _budget_record("2", inn="5678901234")
        calls = []

        def page(_session, params):
            calls.append(params)
            if params.get("filterid"):
                self.assertEqual(params["filterid"], "1")
                return {"data": [own]}
            return {
                "data": [own, foreign],
                "recordCount": 2,
                "version": "10",
                "pageNum": 1,
            }

        with patch("apps.parsers.budget_registry.budget_page", side_effect=page):
            artifact, result = refresh_budget_registry(load_batch=1, session=object())
        self.assertEqual((result.parsed, result.published, result.skipped), (2, 1, 1))
        self.assertEqual(calls[0]["blocks"], "info")
        self.assertEqual(len(calls), 2)
        self.assertEqual(
            ParserStagedRecord.objects.filter(artifact=artifact).count(), 1
        )
        with artifact.file.open("rb") as handle:
            raw = handle.read().decode()
        self.assertNotIn("5678901234", raw)
        record = OrganizationSourceRecord.objects.get(source=BUDGET_SOURCE)
        self.assertEqual(record.status, "inactive")
        self.assertEqual(
            record.payload["upstream"]["unknown_new_block"], [{"raw": "preserved"}]
        )
        self.assertEqual(
            _source_group_queryset("budget_process_registry", export_year=2026).count(),
            1,
        )
        self.assertEqual(Organization.objects.count(), 1)

    def test_budget_incomplete_second_page_preserves_published_snapshot(self):
        own = _budget_record()
        first = {"data": [own], "recordCount": 1, "version": "10", "pageNum": 1}
        with patch(
            "apps.parsers.budget_registry.budget_page",
            side_effect=[first, {"data": [own]}],
        ):
            refresh_budget_registry(load_batch=1, session=object())
        before = OrganizationSourceRecord.objects.get().uid
        first["recordCount"] = 2
        with patch(
            "apps.parsers.budget_registry.budget_page",
            side_effect=[
                first,
                {"data": [own]},
                {"data": [], "recordCount": 2, "version": "10", "pageNum": 2},
            ],
        ), self.assertRaises(SnapshotValidationError):
            refresh_budget_registry(load_batch=2, session=object())
        self.assertEqual(OrganizationSourceRecord.objects.get().uid, before)

    def test_budget_unknown_status_stays_unknown(self):
        own = _budget_record(status="1")
        with patch(
            "apps.parsers.budget_registry.budget_page",
            side_effect=[
                {"data": [own], "recordCount": 1, "version": "10"},
                {"data": [own]},
            ],
        ):
            refresh_budget_registry(load_batch=1, session=object())
        self.assertEqual(OrganizationSourceRecord.objects.get().status, "unknown")


class RegistryTaskLifecycleTest(TestCase):
    def test_revoked_after_claim_does_not_attach_late_batch_metadata(self):
        task = SimpleNamespace(name="task", request=SimpleNamespace(id="revoked-claim"))
        original = ParserLoadLogService.create_load_log_with_next_batch_id

        def create_log(**kwargs):
            value = original(**kwargs)
            BackgroundJob.objects.filter(task_id="revoked-claim").update(
                status=JobStatus.REVOKED
            )
            return value

        with self.assertRaisesMessage(
            SnapshotValidationError, "snapshot_job_no_longer_active"
        ), patch.object(
            ParserLoadLogService,
            "create_load_log_with_next_batch_id",
            side_effect=create_log,
        ), patch(
            "apps.parsers.tasks_registry_snapshots.refresh_budget_registry"
        ) as refresh:
            _run_snapshot(
                task, source=BUDGET_SOURCE, refresh=refresh, requested_by_id=None
            )
        refresh.assert_not_called()
        job = BackgroundJob.objects.get(task_id="revoked-claim")
        self.assertEqual(job.status, JobStatus.REVOKED)
        self.assertNotIn("batch_id", job.meta)

    def test_validation_failure_reports_bounded_reason(self):
        task = SimpleNamespace(name="task", request=SimpleNamespace(id="invalid-page"))
        with self.assertRaises(SnapshotValidationError), patch(
            "apps.parsers.tasks_registry_snapshots.refresh_budget_registry",
            side_effect=SnapshotValidationError("incomplete_budget_snapshot"),
        ) as refresh:
            _run_snapshot(
                task, source=BUDGET_SOURCE, refresh=refresh, requested_by_id=None
            )
        job = BackgroundJob.objects.get(task_id="invalid-page")
        self.assertIn("incomplete_budget_snapshot", job.error)

    def test_duplicate_completed_delivery_returns_result_without_new_batch(self):
        job = BackgroundJob.objects.create(
            task_id="delivery",
            task_name="task",
            status=JobStatus.SUCCESS,
            result={"published": 5},
        )
        task = SimpleNamespace(name="task", request=SimpleNamespace(id=job.task_id))
        with patch(
            "apps.parsers.tasks_registry_snapshots.refresh_sme_support"
        ) as refresh:
            self.assertEqual(
                _run_snapshot(
                    task, source=SME_SOURCE, refresh=refresh, requested_by_id=None
                ),
                {"published": 5},
            )
            refresh.assert_not_called()
        self.assertFalse(ParserLoadLog.objects.exists())

    def test_duplicate_started_delivery_does_not_repeat_import(self):
        BackgroundJob.objects.create(
            task_id="delivery", task_name="task", status=JobStatus.STARTED
        )
        task = SimpleNamespace(name="task", request=SimpleNamespace(id="delivery"))
        with patch(
            "apps.parsers.tasks_registry_snapshots.refresh_sme_support"
        ) as refresh:
            result = _run_snapshot(
                task, source=SME_SOURCE, refresh=refresh, requested_by_id=None
            )
            self.assertTrue(result["duplicate_delivery"])
            refresh.assert_not_called()
        self.assertFalse(ParserLoadLog.objects.exists())

    def test_sequence_failure_marks_precreated_job_failed(self):
        job = BackgroundJob.objects.create(task_id="delivery", task_name="task")
        task = SimpleNamespace(name="task", request=SimpleNamespace(id=job.task_id))
        with self.assertRaises(RuntimeError), patch(
            "apps.parsers.tasks_registry_snapshots.ParserLoadLogService.create_load_log_with_next_batch_id",
            side_effect=RuntimeError("sequence"),
        ):
            _run_snapshot(
                task,
                source=SME_SOURCE,
                refresh=lambda **_: None,
                requested_by_id=None,
            )
        job.refresh_from_db()
        self.assertEqual(job.status, JobStatus.FAILURE)


def test_sme_discovery_selects_latest_official_snapshot():
    url, snapshot = discover_sme_archive(
        b'<a href="https://file.nalog.ru/opendata/7707329152-rsmppp/data-20260715-structure-20230615.zip">old</a><a href="https://file.nalog.ru/opendata/7707329152-rsmppp/data-20260815-structure-20230615.zip">new</a>'
    )
    assert snapshot == "2026-08-15"
    assert "data-20260815" in url


def test_sme_stream_releases_each_measure_before_reading_next_recipient():
    class BoundedReader(BytesIO):
        def read(self, size=-1):
            assert 0 < size <= 65536
            return super().read(size)

    counts = Counter()
    stream = iter_sme_measures(
        BoundedReader(_xml(numbers=tuple(f"measure-{i}" for i in range(1200)))), counts
    )
    _, recipient, previous = next(stream)
    assert recipient["ИННЮЛ"] == "1234567890"
    _, recipient, current = next(stream)
    assert len(previous) == 0
    assert not previous.attrib
    assert current.attrib["НомерПод"] == "measure-1"
    remaining = 0
    for _, recipient, _ in stream:
        assert recipient["ИННЮЛ"] == "1234567890"
        remaining += 1
    assert remaining == 1198
    assert counts == {"documents": 1, "measures": 1200, "declared_documents": 1}