2cfc057e34
All checks were successful
Mostovik Backend CI/CD / Tests and lint (push) Successful in 2m55s
Mostovik Backend CI/CD / Build linux/amd64 release images (push) Successful in 3m48s
Mostovik Backend CI/CD / Deploy and verify internal main (push) Has been skipped
Mostovik Backend CI/CD / Deploy customer main (push) Has been skipped
Mostovik Backend CI/CD / Deploy dev (push) Successful in 1m54s
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""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"<html>upstream schema changed</html>",
|
||
),
|
||
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") == []
|