Aleksandr Meshchriakov
2644822de5
All checks were successful
CI/CD Pipeline / Quality Gate (push) Successful in 32s
CI/CD Pipeline / Build and Push Images (push) Successful in 1s
CI/CD Pipeline / Internal Notify (push) Successful in 1s
CI/CD Pipeline / Deploy Dev via Compose (push) Successful in 1s
CI/CD Pipeline / Quality Gate (pull_request) Successful in 30s
CI/CD Pipeline / Build and Push Images (pull_request) Successful in 1s
CI/CD Pipeline / Internal Notify (pull_request) Successful in 1s
CI/CD Pipeline / Deploy Dev via Compose (pull_request) Successful in 1s
118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
from datetime import date
|
||
|
||
from apps.parsers import tasks as parser_tasks
|
||
from apps.parsers.checko_collection import claim_monthly_collection, finish_collection
|
||
from apps.parsers.models import CheckoCollectionAttempt
|
||
from django.test import TestCase
|
||
|
||
from tests.apps.parsers.organization_helpers import create_directory_organization
|
||
|
||
|
||
class CheckoCollectionClaimTest(TestCase):
|
||
def setUp(self):
|
||
self.organization = create_directory_organization(
|
||
pn_name='ООО "Месячный лимит"',
|
||
mn_ogrn=1027700000199,
|
||
mn_inn=7701000199,
|
||
in_kpp=770101001,
|
||
mn_okpo="12345678",
|
||
)
|
||
|
||
def test_claim_is_unique_per_organization_source_and_calendar_month(self):
|
||
first = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.ARBITRATION,
|
||
at=date(2026, 7, 1),
|
||
)
|
||
|
||
duplicate = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.ARBITRATION,
|
||
at=date(2026, 7, 31),
|
||
)
|
||
other_source = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.INSPECTIONS,
|
||
at=date(2026, 7, 31),
|
||
)
|
||
next_month = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.ARBITRATION,
|
||
at=date(2026, 8, 1),
|
||
)
|
||
|
||
self.assertIsNotNone(first)
|
||
self.assertIsNone(duplicate)
|
||
self.assertIsNotNone(other_source)
|
||
self.assertIsNotNone(next_month)
|
||
|
||
def test_failed_attempt_keeps_monthly_claim(self):
|
||
attempt = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.CONTRACTS,
|
||
at=date(2026, 7, 19),
|
||
)
|
||
assert attempt is not None
|
||
finish_collection(attempt, records_count=0, error="quota exhausted")
|
||
|
||
repeated = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.CONTRACTS,
|
||
at=date(2026, 7, 20),
|
||
)
|
||
|
||
attempt.refresh_from_db()
|
||
self.assertIsNone(repeated)
|
||
self.assertEqual(attempt.status, CheckoCollectionAttempt.Status.FAILED)
|
||
self.assertEqual(attempt.error_message, "quota exhausted")
|
||
|
||
def test_due_filter_is_applied_before_lookup_limit(self):
|
||
self.organization.opk_registry_membership = True
|
||
self.organization.save(update_fields=["opk_registry_membership"])
|
||
second = create_directory_organization(
|
||
pn_name='ООО "Следующая организация"',
|
||
mn_ogrn=1027700000299,
|
||
mn_inn=7701000299,
|
||
in_kpp=770101001,
|
||
mn_okpo="87654321",
|
||
)
|
||
second.opk_registry_membership = True
|
||
second.save(update_fields=["opk_registry_membership"])
|
||
claimed = claim_monthly_collection(
|
||
organization_id=str(self.organization.id),
|
||
source=CheckoCollectionAttempt.Source.INSPECTIONS,
|
||
)
|
||
self.assertIsNotNone(claimed)
|
||
|
||
targets = parser_tasks._active_registry_lookup_targets(
|
||
limit=1,
|
||
checko_source=CheckoCollectionAttempt.Source.INSPECTIONS,
|
||
)
|
||
|
||
self.assertEqual(
|
||
[target.organization_id for target in targets], [str(second.id)]
|
||
)
|
||
|
||
def test_lookup_targets_include_corporation_members_outside_opk(self):
|
||
self.organization.gk_name = 'Госкорпорация "Росатом"'
|
||
self.organization.save(update_fields=["gk_name"])
|
||
outside_scope = create_directory_organization(
|
||
pn_name='ООО "Вне контура обмена"',
|
||
mn_ogrn=1027700000399,
|
||
mn_inn=7701000399,
|
||
in_kpp=770101001,
|
||
mn_okpo="11223344",
|
||
)
|
||
|
||
targets = parser_tasks._active_registry_lookup_targets(
|
||
organization_ids=[
|
||
str(self.organization.id),
|
||
str(outside_scope.id),
|
||
],
|
||
)
|
||
|
||
self.assertEqual(
|
||
[target.organization_id for target in targets],
|
||
[str(self.organization.id)],
|
||
)
|