From 2cfc057e34cf265aa316181810b9f3910b8971dd Mon Sep 17 00:00:00 2001 From: Aleksandr Meshchryakov Date: Tue, 15 Sep 2026 13:25:43 +0200 Subject: [PATCH] fix: publish valid SRO memberships while preserving quarantined history --- docs/sro-membership-source.md | 42 ++- src/apps/core/serializers.py | 3 + src/apps/parsers/registry_snapshots.py | 44 ++- src/apps/parsers/sro_membership.py | 45 ++- src/core/openapi.json | 30 ++ .../parsers/test_snapshot_public_contract.py | 15 + .../test_sro_quarantine_publication.py | 314 ++++++++++++++++++ 7 files changed, 471 insertions(+), 22 deletions(-) create mode 100644 tests/apps/parsers/test_sro_quarantine_publication.py diff --git a/docs/sro-membership-source.md b/docs/sro-membership-source.md index 4a43ef9..905e51e 100644 --- a/docs/sro-membership-source.md +++ b/docs/sro-membership-source.md @@ -120,14 +120,28 @@ static или reverse-proxy document root. Существующая retention-к fallback на public raw нет. Публикация записей, checkpoint организаций, load log и terminal job выполняется -в одной транзакции. Инкрементальная публикация заменяет данные только успешно -проверенного набора организаций; записи остальных организаций сохраняются. -Transport/schema failure или отменённая job откатывает публикацию и checkpoints. -Допустимый пустой ответ удаляет старые членства проверенной организации. -Любая семантически некорректная строка остаётся в quarantine и отклоняет весь -запуск с `sro_incomplete_membership_scan`. Это сохраняет прежние членства и -checkpoints: неизвестный статус или неразрешённая ссылка не доказывают, что -членство исчезло. Организация остаётся кандидатом следующего incremental запуска. +в одной транзакции после завершения сбора всех кандидатов. Transport/schema failure, +нарушение целостности или отменённая job откатывает публикацию и checkpoints целиком. +Незавершённый batch остаётся диагностическим и не становится current snapshot. + +Завершённый сбор может закончиться `success` с quarantine, как в примере истории +исходного контракта. Валидные строки публикуются, включая валидные членства организации, +у которой другая строка оказалась некорректной. Неизвестный статус, отсутствующий +SRO ID или URL, конфликт идентификаторов не исправляются догадками: исходная строка +остаётся в quarantine с причиной. У таких организаций прежние членства, не обновлённые +валидной строкой этого batch, сохраняются с прежними UID, payload и load batch. +Ошибка в строке не доказывает прекращение других членств. Для полностью проверенных +организаций исчезнувшие членства удаляются; допустимый пустой ответ удаляет их все. +При incremental неизменённые, не опрошенные организации также сохраняются. + +Checkpoint полностью проверенной организации обновляется. У организации с quarantine +прежний checkpoint удаляется в той же транзакции, чтобы следующий incremental повторил +lookup даже при неизменившемся fingerprint. Raw artifact и staged audit сохраняются. +Если нет ни одной валидной строки и ни одного подтверждённого пустого lookup `200`, +quarantine отклоняет запуск с `sro_incomplete_membership_scan`; смесь invalid rows +и неоднозначных `404` не считается подтверждённым отсутствием. Для полностью состоящего +из `404` запуска сохраняется отдельный `sro_ambiguous_empty_scan`. В обоих случаях +прежние записи и checkpoints остаются без изменений. Если incremental не нашёл изменившихся организаций, batch имеет нулевую дельту и наследует source registry date/version последней успешной публикации; ссылка @@ -144,8 +158,16 @@ artifacts. Ошибка брокера после commit не превращае общий результат, включая неизменённые организации. `metadata.snapshot_records_count` и `snapshot_organizations_count` дают такие же общие record/distinct organization counts; `batch_published_records_count` и `updated_records_count` — число опубликованных -строк текущей дельты. `parsed_count`, quarantine и candidate/queried counts относятся -к текущей проверке; складывать их с общим published count нельзя. +строк текущей дельты. При full с quarantine общий count также включает сохранённые +старые членства. `retained_records_count` — число старых строк организаций с quarantine, +которые не были обновлены валидным upsert и сохранены от удаления; +`quarantined_organizations_count` — число организаций хотя бы с одной строкой quarantine; +`completed_organizations_count` — число полностью проверенных организаций без quarantine, +включая допустимые пустые результаты. Эти три поля доступны в metadata API. +В приватном artifact `confirmed_empty_lookup_200_count` отдельно фиксирует подтверждённые +пустые `200` для проверки допустимости публикации. `parsed_count`, quarantine и +candidate/queried counts относятся к текущей проверке; складывать их с общим published +count нельзя. Новая дата публикации не изменяет прежнюю lineage сохранённых записей. Источник публикует весь собственный справочник; карточка/list/dashboard для трёх новых источников используют весь published scope. Coverage ОПК показывается diff --git a/src/apps/core/serializers.py b/src/apps/core/serializers.py index bf3004d..746e0af 100644 --- a/src/apps/core/serializers.py +++ b/src/apps/core/serializers.py @@ -85,6 +85,9 @@ class SnapshotRunMetadataSerializer(serializers.Serializer): found_memberships_count = serializers.IntegerField(read_only=True) published_records_count = serializers.IntegerField(read_only=True) quarantined_records_count = serializers.IntegerField(read_only=True) + quarantined_organizations_count = serializers.IntegerField(read_only=True) + completed_organizations_count = serializers.IntegerField(read_only=True) + retained_records_count = serializers.IntegerField(read_only=True) missing_admission_dates_count = serializers.IntegerField(read_only=True) http_errors_count = serializers.IntegerField(read_only=True) parse_errors_count = serializers.IntegerField(read_only=True) diff --git a/src/apps/parsers/registry_snapshots.py b/src/apps/parsers/registry_snapshots.py index ca1963b..f413993 100644 --- a/src/apps/parsers/registry_snapshots.py +++ b/src/apps/parsers/registry_snapshots.py @@ -147,22 +147,41 @@ def _invalidate_committed_snapshot_cache() -> None: ) -def _remove_obsolete_records(source, retained_ids, scope_organization_ids) -> None: +def _remove_obsolete_records( + source, retained_ids, scope_organization_ids, preserve_organization_ids +) -> int: obsolete = OrganizationSourceRecord.objects.filter(source=source) if scope_organization_ids is not None: obsolete = obsolete.filter( extension__organization_id__in=scope_organization_ids ) + retained_count = 0 + if preserve_organization_ids: + protected = obsolete.filter( + extension__organization_id__in=preserve_organization_ids + ) + retained_count = protected.exclude(external_id__in=retained_ids).count() + obsolete = obsolete.exclude( + extension__organization_id__in=preserve_organization_ids + ) obsolete.exclude(external_id__in=retained_ids).delete() + return retained_count -def _set_published_count(artifact, source, published, scope_organization_ids) -> None: +def _set_published_count( + artifact, source, published, scope_organization_ids, retained_count=None +) -> None: artifact.published_count = published - if scope_organization_ids is not None: + if scope_organization_ids is not None or retained_count is not None: artifact.metadata = {**artifact.metadata, "updated_records_count": published} artifact.published_count = OrganizationSourceRecord.objects.filter( source=source ).count() + if retained_count is not None: + artifact.metadata = { + **artifact.metadata, + "retained_records_count": retained_count, + } organizations_count = ( OrganizationSourceRecord.objects.filter(source=source) .order_by() @@ -183,8 +202,13 @@ def publish_snapshot( *, on_publish: Callable[[ParserSourceArtifact, SnapshotResult], None] | None = None, scope_organization_ids: list[uuid.UUID] | None = None, + preserve_organization_ids: list[uuid.UUID] | None = None, ) -> SnapshotResult: - """Publish complete staging in one transaction, preserving stable record IDs.""" + """Publish validated staging atomically; optionally retain uncertain old rows. + + Preservation affects only obsolete deletion for the selected organizations. + Valid staged rows are still upserted, including their existing stable IDs. + """ source = artifact.source descriptor = get_source_group_descriptor(source) staged = ParserStagedRecord.objects.filter( @@ -253,13 +277,21 @@ def publish_snapshot( raise SnapshotValidationError("organization_resolution_changed") published += len(inputs) retained_ids = staged.values_list("external_id", flat=True) - _remove_obsolete_records(source, retained_ids, scope_organization_ids) + retained_count = _remove_obsolete_records( + source, retained_ids, scope_organization_ids, preserve_organization_ids + ) descriptor.extension_model.objects.filter(records__isnull=True).delete() # Deletion can change counts for extensions whose remaining rows were upserted # before old records were removed. _refresh_extension_counts(descriptor) staged.update(disposition=ParserStagedRecord.Disposition.PUBLISHED) - _set_published_count(artifact, source, published, scope_organization_ids) + _set_published_count( + artifact, + source, + published, + scope_organization_ids, + retained_count if preserve_organization_ids is not None else None, + ) artifact.status = ParserSourceArtifact.Status.PUBLISHED artifact.save() result = _artifact_result(artifact) diff --git a/src/apps/parsers/sro_membership.py b/src/apps/parsers/sro_membership.py index b8873f2..9d93d9c 100644 --- a/src/apps/parsers/sro_membership.py +++ b/src/apps/parsers/sro_membership.py @@ -447,6 +447,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress): raw_count, not_found, missing_dates = 0, 0, 0 parse_errors = 0 recognized_lookup_404, successful_lookup_200 = 0, 0 + confirmed_empty_lookup_200 = 0 reasons, dates = Counter(), set() def fetch(url): @@ -480,6 +481,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress): rows, version = parse_sro_lookup(page, expected_lookup_value=value) recognized_lookup_404 += page.status_code == 404 successful_lookup_200 += page.status_code == 200 + confirmed_empty_lookup_200 += page.status_code == 200 and not rows if version: dates.add(version) if len(dates) > 1: @@ -547,6 +549,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress): not_found_organizations_count=not_found, recognized_lookup_404_count=recognized_lookup_404, successful_lookup_200_count=successful_lookup_200, + confirmed_empty_lookup_200_count=confirmed_empty_lookup_200, raw_memberships_count=raw_count, found_memberships_count=raw_count, quarantined_records_count=sum(reasons.values()), @@ -609,6 +612,34 @@ def _save_sro_checkpoints(candidates, artifact) -> None: ) +def _sro_publication_scope(artifact, candidates): + """Keep uncertain old memberships without discarding independently valid rows.""" + staged = artifact.staged_records.all() + if staged.filter(organization_id__isnull=True).exists(): + raise SnapshotValidationError("organization_resolution_changed") + quarantined_ids = set( + staged.filter(disposition=ParserStagedRecord.Disposition.QUARANTINED) + .order_by() + .values_list("organization_id", flat=True) + .distinct() + ) + completed = [org for org in candidates if org.uid not in quarantined_ids] + artifact.metadata.update( + quarantined_organizations_count=len(quarantined_ids), + completed_organizations_count=len(completed), + ) + if ( + quarantined_ids + and not staged.filter( + disposition=ParserStagedRecord.Disposition.STAGED + ).exists() + and not artifact.metadata["confirmed_empty_lookup_200_count"] + ): + # Invalid rows plus an ambiguous 404 do not establish any valid coverage. + raise SnapshotValidationError("sro_incomplete_membership_scan") + return completed, list(quarantined_ids) + + def refresh_sro_membership( *, load_batch: int, @@ -641,11 +672,6 @@ def refresh_sro_membership( ) finally: save_artifact_file(artifact, handle) - if artifact.quarantined_count: - # A semantic failure is not evidence that old memberships disappeared. - # 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) @@ -654,6 +680,7 @@ def refresh_sro_membership( # 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") + completed, quarantined_ids = _sro_publication_scope(artifact, candidates) if mode == "incremental" and not candidates: previous = ( ParserSourceArtifact.objects.filter( @@ -675,7 +702,12 @@ def refresh_sro_membership( artifact.save() def finalize(published_artifact, result): - _save_sro_checkpoints(candidates, published_artifact) + # An unchanged fingerprint must not suppress retry after a full sweep + # found invalid rows. Deletion shares the publication transaction. + SroOrganizationLookup.objects.filter( + organization_id__in=quarantined_ids + ).delete() + _save_sro_checkpoints(completed, published_artifact) if on_publish: on_publish(published_artifact, result) @@ -684,6 +716,7 @@ def refresh_sro_membership( scope_organization_ids=[org.uid for org in candidates] if mode == "incremental" else None, + preserve_organization_ids=quarantined_ids, on_publish=finalize, ) except Exception as exc: diff --git a/src/core/openapi.json b/src/core/openapi.json index 926041b..699cd84 100644 --- a/src/core/openapi.json +++ b/src/core/openapi.json @@ -11787,6 +11787,21 @@ "type": "integer", "readOnly": true }, + "quarantined_organizations_count": { + "title": "Quarantined organizations count", + "type": "integer", + "readOnly": true + }, + "completed_organizations_count": { + "title": "Completed organizations count", + "type": "integer", + "readOnly": true + }, + "retained_records_count": { + "title": "Retained records count", + "type": "integer", + "readOnly": true + }, "missing_admission_dates_count": { "title": "Missing admission dates count", "type": "integer", @@ -11966,6 +11981,21 @@ "type": "integer", "readOnly": true }, + "quarantined_organizations_count": { + "title": "Quarantined organizations count", + "type": "integer", + "readOnly": true + }, + "completed_organizations_count": { + "title": "Completed organizations count", + "type": "integer", + "readOnly": true + }, + "retained_records_count": { + "title": "Retained records count", + "type": "integer", + "readOnly": true + }, "missing_admission_dates_count": { "title": "Missing admission dates count", "type": "integer", diff --git a/tests/apps/parsers/test_snapshot_public_contract.py b/tests/apps/parsers/test_snapshot_public_contract.py index 72cd189..e3cd5a1 100644 --- a/tests/apps/parsers/test_snapshot_public_contract.py +++ b/tests/apps/parsers/test_snapshot_public_contract.py @@ -319,6 +319,9 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client): "found_memberships_count": 2, "snapshot_records_count": 10, "snapshot_organizations_count": 8, + "quarantined_organizations_count": 1, + "completed_organizations_count": 2, + "retained_records_count": 3, "http_errors_count": 1, "parse_errors_count": 2, "upstream_registry_date": None, @@ -334,6 +337,13 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client): assert payload["meta"]["candidate_organizations_count"] == 7 assert payload["meta"]["source"] == "sro_membership_check" assert payload["meta"]["upstream_registry_date"] is None + preservation_counters = { + "quarantined_organizations_count": 1, + "completed_organizations_count": 2, + "retained_records_count": 3, + } + for key, value in preservation_counters.items(): + assert payload["meta"][key] == value assert "private_loader_setting" not in payload["meta"] log = ParserLoadLog.objects.create( source="sro_membership_check", batch_id=1, status="success", records_count=10 @@ -352,6 +362,8 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client): assert response.data["meta"]["http_errors_count"] == 1 assert response.data["meta"]["parse_errors_count"] == 2 assert "private_loader_setting" not in response.data["meta"] + for key, value in preservation_counters.items(): + assert response.data["meta"][key] == value schema = admin_client.get(reverse("schema-swagger-ui"), {"format": "openapi"}).data assert ( schema["definitions"]["SnapshotRunMetadata"]["properties"][ @@ -363,3 +375,6 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client): "organizations_count" in schema["definitions"]["SnapshotJobResult"]["properties"] ) + for key in preservation_counters: + for name in ("SnapshotRunMetadata", "SnapshotJobResult"): + assert schema["definitions"][name]["properties"][key]["type"] == "integer" diff --git a/tests/apps/parsers/test_sro_quarantine_publication.py b/tests/apps/parsers/test_sro_quarantine_publication.py new file mode 100644 index 0000000..c95a58e --- /dev/null +++ b/tests/apps/parsers/test_sro_quarantine_publication.py @@ -0,0 +1,314 @@ +"""Complete collection can publish valid memberships without erasing uncertain ones.""" + +from types import SimpleNamespace +from uuid import UUID, uuid4 + +import pytest +from apps.core.models import BackgroundJob, JobStatus +from apps.parsers.models import ( + ParserLoadLog, + ParserSourceArtifact, + ParserStagedRecord, + SroOrganizationLookup, +) +from apps.parsers.registry_snapshots import SnapshotValidationError +from apps.parsers.source_cards import SourceCardService +from apps.parsers.sro_membership import _sro_candidates, refresh_sro_membership +from apps.parsers.tasks_registry_snapshots import _run_snapshot +from django.utils import timezone +from organizations.models import Organization, OrganizationSourceRecord + +from tests.apps.parsers.test_sro_membership import FixtureSite, lookup_html +from tests.apps.parsers.test_sro_upstream_empty import MixedFixtureSite + + +@pytest.fixture +def organizations(db, settings, tmp_path): + settings.SRO_DEV_COLLECTION_ENABLED = True + settings.MEDIA_ROOT = str(tmp_path / "media") + settings.PARSER_PRIVATE_ARTIFACT_ROOT = str(tmp_path / "private") + return [ + Organization.objects.create( + uid=UUID(int=index), + name=f"АО Смешанная фикстура {index}", + inn=f"{index}234567890", + ogrn=f"{index}027700132195", + okpo=f"0012345{index}", + directory_imported_at=timezone.now(), + opk_registry_membership=True, + ) + for index in (1, 2) + ] + + +def unresolved_html(organization, *, mixed=True): + html = lookup_html( + organization, ("001", "002") if mixed else ("002",), status="Исключен" + ) + return html.replace( + b' href="https://moskva.reestr-sro.ru/sro-v-proektirovanii/sro-id-002/"', + b"", + ) + + +def seed(organizations): + first, _ = organizations + return refresh_sro_membership( + load_batch=1, + mode="full", + client=FixtureSite( + organizations, {first.ogrn: lookup_html(first, ("001", "002", "003"))} + ), + ) + + +def published_rows(): + return list(OrganizationSourceRecord.objects.order_by("uid").values()) + + +def checkpoints(): + return list(SroOrganizationLookup.objects.order_by("organization_id").values()) + + +@pytest.mark.parametrize("mode", ("full", "incremental")) +def test_valid_sibling_upsert_retains_uncertain_old_rows_and_retries_org( + organizations, mode +): + uncertain, empty = organizations + seed(organizations) + old = {row["external_id"]: row for row in published_rows()} + if mode == "incremental": + for organization in organizations: + organization.name += " обновлённая" + organization.save(update_fields=["name"]) + artifact, result = refresh_sro_membership( + load_batch=2, + mode=mode, + client=FixtureSite( + organizations, + { + uncertain.ogrn: unresolved_html(uncertain), + empty.ogrn: lookup_html(empty, ()), + }, + ), + ) + current = {row["external_id"]: row for row in published_rows()} + assert len(current) == result.published == artifact.published_count == 3 + changed = current[f"{uncertain.uid}:001"] + assert changed["uid"] == old[f"{uncertain.uid}:001"]["uid"] + assert changed["status"] == "inactive" + assert changed["load_batch"] == 2 + for identifier in ("002", "003"): + key = f"{uncertain.uid}:{identifier}" + assert current[key] == old[key] + assert not OrganizationSourceRecord.objects.filter( + extension__organization=empty + ).exists() + assert artifact.metadata["retained_records_count"] == 2 + assert artifact.metadata["batch_published_records_count"] == 1 + assert artifact.metadata["snapshot_records_count"] == 3 + assert artifact.metadata["snapshot_organizations_count"] == 1 + assert artifact.metadata["quarantined_organizations_count"] == 1 + assert artifact.metadata["completed_organizations_count"] == 1 + assert artifact.quarantined_count == result.quarantined == 1 + assert ( + artifact.staged_records.get(disposition="quarantined").reason + == "sro_page_unresolved" + ) + assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists() + assert ( + SroOrganizationLookup.objects.get(organization=empty).artifact_id + == artifact.uid + ) + assert [org.uid for org in _sro_candidates("incremental")] == [uncertain.uid] + assert SourceCardService.get_card("sro-membership-check")["records_count"] == 3 + + +def test_first_load_publishes_valid_memberships_of_mixed_organization(organizations): + uncertain, unused = organizations + unused.delete() + artifact, result = refresh_sro_membership( + load_batch=1, + mode="full", + client=FixtureSite([uncertain], {uncertain.ogrn: unresolved_html(uncertain)}), + ) + assert result.published == result.quarantined == 1 + assert artifact.metadata["retained_records_count"] == 0 + assert artifact.metadata["completed_organizations_count"] == 0 + assert artifact.metadata["quarantined_organizations_count"] == 1 + assert OrganizationSourceRecord.objects.get().payload["sro_id"] == "001" + assert not SroOrganizationLookup.objects.exists() + assert [org.uid for org in _sro_candidates("incremental")] == [uncertain.uid] + + +def test_all_invalid_plus_ambiguous_404_is_not_success(organizations): + uncertain, negative = organizations + seed(organizations) + before, before_checkpoints = published_rows(), checkpoints() + site = MixedFixtureSite(organizations, {negative.ogrn}) + site.overrides[uncertain.ogrn] = unresolved_html(uncertain, mixed=False) + with pytest.raises( + SnapshotValidationError, match="^sro_incomplete_membership_scan$" + ): + refresh_sro_membership(load_batch=2, mode="full", client=site) + assert published_rows() == before + assert checkpoints() == before_checkpoints + + +def test_confirmed_zero_with_quarantine_can_publish_preserving_old_memberships( + organizations, +): + uncertain, empty = organizations + seed(organizations) + artifact, result = refresh_sro_membership( + load_batch=2, + mode="full", + client=FixtureSite( + organizations, + { + uncertain.ogrn: unresolved_html(uncertain, mixed=False), + empty.ogrn: lookup_html(empty, ()), + }, + ), + ) + assert result.published == 3 + assert artifact.metadata["batch_published_records_count"] == 0 + assert artifact.metadata["retained_records_count"] == 3 + assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists() + assert ( + SroOrganizationLookup.objects.get(organization=empty).artifact_id + == artifact.uid + ) + + +@pytest.mark.parametrize( + "error", + ( + SnapshotValidationError("sro_upstream_transport_error"), + b"upstream schema changed", + ), + ids=("transport", "schema"), +) +def test_incomplete_collection_does_not_publish_valid_rows_or_clear_checkpoints( + organizations, error +): + uncertain, failed = organizations + seed(organizations) + before, before_checkpoints = published_rows(), checkpoints() + with pytest.raises(SnapshotValidationError): + refresh_sro_membership( + load_batch=2, + mode="full", + client=FixtureSite( + organizations, + {uncertain.ogrn: unresolved_html(uncertain), failed.ogrn: error}, + ), + ) + assert published_rows() == before + assert checkpoints() == before_checkpoints + artifact = ParserSourceArtifact.objects.get(load_batch=2) + assert artifact.status == ParserSourceArtifact.Status.REJECTED + assert not artifact.staged_records.filter(disposition="published").exists() + + +def test_cancelled_finalization_rolls_back_rows_and_quarantined_checkpoint_deletion( + organizations, +): + uncertain, complete = organizations + seed(organizations) + before, before_checkpoints = published_rows(), checkpoints() + + def revoked(artifact, result): + assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists() + assert ( + SroOrganizationLookup.objects.get(organization=complete).artifact_id + == artifact.uid + ) + raise SnapshotValidationError("snapshot_job_no_longer_active") + + with pytest.raises( + SnapshotValidationError, match="^snapshot_job_no_longer_active$" + ): + refresh_sro_membership( + load_batch=2, + mode="full", + client=FixtureSite( + organizations, {uncertain.ogrn: unresolved_html(uncertain)} + ), + on_publish=revoked, + ) + assert published_rows() == before + assert checkpoints() == before_checkpoints + + +def test_task_counts_include_retained_records_but_batch_counts_only_upserts( + organizations, +): + uncertain, _ = organizations + seed(organizations) + task = SimpleNamespace( + request=SimpleNamespace(id=str(uuid4())), + name="parsers.sro_membership_check.refresh", + ) + + def refresh(**kwargs): + return refresh_sro_membership( + mode="full", + client=FixtureSite( + organizations, {uncertain.ogrn: unresolved_html(uncertain)} + ), + **kwargs, + ) + + result = _run_snapshot( + task, source="sro_membership_check", refresh=refresh, requested_by_id=None + ) + assert result["records_count"] == result["published_records_count"] == 4 + assert result["batch_published_records_count"] == 2 + assert result["retained_records_count"] == 2 + assert result["quarantined_records_count"] == 1 + job = BackgroundJob.objects.get(task_id=task.request.id) + assert job.status == JobStatus.SUCCESS + assert job.meta["published_records_count"] == 4 + log = ParserLoadLog.objects.get(pk=result["load_id"]) + assert log.status == ParserLoadLog.Status.SUCCESS + assert log.records_count == 4 + assert ( + ParserStagedRecord.objects.filter( + artifact_id=result["artifact_id"], disposition="published" + ).count() + == 2 + ) + + +def test_successful_incremental_retry_removes_retained_obsolete_memberships( + organizations, +): + uncertain, unchanged = organizations + seed(organizations) + refresh_sro_membership( + load_batch=2, + mode="full", + client=FixtureSite(organizations, {uncertain.ogrn: unresolved_html(uncertain)}), + ) + before = {row["external_id"]: row for row in published_rows()} + assert len(before) == 4 + assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists() + assert [org.uid for org in _sro_candidates("incremental")] == [uncertain.uid] + + artifact, result = refresh_sro_membership( + load_batch=3, mode="incremental", client=FixtureSite([uncertain]) + ) + after = {row["external_id"]: row for row in published_rows()} + assert result.published == len(after) == 2 + assert set(after) == {f"{uncertain.uid}:001", f"{unchanged.uid}:001"} + assert after[f"{uncertain.uid}:001"]["uid"] == before[f"{uncertain.uid}:001"]["uid"] + assert after[f"{uncertain.uid}:001"]["load_batch"] == 3 + assert after[f"{unchanged.uid}:001"] == before[f"{unchanged.uid}:001"] + assert artifact.metadata["retained_records_count"] == 0 + assert artifact.metadata["batch_published_records_count"] == 1 + assert ( + SroOrganizationLookup.objects.get(organization=uncertain).artifact_id + == artifact.uid + ) + assert _sro_candidates("incremental") == []