diff --git a/docs/sro-membership-source.md b/docs/sro-membership-source.md
index b155b69..4a43ef9 100644
--- a/docs/sro-membership-source.md
+++ b/docs/sro-membership-source.md
@@ -7,7 +7,7 @@
## Доступ и пределы запросов
-По умолчанию сбор закрыт. Для открытия необходимы одновременно
+По умолчанию сбор закрыт. Для production необходимы одновременно
`SRO_UPSTREAM_ACCESS_APPROVED=true` и непустой
`SRO_UPSTREAM_APPROVAL_REFERENCE` — ссылка/номер документированного разрешения
владельца сайта на автоматизированные query/search запросы. Согласование разработки
@@ -15,13 +15,30 @@
должны быть совместимы с настройками загрузчика; более строгие пределы задаются
перед включением. Секреты в approval reference хранить нельзя.
+15.09.2026 пользователь явно изменил требование: «Изменить требование и включить
+сбор на dev». Только для dev предусмотрен отдельный opt-in
+`SRO_DEV_COLLECTION_ENABLED=true`. Он разрешает сбор без изменения
+`SRO_UPSTREAM_ACCESS_APPROVED` и `SRO_UPSTREAM_APPROVAL_REFERENCE` и не обозначает
+разрешение владельца upstream. По умолчанию флаг `false`; на production он должен
+оставаться выключенным. Код разрешает запуск при включённом dev-флаге **или** при
+наличии обоих параметров согласованного доступа. Для отключения dev-исключения
+верните флаг в `false` и примените конфигурацию web/worker.
+
+Исходный [source-first контракт](source-integration/sources/sro-membership-check/backend-api.md)
+с ограничением production refresh и
+[документ замечаний](source-integration/source-records-backend-improvements.md)
+с исходным ответом `409` сохранены как происхождение требования. Настоящее
+изменение относится только к включению сбора на dev; ограничения запросов,
+привязки организаций и публикации данных сохраняются.
+
Оба ручных entrypoint проверяют этот gate до enqueue; task и HTTP клиент повторяют
проверку перед внешним IO. Закрытый gate возвращает typed
-`409 upstream_access_not_approved`. Если разрешение отозвано после постановки,
+`409 upstream_access_not_approved`. Если gate закрыт после постановки,
job завершается с ошибкой, без успешной пустой публикации.
| Настройка | По умолчанию | Значение |
| --- | --- | --- |
+| `SRO_DEV_COLLECTION_ENABLED` | `false` | Явное включение сбора только на dev, отдельно от подтверждения разрешения upstream. |
| `SRO_REQUEST_INTERVAL_SECONDS` | `3` | Минимум 3 секунды между запросами, включая redirects/retries; можно увеличить. |
| `SRO_HTTP_MAX_RESPONSE_BYTES` | `2097152` | Предел распакованного тела одного ответа. |
| `SRO_HTTP_TIMEOUT_SECONDS` | `30` | Read timeout; connect timeout 10 секунд. |
@@ -33,7 +50,27 @@ HTTP клиент делает не более трёх попыток при т
ответами; каждый адрес проверяется до обращения. Разрешён только HTTPS без
credentials/нестандартного порта, на `reestr-sro.ru` и его поддоменах. Переменные
окружения HTTP proxy автоматически не используются. HTML/query с идентификаторами
-не выводится в application logs.
+не выводится в application logs. User-Agent — нейтральный
+`Mostovik SRO integration`, без утверждения о разрешении upstream.
+
+Наблюдаемый 15.09.2026 ответ lookup `404` учитывается как отсутствие членства
+только для `/proverka_dopuska/` с одним `q` из 10 или 13 цифр, при точном title
+«Проверка членства в реестре СРО, проверить допуск организации в СРО по ИНН»,
+единственном h1 «Запрашиваемая страница на сайте отсутствует.» и отсутствии
+`table.sro-members`. Конечный `q` после redirects должен совпадать с запрошенным
+идентификатором организации; иначе запуск отклоняется с
+`sro_lookup_identifier_mismatch`. Другой HTML при lookup `404` отклоняется с
+`sro_lookup_response_unrecognized`. На остальных путях `404` сохраняет прежнее
+значение `sro_page_not_found`, в том числе на страницах даты допуска.
+
+Этот шаблон `404` неоднозначен: он сам по себе не доказывает отсутствие членства.
+Поэтому если **все** кандидаты получили такой ответ и ни один lookup не вернул
+успешно разобранный `200`, весь запуск отклоняется с `sro_ambiguous_empty_scan`
+до публикации и изменения checkpoints; прежние записи сохраняются. Смешанный
+запуск с корректным `200` может учитывать распознанные отрицательные ответы.
+Приватный raw manifest сохраняет `status_code`; artifact metadata содержит
+`recognized_lookup_404_count` и `successful_lookup_200_count`. Предел размера
+ответа действует и для lookup `404`. Дата реестра из error-шаблона не извлекается.
## Публикация и обновление
diff --git a/src/apps/parsers/sro_http.py b/src/apps/parsers/sro_http.py
index 823dd37..181a22b 100644
--- a/src/apps/parsers/sro_http.py
+++ b/src/apps/parsers/sro_http.py
@@ -1,12 +1,13 @@
-"""Permission-gated, paced reads of the SRO site; no automatic hidden retries."""
+"""Access-gated SRO reads with explicit dev opt-in, pacing and bounded retries."""
from __future__ import annotations
+import re
import time
from dataclasses import dataclass
from datetime import UTC, datetime
from email.utils import parsedate_to_datetime
-from urllib.parse import urljoin, urlsplit
+from urllib.parse import parse_qs, urljoin, urlsplit
import requests
from apps.core.exceptions import ConflictError
@@ -16,6 +17,8 @@ from django.conf import settings
def require_sro_access_approved() -> None:
+ if settings.SRO_DEV_COLLECTION_ENABLED:
+ return
if not (
settings.SRO_UPSTREAM_ACCESS_APPROVED
and settings.SRO_UPSTREAM_APPROVAL_REFERENCE.strip()
@@ -47,10 +50,26 @@ def safe_sro_url(value: str, *, base: str | None = None) -> str:
return value
+def is_sro_lookup_url(value: str) -> bool:
+ """Only the observed exact identifier lookup may carry a negative 404 body."""
+ try:
+ parts = urlsplit(safe_sro_url(value))
+ query = parse_qs(parts.query, keep_blank_values=True, max_num_fields=20)
+ except (SnapshotValidationError, ValueError):
+ return False
+ identifiers = query.get("q", [])
+ return (
+ parts.path == "/proverka_dopuska/"
+ and len(identifiers) == 1
+ and re.fullmatch(r"[0-9]{10}(?:[0-9]{3})?", identifiers[0]) is not None
+ )
+
+
@dataclass(frozen=True)
class SroPage:
url: str
body: bytes
+ status_code: int = 200
class SroHttpClient:
@@ -62,9 +81,7 @@ class SroHttpClient:
self.owns_session = session is None
if self.owns_session:
self.session.trust_env = False
- self.session.headers[
- "User-Agent"
- ] = "Mostovik SRO integration (approved access)"
+ self.session.headers["User-Agent"] = "Mostovik SRO integration"
self.clock, self.sleep = clock, sleep
self.last_request = None
self.requests_count = 0
@@ -79,11 +96,14 @@ class SroHttpClient:
url = safe_sro_url(response.headers.get("Location", ""), base=url)
continue
if response.status_code == 404:
- raise SnapshotValidationError("sro_page_not_found")
- if response.status_code != 200:
+ if not is_sro_lookup_url(url):
+ raise SnapshotValidationError("sro_page_not_found")
+ elif response.status_code != 200:
raise SnapshotValidationError("sro_upstream_http_error")
return SroPage(
- url, limited_bytes(response, settings.SRO_HTTP_MAX_RESPONSE_BYTES)
+ url,
+ limited_bytes(response, settings.SRO_HTTP_MAX_RESPONSE_BYTES),
+ status_code=response.status_code,
)
except requests.RequestException as exc:
self.http_errors_count += 1
diff --git a/src/apps/parsers/sro_membership.py b/src/apps/parsers/sro_membership.py
index 3c2fe29..b8873f2 100644
--- a/src/apps/parsers/sro_membership.py
+++ b/src/apps/parsers/sro_membership.py
@@ -1,8 +1,4 @@
-"""SRO HTML mapping and complete, permission-gated membership snapshots.
-
-Selectors follow the approved source-first fixtures. Live query/search acceptance
-requires upstream approval and has deliberately not been performed.
-"""
+"""SRO HTML mapping and complete snapshots under configured source access."""
from __future__ import annotations
@@ -13,7 +9,7 @@ import tempfile
import zipfile
from collections import Counter
from datetime import datetime
-from urllib.parse import urlencode, urlsplit
+from urllib.parse import parse_qs, urlencode, urlsplit
from apps.parsers.models import (
ParserSourceArtifact,
@@ -31,6 +27,7 @@ from apps.parsers.registry_snapshots import (
from apps.parsers.sro_http import (
SroHttpClient,
SroPage,
+ is_sro_lookup_url,
require_sro_access_approved,
safe_sro_url,
)
@@ -49,6 +46,10 @@ SRO_STATUSES = {
}
SRO_ID = re.compile(r"/sro-id-(\d+)(?:/|$)")
ZERO_MARKERS = ("ничего не найдено", "сведения не найдены", "найдено 0")
+LOOKUP_NOT_FOUND_TITLE = (
+ "Проверка членства в реестре СРО, проверить допуск организации в СРО по ИНН"
+)
+LOOKUP_NOT_FOUND_HEADING = "Запрашиваемая страница на сайте отсутствует."
MONTHS = {
name: index
for index, name in enumerate(
@@ -158,8 +159,29 @@ def _table_rows(document: BeautifulSoup, *, admission: bool = False) -> list[dic
return result
-def parse_sro_lookup(page: SroPage) -> tuple[list[dict], str | None]:
+def parse_sro_lookup(
+ page: SroPage, *, expected_lookup_value: str | None = None
+) -> tuple[list[dict], str | None]:
document = BeautifulSoup(page.body, "html.parser")
+ if page.status_code == 404:
+ titles, headings = document.find_all("title"), document.find_all("h1")
+ if (
+ is_sro_lookup_url(page.url)
+ and len(titles) == len(headings) == 1
+ and _text(titles[0].get_text(" ", strip=True)) == LOOKUP_NOT_FOUND_TITLE
+ and _text(headings[0].get_text(" ", strip=True)) == LOOKUP_NOT_FOUND_HEADING
+ and document.select_one("table.sro-members") is None
+ ):
+ if expected_lookup_value is not None and parse_qs(
+ urlsplit(page.url).query
+ ).get("q") != [expected_lookup_value]:
+ raise SnapshotValidationError("sro_lookup_identifier_mismatch")
+ # This observed error template is ambiguous in isolation. The scan
+ # requires a valid 200 lookup before it may publish such negatives.
+ return [], None
+ raise SnapshotValidationError("sro_lookup_response_unrecognized")
+ if page.status_code != 200:
+ raise SnapshotValidationError("sro_lookup_response_unrecognized")
result = []
for cells in _table_rows(document):
identity = cells["identity"].get_text(" ", strip=True)
@@ -424,6 +446,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
manifest, pending = [], []
raw_count, not_found, missing_dates = 0, 0, 0
parse_errors = 0
+ recognized_lookup_404, successful_lookup_200 = 0, 0
reasons, dates = Counter(), set()
def fetch(url):
@@ -432,7 +455,9 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
safe_sro_url(page.url)
filename = f"response-{len(manifest) + 1:06d}.html"
raw.writestr(filename, page.body)
- manifest.append({"file": filename, "url": page.url})
+ manifest.append(
+ {"file": filename, "url": page.url, "status_code": page.status_code}
+ )
return page
resolver = SroResolver(fetch)
@@ -452,7 +477,9 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
}
)
page = fetch(f"{SRO_LOOKUP_URL}?{query}")
- rows, version = parse_sro_lookup(page)
+ rows, version = parse_sro_lookup(page, expected_lookup_value=value)
+ recognized_lookup_404 += page.status_code == 404
+ successful_lookup_200 += page.status_code == 200
if version:
dates.add(version)
if len(dates) > 1:
@@ -495,6 +522,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
parse_errors += str(exc).startswith(
(
"sro_members",
+ "sro_lookup_",
"invalid_sro_registry_date",
"sro_registry_changed",
"sro_sitemap_parse_error",
@@ -517,6 +545,8 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
candidate_organizations_count=len(candidates),
queried_organizations_count=completed,
not_found_organizations_count=not_found,
+ recognized_lookup_404_count=recognized_lookup_404,
+ successful_lookup_200_count=successful_lookup_200,
raw_memberships_count=raw_count,
found_memberships_count=raw_count,
quarantined_records_count=sum(reasons.values()),
@@ -546,15 +576,24 @@ def _sro_candidates(mode: str) -> list[Organization]:
.select_related("sro_lookup")
.order_by("uid")
)
- if mode == "full":
- return list(candidates)
- return [
- organization
- for organization in candidates
- if not hasattr(organization, "sro_lookup")
- or organization.sro_lookup.organization_fingerprint
- != _organization_fingerprint(organization)
- ]
+ own_index = OwnOrganizationIndex()
+ selected = []
+ for organization in candidates:
+ resolved, reason = own_index.resolve(
+ inn=organization.inn, ogrn=organization.ogrn
+ )
+ # Shared branch identities belong to the unambiguous canonical head.
+ # Keep conflicts/incomplete identities for the existing strict validation.
+ if not reason and resolved is not None and resolved.uid != organization.uid:
+ continue
+ if (
+ mode == "full"
+ or not hasattr(organization, "sro_lookup")
+ or organization.sro_lookup.organization_fingerprint
+ != _organization_fingerprint(organization)
+ ):
+ selected.append(organization)
+ return selected
def _save_sro_checkpoints(candidates, artifact) -> None:
@@ -607,6 +646,14 @@ def refresh_sro_membership(
# Keep raw/quarantine diagnostics, but never advance any checkpoint or
# replace the previous complete dataset with this incomplete scan.
raise SnapshotValidationError("sro_incomplete_membership_scan")
+ if (
+ candidates
+ and artifact.metadata["recognized_lookup_404_count"] == len(candidates)
+ and not artifact.metadata["successful_lookup_200_count"]
+ ):
+ # An all-404 scan cannot distinguish absence from a site-wide outage.
+ # Raw responses stay available, but previous rows/checkpoints survive.
+ raise SnapshotValidationError("sro_ambiguous_empty_scan")
if mode == "incremental" and not candidates:
previous = (
ParserSourceArtifact.objects.filter(
diff --git a/src/settings/base.py b/src/settings/base.py
index 1b83d1e..24bf6a5 100644
--- a/src/settings/base.py
+++ b/src/settings/base.py
@@ -503,7 +503,8 @@ FNS_FAILED_DIRECTORY = PROJECT_ROOT / "input" / "fns" / "failed"
CHECKO_API_KEY = os.getenv("CHECKO_API_KEY", "")
-# SRO collection stays disabled until the site owner approves automated access.
+# Default closed; the explicit collection override is configured only on dev.
+SRO_DEV_COLLECTION_ENABLED = _env_bool("SRO_DEV_COLLECTION_ENABLED", default=False)
SRO_UPSTREAM_ACCESS_APPROVED = (
os.getenv("SRO_UPSTREAM_ACCESS_APPROVED", "false").lower() == "true"
)
diff --git a/tests/apps/parsers/test_sro_canonical_candidates.py b/tests/apps/parsers/test_sro_canonical_candidates.py
new file mode 100644
index 0000000..e4e91f3
--- /dev/null
+++ b/tests/apps/parsers/test_sro_canonical_candidates.py
@@ -0,0 +1,83 @@
+"""SRO queries use the directory's canonical organization without hiding conflicts."""
+
+from apps.parsers.models import SroOrganizationLookup
+from apps.parsers.sro_membership import _organization_fingerprint, _sro_candidates
+from django.test import TestCase
+from django.utils import timezone
+from organizations.models import Organization
+
+
+class SroCanonicalCandidatesTest(TestCase):
+ def create_organization(self, **overrides):
+ values = {
+ "name": "АО Фикстура",
+ "inn": "1234567890",
+ "ogrn": "1027700132195",
+ "okpo": "00123456",
+ "directory_imported_at": timezone.now(),
+ }
+ values.update(overrides)
+ return Organization.objects.create(**values)
+
+ def candidate_ids(self, mode):
+ return [organization.uid for organization in _sro_candidates(mode)]
+
+ def test_head_and_branches_are_queried_as_one_canonical_organization(self):
+ head = self.create_organization()
+ self.create_organization(name="Первый филиал", is_branch=True)
+ self.create_organization(name="Второй филиал", is_branch=True)
+
+ for mode in ("full", "incremental"):
+ with self.subTest(mode=mode):
+ self.assertEqual(self.candidate_ids(mode), [head.uid])
+
+ def test_ambiguous_heads_are_not_silently_discarded_or_chosen(self):
+ first = self.create_organization()
+ second = self.create_organization(name="Другой головной офис")
+
+ for mode in ("full", "incremental"):
+ with self.subTest(mode=mode):
+ self.assertEqual(set(self.candidate_ids(mode)), {first.uid, second.uid})
+
+ def test_unchanged_head_skips_incremental_and_branches_do_not_retrigger_it(self):
+ head = self.create_organization()
+ self.create_organization(name="Филиал без контрольной точки", is_branch=True)
+ SroOrganizationLookup.objects.create(
+ organization=head,
+ checked_at=timezone.now(),
+ organization_fingerprint=_organization_fingerprint(head),
+ )
+
+ self.assertEqual(self.candidate_ids("incremental"), [])
+ self.assertEqual(self.candidate_ids("full"), [head.uid])
+
+ head.name = "Новое наименование головной организации"
+ head.save(update_fields=["name"])
+ self.assertEqual(self.candidate_ids("incremental"), [head.uid])
+
+ def test_branch_with_distinct_identity_remains_a_candidate(self):
+ head = self.create_organization()
+ branch = self.create_organization(
+ name="Филиал с собственными реквизитами",
+ inn="1234567891",
+ ogrn="1027700132196",
+ is_branch=True,
+ )
+
+ for mode in ("full", "incremental"):
+ with self.subTest(mode=mode):
+ self.assertEqual(set(self.candidate_ids(mode)), {head.uid, branch.uid})
+
+ def test_incomplete_resolution_does_not_discard_the_candidate(self):
+ head = self.create_organization(okpo="")
+ branch = self.create_organization(name="Филиал", is_branch=True)
+
+ for mode in ("full", "incremental"):
+ with self.subTest(mode=mode):
+ self.assertEqual(set(self.candidate_ids(mode)), {head.uid, branch.uid})
+
+ def test_organization_outside_directory_does_not_become_a_candidate(self):
+ head = self.create_organization()
+ self.create_organization(directory_imported_at=None)
+
+ self.assertEqual(self.candidate_ids("full"), [head.uid])
diff --git a/tests/apps/parsers/test_sro_dev_access.py b/tests/apps/parsers/test_sro_dev_access.py
new file mode 100644
index 0000000..d6f75be
--- /dev/null
+++ b/tests/apps/parsers/test_sro_dev_access.py
@@ -0,0 +1,119 @@
+"""Explicit dev collection does not fabricate upstream approval or perform HTTP."""
+
+from importlib import import_module
+from unittest.mock import Mock, patch
+from uuid import UUID
+
+import pytest
+from apps.core.exceptions import ConflictError
+from apps.core.models import BackgroundJob, JobStatus
+from apps.parsers.sro_http import SroHttpClient, require_sro_access_approved
+from core.celery import app as celery_app
+from django.core.cache import cache
+from rest_framework.test import APIClient
+
+from tests.apps.user.factories import UserFactory
+
+
+@pytest.fixture(autouse=True)
+def closed_sro_gate(settings):
+ settings.SRO_DEV_COLLECTION_ENABLED = False
+ settings.SRO_UPSTREAM_ACCESS_APPROVED = False
+ settings.SRO_UPSTREAM_APPROVAL_REFERENCE = ""
+
+
+@pytest.fixture
+def admin_client(db):
+ cache.clear()
+ client = APIClient()
+ client.force_authenticate(UserFactory.create_user(is_staff=True))
+ return client
+
+
+@pytest.mark.parametrize(
+ "dev_enabled,approved,reference,allowed",
+ (
+ (False, False, "", False),
+ (False, True, "", False),
+ (False, False, "written approval", False),
+ (False, True, "written approval", True),
+ (True, False, "", True),
+ (True, True, "", True),
+ ),
+)
+def test_gate_keeps_dev_override_separate_from_upstream_approval(
+ settings, dev_enabled, approved, reference, allowed
+):
+ settings.SRO_DEV_COLLECTION_ENABLED = dev_enabled
+ settings.SRO_UPSTREAM_ACCESS_APPROVED = approved
+ settings.SRO_UPSTREAM_APPROVAL_REFERENCE = reference
+ if allowed:
+ require_sro_access_approved()
+ else:
+ with pytest.raises(ConflictError) as error:
+ require_sro_access_approved()
+ assert error.value.code == "upstream_access_not_approved"
+ assert settings.SRO_UPSTREAM_ACCESS_APPROVED is approved
+ assert reference == settings.SRO_UPSTREAM_APPROVAL_REFERENCE
+
+
+@pytest.mark.parametrize(
+ "url",
+ (
+ "/api/v1/sources/sro-membership-check/refresh/",
+ "/api/v1/parsers/run/sro_membership_check/",
+ ),
+)
+@pytest.mark.parametrize("enabled", (False, True))
+def test_dev_refresh_entrypoints_enqueue_only_with_explicit_opt_in(
+ admin_client, settings, url, enabled
+):
+ settings.SRO_DEV_COLLECTION_ENABLED = enabled
+ import_module("apps.parsers.tasks")
+ task = celery_app.tasks["parsers.sro_membership_check.refresh"]
+ with (
+ patch.object(task, "apply_async") as dispatch,
+ patch("apps.parsers.sro_http.requests.Session") as http_session,
+ ):
+ response = admin_client.post(url, {}, format="json")
+ http_session.assert_not_called()
+ if not enabled:
+ assert response.status_code == 409
+ assert response.data["errors"][0]["code"] == "upstream_access_not_approved"
+ assert not BackgroundJob.objects.exists()
+ dispatch.assert_not_called()
+ return
+ assert response.status_code == 202
+ payload = response.data.get("data", response.data)
+ assert payload["status"] == "queued"
+ assert payload["task_ids"] == [payload["task_id"]]
+ assert str(UUID(payload["task_id"])) == payload["task_id"]
+ job = BackgroundJob.objects.get(task_id=payload["task_id"])
+ assert job.status == JobStatus.PENDING
+ assert job.task_name == "parsers.sro_membership_check.refresh"
+ dispatch.assert_called_once()
+ assert settings.SRO_UPSTREAM_ACCESS_APPROVED is False
+ assert settings.SRO_UPSTREAM_APPROVAL_REFERENCE == ""
+
+
+def test_dev_http_client_uses_neutral_user_agent(settings):
+ settings.SRO_DEV_COLLECTION_ENABLED = True
+ session = Mock(headers={})
+ with patch("apps.parsers.sro_http.requests.Session", return_value=session):
+ client = SroHttpClient()
+ client.close()
+ assert session.headers["User-Agent"] == "Mostovik SRO integration"
+ assert session.trust_env is False
+ session.get.assert_not_called()
+ session.close.assert_called_once()
+
+
+def test_revoking_dev_override_stops_before_http(settings):
+ settings.SRO_DEV_COLLECTION_ENABLED = True
+ session = Mock()
+ client = SroHttpClient(session=session)
+ settings.SRO_DEV_COLLECTION_ENABLED = False
+ with pytest.raises(ConflictError) as error:
+ client.get("https://reestr-sro.ru/")
+ assert error.value.code == "upstream_access_not_approved"
+ session.get.assert_not_called()
diff --git a/tests/apps/parsers/test_sro_upstream_empty.py b/tests/apps/parsers/test_sro_upstream_empty.py
new file mode 100644
index 0000000..f0b1e43
--- /dev/null
+++ b/tests/apps/parsers/test_sro_upstream_empty.py
@@ -0,0 +1,260 @@
+"""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=１２３４５６７８９０",
+ ),
+)
+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"LoginSign 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"