"""Recognize the observed lookup 404 template without publishing an outage as empty.""" import json import zipfile from unittest.mock import Mock from urllib.parse import parse_qs, urlsplit import pytest from apps.parsers.models import ParserSourceArtifact, SroOrganizationLookup from apps.parsers.registry_snapshots import SnapshotValidationError from apps.parsers.sro_http import SroHttpClient, SroPage from apps.parsers.sro_membership import ( SRO_LOOKUP_URL, parse_sro_lookup, refresh_sro_membership, ) from django.utils import timezone from organizations.models import Organization, OrganizationSourceRecord from tests.apps.parsers.test_sro_membership import FixtureSite LOOKUP_TITLE = ( "Проверка членства в реестре СРО, проверить допуск организации в СРО по ИНН" ) NOT_FOUND_HEADING = "Запрашиваемая страница на сайте отсутствует." LOOKUP_EMPTY = ( f"{LOOKUP_TITLE}" f"

{NOT_FOUND_HEADING}

" ).encode() @pytest.fixture(autouse=True) def sro_settings(settings): settings.SRO_DEV_COLLECTION_ENABLED = True def response_404(body=LOOKUP_EMPTY): response = Mock(status_code=404, is_redirect=False) response.iter_content.return_value = [body] return response @pytest.mark.parametrize("query", ("1234567890", "1027700132195")) def test_http_returns_bounded_lookup_404_with_status(query): response = response_404() session = Mock() session.get.return_value = response url = f"{SRO_LOOKUP_URL}?q={query}&sro_name=&search=Find" page = SroHttpClient(session=session).get(url) assert (page.url, page.body, page.status_code) == (url, LOOKUP_EMPTY, 404) session.get.assert_called_once() response.close.assert_called_once() @pytest.mark.parametrize( "suffix", ( "/", "/members/?q=1234567890", "/proverka_dopuska/extra/?q=1234567890", "/proverka_dopuska/", "/proverka_dopuska/?q=12345678901", "/proverka_dopuska/?q=1234567890&q=", "/proverka_dopuska/?q=1234567890&q=1027700132195", "/proverka_dopuska/?q=not-an-identifier", "/proverka_dopuska/?q=1234567890", ), ) def test_other_404s_are_not_reclassified_or_read(suffix): response = response_404() session = Mock() session.get.return_value = response with pytest.raises(SnapshotValidationError, match="^sro_page_not_found$"): SroHttpClient(session=session).get(f"https://www.reestr-sro.ru{suffix}") response.iter_content.assert_not_called() response.close.assert_called_once() def test_lookup_404_response_body_size_limit(settings): settings.SRO_HTTP_MAX_RESPONSE_BYTES = 4 response = response_404(b"12345") session = Mock() session.get.return_value = response with pytest.raises(SnapshotValidationError, match="^source_response_too_large$"): SroHttpClient(session=session).get(f"{SRO_LOOKUP_URL}?q=1234567890") response.close.assert_called_once() def test_parser_recognizes_only_observed_negative_404(): page = SroPage(f"{SRO_LOOKUP_URL}?q=1234567890", LOOKUP_EMPTY, status_code=404) assert parse_sro_lookup(page) == ([], None) assert SroPage(page.url, b"fixture").status_code == 200 @pytest.mark.parametrize( "body", ( b"

404 Not Found

", b"Login

Sign in

", "Ничего не найдено".encode(), LOOKUP_EMPTY.replace(LOOKUP_TITLE.encode(), b"Other source"), LOOKUP_EMPTY.replace(NOT_FOUND_HEADING.encode(), b"Captcha"), LOOKUP_EMPTY.replace(b"", b"

Other

"), LOOKUP_EMPTY.replace(b"", b'
'), ), ) def test_parser_rejects_unrecognized_404_even_with_empty_markers(body): page = SroPage(f"{SRO_LOOKUP_URL}?q=1234567890", body, status_code=404) with pytest.raises( SnapshotValidationError, match="^sro_lookup_response_unrecognized$" ): parse_sro_lookup(page) @pytest.mark.parametrize( "url,status", ( (f"{SRO_LOOKUP_URL}?q=1234567890", 200), (f"{SRO_LOOKUP_URL}?q=1234567890", 503), ("https://www.reestr-sro.ru/members/?q=1234567890", 404), ("https://other.invalid/proverka_dopuska/?q=1234567890", 404), ), ) def test_template_is_not_empty_for_other_status_or_url(url, status): with pytest.raises(SnapshotValidationError): parse_sro_lookup(SroPage(url, LOOKUP_EMPTY, status_code=status)) class MixedFixtureSite(FixtureSite): def __init__(self, organizations, negative_ids): super().__init__(organizations) self.negative_ids = negative_ids def get(self, url): parts = urlsplit(url) query = parse_qs(parts.query).get("q", [None])[0] if parts.path == "/proverka_dopuska/" and query in self.negative_ids: self.urls.append(url) return SroPage(url, LOOKUP_EMPTY, status_code=404) return super().get(url) @pytest.fixture def organizations(db, settings, tmp_path): settings.MEDIA_ROOT = str(tmp_path / "media") settings.PARSER_PRIVATE_ARTIFACT_ROOT = str(tmp_path / "private") return [ Organization.objects.create( name=f"АО Фикстура {index}", inn=f"{index}234567890", ogrn=f"{index}027700132195", okpo=f"0012345{index}", directory_imported_at=timezone.now(), ) for index in (1, 2) ] def test_mixed_positive_and_known_404_publishes_with_status_provenance(organizations): positive, negative = organizations refresh_sro_membership(load_batch=1, mode="full", client=FixtureSite(organizations)) artifact, result = refresh_sro_membership( load_batch=2, mode="full", client=MixedFixtureSite(organizations, {negative.ogrn}), ) assert result.published == 1 assert ( OrganizationSourceRecord.objects.get().extension.organization_id == positive.uid ) assert artifact.metadata["recognized_lookup_404_count"] == 1 assert artifact.metadata["successful_lookup_200_count"] == 1 assert artifact.metadata["not_found_organizations_count"] == 1 assert ( SroOrganizationLookup.objects.get(organization=negative).artifact_id == artifact.uid ) with zipfile.ZipFile(artifact.file.path) as raw: assert raw.testzip() is None manifest = json.loads(raw.read("manifest.json")) assert sorted(item["status_code"] for item in manifest) == [200, 200, 404] negative_entry = next(item for item in manifest if item["status_code"] == 404) assert raw.read(negative_entry["file"]) == LOOKUP_EMPTY @pytest.mark.parametrize("mode", ("full", "incremental")) def test_all_known_404_rejects_without_changing_previous_snapshot_or_checkpoints( organizations, mode ): previous, _ = refresh_sro_membership( load_batch=1, mode="full", client=FixtureSite(organizations) ) records = list(OrganizationSourceRecord.objects.order_by("uid").values()) checkpoints = list( SroOrganizationLookup.objects.order_by("organization_id").values() ) for organization in organizations: organization.name += " изменённая" organization.save(update_fields=["name"]) on_publish = Mock() with pytest.raises(SnapshotValidationError, match="^sro_ambiguous_empty_scan$"): refresh_sro_membership( load_batch=2, mode=mode, client=MixedFixtureSite(organizations, {org.ogrn for org in organizations}), on_publish=on_publish, ) on_publish.assert_not_called() assert list(OrganizationSourceRecord.objects.order_by("uid").values()) == records assert ( list(SroOrganizationLookup.objects.order_by("organization_id").values()) == checkpoints ) previous.refresh_from_db() assert previous.status == ParserSourceArtifact.Status.PUBLISHED rejected = ParserSourceArtifact.objects.get(load_batch=2) assert rejected.status == ParserSourceArtifact.Status.REJECTED assert rejected.metadata["recognized_lookup_404_count"] == 2 assert rejected.metadata["successful_lookup_200_count"] == 0 with zipfile.ZipFile(rejected.file.path) as raw: assert raw.testzip() is None assert [ item["status_code"] for item in json.loads(raw.read("manifest.json")) ] == [404, 404] def test_negative_redirect_to_other_identifier_preserves_snapshot(organizations): positive, negative = organizations refresh_sro_membership(load_batch=1, mode="full", client=FixtureSite(organizations)) records = list(OrganizationSourceRecord.objects.order_by("uid").values()) checkpoints = list( SroOrganizationLookup.objects.order_by("organization_id").values() ) class RedirectedSite(MixedFixtureSite): def get(self, url): page = super().get(url) if page.status_code == 404: return SroPage( f"{SRO_LOOKUP_URL}?q={positive.ogrn}", page.body, status_code=404 ) return page with pytest.raises( SnapshotValidationError, match="^sro_lookup_identifier_mismatch$" ): refresh_sro_membership( load_batch=2, mode="full", client=RedirectedSite(organizations, {negative.ogrn}), ) assert list(OrganizationSourceRecord.objects.order_by("uid").values()) == records assert ( list(SroOrganizationLookup.objects.order_by("organization_id").values()) == checkpoints ) rejected = ParserSourceArtifact.objects.get(load_batch=2) assert rejected.status == ParserSourceArtifact.Status.REJECTED assert rejected.metadata["recognized_lookup_404_count"] == 0 assert rejected.metadata["parse_errors_count"] == 1