bito-code-review[bot] commented on code in PR #43490: URL: https://github.com/apache/superset/pull/43490#discussion_r3848911174
########## superset/migrations/versions/2026-08-24_15-50_a6c21e5b4d93_index_purge_audit_pruning.py: ########## @@ -0,0 +1,46 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Index purge-audit pruning predicates. + +Revision ID: a6c21e5b4d93 +Revises: 1072de5ed955 +Create Date: 2026-08-24 15:50:00.000000 + +""" + +from superset.migrations.shared.utils import create_index, drop_index + +# revision identifiers, used by Alembic. +revision: str = "a6c21e5b4d93" +down_revision: str = "1072de5ed955" + +_TABLE_NAME: str = "purge_audit_log" +_INDEX_NAME: str = "ix_purge_audit_log_pruning" + + +def upgrade() -> None: + """Add an index matching recurring pruning access patterns.""" + create_index( + _TABLE_NAME, + _INDEX_NAME, + ["status", "entity_type", "entity_uuid", "created_on"], + ) + + +def downgrade() -> None: + """Remove the purge-audit pruning index.""" + drop_index(_TABLE_NAME, _INDEX_NAME) Review Comment: <div> <div id="suggestion"> <div id="issue"><b>Missing unit test coverage for new migration</b></div> <div id="fix"> The migration adds a composite index for pruning queries but lacks unit test coverage. The predecessor migration (`b8d2f4a6c901`) has a dedicated test file (`test_purge_audit_predecessor_index.py`) validating upgrade/downgrade and idempotency; this migration should have equivalent coverage per project standards for new tools. </div> </div> <small><i>Code Review Run #398665</i></small> </div> --- Should Bito avoid suggestions like this for future reviews? (<a href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>) - [ ] Yes, avoid them ########## tests/integration_tests/deletion_retention/prune_audit_tests.py: ########## @@ -0,0 +1,484 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Integration tests for purge-audit pruning against the real metadata DB. + +Rows are seeded directly (pruning reads only ``purge_audit_log``); every +test's rows carry a distinctive ``entity_type`` prefix and are removed in +teardown, so no other suite ever sees them. Scoping on ``entity_type`` +rather than ``entity_uuid`` matters: some rows deliberately carry a NULL +uuid, which a ``LIKE`` on the uuid column would never match. +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import timedelta +from functools import partial +from typing import Any +from unittest.mock import patch +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from flask import current_app +from sqlalchemy.orm import Query + +from superset import db +from superset.commands.deletion_retention import audit, prune_audit +from superset.commands.deletion_retention.prune_audit import ( + EVIDENCE_RETENTION_KEY, + OPERATIONAL_RETENTION_KEY, +) +from superset.models.purge_audit_log import ( + PurgeAuditLog, + STATUS_BLOCKED, + STATUS_CONFIRMED, + STATUS_FAILED, + STATUS_PENDING, + STATUS_TARGET_ABSENT, +) +from tests.integration_tests.base_tests import SupersetTestCase +from tests.integration_tests.deletion_retention._base import ( + ensure_purge_audit_coordination, +) + +_PREFIX: str = "prune_audit_it_" +_ENTITY_TYPE: str = f"{_PREFIX}slices" + + +class TestPruneAudit(SupersetTestCase): + """Behavioral coverage for ``prune_audit.run_prune``.""" + + _seq: int + + def setUp(self) -> None: + super().setUp() + ensure_purge_audit_coordination() + self._seq = 0 + self._cleanup() + + def tearDown(self) -> None: + self._cleanup() + super().tearDown() + + def _cleanup(self) -> None: + db.session.rollback() + db.session.execute( + sa.delete(PurgeAuditLog.__table__).where( + PurgeAuditLog.__table__.c.entity_type == _ENTITY_TYPE + ) + ) + db.session.commit() + + def add_row( + self, + status: str, + entity: str | None = "e1", + age_days: float = 0, + trigger: str = audit.TRIGGER_RETENTION, + ) -> UUID: + """Seed one audit row and return its id. + + Ids rather than instances: an instance whose row a later prune + deletes raises ObjectDeletedError on attribute access. ``age_days`` + may be fractional; a strictly increasing microsecond sequence keeps + every ``created_on`` unique so streak ordering is deterministic. + ``entity=None`` seeds a row with no ``entity_uuid``. + """ + self._seq += 1 + row: PurgeAuditLog = PurgeAuditLog( + id=uuid4(), + status=status, + trigger=trigger, + actor=audit.ACTOR_SYSTEM, + entity_type=_ENTITY_TYPE, + entity_uuid=None if entity is None else f"{_PREFIX}{entity}", + created_on=audit.utc_now() + - timedelta(days=age_days) + + timedelta(microseconds=self._seq), + ) + db.session.add(row) + db.session.commit() + return row.id + + def remaining_ids(self, entity: str | None = "__any__") -> list[UUID]: + """Ids of surviving seeded rows, oldest first.""" + query: Query[PurgeAuditLog] = db.session.query(PurgeAuditLog).filter( + PurgeAuditLog.entity_type == _ENTITY_TYPE + ) + if entity != "__any__": + query = query.filter( + PurgeAuditLog.entity_uuid.is_(None) + if entity is None + else PurgeAuditLog.entity_uuid == f"{_PREFIX}{entity}" + ) + return [r.id for r in query.order_by(PurgeAuditLog.created_on).all()] + + def run_prune(self, **config: Any) -> prune_audit.PruneRunResult: + with patch.dict(current_app.config, config): + return prune_audit.run_prune() + + # -- US1: bounded blockage history ------------------------------------- + + def test_duplicates_reduce_to_the_earliest_survivor_regardless_of_age(self) -> None: + """Keep only the earliest row in a current blockage streak.""" + ids: list[UUID] = [ + self.add_row(STATUS_BLOCKED, age_days=5 - i / 100) for i in range(50) + ] + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 49 + assert result.carried_over is False + assert self.remaining_ids("e1") == [ids[0]] + + def test_backlog_converges_over_bounded_runs(self) -> None: + """Drain oversized backlogs across bounded successive runs.""" + for i in range(56): + self.add_row(STATUS_BLOCKED, age_days=3 - i / 1000) + + removed_per_run: list[int] = [] + with ( + patch.object(prune_audit, "BATCH_SIZE", 10), + patch.object(prune_audit, "MAX_BATCHES_PER_RUN", 2), + ): + while True: + result: prune_audit.PruneRunResult = self.run_prune() + removed_per_run.append(result.blocked_duplicates) + if not result.carried_over: + break + + assert sum(removed_per_run) == 55 # every duplicate, none of the survivor + assert all(n <= 20 for n in removed_per_run) # never exceeds the budget + assert removed_per_run[-1] > 0 # carryover is not reported spuriously + assert len(self.remaining_ids("e1")) == 1 # the survivor + + def test_dedup_never_crosses_entities(self) -> None: + a_ids: list[UUID] = [ + self.add_row(STATUS_BLOCKED, entity="a", age_days=2) for _ in range(3) + ] + b_ids: list[UUID] = [ + self.add_row(STATUS_BLOCKED, entity="b", age_days=2) for _ in range(4) + ] + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 5 + assert self.remaining_ids("a") == [a_ids[0]] + assert self.remaining_ids("b") == [b_ids[0]] + + def test_resolved_streak_has_no_survivor_and_ages_out(self) -> None: + """Expire blocked rows after newer evidence resolves their streak.""" + for _ in range(3): + self.add_row(STATUS_BLOCKED, age_days=200) + confirmed: UUID = self.add_row(STATUS_CONFIRMED, age_days=100) + new_survivor: UUID = self.add_row(STATUS_BLOCKED, age_days=50) + self.add_row(STATUS_BLOCKED, age_days=40) # duplicate in the new streak + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 1 + assert result.operational_expired == 3 + assert set(self.remaining_ids("e1")) == {confirmed, new_survivor} + + def test_a_failed_attempt_does_not_reset_the_blocked_since_survivor(self) -> None: + """Keep one blockage streak across a failed purge attempt.""" + blocked_since: UUID = self.add_row(STATUS_BLOCKED, age_days=200) + self.add_row(STATUS_FAILED, age_days=150) # ages out as operational + self.add_row(STATUS_BLOCKED, age_days=100) + self.add_row(STATUS_BLOCKED, age_days=50) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 2 + assert result.operational_expired == 1 + assert self.remaining_ids("e1") == [blocked_since] + + # -- US2: evidence survives by default ---------------------------------- + + def test_defaults_leave_completed_destruction_evidence_untouched(self) -> None: + evidence: list[UUID] = [ + self.add_row(STATUS_CONFIRMED, entity="ev", age_days=3650), + self.add_row(STATUS_TARGET_ABSENT, entity="ev", age_days=1000), + self.add_row(STATUS_CONFIRMED, entity="ev", age_days=1), + ] + self.add_row(STATUS_FAILED, entity="ev", age_days=365) # ages out + + for _ in range(3): # any sequence of runs (SC-002) + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.evidence_expired == 0 + assert set(self.remaining_ids("ev")) == set(evidence) + + def test_evidence_opt_in_expires_only_rows_older_than_its_window(self) -> None: + old: UUID = self.add_row(STATUS_CONFIRMED, entity="ev", age_days=400) + young: UUID = self.add_row(STATUS_TARGET_ABSENT, entity="ev", age_days=300) + + result: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: 365} + ) + + assert result.evidence_expired == 1 + assert self.remaining_ids("ev") == [young] + assert old not in self.remaining_ids("ev") + + def test_opt_in_disabled_again_removes_no_further_evidence(self) -> None: + self.add_row(STATUS_CONFIRMED, entity="ev", age_days=400) + survivor: UUID = self.add_row(STATUS_CONFIRMED, entity="ev", age_days=390) + + first: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: 395} + ) + assert first.evidence_expired == 1 + + second: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: None} + ) + assert second.evidence_expired == 0 + assert self.remaining_ids("ev") == [survivor] + + def test_pending_and_future_rows_survive_every_configuration(self) -> None: + pending: UUID = self.add_row(STATUS_PENDING, entity="px", age_days=3650) + future_blocked: UUID = self.add_row(STATUS_BLOCKED, entity="px", age_days=-1) + future_confirmed: UUID = self.add_row( + STATUS_CONFIRMED, entity="px", age_days=-2 + ) + + result: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 1, EVIDENCE_RETENTION_KEY: 1} + ) + + assert result.total_removed == 0 + assert set(self.remaining_ids("px")) == { + pending, + future_blocked, + future_confirmed, + } + + def test_a_future_dated_outcome_cannot_resolve_a_live_streak(self) -> None: + """Exclude future-dated evidence from current streak boundaries.""" + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="sk", age_days=500) + self.add_row(STATUS_BLOCKED, entity="sk", age_days=400) + skewed: UUID = self.add_row(STATUS_CONFIRMED, entity="sk", age_days=-30) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 1 + assert result.operational_expired == 0 + assert set(self.remaining_ids("sk")) == {blocked_since, skewed} + + # -- Survivor invariant under boundary removal (the review's HIGH) ------ + + def test_evidence_expiry_spares_a_boundary_that_still_bounds_blocked_rows( + self, + ) -> None: + """Keep evidence while it still bounds surviving blocked rows.""" + blocked: list[UUID] = [ + self.add_row(STATUS_BLOCKED, entity="bd", age_days=30) for _ in range(3) + ] + boundary: UUID = self.add_row(STATUS_CONFIRMED, entity="bd", age_days=20) + + # Evidence window far shorter than the operational one: the boundary + # is past its cutoff, the blocked rows it resolved are not. + first: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 90, EVIDENCE_RETENTION_KEY: 1} + ) + + assert first.evidence_expired == 0 + assert set(self.remaining_ids("bd")) == {*blocked, boundary} + + # Once the blocked rows age out, the boundary bounds nothing and + # becomes expirable — the guard defers, it does not immortalize. + # Categories drain in order within one run, so the blocked rows go + # first and the boundary follows in the same pass. + second: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 10, EVIDENCE_RETENTION_KEY: 1} + ) + assert second.operational_expired == 3 + assert second.evidence_expired == 1 + assert self.remaining_ids("bd") == [] + + def test_a_block_after_an_unresolved_attempt_is_not_treated_as_a_duplicate( + self, + ) -> None: + """An in-flight attempt is a boundary waiting to happen. + + Finalizing a ``pending`` row resolves it *in place*, keeping its + original timestamp — the one way a boundary can appear in the middle + of history. The blocked row after it would become the new streak's + survivor, so it must not be classified as a duplicate while the + attempt is unresolved. Deleting it would destroy the only record + that the entity was still blocked after that attempt. + """ + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="ua", age_days=100) + attempt: UUID = self.add_row(STATUS_PENDING, entity="ua", age_days=50) + later_block: UUID = self.add_row(STATUS_BLOCKED, entity="ua", age_days=10) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 0 + assert set(self.remaining_ids("ua")) == {blocked_since, attempt, later_block} + + # Once the attempt finalizes, the boundary is real: the later block + # is the new streak's survivor and stays; the older block is now a + # resolved-streak row that ages out on the normal window. + db.session.execute( + sa.update(PurgeAuditLog.__table__) + .where(PurgeAuditLog.__table__.c.id == attempt) + .values(status=STATUS_CONFIRMED) + ) + db.session.commit() + + after: prune_audit.PruneRunResult = self.run_prune() + + assert after.blocked_duplicates == 0 + assert after.operational_expired == 1 # the pre-attempt block + assert set(self.remaining_ids("ua")) == {attempt, later_block} + + def test_age_does_not_make_an_unstable_block_expirable(self) -> None: + """Keep aged blocked rows whose classification remains unstable. + + Seeded entirely outside the retention window, so only the guard — + not the cutoff — can save the row. + """ + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="ao", age_days=100) + attempt: UUID = self.add_row(STATUS_PENDING, entity="ao", age_days=98) + later_block: UUID = self.add_row(STATUS_BLOCKED, entity="ao", age_days=95) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 0 + assert result.operational_expired == 0 + assert set(self.remaining_ids("ao")) == {blocked_since, attempt, later_block} + + # Resolving the attempt makes the boundary real: the later block is + # the current streak's survivor and is exempt regardless of age, + # while the pre-attempt block ages out. + db.session.execute( + sa.update(PurgeAuditLog.__table__) + .where(PurgeAuditLog.__table__.c.id == attempt) + .values(status=STATUS_TARGET_ABSENT) + ) + db.session.commit() + + after: prune_audit.PruneRunResult = self.run_prune() + + assert after.operational_expired == 1 + assert set(self.remaining_ids("ao")) == {attempt, later_block} + + def test_evidence_guard_also_defers_to_an_unresolved_older_attempt(self) -> None: + """Treat an older pending attempt as potential blocked evidence.""" + attempt: UUID = self.add_row(STATUS_PENDING, entity="ug", age_days=400) + boundary: UUID = self.add_row(STATUS_CONFIRMED, entity="ug", age_days=300) + + result: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: 200} + ) + + assert result.evidence_expired == 0 + assert set(self.remaining_ids("ug")) == {attempt, boundary} + + def test_blocked_rows_without_an_entity_uuid_are_never_aged_out(self) -> None: + """Keep UUID-less blocked rows that cannot be proven redundant.""" + anonymous_block: UUID = self.add_row(STATUS_BLOCKED, entity=None, age_days=1000) + self.add_row(STATUS_FAILED, entity=None, age_days=1000) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.operational_expired == 1 + assert self.remaining_ids(None) == [anonymous_block] + + # -- US3: operator controls and observability --------------------------- + + def test_configured_retention_window_is_honored(self) -> None: + kept: UUID = self.add_row(STATUS_FAILED, entity="w", age_days=5) + self.add_row(STATUS_FAILED, entity="w", age_days=15) + + result: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 10} + ) + + assert result.operational_expired == 1 + assert self.remaining_ids("w") == [kept] + + def test_invalid_operational_window_skips_the_category_not_widens(self) -> None: + """Fail closed when the operational retention window is invalid.""" + old_failed: UUID = self.add_row(STATUS_FAILED, entity="iv", age_days=1000) + self.add_row(STATUS_BLOCKED, entity="iv", age_days=3) + self.add_row(STATUS_BLOCKED, entity="iv", age_days=2) + + result: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 0} + ) + + assert result.operational_expired == 0 + assert result.blocked_duplicates == 1 + assert result.invalid_config_keys == [OPERATIONAL_RETENTION_KEY] + assert old_failed in self.remaining_ids("iv") + + def test_second_run_over_the_same_candidates_removes_and_reports_zero(self) -> None: + """Report zero when rerunning over an already-pruned history.""" + for _ in range(4): + self.add_row(STATUS_BLOCKED, entity="cc", age_days=2) + self.add_row(STATUS_FAILED, entity="cc", age_days=100) + + first: prune_audit.PruneRunResult = self.run_prune() + assert first.blocked_duplicates == 3 + assert first.operational_expired == 1 + + second: prune_audit.PruneRunResult = self.run_prune() + assert second.total_removed == 0 + assert len(self.remaining_ids("cc")) == 1 + + def test_a_duplicate_backlog_cannot_starve_the_age_based_categories(self) -> None: + """Prevent duplicate backlogs from starving age-based categories.""" + for _ in range(20): + self.add_row(STATUS_BLOCKED, entity="st", age_days=2) + old_failed: UUID = self.add_row(STATUS_FAILED, entity="st", age_days=1000) + + with ( + patch.object(prune_audit, "BATCH_SIZE", 1), + patch.object(prune_audit, "MAX_BATCHES_PER_RUN", 2), + ): + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 1 # budget-limited, as expected + assert result.operational_expired == 1 # but age-out still progressed + assert result.carried_over is True + assert old_failed not in self.remaining_ids("st") + + def test_delete_rechecks_survivor_after_a_pending_attempt_appears(self) -> None: + """Evaluate survivor safety inside the DELETE statement. + + Constructing the candidate query must not freeze its result. A pending + attempt committed before execution can become a mid-history boundary, + so the later blocked row must remain available as its future survivor. + """ + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="rc", age_days=3) + later_block: UUID = self.add_row(STATUS_BLOCKED, entity="rc", age_days=1) + select_candidates: Callable[[int], sa.sql.Select] = partial( + prune_audit._duplicate_candidates, audit.utc_now() Review Comment: <div> <div id="suggestion"> <div id="issue"><b>Replace private API with public interface</b></div> <div id="fix"> Accessing private member `_duplicate_candidates` may break if internal implementation changes; consider using a public interface or clearly document the intentional coupling. </div> </div> <small><i>Code Review Run #398665</i></small> </div> --- Should Bito avoid suggestions like this for future reviews? (<a href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>) - [ ] Yes, avoid them ########## tests/unit_tests/commands/deletion_retention/test_prune_audit.py: ########## @@ -0,0 +1,294 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Unit tests for the pure parts of purge-audit pruning. + +Query behavior against real rows (streak survivors, batching, protection +invariants) is covered by +``tests/integration_tests/deletion_retention/prune_audit_tests.py``. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from datetime import datetime +from typing import Any, Iterator +from unittest.mock import patch + +import pytest +import sqlalchemy as sa +from flask import current_app +from sqlalchemy.dialects import mysql + +from superset.commands.deletion_retention import prune_audit +from superset.commands.deletion_retention.prune_audit import ( + EVIDENCE_RETENTION_KEY, + OPERATIONAL_RETENTION_KEY, + OPERATIONAL_STATUSES, + PROTECTED_STATUSES, + PruneRunResult, + resolve_evidence_retention_days, + resolve_operational_retention_days, +) +from superset.models.purge_audit_log import ( + ALL_STATUSES, + STATUS_BLOCKED, + STATUS_CONFIRMED, + STATUS_FAILED, + STATUS_PENDING, + STATUS_TARGET_ABSENT, +) + +_METRIC_PREFIX: str = "deletion_retention.prune_purge_audit" + + +@contextmanager +def without_config(key: str) -> Iterator[None]: + """Remove a config key for the duration of the block, then restore it. + + ``patch.dict`` cannot express key *removal*, and the unit-test app + fixture is module-scoped, so a bare ``pop`` would leak to later tests. + """ + with patch.dict(current_app.config): + current_app.config.pop(key, None) + yield + + +def test_retention_categories_partition_every_status() -> None: + """Require every status to belong to a retention category. + + A new status added to the model without a category would silently never + be pruned; this assertion catches that omission. + """ + assert OPERATIONAL_STATUSES == {STATUS_BLOCKED, STATUS_FAILED} + assert PROTECTED_STATUSES == {STATUS_CONFIRMED, STATUS_TARGET_ABSENT} + assert not OPERATIONAL_STATUSES & PROTECTED_STATUSES + assert OPERATIONAL_STATUSES | PROTECTED_STATUSES | {STATUS_PENDING} == ALL_STATUSES + + +def test_only_proof_of_destruction_breaks_a_blockage_streak() -> None: + """Require destruction evidence to break a blockage streak. + + A failed attempt is an infrastructure outcome, and pending is provisional. + Neither proves that the blockage cleared. + """ + assert prune_audit._STREAK_BREAKING_STATUSES == { + STATUS_CONFIRMED, + STATUS_TARGET_ABSENT, + } + assert STATUS_FAILED not in prune_audit._STREAK_BREAKING_STATUSES + assert STATUS_PENDING not in prune_audit._STREAK_BREAKING_STATUSES + + +def test_operational_retention_defaults_to_ninety_days() -> None: + """Use the shipped 90-day operational retention default.""" + assert current_app.config[OPERATIONAL_RETENTION_KEY] == 90 + assert resolve_operational_retention_days().days == 90 + + [email protected]("value", [30, 1, 36500]) +def test_operational_retention_accepts_positive_days(value: int) -> None: + with patch.dict(current_app.config, {OPERATIONAL_RETENTION_KEY: value}): + assert resolve_operational_retention_days() == (value, None) + + [email protected]("value", [0, -5, True, False, "ninety", None, 1.5]) +def test_operational_retention_fails_closed_on_invalid_values(value: Any) -> None: + """Disable a category and identify its invalid configuration key.""" + with patch.dict(current_app.config, {OPERATIONAL_RETENTION_KEY: value}): + window: prune_audit.ResolvedWindow = resolve_operational_retention_days() + assert window.days is None + assert window.invalid_key == OPERATIONAL_RETENTION_KEY + + +def test_missing_operational_key_is_reported_as_invalid_not_assumed() -> None: + """A popped key is operator error, not a silent 90-day assumption.""" + with without_config(OPERATIONAL_RETENTION_KEY): + window: prune_audit.ResolvedWindow = resolve_operational_retention_days() + assert window.days is None + assert window.invalid_key == OPERATIONAL_RETENTION_KEY + + +def test_evidence_retention_defaults_to_off_without_warning() -> None: + """Unset is the documented never-expire default (FR-006), not an error: + disabled, no warning, and no key reported as invalid.""" + with without_config(EVIDENCE_RETENTION_KEY): + with patch.object(prune_audit, "logger") as mock_logger: + window: prune_audit.ResolvedWindow = resolve_evidence_retention_days() + mock_logger.warning.assert_not_called() + assert window == (None, None) + + +def test_evidence_retention_accepts_the_explicit_opt_in() -> None: + with patch.dict(current_app.config, {EVIDENCE_RETENTION_KEY: 3650}): + window: prune_audit.ResolvedWindow = resolve_evidence_retention_days() + assert window.days == 3650 + + [email protected]("value", [0, -1, True, "forever"]) +def test_evidence_retention_fails_closed_on_invalid_values(value: Any) -> None: + with patch.dict(current_app.config, {EVIDENCE_RETENTION_KEY: value}): + with patch.object(prune_audit, "logger") as mock_logger: + window: prune_audit.ResolvedWindow = resolve_evidence_retention_days() + mock_logger.warning.assert_called_once() + assert window.days is None + assert window.invalid_key == EVIDENCE_RETENTION_KEY + + +def test_prune_run_result_totals_and_dict_shape() -> None: + result: PruneRunResult = PruneRunResult( + blocked_duplicates=3, operational_expired=2, evidence_expired=1 + ) + assert result.total_removed == 6 + assert result.as_dict() == { + "removed": { + "blocked_duplicates": 3, + "operational_expired": 2, + "evidence_expired": 1, + }, + "carried_over": False, + "invalid_config_keys": [], + } + + +def test_disabled_task_reports_itself_and_removes_nothing() -> None: + """Report a disabled run without reaching the prune implementation.""" + from superset.tasks import deletion_retention as task_module + + with patch.dict(current_app.config, {"PURGE_AUDIT_PRUNING_ENABLED": False}): + with ( + patch.object(task_module, "stats_logger_manager") as mock_stats, + patch.object(task_module, "logger") as mock_logger, + patch.object(task_module.prune_audit, "run_prune") as mock_run, + ): + outcome: dict[str, Any] = task_module.prune_purge_audit() + assert outcome == {"skipped_disabled": 1} + mock_run.assert_not_called() + mock_stats.instance.incr.assert_called_once_with( + f"{_METRIC_PREFIX}.skipped_disabled" + ) + assert mock_logger.info.called + + [email protected]("value", ["false", "0", 1, None]) +def test_non_boolean_master_switch_fails_closed(value: Any) -> None: + """Never interpret truthy strings or numeric values as deletion opt-in.""" + from superset.tasks import deletion_retention as task_module + + with patch.dict(current_app.config, {"PURGE_AUDIT_PRUNING_ENABLED": value}): + with ( + patch.object(task_module, "stats_logger_manager") as mock_stats, + patch.object(task_module.prune_audit, "run_prune") as mock_run, + ): + outcome: dict[str, Any] = task_module.prune_purge_audit() + + assert outcome == {"skipped_invalid_config": 1} + mock_run.assert_not_called() + mock_stats.instance.incr.assert_called_once_with( + f"{_METRIC_PREFIX}.skipped_invalid_config" + ) + + +def test_failed_run_is_isolated_rolled_back_and_distinguishable() -> None: + """Report and isolate a failed pruning run. + + The task rolls back and returns an error marker that cannot be mistaken + for a successful run that removed nothing. + """ + from superset.tasks import deletion_retention as task_module + + with patch.dict(current_app.config, {"PURGE_AUDIT_PRUNING_ENABLED": True}): + with ( + patch.object(task_module, "stats_logger_manager") as mock_stats, + patch.object(task_module, "db") as mock_db, + patch.object( + task_module.prune_audit, + "run_prune", + side_effect=RuntimeError("boom"), + ), + ): + outcome: dict[str, Any] = task_module.prune_purge_audit() + assert outcome == {"error": 1} + mock_db.session.rollback.assert_called_once() + mock_stats.instance.incr.assert_called_once_with(f"{_METRIC_PREFIX}.failed") + + +def test_successful_run_mirrors_counts_and_carryover_into_metrics() -> None: + """Expose category counts and convergence through metrics.""" + from superset.tasks import deletion_retention as task_module + + fake: PruneRunResult = PruneRunResult( + blocked_duplicates=7, + operational_expired=4, + evidence_expired=0, + carried_over=True, + ) Review Comment: <div> <div id="suggestion"> <div id="issue"><b>Duplicate test setup code</b></div> <div id="fix"> The test file contains syntactic duplication (9 lines repeated at lines 238-246 and 264-272) involving identical patch contexts for stats_logger_manager and run_prune. Consider extracting this into a shared fixture to improve maintainability. No changes required from this diff as no modifications are present. </div> </div> <small><i>Code Review Run #398665</i></small> </div> --- Should Bito avoid suggestions like this for future reviews? (<a href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>) - [ ] Yes, avoid them ########## tests/integration_tests/deletion_retention/prune_audit_tests.py: ########## @@ -0,0 +1,484 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Integration tests for purge-audit pruning against the real metadata DB. + +Rows are seeded directly (pruning reads only ``purge_audit_log``); every +test's rows carry a distinctive ``entity_type`` prefix and are removed in +teardown, so no other suite ever sees them. Scoping on ``entity_type`` +rather than ``entity_uuid`` matters: some rows deliberately carry a NULL +uuid, which a ``LIKE`` on the uuid column would never match. +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import timedelta +from functools import partial +from typing import Any +from unittest.mock import patch +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from flask import current_app +from sqlalchemy.orm import Query + +from superset import db +from superset.commands.deletion_retention import audit, prune_audit +from superset.commands.deletion_retention.prune_audit import ( + EVIDENCE_RETENTION_KEY, + OPERATIONAL_RETENTION_KEY, +) +from superset.models.purge_audit_log import ( + PurgeAuditLog, + STATUS_BLOCKED, + STATUS_CONFIRMED, + STATUS_FAILED, + STATUS_PENDING, + STATUS_TARGET_ABSENT, +) +from tests.integration_tests.base_tests import SupersetTestCase +from tests.integration_tests.deletion_retention._base import ( + ensure_purge_audit_coordination, +) + +_PREFIX: str = "prune_audit_it_" +_ENTITY_TYPE: str = f"{_PREFIX}slices" + + +class TestPruneAudit(SupersetTestCase): + """Behavioral coverage for ``prune_audit.run_prune``.""" + + _seq: int + + def setUp(self) -> None: + super().setUp() + ensure_purge_audit_coordination() + self._seq = 0 + self._cleanup() + + def tearDown(self) -> None: + self._cleanup() + super().tearDown() + + def _cleanup(self) -> None: + db.session.rollback() + db.session.execute( + sa.delete(PurgeAuditLog.__table__).where( + PurgeAuditLog.__table__.c.entity_type == _ENTITY_TYPE + ) + ) + db.session.commit() + + def add_row( + self, + status: str, + entity: str | None = "e1", + age_days: float = 0, + trigger: str = audit.TRIGGER_RETENTION, + ) -> UUID: + """Seed one audit row and return its id. + + Ids rather than instances: an instance whose row a later prune + deletes raises ObjectDeletedError on attribute access. ``age_days`` + may be fractional; a strictly increasing microsecond sequence keeps + every ``created_on`` unique so streak ordering is deterministic. + ``entity=None`` seeds a row with no ``entity_uuid``. + """ + self._seq += 1 + row: PurgeAuditLog = PurgeAuditLog( + id=uuid4(), + status=status, + trigger=trigger, + actor=audit.ACTOR_SYSTEM, + entity_type=_ENTITY_TYPE, + entity_uuid=None if entity is None else f"{_PREFIX}{entity}", + created_on=audit.utc_now() + - timedelta(days=age_days) + + timedelta(microseconds=self._seq), + ) + db.session.add(row) + db.session.commit() + return row.id + + def remaining_ids(self, entity: str | None = "__any__") -> list[UUID]: + """Ids of surviving seeded rows, oldest first.""" + query: Query[PurgeAuditLog] = db.session.query(PurgeAuditLog).filter( + PurgeAuditLog.entity_type == _ENTITY_TYPE + ) + if entity != "__any__": + query = query.filter( + PurgeAuditLog.entity_uuid.is_(None) + if entity is None + else PurgeAuditLog.entity_uuid == f"{_PREFIX}{entity}" + ) + return [r.id for r in query.order_by(PurgeAuditLog.created_on).all()] + + def run_prune(self, **config: Any) -> prune_audit.PruneRunResult: + with patch.dict(current_app.config, config): + return prune_audit.run_prune() + + # -- US1: bounded blockage history ------------------------------------- + + def test_duplicates_reduce_to_the_earliest_survivor_regardless_of_age(self) -> None: + """Keep only the earliest row in a current blockage streak.""" + ids: list[UUID] = [ + self.add_row(STATUS_BLOCKED, age_days=5 - i / 100) for i in range(50) + ] + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 49 + assert result.carried_over is False + assert self.remaining_ids("e1") == [ids[0]] + + def test_backlog_converges_over_bounded_runs(self) -> None: + """Drain oversized backlogs across bounded successive runs.""" + for i in range(56): + self.add_row(STATUS_BLOCKED, age_days=3 - i / 1000) + + removed_per_run: list[int] = [] + with ( + patch.object(prune_audit, "BATCH_SIZE", 10), + patch.object(prune_audit, "MAX_BATCHES_PER_RUN", 2), + ): + while True: + result: prune_audit.PruneRunResult = self.run_prune() + removed_per_run.append(result.blocked_duplicates) + if not result.carried_over: + break + + assert sum(removed_per_run) == 55 # every duplicate, none of the survivor + assert all(n <= 20 for n in removed_per_run) # never exceeds the budget + assert removed_per_run[-1] > 0 # carryover is not reported spuriously + assert len(self.remaining_ids("e1")) == 1 # the survivor + + def test_dedup_never_crosses_entities(self) -> None: + a_ids: list[UUID] = [ + self.add_row(STATUS_BLOCKED, entity="a", age_days=2) for _ in range(3) + ] + b_ids: list[UUID] = [ + self.add_row(STATUS_BLOCKED, entity="b", age_days=2) for _ in range(4) + ] + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 5 + assert self.remaining_ids("a") == [a_ids[0]] + assert self.remaining_ids("b") == [b_ids[0]] + + def test_resolved_streak_has_no_survivor_and_ages_out(self) -> None: + """Expire blocked rows after newer evidence resolves their streak.""" + for _ in range(3): + self.add_row(STATUS_BLOCKED, age_days=200) + confirmed: UUID = self.add_row(STATUS_CONFIRMED, age_days=100) + new_survivor: UUID = self.add_row(STATUS_BLOCKED, age_days=50) + self.add_row(STATUS_BLOCKED, age_days=40) # duplicate in the new streak + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 1 + assert result.operational_expired == 3 + assert set(self.remaining_ids("e1")) == {confirmed, new_survivor} + + def test_a_failed_attempt_does_not_reset_the_blocked_since_survivor(self) -> None: + """Keep one blockage streak across a failed purge attempt.""" + blocked_since: UUID = self.add_row(STATUS_BLOCKED, age_days=200) + self.add_row(STATUS_FAILED, age_days=150) # ages out as operational + self.add_row(STATUS_BLOCKED, age_days=100) + self.add_row(STATUS_BLOCKED, age_days=50) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 2 + assert result.operational_expired == 1 + assert self.remaining_ids("e1") == [blocked_since] + + # -- US2: evidence survives by default ---------------------------------- + + def test_defaults_leave_completed_destruction_evidence_untouched(self) -> None: + evidence: list[UUID] = [ + self.add_row(STATUS_CONFIRMED, entity="ev", age_days=3650), + self.add_row(STATUS_TARGET_ABSENT, entity="ev", age_days=1000), + self.add_row(STATUS_CONFIRMED, entity="ev", age_days=1), + ] + self.add_row(STATUS_FAILED, entity="ev", age_days=365) # ages out + + for _ in range(3): # any sequence of runs (SC-002) + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.evidence_expired == 0 + assert set(self.remaining_ids("ev")) == set(evidence) + + def test_evidence_opt_in_expires_only_rows_older_than_its_window(self) -> None: + old: UUID = self.add_row(STATUS_CONFIRMED, entity="ev", age_days=400) + young: UUID = self.add_row(STATUS_TARGET_ABSENT, entity="ev", age_days=300) + + result: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: 365} + ) + + assert result.evidence_expired == 1 + assert self.remaining_ids("ev") == [young] + assert old not in self.remaining_ids("ev") + + def test_opt_in_disabled_again_removes_no_further_evidence(self) -> None: + self.add_row(STATUS_CONFIRMED, entity="ev", age_days=400) + survivor: UUID = self.add_row(STATUS_CONFIRMED, entity="ev", age_days=390) + + first: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: 395} + ) + assert first.evidence_expired == 1 + + second: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: None} + ) + assert second.evidence_expired == 0 + assert self.remaining_ids("ev") == [survivor] + + def test_pending_and_future_rows_survive_every_configuration(self) -> None: + pending: UUID = self.add_row(STATUS_PENDING, entity="px", age_days=3650) + future_blocked: UUID = self.add_row(STATUS_BLOCKED, entity="px", age_days=-1) + future_confirmed: UUID = self.add_row( + STATUS_CONFIRMED, entity="px", age_days=-2 + ) + + result: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 1, EVIDENCE_RETENTION_KEY: 1} + ) + + assert result.total_removed == 0 + assert set(self.remaining_ids("px")) == { + pending, + future_blocked, + future_confirmed, + } + + def test_a_future_dated_outcome_cannot_resolve_a_live_streak(self) -> None: + """Exclude future-dated evidence from current streak boundaries.""" + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="sk", age_days=500) + self.add_row(STATUS_BLOCKED, entity="sk", age_days=400) + skewed: UUID = self.add_row(STATUS_CONFIRMED, entity="sk", age_days=-30) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 1 + assert result.operational_expired == 0 + assert set(self.remaining_ids("sk")) == {blocked_since, skewed} + + # -- Survivor invariant under boundary removal (the review's HIGH) ------ + + def test_evidence_expiry_spares_a_boundary_that_still_bounds_blocked_rows( + self, + ) -> None: + """Keep evidence while it still bounds surviving blocked rows.""" + blocked: list[UUID] = [ + self.add_row(STATUS_BLOCKED, entity="bd", age_days=30) for _ in range(3) + ] + boundary: UUID = self.add_row(STATUS_CONFIRMED, entity="bd", age_days=20) + + # Evidence window far shorter than the operational one: the boundary + # is past its cutoff, the blocked rows it resolved are not. + first: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 90, EVIDENCE_RETENTION_KEY: 1} + ) + + assert first.evidence_expired == 0 + assert set(self.remaining_ids("bd")) == {*blocked, boundary} + + # Once the blocked rows age out, the boundary bounds nothing and + # becomes expirable — the guard defers, it does not immortalize. + # Categories drain in order within one run, so the blocked rows go + # first and the boundary follows in the same pass. + second: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 10, EVIDENCE_RETENTION_KEY: 1} + ) + assert second.operational_expired == 3 + assert second.evidence_expired == 1 + assert self.remaining_ids("bd") == [] + + def test_a_block_after_an_unresolved_attempt_is_not_treated_as_a_duplicate( + self, + ) -> None: + """An in-flight attempt is a boundary waiting to happen. + + Finalizing a ``pending`` row resolves it *in place*, keeping its + original timestamp — the one way a boundary can appear in the middle + of history. The blocked row after it would become the new streak's + survivor, so it must not be classified as a duplicate while the + attempt is unresolved. Deleting it would destroy the only record + that the entity was still blocked after that attempt. + """ + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="ua", age_days=100) + attempt: UUID = self.add_row(STATUS_PENDING, entity="ua", age_days=50) + later_block: UUID = self.add_row(STATUS_BLOCKED, entity="ua", age_days=10) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 0 + assert set(self.remaining_ids("ua")) == {blocked_since, attempt, later_block} + + # Once the attempt finalizes, the boundary is real: the later block + # is the new streak's survivor and stays; the older block is now a + # resolved-streak row that ages out on the normal window. + db.session.execute( + sa.update(PurgeAuditLog.__table__) + .where(PurgeAuditLog.__table__.c.id == attempt) + .values(status=STATUS_CONFIRMED) + ) + db.session.commit() + + after: prune_audit.PruneRunResult = self.run_prune() + + assert after.blocked_duplicates == 0 + assert after.operational_expired == 1 # the pre-attempt block + assert set(self.remaining_ids("ua")) == {attempt, later_block} + + def test_age_does_not_make_an_unstable_block_expirable(self) -> None: + """Keep aged blocked rows whose classification remains unstable. + + Seeded entirely outside the retention window, so only the guard — + not the cutoff — can save the row. + """ + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="ao", age_days=100) + attempt: UUID = self.add_row(STATUS_PENDING, entity="ao", age_days=98) + later_block: UUID = self.add_row(STATUS_BLOCKED, entity="ao", age_days=95) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 0 + assert result.operational_expired == 0 + assert set(self.remaining_ids("ao")) == {blocked_since, attempt, later_block} + + # Resolving the attempt makes the boundary real: the later block is + # the current streak's survivor and is exempt regardless of age, + # while the pre-attempt block ages out. + db.session.execute( + sa.update(PurgeAuditLog.__table__) + .where(PurgeAuditLog.__table__.c.id == attempt) + .values(status=STATUS_TARGET_ABSENT) + ) + db.session.commit() + + after: prune_audit.PruneRunResult = self.run_prune() + + assert after.operational_expired == 1 + assert set(self.remaining_ids("ao")) == {attempt, later_block} + + def test_evidence_guard_also_defers_to_an_unresolved_older_attempt(self) -> None: + """Treat an older pending attempt as potential blocked evidence.""" + attempt: UUID = self.add_row(STATUS_PENDING, entity="ug", age_days=400) + boundary: UUID = self.add_row(STATUS_CONFIRMED, entity="ug", age_days=300) + + result: prune_audit.PruneRunResult = self.run_prune( + **{EVIDENCE_RETENTION_KEY: 200} + ) + + assert result.evidence_expired == 0 + assert set(self.remaining_ids("ug")) == {attempt, boundary} + + def test_blocked_rows_without_an_entity_uuid_are_never_aged_out(self) -> None: + """Keep UUID-less blocked rows that cannot be proven redundant.""" + anonymous_block: UUID = self.add_row(STATUS_BLOCKED, entity=None, age_days=1000) + self.add_row(STATUS_FAILED, entity=None, age_days=1000) + + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.operational_expired == 1 + assert self.remaining_ids(None) == [anonymous_block] + + # -- US3: operator controls and observability --------------------------- + + def test_configured_retention_window_is_honored(self) -> None: + kept: UUID = self.add_row(STATUS_FAILED, entity="w", age_days=5) + self.add_row(STATUS_FAILED, entity="w", age_days=15) + + result: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 10} + ) + + assert result.operational_expired == 1 + assert self.remaining_ids("w") == [kept] + + def test_invalid_operational_window_skips_the_category_not_widens(self) -> None: + """Fail closed when the operational retention window is invalid.""" + old_failed: UUID = self.add_row(STATUS_FAILED, entity="iv", age_days=1000) + self.add_row(STATUS_BLOCKED, entity="iv", age_days=3) + self.add_row(STATUS_BLOCKED, entity="iv", age_days=2) + + result: prune_audit.PruneRunResult = self.run_prune( + **{OPERATIONAL_RETENTION_KEY: 0} + ) + + assert result.operational_expired == 0 + assert result.blocked_duplicates == 1 + assert result.invalid_config_keys == [OPERATIONAL_RETENTION_KEY] + assert old_failed in self.remaining_ids("iv") + + def test_second_run_over_the_same_candidates_removes_and_reports_zero(self) -> None: + """Report zero when rerunning over an already-pruned history.""" + for _ in range(4): + self.add_row(STATUS_BLOCKED, entity="cc", age_days=2) + self.add_row(STATUS_FAILED, entity="cc", age_days=100) + + first: prune_audit.PruneRunResult = self.run_prune() + assert first.blocked_duplicates == 3 + assert first.operational_expired == 1 + + second: prune_audit.PruneRunResult = self.run_prune() + assert second.total_removed == 0 + assert len(self.remaining_ids("cc")) == 1 + + def test_a_duplicate_backlog_cannot_starve_the_age_based_categories(self) -> None: + """Prevent duplicate backlogs from starving age-based categories.""" + for _ in range(20): + self.add_row(STATUS_BLOCKED, entity="st", age_days=2) + old_failed: UUID = self.add_row(STATUS_FAILED, entity="st", age_days=1000) + + with ( + patch.object(prune_audit, "BATCH_SIZE", 1), + patch.object(prune_audit, "MAX_BATCHES_PER_RUN", 2), + ): + result: prune_audit.PruneRunResult = self.run_prune() + + assert result.blocked_duplicates == 1 # budget-limited, as expected + assert result.operational_expired == 1 # but age-out still progressed + assert result.carried_over is True + assert old_failed not in self.remaining_ids("st") + + def test_delete_rechecks_survivor_after_a_pending_attempt_appears(self) -> None: + """Evaluate survivor safety inside the DELETE statement. + + Constructing the candidate query must not freeze its result. A pending + attempt committed before execution can become a mid-history boundary, + so the later blocked row must remain available as its future survivor. + """ + blocked_since: UUID = self.add_row(STATUS_BLOCKED, entity="rc", age_days=3) + later_block: UUID = self.add_row(STATUS_BLOCKED, entity="rc", age_days=1) + select_candidates: Callable[[int], sa.sql.Select] = partial( + prune_audit._duplicate_candidates, audit.utc_now() + ) + + attempt: UUID = self.add_row(STATUS_PENDING, entity="rc", age_days=2) + removed: int = prune_audit._delete_batch(select_candidates) Review Comment: <div> <div id="suggestion"> <div id="issue"><b>Replace private API with public interface</b></div> <div id="fix"> Accessing private member `_delete_batch` may break if internal implementation changes; consider using a public interface or clearly document the intentional coupling. </div> </div> <small><i>Code Review Run #398665</i></small> </div> --- Should Bito avoid suggestions like this for future reviews? (<a href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>) - [ ] Yes, avoid them -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
