sadpandajoe commented on code in PR #44892:
URL: https://github.com/apache/superset/pull/44892#discussion_r4199522861
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1398,283 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ The shared cleanup empties associations before owned children, while an
+ owned table's predicate selects its rows *through* its ownership path. If
+ a hop on that path is an association, its rows are already gone when the
+ owned delete runs, so the statement matches nothing.
+
+ Kept where the rest of the ordering rules were dropped because this one
+ can be silent: where foreign keys are enforced the association delete
+ fails loudly, but where they are not -- SQLite -- the root is purged and
+ its descendants are left orphaned with nothing reported. The remedy is to
+ classify the intermediate table as owned, which places it in the same
+ phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
- raise RuntimeError(
- f"Missing listener action for {dependency.key.describe()}"
- )
- if (
- dependency.classification is DependencyClassification.VERSION_OWNED
- and dependency.version_column is None
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage, and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _validate_recursive_ownership(policy: PurgeEntityPolicy) -> None:
+ """Reject a self-referencing owned table under one-level cleanup.
+
+ Cleanup that issues one statement per declared edge prunes a table that
+ owns itself one level deep. Where foreign keys are enforced the root's own
+ delete then fails and rolls back; where they are not -- SQLite -- the root
+ is purged and its grandchildren are left behind with a dangling parent id
+ and nothing reported. A policy declaring a tree walks it itself and says
+ so with ``walks_own_subtree``.
+ """
+ for dependency in policy.dependencies:
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ key: DependencyKey = dependency.key
+ if key.owner_table != key.related_table:
+ continue
+ raise RuntimeError(
+ f"Owned dependency {key.describe()} is self-referencing; the stock
"
+ "owned-child cleanup deletes one level, so this policy must supply
"
+ "its own delete_owned_children"
+ )
+
+
+def _validate_scanner_requirements(policy: PurgeEntityPolicy) -> None:
+ """Reject a root the scheduled scan cannot page through.
+
+ The retention task selects, windows and orders eligible rows by ``id``,
+ and the cascade pins each row by it. A root keyed on something else -- a
+ UUID primary key with no ``id`` column -- raises inside the scan, outside
+ the per-row error handling, so the run aborts before the remaining roots
+ are reached.
+ """
+ mapper: Mapper[Any] = sa.inspect(policy.model)
+ table: sa.Table = mapper.local_table
+ if "id" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'id' column; the "
+ "scheduled scan pages eligible rows by id"
+ )
+ primary_key: tuple[sa.Column[Any], ...] = tuple(mapper.primary_key)
Review Comment:
A host root with an integer `id`, `deleted_at`, and an ordinary `sa.Uuid()`
`uuid` column clears this scanner validation, but `_identity_predicates` in
`purge_cascade.py` then binds `uuid_column == str(entity.uuid)`. On
SQLite/MySQL the non-native `Uuid` bind processor calls `.hex` on the value, so
every eligible row raises a `StatementError`, stays archived, and a dry run
still counts it as purgeable. Could admission also check that a `uuid` column,
when present, accepts the string the identity predicate binds, or could the
predicate bind the column's own Python type?
##########
tests/unit_tests/commands/deletion_retention/test_purge_policy.py:
##########
@@ -568,3 +580,1123 @@ def
test_core_delete_actions_compile_for_supported_dialects(dialect: str) -> Non
assert compiled
assert all(statement.startswith(("SELECT", "DELETE")) for statement in
compiled)
+
+
+#: Throwaway mapper registries created by the fixtures below, disposed after
+#: each test. Left in place they stay in SQLAlchemy's global mapper state,
+#: where ``configure_mappers()`` in the real-graph tests would walk them.
+_HOST_MAPPERS: list[registry] = []
+
+
[email protected](autouse=True)
+def _dispose_host_mappers() -> Iterator[None]:
+ """Drop the mappers a test mapped, as the deep-path test does."""
+ yield
+ while _HOST_MAPPERS:
+ _HOST_MAPPERS.pop().dispose()
+
+
[email protected]
+def versioned_host_root(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Make any root resolve a version class.
+
+ The root-relative version rules only come into play once the cascade
+ reaches the root at all, which it does by resolving the root's own version
+ class. A throwaway root has none, so these tests stand one in.
+ """
+ import sqlalchemy_continuum
+
+ monkeypatch.setattr(sqlalchemy_continuum, "version_class", lambda model:
model)
+
+
+def _map_host_root(model: type[Any], table: sa.Table) -> type[Any]:
+ """Map *model* onto *table* for the duration of one test."""
+ mapper_registry: registry = registry()
+ _HOST_MAPPERS.append(mapper_registry)
+ mapper_registry.map_imperatively(model, table)
+ return model
+
+
+def _host_root(prefix: str) -> type[Any]:
+ """Map a throwaway root in its own ``MetaData``.
+
+ The hook is indifferent to a host root's shape, so this is the smallest
+ graph with anything to declare: one outbound foreign key, the same edge
+ every built-in policy preserves for its ``ab_user`` columns.
+ """
+ metadata: sa.MetaData = sa.MetaData()
+ sa.Table(
+ f"{prefix}_owner",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ )
+ root_table: sa.Table = sa.Table(
+ f"{prefix}_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("owner_id", sa.Integer, sa.ForeignKey(f"{prefix}_owner.id")),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+
+ class HostRoot:
+ """Temporary mapped root standing in for an entity core does not
own."""
+
+ return _map_host_root(HostRoot, root_table)
+
+
+def _host_edge(prefix: str) -> DependencyPolicy:
+ """The one dependency ``discover_dependencies`` finds for that root."""
+ return DependencyPolicy(
+ DependencyKey(
+ "foreign_key",
+ f"{prefix}_entity",
+ f"{prefix}_owner",
+ ("owner_id",),
+ ("id",),
+ "outbound",
+ ),
+ DependencyClassification.PRESERVE,
+ )
+
+
+def _host_policy(
+ model: type[Any], dependencies: tuple[DependencyPolicy, ...]
+) -> PurgeEntityPolicy:
+ """Borrow the chart policy's actions for a host root.
+
+ The documented clone pattern, kept as the fixture so the tests exercise
+ what an integrator would actually write. The chart-specific snapshots it
+ carries are inert on a host root: each resolves from an entity type a
+ host cannot claim.
+ """
+ return replace(
+ get_purge_policy(Slice),
+ model=model,
+ entity_type="host_root",
+ dependencies=dependencies,
+ )
+
+
+@contextmanager
+def _installed(provider: Any) -> Iterator[None]:
+ """Install a host policy provider for the duration of one test.
+
+ Typed loosely on purpose: some tests install a value that is not a
+ provider at all, which is what the host boundary has to cope with.
+ """
+ previous: Any = current_app.config.get(HOST_POLICIES_CONFIG_KEY)
+ current_app.config[HOST_POLICIES_CONFIG_KEY] = provider
+ try:
+ yield
+ finally:
+ current_app.config[HOST_POLICIES_CONFIG_KEY] = previous
+
+
+def _assert_rejected(
+ model: type[Any],
+ policy: PurgeEntityPolicy,
+ reason: str,
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """A rejected declaration costs the host its root, with the reason logged.
+
+ Rejection happens where the policy is admitted, so the root simply reads
+ as unsupported -- the shape the retention task already reports -- rather
+ than raising once per eligible row.
+ """
+ with _installed(lambda: [policy]), caplog.at_level(logging.ERROR):
+ assert model not in purge_policy_registry()
+ with pytest.raises(ValueError, match="Unsupported purge model"):
+ get_purge_policy(model)
+ assert reason in caplog.text
+
+
+def test_host_policy_extends_the_registry() -> None:
+ """A complete host declaration resolves like a built-in root."""
+ model: type[Any] = _host_root("extend")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("extend"),))
+
+ with _installed(lambda: [policy]):
+ assert get_purge_policy(model) is policy
+ assert {Slice, Dashboard, SqlaTable} <= set(purge_policy_registry())
+
+
+def test_host_policy_is_held_to_the_discovered_graph(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """An incomplete host declaration is rejected, not silently honored."""
+ model: type[Any] = _host_root("incomplete")
+
+ _assert_rejected(model, _host_policy(model, ()), "Incomplete purge
policy", caplog)
+
+
[email protected](
+ "provider",
+ [
+ pytest.param(lambda: 1 // 0, id="raises"),
+ pytest.param(lambda: "not-a-sequence", id="wrong_type"),
+ pytest.param(lambda: [object()], id="wrong_member"),
+ ],
+)
+def test_malformed_host_provider_leaves_builtin_roots_intact(
+ provider: Callable[[], Any],
+) -> None:
+ """A broken host boundary costs the host its roots, not the purge."""
+ with _installed(provider):
+ assert set(purge_policy_registry()) == {Slice, Dashboard, SqlaTable}
+
+
+def test_host_policies_need_an_app_context() -> None:
+ """Off an app context the host's roots drop out; the built-ins remain.
+
+ The provider lives in config, so there is nothing to read without an
+ application context. Dropping the host's roots leaves them reported as
+ unsupported, where propagating the Flask error would abort the whole
+ scheduled run -- including the roots that do have policies.
+ """
+ model: type[Any] = _host_root("contextless")
+ policy: PurgeEntityPolicy = _host_policy(model,
(_host_edge("contextless"),))
+
+ with (
+ _installed(lambda: [policy]),
+ patch(
+
"superset.commands.deletion_retention.purge_policy.has_app_context",
+ return_value=False,
+ ),
+ ):
+ assert set(purge_policy_registry()) == {Slice, Dashboard, SqlaTable}
+
+
+def test_duplicate_host_declarations_do_not_abort_the_registry() -> None:
+ """Two policies for one host root drop that root, not the whole index."""
+ model: type[Any] = _host_root("duplicate")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("duplicate"),))
+
+ with _installed(lambda: [policy, policy]):
+ assert set(purge_policy_registry()) == {Slice, Dashboard, SqlaTable}
+ with pytest.raises(ValueError, match="Unsupported purge model"):
+ get_purge_policy(model)
+
+
+def test_host_policy_cannot_redeclare_a_builtin_root() -> None:
+ """A host cannot redefine how a chart, dashboard or dataset is purged."""
+ builtin: PurgeEntityPolicy = get_purge_policy(Slice)
+
+ with _installed(lambda: [replace(builtin, dependencies=())]):
+ assert purge_policy_registry()[Slice] is builtin
+
+
+def _host_chain(prefix: str) -> type[Any]:
+ """Map a root that reaches a detail table only through a link table."""
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ f"{prefix}_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+ sa.Table(
+ f"{prefix}_link",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("entity_id", sa.Integer,
sa.ForeignKey(f"{prefix}_entity.id")),
+ )
+ sa.Table(
+ f"{prefix}_detail",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("link_id", sa.Integer, sa.ForeignKey(f"{prefix}_link.id")),
+ )
+
+ class HostChainRoot:
+ """Temporary mapped root with a two-hop ownership path."""
+
+ return _map_host_root(HostChainRoot, root_table)
+
+
+def _host_chain_edges(
+ prefix: str, link: DependencyClassification
+) -> tuple[DependencyPolicy, ...]:
+ """Classify the chain, varying only how the intermediate hop is
declared."""
+ entity: str = f"{prefix}_entity"
+ link_table: str = f"{prefix}_link"
+ detail: str = f"{prefix}_detail"
+ link_phase: ExecutionPhase = (
+ ExecutionPhase.ASSOCIATIONS
+ if link is DependencyClassification.ASSOCIATION
+ else ExecutionPhase.OWNED
+ )
+ return (
+ DependencyPolicy(
+ DependencyKey(
+ "foreign_key", entity, link_table, ("id",), ("entity_id",),
"inbound"
+ ),
+ link,
+ link_phase,
+ ),
+ DependencyPolicy(
+ DependencyKey(
+ "foreign_key", link_table, detail, ("id",), ("link_id",),
"inbound"
+ ),
+ DependencyClassification.OWNED,
+ ExecutionPhase.OWNED,
+ ),
+ DependencyPolicy(
+ DependencyKey(
+ "foreign_key", link_table, entity, ("entity_id",), ("id",),
"outbound"
+ ),
+ DependencyClassification.PRESERVE,
+ ),
+ DependencyPolicy(
+ DependencyKey(
+ "foreign_key", detail, link_table, ("link_id",), ("id",),
"outbound"
+ ),
+ DependencyClassification.PRESERVE,
+ ),
+ )
+
+
+def test_owned_table_behind_an_owned_link_is_accepted() -> None:
+ """Declaring the intermediate hop owned puts both in the same phase."""
+ model: type[Any] = _host_chain("owned")
+ policy: PurgeEntityPolicy = _host_policy(
+ model, _host_chain_edges("owned", DependencyClassification.OWNED)
+ )
+
+ with _installed(lambda: [policy]):
+ assert get_purge_policy(model) is policy
+
+
+def _host_bare_root(prefix: str, *columns: str) -> type[Any]:
+ """Map a host root carrying only *columns*, to probe scan requirements."""
+ available: dict[str, sa.Column[Any]] = {
+ "id": sa.Column("id", sa.Integer, primary_key=True),
+ "uuid": sa.Column("uuid", sa.String(36), primary_key=True),
+ "deleted_at": sa.Column("deleted_at", sa.DateTime, nullable=True),
+ }
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ f"{prefix}_entity", metadata, *(available[name] for name in columns)
+ )
+
+ class HostBareRoot:
+ """Temporary mapped root missing a column some purge path needs."""
+
+ return _map_host_root(HostBareRoot, root_table)
+
+
+def _host_referenced(prefix: str) -> type[Any]:
+ """Map a host root with one inbound foreign key and nothing else."""
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ f"{prefix}_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+ sa.Table(
+ f"{prefix}_ref",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("entity_id", sa.Integer,
sa.ForeignKey(f"{prefix}_entity.id")),
+ )
+
+ class HostReferencedRoot:
+ """Temporary mapped root referenced by one other table."""
+
+ return _map_host_root(HostReferencedRoot, root_table)
+
+
+def _inbound_ref_key(prefix: str) -> DependencyKey:
+ return DependencyKey(
+ "foreign_key",
+ f"{prefix}_entity",
+ f"{prefix}_ref",
+ ("id",),
+ ("entity_id",),
+ "inbound",
+ )
+
+
+def _host_tree(prefix: str) -> type[Any]:
+ """Map a host root whose table owns itself through a parent column."""
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ f"{prefix}_node",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("parent_id", sa.Integer, sa.ForeignKey(f"{prefix}_node.id")),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+
+ class HostTreeRoot:
+ """Temporary mapped root with a recursive sub-tree."""
+
+ return _map_host_root(HostTreeRoot, root_table)
+
+
+def _host_tree_edges(prefix: str) -> tuple[DependencyPolicy, ...]:
+ node: str = f"{prefix}_node"
+ return (
+ DependencyPolicy(
+ DependencyKey(
+ "foreign_key", node, node, ("id",), ("parent_id",), "inbound"
+ ),
+ DependencyClassification.OWNED,
+ ExecutionPhase.OWNED,
+ ),
+ DependencyPolicy(
+ DependencyKey(
+ "foreign_key", node, node, ("parent_id",), ("id",), "outbound"
+ ),
+ DependencyClassification.PRESERVE,
+ ),
+ )
+
+
+def test_root_without_an_id_column_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """The scheduled scan pages by id, so a root must have one."""
+ model: type[Any] = _host_bare_root("uuidonly", "uuid")
+
+ _assert_rejected(model, _host_policy(model, ()), "has no 'id' column",
caplog)
+
+
+def test_self_referencing_owned_table_accepts_declared_subtree_cleanup() ->
None:
+ """A policy that says it walks the sub-tree may declare the shape."""
+ model: type[Any] = _host_tree("customtree")
+ policy: PurgeEntityPolicy = replace(
+ _host_policy(model, _host_tree_edges("customtree")),
+ delete_owned_children=lambda session, policy, entity_id: None,
+ walks_own_subtree=True,
+ )
+
+ with _installed(lambda: [policy]):
+ assert get_purge_policy(model) is policy
+
+
+def test_discovery_walks_each_recursive_table_once(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """Traversal is per table, not per permutation of tables.
+
+ Five link tables walked per permutation is already hundreds of visits;
+ a dozen is billions, which a host root could reach before any row is
+ purged.
+ """
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ "wide_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ )
+ link_names: list[str] = [f"wide_link{index}" for index in range(5)]
+ for name in link_names:
+ sa.Table(
+ name,
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("entity_id", sa.Integer,
sa.ForeignKey("wide_entity.id")),
+ )
+
+ class WideRoot:
+ """Temporary mapped root owning several independent link tables."""
+
+ _map_host_root(WideRoot, root_table)
+
+ walked: list[str] = []
+ original = purge_policy_module._discover_table_dependencies # noqa: SLF001
+
+ def counting(table: sa.Table, recursive_tables: Any, seen: Any) -> Any:
+ walked.append(table.name)
+ return original(table, recursive_tables, seen)
+
+ monkeypatch.setattr(purge_policy_module, "_discover_table_dependencies",
counting)
+
+ discover_dependencies(sa.inspect(WideRoot),
recursive_tables=frozenset(link_names))
+
+ assert sorted(walked) == sorted(link_names)
+
+
+def test_root_without_a_deleted_at_column_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Every purge path requires the row to be archived first."""
+ model: type[Any] = _host_bare_root("noflag", "id")
+
+ _assert_rejected(model, _host_policy(model, ()), "no 'deleted_at' column",
caplog)
+
+
+def test_discovery_keeps_a_child_version_shadow_reached_through_its_mapper()
-> None:
+ """A table walked plainly sees fewer edges than one walked as a mapper.
+
+ ``sql_metrics`` is reachable both ways from the dataset root. Tracking the
+ two kinds of walk together, rather than separately, dropped its version
+ shadow -- which then reads as a stale declaration on the dataset policy.
+ This pins the edge the split exists to preserve.
+ """
+ configure_mappers()
+ policy: PurgeEntityPolicy = get_purge_policy(SqlaTable)
+ recursive_tables: frozenset[str] = frozenset(
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {
+ DependencyClassification.OWNED,
+ DependencyClassification.ASSOCIATION,
+ }
+ )
+
+ discovered: frozenset[DependencyKey] = discover_dependencies(
+ sa.inspect(SqlaTable), recursive_tables=recursive_tables
+ )
+
+ assert (
+ DependencyKey(
+ kind="relationship",
+ owner_table="sql_metrics",
+ related_table="sql_metrics_version",
+ direction="onetomany",
+ relationship="versions",
+ )
+ in discovered
+ )
+
+
+def test_provider_is_invoked_once_however_many_roots_resolve() -> None:
+ """The index is cached per provider, so a slow provider runs once."""
+ model: type[Any] = _host_root("once")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("once"),))
+ invocations: list[int] = []
+
+ def provider() -> list[PurgeEntityPolicy]:
+ invocations.append(1)
+ return [policy]
+
+ with _installed(provider):
+ for _ in range(5):
+ assert get_purge_policy(model) is policy
+ assert get_purge_policy(Slice) is not None
+
+ assert len(invocations) == 1
+
+
+def test_host_policy_claiming_a_built_in_entity_type_is_dropped(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Core branches on entity_type, so its names are reserved."""
+ model: type[Any] = _host_root("claimed")
+ policy: PurgeEntityPolicy = replace(
+ _host_policy(model, (_host_edge("claimed"),)), entity_type="dataset"
+ )
+
+ _assert_rejected(model, policy, "reserved entity type", caplog)
+
+
[email protected](
+ "root",
+ [
+ pytest.param(
+ sa.Table(
+ "host_not_a_class",
+ sa.MetaData(),
+ sa.Column("id", sa.Integer, primary_key=True),
+ ),
+ id="table",
+ ),
+ pytest.param([], id="unhashable"),
+ ],
+)
+def test_host_policy_with_a_non_class_root_is_dropped(
+ root: Any, caplog: pytest.LogCaptureFixture
+) -> None:
+ """A root that is not a class is dropped without touching the index.
+
+ Reading or hashing it must not happen before the boundary can refuse it:
+ an exception escaping here would break the registry for the built-in
+ roots too.
+ """
+ policy: PurgeEntityPolicy = replace(
+ get_purge_policy(Slice), model=root, entity_type="host_root",
dependencies=()
+ )
+
+ with _installed(lambda: [policy]), caplog.at_level(logging.ERROR):
+ assert set(purge_policy_registry()) == {Slice, Dashboard, SqlaTable}
+
+ assert "is not a class" in caplog.text
+
+
+def test_host_policies_may_arrive_as_any_iterable() -> None:
+ """The contract is a sequence of policies, not one particular container."""
+ model: type[Any] = _host_root("generated")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("generated"),))
+
+ with _installed(lambda: (item for item in [policy])):
+ assert get_purge_policy(model) is policy
+
+
+def test_root_with_a_composite_primary_key_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """The scan pages by id and then fetches the row with a scalar lookup.
+
+ A composite key matches nothing, so every row fails while a dry run still
+ counts it as purgeable.
+ """
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ "composite_entity",
+ metadata,
+ sa.Column("tenant_id", sa.Integer, primary_key=True),
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+
+ class CompositeRoot:
+ """Temporary mapped root keyed on two columns."""
+
+ _map_host_root(CompositeRoot, root_table)
+
+ _assert_rejected(
+ CompositeRoot, _host_policy(CompositeRoot, ()), "is keyed on", caplog
+ )
+
+
+def test_version_target_on_an_intermediate_owner_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+ versioned_host_root: None,
+) -> None:
+ """Version cleanup compares the declared column with the root's id.
+
+ A column that is not root-relative matches another root's history and
+ leaves this root's behind.
+ """
+ model: type[Any] = _host_referenced("versioned")
+ declared: tuple[DependencyPolicy, ...] = (
+ DependencyPolicy(
+ DependencyKey(
+ "relationship",
+ "versioned_ref",
+ "versioned_ref_version",
+ direction="onetomany",
+ relationship="versions",
+ ),
+ DependencyClassification.VERSION_OWNED,
+ ExecutionPhase.VERSION,
+ version_column="id",
+ ),
+ )
+
+ _assert_rejected(
+ model, _host_policy(model, declared), "is not root-relative", caplog
+ )
+
+
+def test_root_whose_table_name_is_ambiguous_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Identities are recorded by bare name while metadata keys them by schema.
+
+ With the same bare name in two schemas, cleanup would resolve to the
+ default-schema table and delete rows belonging to unrelated roots.
+ """
+ metadata: sa.MetaData = sa.MetaData()
+ sa.Table(
+ "ambiguous_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+ qualified: sa.Table = sa.Table(
+ "ambiguous_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ schema="host",
+ )
+
+ class QualifiedRoot:
+ """Temporary mapped root living in a named schema."""
+
+ _map_host_root(QualifiedRoot, qualified)
+
+ _assert_rejected(
+ QualifiedRoot,
+ _host_policy(QualifiedRoot, ()),
+ "unambiguously by name",
+ caplog,
+ )
+
+
+def test_provider_may_build_on_a_built_in_policy() -> None:
+ """Borrowing a built-in policy inside the provider must not stall.
+
+ ``replace(get_purge_policy(Slice), ...)`` is the obvious way for a host to
+ pick up the stock callbacks, and it re-enters resolution while the first
+ call is still inside the provider. Run on a thread so a regression here
+ surfaces as a timeout rather than hanging the suite.
+ """
+ app = current_app._get_current_object() # noqa: SLF001
+ model: type[Any] = _host_root("reentrant")
+
+ def provider() -> list[PurgeEntityPolicy]:
+ # _host_policy itself resolves the chart policy, which is the
+ # re-entrant call under test.
+ return [_host_policy(model, (_host_edge("reentrant"),))]
+
+ resolved: list[PurgeEntityPolicy] = []
+
+ def resolve() -> None:
+ with app.app_context():
+ resolved.append(get_purge_policy(model))
+
+ with _installed(provider):
+ thread = threading.Thread(target=resolve)
+ thread.start()
+ thread.join(timeout=10)
+
+ assert not thread.is_alive(), "resolution did not finish: re-entry stalled"
+ assert len(resolved) == 1
+ assert resolved[0].model is model
+
+
+def test_concurrent_first_use_publishes_a_whole_index() -> None:
+ """Each caller sees a complete index, never a half-built one.
+
+ Two first uses may each invoke the provider -- the deliberate trade for
+ not holding a lock across host code -- but publishing is a single rebind,
+ so neither sees a partial result.
+ """
+ app = current_app._get_current_object() # noqa: SLF001
+ model: type[Any] = _host_root("concurrent")
+ policy: PurgeEntityPolicy = _host_policy(model,
(_host_edge("concurrent"),))
+ both_ready: threading.Barrier = threading.Barrier(2)
+
+ def provider() -> list[PurgeEntityPolicy]:
+ time.sleep(0.05)
+ return [policy]
+
+ seen: list[tuple[PurgeEntityPolicy, set[type[Any]]]] = []
+
+ def resolve() -> None:
+ with app.app_context():
+ both_ready.wait(timeout=5)
+ seen.append((get_purge_policy(model),
set(purge_policy_registry())))
+
+ with _installed(provider):
+ threads: list[threading.Thread] = [
+ threading.Thread(target=resolve) for _ in range(2)
+ ]
+ for thread in threads:
+ thread.start()
+ for thread in threads:
+ thread.join(timeout=10)
+
+ assert len(seen) == 2
+ for found, roots in seen:
+ assert found is policy
+ assert {Slice, Dashboard, SqlaTable} <= roots
+
+
+def test_root_with_a_non_integer_id_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """The scan pages an integer watermark, starting from zero."""
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ "stringid_entity",
+ metadata,
+ sa.Column("id", sa.String(36), primary_key=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+
+ class StringIdRoot:
+ """Temporary mapped root keyed on a string id."""
+
+ _map_host_root(StringIdRoot, root_table)
+
+ _assert_rejected(
+ StringIdRoot, _host_policy(StringIdRoot, ()), "integer 'id'", caplog
+ )
+
+
+def test_version_target_keyed_on_a_non_primary_root_column_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+ versioned_host_root: None,
+) -> None:
+ """Cleanup compares the target with the root's id, not another column.
+
+ A child keyed on some other root column matches rows whose value happens
+ to equal this root's id -- another root's history.
+ """
+ metadata: sa.MetaData = sa.MetaData()
+ root_table: sa.Table = sa.Table(
+ "coded_entity",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("code", sa.Integer, unique=True),
+ sa.Column("deleted_at", sa.DateTime, nullable=True),
+ )
+ sa.Table(
+ "coded_child",
+ metadata,
+ sa.Column("id", sa.Integer, primary_key=True),
+ sa.Column("root_code", sa.Integer, sa.ForeignKey("coded_entity.code")),
+ )
+
+ class CodedRoot:
+ """Temporary mapped root whose child is keyed on a non-primary
column."""
+
+ _map_host_root(CodedRoot, root_table)
+ declared: tuple[DependencyPolicy, ...] = (
+ DependencyPolicy(
+ DependencyKey(
+ "relationship",
+ "coded_child",
+ "coded_child_version",
+ direction="onetomany",
+ relationship="versions",
+ ),
+ DependencyClassification.VERSION_OWNED,
+ ExecutionPhase.VERSION,
+ version_column="root_code",
+ ),
+ )
+
+ _assert_rejected(
+ CodedRoot, _host_policy(CodedRoot, declared), "is not root-relative",
caplog
+ )
+
+
[email protected](
+ "classification",
+ [
+ pytest.param(DependencyClassification.BLOCK, id="block"),
+ pytest.param(DependencyClassification.LISTENER_EFFECT,
id="listener_effect"),
+ ],
+)
+def test_host_policy_declaring_core_only_dependencies_is_dropped(
+ classification: DependencyClassification, caplog: pytest.LogCaptureFixture
+) -> None:
+ """Both resolve from core's own identities, so a host declaration is
silent.
+
+ The stock listener actions decide what to delete from a built-in entity
+ type, and blockers are applied by the stock validator. Declared by a host
+ either one is accepted and then never carried out -- the one outcome worth
+ refusing, since nothing announces it. A host refuses a purge by raising
+ PurgeBlockedError from its own validator instead.
+ """
+ model: type[Any] = _host_referenced("coreonly")
+ declared: tuple[DependencyPolicy, ...] = (
+ DependencyPolicy(
+ _inbound_ref_key("coreonly"),
+ classification,
+ ExecutionPhase.VALIDATE,
+ blocker=BlockerReason("host_reference", "a reference exists"),
+ listener_action=ListenerAction.DELETE_TAGGED_OBJECTS,
+ ),
+ )
+
+ _assert_rejected(
+ model, _host_policy(model, declared), "not available to a host root",
caplog
+ )
+
+
+def test_owned_table_behind_an_association_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Associations are emptied first, so owning through one cannot execute.
+
+ Kept where the other ordering rules were dropped because this one can be
+ silent: without enforced foreign keys the root purges and its descendants
+ are orphaned with nothing reported.
+ """
+ model: type[Any] = _host_chain("behind")
+ policy: PurgeEntityPolicy = _host_policy(
+ model, _host_chain_edges("behind",
DependencyClassification.ASSOCIATION)
+ )
+
+ _assert_rejected(model, policy, "associations are deleted first", caplog)
+
+
+def test_a_failing_provider_is_not_called_again_on_every_read() -> None:
+ """A failure answers reads for a while instead of re-running the provider.
+
+ The registry is read roughly three times per purged entity, so retrying
+ per read meant tens of thousands of provider calls -- and tracebacks --
+ in a single pass.
+ """
+ calls: list[int] = []
+
+ def provider() -> list[PurgeEntityPolicy]:
+ calls.append(1)
+ raise RuntimeError("manager unreachable")
+
+ with _installed(provider):
+ for _ in range(4):
+ assert set(purge_policy_registry()) == {Slice, Dashboard,
SqlaTable}
+
+ assert len(calls) == 1
+
+
+def test_a_failing_provider_is_tried_again_once_its_window_passes(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """And not remembered for the life of the process either.
+
+ The window is set to zero here so the published failure is already stale,
+ which is what a later purge sees after a transient outage.
+ """
+ monkeypatch.setattr(purge_policy_module, "_PROVIDER_RETRY_SECONDS", 0.0)
+ model: type[Any] = _host_root("transient")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("transient"),))
+ calls: list[int] = []
+
+ def provider() -> list[PurgeEntityPolicy]:
+ calls.append(1)
+ if len(calls) == 1:
+ raise RuntimeError("manager unreachable")
+ return [policy]
+
+ with _installed(provider):
+ assert model not in purge_policy_registry()
+ assert get_purge_policy(model) is policy
+
+ assert len(calls) == 2
+
+
+def test_replacing_only_the_association_delete_still_refuses_the_shape(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """The orphaning hazard belongs to the stock owned cleanup alone.
+
+ Whoever empties the association rows, the stock owned delete still
+ traverses an ownership path that is empty by the time it runs.
+ """
+ model: type[Any] = _host_chain("assoconly")
+ policy: PurgeEntityPolicy = replace(
+ _host_policy(
+ model, _host_chain_edges("assoconly",
DependencyClassification.ASSOCIATION)
+ ),
+ delete_associations=lambda session, policy, entity_id: None,
+ )
+
+ _assert_rejected(model, policy, "associations are deleted first", caplog)
+
+
+def test_self_referencing_owned_table_rejects_stock_cleanup(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Stock cleanup prunes one level, which orphans a deeper tree.
+
+ Loud where foreign keys are enforced, silent on SQLite: the root is
+ purged and its grandchildren keep a dangling parent id.
+ """
+ model: type[Any] = _host_tree("stocktree")
+ policy: PurgeEntityPolicy = _host_policy(model,
_host_tree_edges("stocktree"))
+
+ _assert_rejected(model, policy, "must supply its own", caplog)
+
+
+def test_self_referencing_version_target_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+ versioned_host_root: None,
+) -> None:
+ """A parent column on the root's own shadow is not the root's identity.
+
+ Cleanup compares it with the purged row's id, so it deletes that row's
+ children's history and leaves its own behind.
+ """
+ model: type[Any] = _host_tree("selfversion")
+ declared: tuple[DependencyPolicy, ...] = (
+ *_host_tree_edges("selfversion"),
+ DependencyPolicy(
+ DependencyKey(
+ "relationship",
+ "selfversion_node",
+ "selfversion_node_version",
+ direction="onetomany",
+ relationship="versions",
+ ),
+ DependencyClassification.VERSION_OWNED,
+ ExecutionPhase.VERSION,
+ version_column="parent_id",
+ ),
+ )
+ policy: PurgeEntityPolicy = replace(
+ _host_policy(model, declared),
+ delete_owned_children=lambda session, policy, entity_id: None,
+ )
+
+ _assert_rejected(model, policy, "is not root-relative", caplog)
+
+
+def test_each_app_resolves_its_own_index(app_context: None) -> None:
+ """Two apps in one process do not thrash a single global snapshot.
+
+ Without per-app state each read from either app would see the other's
+ provider, re-resolve, and call the provider again -- once per root.
+ """
+ from superset.app import SupersetApp
+
+ first = current_app._get_current_object() # noqa: SLF001
+ model: type[Any] = _host_root("perapp")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("perapp"),))
+ calls: list[int] = []
+
+ def provider() -> list[PurgeEntityPolicy]:
+ calls.append(1)
+ return [policy]
+
+ second = SupersetApp(__name__)
+ second.config.update(first.config)
+ second.config[HOST_POLICIES_CONFIG_KEY] = None
+
+ with _installed(provider):
+ assert get_purge_policy(model) is policy
+ with second.app_context():
+ # The second app has no provider, so its index carries only the
+ # built-in roots -- and resolving it must not disturb the first.
+ assert model not in purge_policy_registry()
+ assert get_purge_policy(model) is policy
+
+ assert len(calls) == 1
+
+
+def test_a_lazy_payload_that_raises_is_retried_like_the_call(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """Walking a generator runs host code, so it earns the provider's retry.
+
+ Reading a lazy payload can fail for the same transient reasons calling
+ the provider can; only the shape of a payload that arrived intact is
+ settled for good.
+ """
+ monkeypatch.setattr(purge_policy_module, "_PROVIDER_RETRY_SECONDS", 0.0)
+ model: type[Any] = _host_root("lazy")
+ policy: PurgeEntityPolicy = _host_policy(model, (_host_edge("lazy"),))
+ calls: list[int] = []
+
+ def provider() -> Iterator[PurgeEntityPolicy]:
+ calls.append(1)
+ if len(calls) == 1:
+ raise RuntimeError("manager unreachable")
+ yield policy # pragma: no cover - unreachable, keeps this a
generator
+ yield policy
+
+ with _installed(provider):
+ assert model not in purge_policy_registry()
+ assert get_purge_policy(model) is policy
+
+ assert len(calls) == 2
+
+
+def test_version_target_on_an_unversioned_root_is_rejected(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """The cascade stops before a declared target when the root has none.
+
+ It resolves the root's own version class first, so for an unversioned root
+ a target on a child's shadow is never reached: the root and its live child
+ rows are purged while the child's history survives, unreported.
+ """
+ model: type[Any] = _host_referenced("unversioned")
+ declared: tuple[DependencyPolicy, ...] = (
+ DependencyPolicy(
+ DependencyKey(
+ "relationship",
+ "unversioned_ref",
+ "unversioned_ref_version",
+ direction="onetomany",
+ relationship="versions",
+ ),
+ DependencyClassification.VERSION_OWNED,
+ ExecutionPhase.VERSION,
+ version_column="entity_id",
+ ),
+ )
+
+ _assert_rejected(
+ model, _host_policy(model, declared), "has no version class", caplog
+ )
+
+
+def test_host_policy_mapped_onto_a_built_in_table_is_dropped(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """A different class on a built-in table is still this package's rows.
+
+ It is not the built-in root by identity, so the class check alone admits
+ it -- and then host cleanup runs against ``slices``. Inheriting the mixin
+ also puts it in the soft-delete registry, so the scheduled task would
+ scan it.
+ """
+
+ class HostSliceClone(Slice):
Review Comment:
Defining `HostSliceClone(Slice)` here appends it to the real
`SoftDeleteMixin._registered_subclasses` through `__init_subclass__`, and
neither this test nor `_dispose_host_mappers` removes it afterward. Any later
test in the same process that scans the live registry will then see a second
`slices` model and report `unsupported_models == {"slices": 1}` even though the
chart policy is installed. Could this snapshot and restore the registry list
the way the `NoTableName` test does?
--
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]
