diff --git a/api/scripts/purge_statements.py b/api/scripts/purge_statements.py index fade70e..062916d 100644 --- a/api/scripts/purge_statements.py +++ b/api/scripts/purge_statements.py @@ -91,11 +91,13 @@ class PurgePlan: class PurgeRefusedError(Exception): - """Очистка отклонена: есть документы, замороженные в верифицированных версиях.""" + """Очистка отклонена: есть документы, замороженные в верифицированных версиях. `plan` — свежий план, + построенный под блокировками (его счётчики и `blocking` описывают состояние базы на момент отказа).""" - def __init__(self, blocking: Sequence[BlockingVersion]) -> None: + def __init__(self, plan: PurgePlan) -> None: super().__init__("очистка отклонена") - self.blocking = list(blocking) + self.plan = plan + self.blocking = list(plan.blocking) def _frozen_version() -> ColumnElement[bool]: @@ -258,15 +260,22 @@ async def apply_purge(session: AsyncSession, plan: PurgePlan, *, now: datetime) Блокирует файлы плана (по `id`), заново проверяет отказ под блокировками и только потом меняет данные: при `PurgeRefusedError` база не изменена.""" - whole = list(plan.form_only_files) + (list(plan.mixed_files) if plan.include_mixed else []) - partial = [] if plan.include_mixed else list(plan.mixed_files) - for file_id in sorted([*whole, *partial]): - await files_repo.lock_jobs(session, file_id) - await files_repo.lock_file(session, file_id) - await files_repo.lock_file_documents(session, file_id) - fresh = await plan_purge(session, include_mixed=plan.include_mixed) + locked: set[uuid.UUID] = set() + + async def lock(file_ids: Sequence[uuid.UUID]) -> None: + for file_id in sorted(set(file_ids) - locked): + await files_repo.lock_jobs(session, file_id) + await files_repo.lock_file(session, file_id) + await files_repo.lock_file_documents(session, file_id) + locked.add(file_id) + + await lock([*plan.form_only_files, *plan.mixed_files]) + fresh = await plan_purge(session, include_mixed=plan.include_mixed) # T4.2: списки файлов — из свежего плана + await lock([*fresh.form_only_files, *fresh.mixed_files]) # файлы, появившиеся после первого плана if fresh.blocking: - raise PurgeRefusedError(fresh.blocking) + raise PurgeRefusedError(fresh) + whole = list(fresh.form_only_files) + (list(fresh.mixed_files) if fresh.include_mixed else []) + partial = [] if fresh.include_mixed else list(fresh.mixed_files) audit = AuditService(session) for file_id in sorted(whole): await _purge_whole_file(session, audit, file_id, now) @@ -323,7 +332,7 @@ async def run(argv: Sequence[str], *, transaction: Transaction = session_transac return EXIT_REFUSED await apply_purge(session, plan, now=utc_now()) except PurgeRefusedError as error: - print(render_report(PurgePlan(blocking=error.blocking)), file=sys.stderr) + print(render_report(error.plan), file=sys.stderr) # T4.4: свежий план с реальными счётчиками print("очистка отклонена, база не изменена", file=sys.stderr) return EXIT_REFUSED print("очистка выполнена") diff --git a/api/tests/test_purge_statements.py b/api/tests/test_purge_statements.py index 594c381..720186f 100644 --- a/api/tests/test_purge_statements.py +++ b/api/tests/test_purge_statements.py @@ -458,3 +458,33 @@ async def test_reupload_after_purge_not_duplicate(world: World, seeded: AsyncSes service = PrimaryFilesService.__new__(PrimaryFilesService) service._session = session # type: ignore[attr-defined] # find_duplicate читает только сессию assert await service.find_duplicate(entity_id, "ab" * 32) is None + + +async def test_apply_uses_fresh_plan_for_file_lists(world: World) -> None: + """T4.2: файл, появившийся после построения плана, удаляется по свежему плану под блокировками.""" + session = world.session + first, first_job = await make_stored_file(session, entity=world.entity, original_name="first.pdf") + await world.form_document(first, first_job, DocumentType.BALANCE) + stale = await purge_statements.plan_purge(session, include_mixed=False) + late, late_job = await make_stored_file(session, entity=world.entity, original_name="late.pdf") + await world.form_document(late, late_job, DocumentType.BALANCE) + first_id, late_id = first.id, late.id + + await purge_statements.apply_purge(session, stale, now=datetime.datetime.now(datetime.UTC)) + + assert await _count(session, StoredFile, StoredFile.id.in_([first_id, late_id])) == 0 + + +async def test_refusal_carries_fresh_plan_counters(world: World) -> None: + """T4.4: отказ под блокировками несёт свежий план — счётчики отчёта не нулевые.""" + objects = await _mixed_and_form_only(world) + stale = await purge_statements.plan_purge(world.session, include_mixed=False) + version = await world.version(ConsolidationStatus.VERIFIED) + await world.bind(objects["income"], version) # type: ignore[arg-type] + + with pytest.raises(purge_statements.PurgeRefusedError) as raised: + await purge_statements.apply_purge(world.session, stale, now=datetime.datetime.now(datetime.UTC)) + + refused = raised.value.plan + assert refused.form_documents > 0 and refused.blocking + assert "документов форм: 3" in purge_statements.render_report(refused)