18971d33ec
All checks were successful
Mostovik Backend CI/CD / Tests and lint (push) Successful in 3m55s
Mostovik Backend CI/CD / Build linux/amd64 release images (push) Successful in 3m43s
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 1m45s
401 lines
14 KiB
Python
401 lines
14 KiB
Python
"""Source-specific filters and pagination operate on the published dataset."""
|
||
|
||
from uuid import UUID
|
||
|
||
import pytest
|
||
from django.urls import reverse
|
||
from organizations.models import (
|
||
BudgetProcessRegistryExtension,
|
||
GovernmentSupportExtension,
|
||
Organization,
|
||
OrganizationSourceRecord,
|
||
SroMembershipExtension,
|
||
)
|
||
from rest_framework.test import APIClient
|
||
|
||
from tests.apps.user.factories import UserFactory
|
||
|
||
|
||
@pytest.fixture
|
||
def registry_query(db):
|
||
client = APIClient()
|
||
client.force_authenticate(UserFactory.create_user())
|
||
organization = Organization.objects.create(
|
||
name="Организация реестра",
|
||
inn="7707083810",
|
||
ogrn="1027700132195",
|
||
okpo="00123456",
|
||
opk_registry_membership=False,
|
||
)
|
||
budget = BudgetProcessRegistryExtension.objects.create(organization=organization)
|
||
support = GovernmentSupportExtension.objects.create(organization=organization)
|
||
records = []
|
||
for i, region, branch in ((1, "77", True), (2, "50", False), (3, None, None)):
|
||
records.append(
|
||
OrganizationSourceRecord.objects.create(
|
||
uid=UUID(int=i),
|
||
extension=budget,
|
||
source="budget_ubpandnubp",
|
||
record_type="budget_registry_organization",
|
||
external_id=f"000{i}",
|
||
title="Бюджетная организация",
|
||
record_date="2026-09-14",
|
||
status="active" if i == 1 else "unknown",
|
||
payload={
|
||
"registry": {"code": f"REG-{i}"},
|
||
"address": {"region": {"code": region, "name": region}},
|
||
"classification": {
|
||
"organization_type": {"code": "03", "name": "Казённое"}
|
||
},
|
||
"budget": {"level": {"code": "10", "name": "Федеральный"}},
|
||
"is_separate_division": branch,
|
||
"summary": {"has_procurement_permission": i == 1},
|
||
},
|
||
)
|
||
)
|
||
for i, amount, unit in (
|
||
(4, "120.50", "RUB"),
|
||
(5, None, "hour"),
|
||
(6, "900.00", "RUB"),
|
||
):
|
||
records.append(
|
||
OrganizationSourceRecord.objects.create(
|
||
uid=UUID(int=i),
|
||
extension=support,
|
||
source="fns_sme_support_recipients",
|
||
record_type="sme_support_measure",
|
||
external_id=f"MEASURE-{i}",
|
||
title="Поддержка",
|
||
amount=amount,
|
||
record_date="2026-09-14",
|
||
payload={
|
||
"support_registry_number": f"000{i}",
|
||
"recipient_type": "1",
|
||
"sme_category": {"code": "1", "name": "Микропредприятие"},
|
||
"support_form": {"code": "01", "name": "Финансовая"},
|
||
"support_kind": {"code": "07", "name": "Субсидия"},
|
||
"provider": {"inn": "7707083810", "name": "Поставщик помощи"},
|
||
"has_violation": i == 4,
|
||
"support_sizes": [
|
||
{
|
||
"unit_code": "1" if unit == "RUB" else "3",
|
||
"unit": unit,
|
||
"value": amount or "8.00",
|
||
}
|
||
],
|
||
"registry_entry_date": "2026-09-14",
|
||
"decision_date": "2026-09-14",
|
||
"source_snapshot_date": "2026-09-14",
|
||
"support_until": "2027-09-14",
|
||
"termination_date": None,
|
||
},
|
||
)
|
||
)
|
||
return client, records
|
||
|
||
|
||
def get_records(client, **params):
|
||
return client.get(
|
||
reverse("api_v2:organizations:organization-source-records-list"), params
|
||
)
|
||
|
||
|
||
def ids(response):
|
||
assert response.status_code == 200, response.data
|
||
return [UUID(row["uid"]).int for row in response.data["data"]]
|
||
|
||
|
||
def test_registry_scope_includes_own_non_opk_and_typed_pagination(registry_query):
|
||
client, _ = registry_query
|
||
response = get_records(
|
||
client, source_group="budget_process_registry", ordering="uid"
|
||
)
|
||
assert ids(response) == [1, 2, 3]
|
||
assert response.data["meta"]["pagination"] == {
|
||
"page": 1,
|
||
"page_size": 50,
|
||
"total_count": 3,
|
||
"total_pages": 1,
|
||
"has_next": False,
|
||
"has_previous": False,
|
||
}
|
||
empty = get_records(
|
||
client, source_group="budget_process_registry", region_code="99"
|
||
)
|
||
assert ids(empty) == []
|
||
assert empty.data["meta"]["pagination"]["total_pages"] == 0
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("params", "expected"),
|
||
[
|
||
({"region_code": "77"}, [1]),
|
||
({"is_branch": "false"}, [2]),
|
||
({"has_procurement_permission": "true"}, [1]),
|
||
({"organization_type": "03", "budget_level": "10"}, [1, 2, 3]),
|
||
({"search": "REG-2"}, [2]),
|
||
],
|
||
)
|
||
def test_budget_filters_and_registry_search(registry_query, params, expected):
|
||
client, _ = registry_query
|
||
assert (
|
||
ids(
|
||
get_records(
|
||
client, source_group="budget_process_registry", ordering="uid", **params
|
||
)
|
||
)
|
||
== expected
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("params", "expected"),
|
||
[
|
||
({"support_unit": "hour"}, [5]),
|
||
({"amount_from": "120.50", "amount_to": "120.50"}, [4]),
|
||
({"has_violation": "true"}, [4]),
|
||
(
|
||
{"sme_category": "1", "support_form_code": "01", "support_kind_code": "07"},
|
||
[4, 5, 6],
|
||
),
|
||
(
|
||
{"support_form": "01", "support_kind": "07", "provider_inn": "7707083810"},
|
||
[4, 5, 6],
|
||
),
|
||
(
|
||
{"registry_entry_from": "2026-09-14", "support_until_to": "2027-09-14"},
|
||
[4, 5, 6],
|
||
),
|
||
({"search": "Поставщик помощи"}, [4, 5, 6]),
|
||
({"termination_from": "2020-01-01"}, []),
|
||
],
|
||
)
|
||
def test_sme_filters_units_amounts_dates_and_search(registry_query, params, expected):
|
||
client, _ = registry_query
|
||
assert (
|
||
ids(
|
||
get_records(
|
||
client, source_group="government_support", ordering="uid", **params
|
||
)
|
||
)
|
||
== expected
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"ordering", ["payload__is_separate_division", "-payload__is_separate_division"]
|
||
)
|
||
def test_budget_boolean_ordering_null_last_and_uid_ties(registry_query, ordering):
|
||
client, _ = registry_query
|
||
expected = [2, 1, 3] if not ordering.startswith("-") else [1, 2, 3]
|
||
response = get_records(
|
||
client, source_group="budget_process_registry", ordering=ordering
|
||
)
|
||
assert ids(response) == expected
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"params",
|
||
[
|
||
{"unknown_filter": "yes"},
|
||
{"is_branch": "perhaps"},
|
||
{"provider_inn": "123"},
|
||
{"organization": "not-a-uuid"},
|
||
{"search": "x" * 201},
|
||
{"amount_from": "NaN"},
|
||
{"amount_from": "2", "amount_to": "1"},
|
||
{"registry_entry_from": "2026-02-30"},
|
||
{"support_until_from": "2026-02-01", "support_until_to": "2026-01-01"},
|
||
{"support_form_code": "1", "support_form": "2"},
|
||
{"support_unit": "dollar"},
|
||
{"page_size": "101"},
|
||
{"ordering": "payload__secret"},
|
||
],
|
||
)
|
||
def test_invalid_filters_return_machine_readable_400(registry_query, params):
|
||
client, _ = registry_query
|
||
response = get_records(client, source_group="government_support", **params)
|
||
assert response.status_code == 400, response.data
|
||
assert response.data["errors"][0]["code"]
|
||
|
||
|
||
def test_filtering_and_ordering_precede_pagination(registry_query):
|
||
client, _ = registry_query
|
||
response = get_records(
|
||
client,
|
||
source_group="government_support",
|
||
support_unit="RUB",
|
||
ordering="amount",
|
||
page_size=1,
|
||
page=2,
|
||
)
|
||
assert ids(response) == [6]
|
||
assert response.data["meta"]["pagination"]["total_count"] == 2
|
||
|
||
|
||
@pytest.mark.parametrize("field", ["name", "full_name", "inn", "ogrn", "okpo"])
|
||
@pytest.mark.parametrize("prefix", ["organization__", "extension__organization__"])
|
||
@pytest.mark.parametrize("descending", [False, True])
|
||
def test_organization_ordering_keeps_public_nulls_last(
|
||
registry_query, field, prefix, descending
|
||
):
|
||
client, records = registry_query
|
||
Organization.objects.filter(pk=records[0].extension.organization_id).update(
|
||
**{field: ""}
|
||
)
|
||
organization = Organization.objects.create(
|
||
name="Заполненная организация",
|
||
full_name="Полное наименование",
|
||
inn="7700000001",
|
||
ogrn="1027700000001",
|
||
okpo="00000001",
|
||
)
|
||
extension = BudgetProcessRegistryExtension.objects.create(organization=organization)
|
||
OrganizationSourceRecord.objects.create(
|
||
uid=UUID(int=100),
|
||
extension=extension,
|
||
source="budget_ubpandnubp",
|
||
record_type="budget_registry_organization",
|
||
external_id="filled",
|
||
payload={},
|
||
)
|
||
response = get_records(
|
||
client,
|
||
source_group="budget_process_registry",
|
||
ordering=("-" if descending else "") + prefix + field,
|
||
)
|
||
assert ids(response) == [100, 1, 2, 3]
|
||
assert response.data["data"][0]["organization"][field]
|
||
assert [row["organization"][field] for row in response.data["data"][1:]] == [
|
||
None,
|
||
None,
|
||
None,
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("field", ["name", "full_name"])
|
||
@pytest.mark.parametrize("prefix", ["organization__", "extension__organization__"])
|
||
@pytest.mark.parametrize("descending", [False, True])
|
||
def test_sro_ordering_uses_displayed_names_before_pagination(
|
||
registry_query, field, prefix, descending
|
||
):
|
||
client, _ = registry_query
|
||
for uid, canonical, displayed in (
|
||
(101, "Z", "Alpha"),
|
||
(102, "A", "Zulu"),
|
||
(103, "Middle", ""),
|
||
(104, "", None),
|
||
(105, "Y", "Alpha"),
|
||
):
|
||
organization = Organization.objects.create(
|
||
name=canonical or "Temporary name", full_name=canonical
|
||
)
|
||
Organization.objects.filter(pk=organization.pk).update(**{field: canonical})
|
||
extension = SroMembershipExtension.objects.create(organization=organization)
|
||
OrganizationSourceRecord.objects.create(
|
||
uid=UUID(int=uid),
|
||
extension=extension,
|
||
source="sro_membership_check",
|
||
record_type="sro_membership",
|
||
external_id=str(uid),
|
||
payload={
|
||
"source_organization": {field: displayed},
|
||
"membership_status": "active",
|
||
"membership_status_raw": "active",
|
||
"region": "Москва",
|
||
"sro_id": str(uid),
|
||
"sro_name": "СРО",
|
||
"sro_url": f"https://reestr-sro.ru/sro/{uid}",
|
||
"admission_date": None,
|
||
"source_registry_date": None,
|
||
},
|
||
)
|
||
actual_ids, names = [], []
|
||
for page in (1, 2, 3):
|
||
response = get_records(
|
||
client,
|
||
source_group="sro_membership",
|
||
ordering=("-" if descending else "") + prefix + field,
|
||
page_size=2,
|
||
page=page,
|
||
)
|
||
actual_ids.extend(ids(response))
|
||
names.extend(row["organization"][field] for row in response.data["data"])
|
||
assert actual_ids == (
|
||
[102, 103, 101, 105, 104] if descending else [101, 105, 103, 102, 104]
|
||
)
|
||
assert names == (
|
||
["Zulu", "Middle", "Alpha", "Alpha", None]
|
||
if descending
|
||
else ["Alpha", "Alpha", "Middle", "Zulu", None]
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("params", "code"),
|
||
[
|
||
({"organization": "not-a-uuid"}, "invalid_uuid"),
|
||
({"unknown_filter": "yes"}, "unknown_filter"),
|
||
({"ordering": "payload__secret"}, "invalid_ordering"),
|
||
({"is_branch": "perhaps"}, "invalid_boolean"),
|
||
],
|
||
)
|
||
def test_detail_rejects_invalid_query_with_typed_400(registry_query, params, code):
|
||
client, records = registry_query
|
||
response = client.get(
|
||
reverse(
|
||
"api_v2:organizations:organization-source-records-detail",
|
||
kwargs={"uid": records[0].uid},
|
||
),
|
||
params,
|
||
)
|
||
assert response.status_code == 400, response.data
|
||
assert response.data["success"] is False
|
||
assert response.data["data"] is None
|
||
assert response.data["errors"][0]["code"] == code
|
||
|
||
|
||
def test_missing_detail_returns_source_record_not_found(registry_query):
|
||
client, _ = registry_query
|
||
response = client.get(
|
||
reverse(
|
||
"api_v2:organizations:organization-source-records-detail",
|
||
kwargs={"uid": UUID(int=999)},
|
||
)
|
||
)
|
||
assert response.status_code == 404, response.data
|
||
assert response.data["data"] is None
|
||
assert response.data["errors"][0]["code"] == "source_record_not_found"
|
||
|
||
|
||
def test_sme_published_filter_includes_legacy_empty_status_only_for_sme(registry_query):
|
||
client, records = registry_query
|
||
OrganizationSourceRecord.objects.filter(uid=records[3].uid).update(
|
||
status="published"
|
||
)
|
||
OrganizationSourceRecord.objects.filter(uid=records[0].uid).update(status="")
|
||
response = get_records(client, status="published", ordering="uid")
|
||
assert ids(response) == [4, 5, 6]
|
||
assert {row["status"] for row in response.data["data"]} == {"published"}
|
||
|
||
|
||
def test_legacy_source_preserves_empty_text_ordering(registry_query):
|
||
client, records = registry_query
|
||
organization = Organization.objects.create(name="Other", full_name="Filled")
|
||
extension = BudgetProcessRegistryExtension.objects.create(organization=organization)
|
||
OrganizationSourceRecord.objects.create(
|
||
uid=UUID(int=100),
|
||
extension=extension,
|
||
source="legacy",
|
||
record_type="legacy",
|
||
external_id="filled",
|
||
payload={},
|
||
)
|
||
OrganizationSourceRecord.objects.filter(uid=records[0].uid).update(source="legacy")
|
||
response = get_records(client, source="legacy", ordering="organization__full_name")
|
||
assert ids(response) == [1, 100]
|
||
assert [row["organization"]["full_name"] for row in response.data["data"]] == [
|
||
"",
|
||
"Filled",
|
||
]
|