From a198f7e960694b6888e695a9238b9c11b4a02999 Mon Sep 17 00:00:00 2001 From: Aleksandr Meshchriakov Date: Sun, 9 Aug 2026 12:23:36 +0200 Subject: [PATCH] feat: update finance exchange and source refresh jobs --- src/apps/core/models.py | 4 ++ src/apps/core/services.py | 18 +++---- src/apps/exchange/state_corp_services.py | 5 ++ src/apps/parsers/frontend_compat.py | 35 ++++++++++--- .../0030_extend_stale_parser_job_timeouts.py | 39 +++++++++++++++ src/apps/parsers/serializers.py | 1 + src/apps/parsers/services.py | 19 +++++-- src/apps/parsers/source_cards.py | 45 ++++++++++++----- src/apps/parsers/tasks.py | 19 +++++-- src/apps/parsers/views.py | 14 ++++-- src/organizations/test_companies.py | 5 +- tests/apps/core/test_background_jobs.py | 50 ++++++++++++++++++- .../apps/exchange/test_state_corp_services.py | 6 +++ .../test_test_companies_commands.py | 9 ++++ tests/apps/parsers/test_services.py | 10 ++-- .../apps/parsers/test_source_cards_service.py | 7 +-- tests/apps/parsers/test_sources_api_e2e.py | 5 ++ tests/apps/parsers/test_tasks.py | 10 +++- tests/apps/parsers/test_views.py | 17 +++++++ 19 files changed, 266 insertions(+), 52 deletions(-) create mode 100644 src/apps/parsers/migrations/0030_extend_stale_parser_job_timeouts.py diff --git a/src/apps/core/models.py b/src/apps/core/models.py index 9b4cb7c..b258920 100644 --- a/src/apps/core/models.py +++ b/src/apps/core/models.py @@ -169,12 +169,16 @@ class BackgroundJob(TimestampMixin, models.Model): self.status = JobStatus.SUCCESS self.progress = 100 self.result = result + self.error = "" + self.traceback = "" self.completed_at = timezone.now() self.save( update_fields=[ "status", "progress", "result", + "error", + "traceback", "completed_at", "updated_at", ] diff --git a/src/apps/core/services.py b/src/apps/core/services.py index 1d2cd73..54c5821 100644 --- a/src/apps/core/services.py +++ b/src/apps/core/services.py @@ -752,21 +752,21 @@ class BackgroundJobService(BaseReadOnlyService): cls, *, max_age_minutes: int, + pending_max_age_minutes: int = 24 * 60, task_names: set[str] | None = None, meta_sources: set[str] | None = None, ) -> int: """Mark old active jobs as failed after worker restarts or hard kills.""" from apps.core.models import JobStatus - cutoff = timezone.now() - timedelta(minutes=max_age_minutes) - queryset = ( - cls.get_queryset() - .filter( - status__in=[JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY], - ) - .filter( - Q(started_at__isnull=False, started_at__lt=cutoff) - | Q(started_at__isnull=True, created_at__lt=cutoff) + now = timezone.now() + cutoff = now - timedelta(minutes=max_age_minutes) + pending_cutoff = now - timedelta(minutes=pending_max_age_minutes) + queryset = cls.get_queryset().filter( + Q(status=JobStatus.PENDING, created_at__lt=pending_cutoff) + | Q( + status__in=[JobStatus.STARTED, JobStatus.RETRY], + updated_at__lt=cutoff, ) ) if task_names: diff --git a/src/apps/exchange/state_corp_services.py b/src/apps/exchange/state_corp_services.py index a6ecf75..3da2794 100644 --- a/src/apps/exchange/state_corp_services.py +++ b/src/apps/exchange/state_corp_services.py @@ -785,6 +785,10 @@ class StateCorpExchangeService: if not case_number or decision_date is None: continue + claim_amount = cls._serialize_decimal(record.amount) + if claim_amount is None: + claim_amount = str(payload.get("claim_amount") or "").strip() or None + items.append( { "organization_inn": cls._digits(record.inn), @@ -795,6 +799,7 @@ class StateCorpExchangeService: ).strip(), "status": str(payload.get("status") or record.status or "").strip(), "decision_date": decision_date.isoformat(), + "claim_amount": claim_amount, } ) return items diff --git a/src/apps/parsers/frontend_compat.py b/src/apps/parsers/frontend_compat.py index 40fad2f..0f8b67a 100644 --- a/src/apps/parsers/frontend_compat.py +++ b/src/apps/parsers/frontend_compat.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from datetime import timedelta from typing import Any +from apps.core.models import JobStatus from apps.core.response import api_error_response, api_response from apps.core.services import BackgroundJobService from apps.parsers.models import ( @@ -46,7 +47,8 @@ SYSTEM_LOGS_TAG = "System Logs" ACTIVE_JOB_STATUSES = {"pending", "started", "retry"} SUCCESS_LOAD_STATUSES = {"success", "skipped"} ERROR_LOAD_STATUSES = {"failed", "failure", "error"} -STALE_ACTIVE_MAX_AGE_MINUTES = 90 +STALE_ACTIVE_MAX_AGE_MINUTES = 4 * 60 +STALE_PENDING_MAX_AGE_MINUTES = 24 * 60 PARSING_SETTINGS_CACHE_KEY = "parsers:frontend_compat:parsing_settings" PARSING_SETTINGS_FIELDS = { @@ -297,15 +299,34 @@ def _active_tasks_for_definition( for source_key in definition.source_keys if source_key in PARSER_SOURCES ] + now = timezone.now() + active_cutoff = now - timedelta( + minutes=int( + getattr( + settings, + "PARSER_STALE_LOAD_MAX_AGE_MINUTES", + STALE_ACTIVE_MAX_AGE_MINUTES, + ) + ) + ) + pending_cutoff = now - timedelta( + minutes=int( + getattr( + settings, + "PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES", + STALE_PENDING_MAX_AGE_MINUTES, + ) + ) + ) queryset = ( BackgroundJobService.get_queryset() + .filter(task_name__in=task_names) .filter( - task_name__in=task_names, - status__in=ACTIVE_JOB_STATUSES, - ) - .filter( - Q(started_at__isnull=False, started_at__gte=_stale_cutoff()) - | Q(started_at__isnull=True, created_at__gte=_stale_cutoff()) + Q(status=JobStatus.PENDING, created_at__gte=pending_cutoff) + | Q( + status__in=[JobStatus.STARTED, JobStatus.RETRY], + updated_at__gte=active_cutoff, + ) ) ) return [_serialize_active_job(job) for job in queryset.order_by("-created_at")[:10]] diff --git a/src/apps/parsers/migrations/0030_extend_stale_parser_job_timeouts.py b/src/apps/parsers/migrations/0030_extend_stale_parser_job_timeouts.py new file mode 100644 index 0000000..8431b27 --- /dev/null +++ b/src/apps/parsers/migrations/0030_extend_stale_parser_job_timeouts.py @@ -0,0 +1,39 @@ +import json + +from django.db import migrations + +CLEANUP_TASK_NAME = "parser:cleanup-stale-loads" +ACTIVE_MAX_AGE_MINUTES = 4 * 60 +PENDING_MAX_AGE_MINUTES = 24 * 60 + + +def extend_stale_parser_job_timeouts(apps, schema_editor): + PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + PeriodicTask.objects.filter(name=CLEANUP_TASK_NAME).update( + kwargs=json.dumps( + { + "max_age_minutes": ACTIVE_MAX_AGE_MINUTES, + "pending_max_age_minutes": PENDING_MAX_AGE_MINUTES, + } + ) + ) + + +def restore_stale_parser_job_timeouts(apps, schema_editor): + PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask") + PeriodicTask.objects.filter(name=CLEANUP_TASK_NAME).update( + kwargs=json.dumps({"max_age_minutes": 90}) + ) + + +class Migration(migrations.Migration): + dependencies = [ + ("parsers", "0029_neutralize_external_collection_labels"), + ] + + operations = [ + migrations.RunPython( + extend_stale_parser_job_timeouts, + reverse_code=restore_stale_parser_job_timeouts, + ), + ] diff --git a/src/apps/parsers/serializers.py b/src/apps/parsers/serializers.py index 430a173..092c831 100644 --- a/src/apps/parsers/serializers.py +++ b/src/apps/parsers/serializers.py @@ -945,6 +945,7 @@ class SourceCardItemSerializer(serializers.Serializer): """Подисточник внутри агрегированной карточки.""" code = serializers.CharField(read_only=True) + refresh_key = serializers.CharField(read_only=True) title = serializers.CharField(read_only=True) description = serializers.CharField(read_only=True) parser_source = serializers.CharField(read_only=True, allow_null=True) diff --git a/src/apps/parsers/services.py b/src/apps/parsers/services.py index 88f5609..0434992 100644 --- a/src/apps/parsers/services.py +++ b/src/apps/parsers/services.py @@ -419,11 +419,14 @@ class ParserLoadLogService(BaseService[ParserLoadLog]): cls, *, max_age_minutes: int, + pending_max_age_minutes: int = 24 * 60, ) -> int: """Закрыть зависшие in_progress логи без живой свежей BackgroundJob.""" from apps.core.models import BackgroundJob, JobStatus - cutoff = timezone.now() - timedelta(minutes=max_age_minutes) + now = timezone.now() + cutoff = now - timedelta(minutes=max_age_minutes) + pending_cutoff = now - timedelta(minutes=pending_max_age_minutes) stale_logs = list( cls.model.objects.filter( status=ParserLoadLog.Status.IN_PROGRESS, @@ -442,8 +445,11 @@ class ParserLoadLogService(BaseService[ParserLoadLog]): meta__source=log.source, ).filter(Q(meta__batch_id=log.batch_id) | Q(meta__batch_id__isnull=True)) fresh_jobs = active_jobs.filter( - Q(started_at__isnull=False, started_at__gte=cutoff) - | Q(started_at__isnull=True, created_at__gte=cutoff) + Q(status=JobStatus.PENDING, created_at__gte=pending_cutoff) + | Q( + status__in=[JobStatus.STARTED, JobStatus.RETRY], + updated_at__gte=cutoff, + ) ) if fresh_jobs.exists(): continue @@ -451,8 +457,11 @@ class ParserLoadLogService(BaseService[ParserLoadLog]): cls.mark_failed(log, stale_message) updated += 1 stale_jobs = active_jobs.filter( - Q(started_at__isnull=False, started_at__lt=cutoff) - | Q(started_at__isnull=True, created_at__lt=cutoff) + Q(status=JobStatus.PENDING, created_at__lt=pending_cutoff) + | Q( + status__in=[JobStatus.STARTED, JobStatus.RETRY], + updated_at__lt=cutoff, + ) ) for job in stale_jobs.order_by("created_at"): job.fail(error=stale_message) diff --git a/src/apps/parsers/source_cards.py b/src/apps/parsers/source_cards.py index 29a7ac6..8fe2225 100644 --- a/src/apps/parsers/source_cards.py +++ b/src/apps/parsers/source_cards.py @@ -31,7 +31,8 @@ from rest_framework.exceptions import ValidationError SUCCESSFUL_LOAD_STATUSES = {"success", "skipped"} ACTIVE_JOB_STATUSES = [JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY] -STALE_ACTIVE_MAX_AGE_MINUTES = 90 +STALE_ACTIVE_MAX_AGE_MINUTES = 4 * 60 +STALE_PENDING_MAX_AGE_MINUTES = 24 * 60 SOURCE_CARD_STATS_CACHE_TIMEOUT_SECONDS = 7 * 24 * 60 * 60 @@ -55,6 +56,7 @@ class SourceItemDefinition: title: str description: str parser_source: str | None = None + refresh_key: str | None = None @dataclass(frozen=True) @@ -218,6 +220,7 @@ SOURCE_CARD_DEFINITIONS: tuple[SourceCardDefinition, ...] = ( "Реестр промышленной продукции, произведенной на территории РФ." ), parser_source=ParserLoadLog.Source.INDUSTRIAL_PRODUCTS, + refresh_key="mpt_products", ), SourceItemDefinition( code="manufactures", @@ -769,17 +772,12 @@ class SourceCardService: active_tasks_by_slug: dict[str, list[dict[str, Any]]] = { slug: [] for slug in task_names_by_slug } - cutoff = cls._stale_cutoff() queryset = ( BackgroundJobService.get_queryset() .filter( task_name__in=list(slugs_by_task_name), - status__in=ACTIVE_JOB_STATUSES, - ) - .filter( - Q(started_at__isnull=False, started_at__gte=cutoff) - | Q(started_at__isnull=True, created_at__gte=cutoff) ) + .filter(cls._fresh_active_job_filter()) .order_by("-created_at") ) @@ -1157,6 +1155,7 @@ class SourceCardService: return { "code": item.code, + "refresh_key": item.refresh_key or item.code, "title": item.title, "description": item.description, "parser_source": item.parser_source, @@ -1288,17 +1287,12 @@ class SourceCardService: def _get_active_tasks( cls, definition: SourceCardDefinition ) -> list[dict[str, Any]]: - cutoff = cls._stale_cutoff() queryset = ( BackgroundJobService.get_queryset() .filter( task_name__in=definition.task_names, - status__in=ACTIVE_JOB_STATUSES, - ) - .filter( - Q(started_at__isnull=False, started_at__gte=cutoff) - | Q(started_at__isnull=True, created_at__gte=cutoff) ) + .filter(cls._fresh_active_job_filter()) ) return [ cls._serialize_job(job) for job in queryset.order_by("-created_at")[:10] @@ -1349,6 +1343,31 @@ class SourceCardService: ) return timezone.now() - timedelta(minutes=max_age_minutes) + @classmethod + def _fresh_active_job_filter(cls) -> Q: + now = timezone.now() + active_max_age_minutes = int( + getattr( + settings, + "PARSER_STALE_LOAD_MAX_AGE_MINUTES", + STALE_ACTIVE_MAX_AGE_MINUTES, + ) + ) + pending_max_age_minutes = int( + getattr( + settings, + "PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES", + STALE_PENDING_MAX_AGE_MINUTES, + ) + ) + return Q( + status=JobStatus.PENDING, + created_at__gte=now - timedelta(minutes=pending_max_age_minutes), + ) | Q( + status__in=[JobStatus.STARTED, JobStatus.RETRY], + updated_at__gte=now - timedelta(minutes=active_max_age_minutes), + ) + @classmethod def _is_stale_load(cls, latest_load: ParserLoadLog | None) -> bool: if latest_load is None or latest_load.status != "in_progress": diff --git a/src/apps/parsers/tasks.py b/src/apps/parsers/tasks.py index f129492..21bada0 100644 --- a/src/apps/parsers/tasks.py +++ b/src/apps/parsers/tasks.py @@ -88,7 +88,8 @@ REGISTRY_INSPECTIONS_CHECKO_LIMIT = 1000 REGISTRY_CONTRACTS_CHECKO_LIMIT = 1000 REGISTRY_ENRICHMENT_BATCH_SIZE = 250 FSTEC_CHECKO_IDENTITY_LOOKUP_LIMIT = 1000 -PARSER_STALE_LOAD_MAX_AGE_MINUTES = 90 +PARSER_STALE_LOAD_MAX_AGE_MINUTES = 4 * 60 +PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES = 24 * 60 PARSER_SOFT_TIME_LIMIT_SECONDS = 15 * 60 PARSER_TIME_LIMIT_SECONDS = 20 * 60 INDUSTRIAL_PRODUCTS_SOFT_TIME_LIMIT_SECONDS = 45 * 60 @@ -3604,7 +3605,10 @@ def parse_fstec_registers( @shared_task -def cleanup_stale_parser_loads(max_age_minutes: int | None = None) -> dict: +def cleanup_stale_parser_loads( + max_age_minutes: int | None = None, + pending_max_age_minutes: int | None = None, +) -> dict: """Закрыть stale in_progress загрузки и jobs после рестартов worker/deploy.""" if max_age_minutes is None: max_age_minutes = getattr( @@ -3612,14 +3616,22 @@ def cleanup_stale_parser_loads(max_age_minutes: int | None = None) -> dict: "PARSER_STALE_LOAD_MAX_AGE_MINUTES", PARSER_STALE_LOAD_MAX_AGE_MINUTES, ) + if pending_max_age_minutes is None: + pending_max_age_minutes = getattr( + settings, + "PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES", + PARSER_STALE_PENDING_JOB_MAX_AGE_MINUTES, + ) source_values = {descriptor.source for descriptor in PARSER_SOURCES.values()} task_names = {descriptor.task_name for descriptor in PARSER_SOURCES.values()} task_names.add("apps.parsers.tasks.scan_fns_directory") marked_failed = ParserLoadLogService.mark_stale_in_progress_failed( - max_age_minutes=int(max_age_minutes) + max_age_minutes=int(max_age_minutes), + pending_max_age_minutes=int(pending_max_age_minutes), ) marked_jobs_failed = BackgroundJobService.mark_stale_active_jobs_failed( max_age_minutes=int(max_age_minutes), + pending_max_age_minutes=int(pending_max_age_minutes), task_names=task_names, meta_sources=source_values, ) @@ -3628,6 +3640,7 @@ def cleanup_stale_parser_loads(max_age_minutes: int | None = None) -> dict: "marked_failed": marked_failed, "marked_jobs_failed": marked_jobs_failed, "max_age_minutes": int(max_age_minutes), + "pending_max_age_minutes": int(pending_max_age_minutes), } diff --git a/src/apps/parsers/views.py b/src/apps/parsers/views.py index e5ab40f..89fff37 100644 --- a/src/apps/parsers/views.py +++ b/src/apps/parsers/views.py @@ -151,6 +151,10 @@ TASKS_BY_NAME = { "apps.parsers.tasks.parse_trudvsem_vacancies": tasks.parse_trudvsem_vacancies, } +PARSER_SOURCE_ALIASES = { + "industrial_products": "mpt_products", +} + class MultipartFormSwaggerAutoSchema(SwaggerAutoSchema): """Document mixed JSON/multipart upload endpoints as form-data in Swagger.""" @@ -2642,21 +2646,25 @@ class ParserRunView(APIView): permission_classes = [IsAuthenticated] def post(self, request: Request, source_key: str): - descriptor = PARSER_SOURCES.get(source_key) + canonical_source_key = PARSER_SOURCE_ALIASES.get(source_key, source_key) + descriptor = PARSER_SOURCES.get(canonical_source_key) if descriptor is None: return _source_not_found_response(source_key) serializer = ParserRunRequestSerializer(data=request.data) serializer.is_valid(raise_exception=True) task = TASKS_BY_NAME[descriptor.task_name] task_kwargs = build_task_kwargs( - source_key, serializer.validated_data, request.user.id + canonical_source_key, serializer.validated_data, request.user.id ) task_id = str(uuid.uuid4()) BackgroundJobService.create_job( task_id=task_id, task_name=descriptor.task_name, user_id=request.user.id, - meta={"source_key": source_key, "source": descriptor.source}, + meta={ + "source_key": canonical_source_key, + "source": descriptor.source, + }, ) async_result = task.apply_async(kwargs=task_kwargs, task_id=task_id) return api_response( diff --git a/src/organizations/test_companies.py b/src/organizations/test_companies.py index 8a48023..09353cc 100644 --- a/src/organizations/test_companies.py +++ b/src/organizations/test_companies.py @@ -661,6 +661,7 @@ class TestCompanyDatasetService: @staticmethod def _refresh_financial_lines(*, record, index: int) -> None: + report_year = timezone.localdate().year expected = { ("1", "1600", "Баланс (актив)", 10_000_000 + index * 100_000), ("1", "1300", "Капитал и резервы", 4_000_000 + index * 50_000), @@ -669,12 +670,12 @@ class TestCompanyDatasetService: } expected_keys = [] for form_code, line_code, line_name, period_end in expected: - expected_keys.append((form_code, line_code, 2025)) + expected_keys.append((form_code, line_code, report_year)) OrganizationSourceFinancialLine.objects.update_or_create( source_record=record, form_code=form_code, line_code=line_code, - year=2025, + year=report_year, defaults={ "line_name": line_name, "period_start": period_end - 100_000, diff --git a/tests/apps/core/test_background_jobs.py b/tests/apps/core/test_background_jobs.py index a56d022..9ee83b8 100644 --- a/tests/apps/core/test_background_jobs.py +++ b/tests/apps/core/test_background_jobs.py @@ -61,6 +61,20 @@ class BackgroundJobModelTest(TestCase): self.assertEqual(job.result, result) self.assertIsNotNone(job.completed_at) + def test_complete_clears_previous_failure_details(self): + job = BackgroundJob.objects.create( + task_id=fake.uuid4(), + task_name="test.task", + error="stale error", + traceback="stale traceback", + ) + + job.complete(result={"processed": 1}) + + self.assertEqual(job.status, JobStatus.SUCCESS) + self.assertEqual(job.error, "") + self.assertEqual(job.traceback, "") + def test_fail(self): """Тест завершения с ошибкой.""" job = BackgroundJob.objects.create( @@ -288,7 +302,7 @@ class BackgroundJobServiceTest(TestCase): task_name="apps.other.tasks.task", meta={"source": "industrial_products"}, ) - old_timestamp = timezone.now() - timedelta(hours=3) + old_timestamp = timezone.now() - timedelta(hours=25) BackgroundJob.objects.filter( task_id__in=[stale_job.task_id, unrelated_job.task_id] ).update(created_at=old_timestamp, updated_at=timezone.now()) @@ -307,3 +321,37 @@ class BackgroundJobServiceTest(TestCase): self.assertIn("Stale background job", stale_job.error) self.assertEqual(fresh_job.status, JobStatus.PENDING) self.assertEqual(unrelated_job.status, JobStatus.PENDING) + + def test_mark_stale_active_jobs_uses_updated_at_for_started_job(self): + job = BackgroundJobService.create_job( + task_id="job-heartbeat", + task_name="apps.parsers.tasks.parse_industrial_products", + meta={"source": "industrial_products"}, + ) + job.mark_started() + old_timestamp = timezone.now() - timedelta(hours=5) + BackgroundJob.objects.filter(pk=job.pk).update( + started_at=old_timestamp, + updated_at=timezone.now(), + ) + + updated = BackgroundJobService.mark_stale_active_jobs_failed( + max_age_minutes=240, + task_names={"apps.parsers.tasks.parse_industrial_products"}, + meta_sources={"industrial_products"}, + ) + + self.assertEqual(updated, 0) + job.refresh_from_db() + self.assertEqual(job.status, JobStatus.STARTED) + + BackgroundJob.objects.filter(pk=job.pk).update(updated_at=old_timestamp) + updated = BackgroundJobService.mark_stale_active_jobs_failed( + max_age_minutes=240, + task_names={"apps.parsers.tasks.parse_industrial_products"}, + meta_sources={"industrial_products"}, + ) + + self.assertEqual(updated, 1) + job.refresh_from_db() + self.assertEqual(job.status, JobStatus.FAILURE) diff --git a/tests/apps/exchange/test_state_corp_services.py b/tests/apps/exchange/test_state_corp_services.py index 7b60287..7c6d4bd 100644 --- a/tests/apps/exchange/test_state_corp_services.py +++ b/tests/apps/exchange/test_state_corp_services.py @@ -249,10 +249,12 @@ class StateCorpExchangeServiceTest(TestCase): ogrn=organization.ogrn, title="А40-1/2026", record_date="2026-03-25", + amount="1250000.50", status="in_progress", payload={ "case_number": "А40-1/2026", "court_name": "АС города Москвы", + "claim_amount": "1.00", "target": {"role": "ответчик"}, }, ) @@ -429,6 +431,10 @@ class StateCorpExchangeServiceTest(TestCase): payload["data"]["arbitration_cases"][0]["case_number"], "А40-1/2026", ) + self.assertEqual( + payload["data"]["arbitration_cases"][0]["claim_amount"], + "1250000.50", + ) self.assertEqual( payload["data"]["bankruptcy_procedures"][0]["case_number"], "А40-555/2026", diff --git a/tests/apps/organizations/test_test_companies_commands.py b/tests/apps/organizations/test_test_companies_commands.py index 7d149a7..84ca974 100644 --- a/tests/apps/organizations/test_test_companies_commands.py +++ b/tests/apps/organizations/test_test_companies_commands.py @@ -15,6 +15,7 @@ from apps.parsers.models import ( ) from django.core.management import call_command from django.test import TestCase, override_settings +from django.utils import timezone from organizations.models import ( Organization, OrganizationSourceExtension, @@ -88,6 +89,14 @@ class TestCompaniesCommandsTest(TestCase): ).count(), 20 * 4, ) + self.assertEqual( + set( + OrganizationSourceFinancialLine.objects.filter( + source_record__extension__organization__in=companies + ).values_list("year", flat=True) + ), + {timezone.localdate().year}, + ) self.assertEqual(IndustrialCertificateRecord.objects.count(), 20) self.assertEqual(IndustrialProductRecord.objects.count(), 20) self.assertEqual(ManufacturerRecord.objects.count(), 20) diff --git a/tests/apps/parsers/test_services.py b/tests/apps/parsers/test_services.py index f4b021b..f1e07e1 100644 --- a/tests/apps/parsers/test_services.py +++ b/tests/apps/parsers/test_services.py @@ -441,8 +441,8 @@ class ParserLoadLogServiceTest(TestCase): self.assertEqual(updated, 0) self.assertEqual(log.status, ParserLoadLog.Status.IN_PROGRESS) - def test_mark_stale_in_progress_failed_closes_precreated_job_without_batch(self): - """Pre-created source-card jobs without batch_id are still linked by source.""" + def test_mark_stale_in_progress_failed_keeps_recent_heartbeat_without_batch(self): + """A recent heartbeat keeps a pre-created source-card job alive.""" log = ParserLoadLogFactory( source=ParserLoadLog.Source.INDUSTRIAL_PRODUCTS, batch_id=2, @@ -465,9 +465,9 @@ class ParserLoadLogServiceTest(TestCase): log.refresh_from_db() job.refresh_from_db() - self.assertEqual(updated, 1) - self.assertEqual(log.status, ParserLoadLog.Status.FAILED) - self.assertEqual(job.status, JobStatus.FAILURE) + self.assertEqual(updated, 0) + self.assertEqual(log.status, ParserLoadLog.Status.IN_PROGRESS) + self.assertEqual(job.status, JobStatus.STARTED) @unittest.skip( diff --git a/tests/apps/parsers/test_source_cards_service.py b/tests/apps/parsers/test_source_cards_service.py index 68fb3d5..1026673 100644 --- a/tests/apps/parsers/test_source_cards_service.py +++ b/tests/apps/parsers/test_source_cards_service.py @@ -392,7 +392,7 @@ class SourceCardServiceUnitTest(SimpleTestCase): ) stale_in_progress_load = SimpleNamespace( status="in_progress", - updated_at=timezone.now() - timedelta(hours=3), + updated_at=timezone.now() - timedelta(hours=5), ) self.assertEqual( SourceCardService._get_status( @@ -618,7 +618,7 @@ class SourceCardServiceDatabaseTest(TestCase): ) self.assertEqual(procurements_card["records_count"], 1) - def test_get_active_tasks_ignores_old_jobs_even_when_updated_recently(self): + def test_get_active_tasks_keeps_old_job_with_recent_heartbeat(self): job = BackgroundJob.objects.create( task_id="old-source-task", task_name="apps.parsers.tasks.parse_industrial_products", @@ -637,7 +637,8 @@ class SourceCardServiceDatabaseTest(TestCase): SourceCardService.get_definition("manufacturers-and-products") ) - self.assertEqual(tasks, []) + self.assertEqual(len(tasks), 1) + self.assertEqual(tasks[0]["task_id"], job.task_id) def test_get_active_tasks_keeps_recent_pending_jobs(self): BackgroundJob.objects.create( diff --git a/tests/apps/parsers/test_sources_api_e2e.py b/tests/apps/parsers/test_sources_api_e2e.py index cf74476..5fbeb5e 100644 --- a/tests/apps/parsers/test_sources_api_e2e.py +++ b/tests/apps/parsers/test_sources_api_e2e.py @@ -133,6 +133,11 @@ class SourcesApiE2ETest(APITestCase): self.assertEqual(minprom_card["records_count"], 3) self.assertEqual(minprom_card["organizations_count"], 1) self.assertEqual(len(minprom_card["source_items"]), 3) + minprom_items = {item["code"]: item for item in minprom_card["source_items"]} + self.assertEqual( + minprom_items["industrial_products"]["refresh_key"], + "mpt_products", + ) statuses = {item["slug"]: item for item in statuses_response.data["data"]} self.assertEqual(statuses["planned-inspections"]["progress"], 55) diff --git a/tests/apps/parsers/test_tasks.py b/tests/apps/parsers/test_tasks.py index 1a4e52f..0eb70c2 100644 --- a/tests/apps/parsers/test_tasks.py +++ b/tests/apps/parsers/test_tasks.py @@ -1086,9 +1086,16 @@ class GenericSourceFetchTestCase(TestCase): ) as jobs_cleanup_mock: result = parser_tasks.cleanup_stale_parser_loads(max_age_minutes=45) - cleanup_mock.assert_called_once_with(max_age_minutes=45) + cleanup_mock.assert_called_once_with( + max_age_minutes=45, + pending_max_age_minutes=24 * 60, + ) jobs_cleanup_mock.assert_called_once() self.assertEqual(jobs_cleanup_mock.call_args.kwargs["max_age_minutes"], 45) + self.assertEqual( + jobs_cleanup_mock.call_args.kwargs["pending_max_age_minutes"], + 24 * 60, + ) self.assertIn( "apps.parsers.tasks.scan_fns_directory", jobs_cleanup_mock.call_args.kwargs["task_names"], @@ -1101,6 +1108,7 @@ class GenericSourceFetchTestCase(TestCase): self.assertEqual(result["marked_failed"], 2) self.assertEqual(result["marked_jobs_failed"], 3) self.assertEqual(result["max_age_minutes"], 45) + self.assertEqual(result["pending_max_age_minutes"], 24 * 60) def test_get_or_create_background_job_merges_meta_for_precreated_job(self): BackgroundJobService.create_job( diff --git a/tests/apps/parsers/test_views.py b/tests/apps/parsers/test_views.py index 0f0dc6b..06c51ba 100644 --- a/tests/apps/parsers/test_views.py +++ b/tests/apps/parsers/test_views.py @@ -11,6 +11,7 @@ import zipfile from datetime import date from unittest.mock import Mock, patch +from apps.core.models import BackgroundJob from apps.parsers.models import ( FinancialReport, FinancialReportLine, @@ -1411,3 +1412,19 @@ class ParsersViewSetTest(APITestCase): for key, value in payload.items(): self.assertEqual(task_kwargs[key], value) self.assertEqual(task_kwargs["requested_by_id"], self.user.id) + + def test_run_industrial_products_uses_legacy_alias(self): + self.client.force_authenticate(self.user) + url = reverse("api_v1:parsers:run-parser", args=["industrial_products"]) + + with patch( + "apps.parsers.views.tasks.parse_industrial_products.apply_async", + return_value=Mock(id="task-products"), + ) as apply_async_mock: + response = self.client.post(url, {}, format="json") + + self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) + apply_async_mock.assert_called_once() + queued_task_id = apply_async_mock.call_args.kwargs["task_id"] + job = BackgroundJob.objects.get(task_id=queued_task_id) + self.assertEqual(job.meta["source_key"], "mpt_products")