diff --git a/changelog/572.fixed.md b/changelog/572.fixed.md new file mode 100644 index 00000000..8193f10f --- /dev/null +++ b/changelog/572.fixed.md @@ -0,0 +1,7 @@ +Tracking groups now reconcile correctly when a run saves no nodes at all. Previously `update_group()` returned early on an empty member list, so a generator that produced nothing (a decommissioning run) or a repository whose last object file was removed left every previously tracked node behind as an orphan, still listed in the group. A run that tracks nothing but has an existing group now prunes it; a run that tracks nothing and has no group still creates none. + +Cleanup is also no longer aborted by a single refused delete. `delete_unused()` attempts every unused member and reports the refusals together as `TrackingGroupCleanupError` instead of propagating the first `GraphQLError` and silently skipping the rest. Members that could not be deleted are kept in the tracking group so a later run retries them, and the sync client now has the same error tolerance as the async one. + +A failure that is not a refusal, such as a timeout or an expired token, is not a fact about the member being deleted: it stops the cleanup and is reported as itself rather than recorded against every remaining member in turn. The tracking group is still written before that failure surfaces, listing the members the cleanup did not get through along with the nodes the run created, so an interrupted cleanup can no longer leave those nodes in no group at all. `delete_unused()` returns a `ReapResult` describing all three outcomes rather than a bare mapping of failures. + +`TrackingGroupCleanupError` reports only the server's reason for each member rather than the full mutation that was attempted, `infrahubctl` renders it as a table instead of a traceback, and the exception can be serialized so it survives a task orchestrator's failure handling. diff --git a/infrahub_sdk/client.py b/infrahub_sdk/client.py index 51d3eca6..c47ff3a3 100644 --- a/infrahub_sdk/client.py +++ b/infrahub_sdk/client.py @@ -2295,10 +2295,13 @@ async def __aexit__( exc_value: BaseException | None, traceback: TracebackType | None, ) -> None: - if exc_type is None and self.mode == InfrahubClientMode.TRACKING: - await self.group_context.update_group() - - self.mode = InfrahubClientMode.DEFAULT + try: + if exc_type is None and self.mode == InfrahubClientMode.TRACKING: + await self.group_context.update_group() + finally: + # update_group() can raise, and leaving the client in tracking mode would + # silently enroll every later save into the stale context. + self.mode = InfrahubClientMode.DEFAULT async def convert_object_type( self, @@ -4090,10 +4093,13 @@ def __exit__( exc_value: BaseException | None, traceback: TracebackType | None, ) -> None: - if exc_type is None and self.mode == InfrahubClientMode.TRACKING: - self.group_context.update_group() - - self.mode = InfrahubClientMode.DEFAULT + try: + if exc_type is None and self.mode == InfrahubClientMode.TRACKING: + self.group_context.update_group() + finally: + # update_group() can raise, and leaving the client in tracking mode would + # silently enroll every later save into the stale context. + self.mode = InfrahubClientMode.DEFAULT def convert_object_type( self, diff --git a/infrahub_sdk/ctl/utils.py b/infrahub_sdk/ctl/utils.py index 3898b7d9..07d37822 100644 --- a/infrahub_sdk/ctl/utils.py +++ b/infrahub_sdk/ctl/utils.py @@ -13,6 +13,7 @@ from rich.console import Console from rich.logging import RichHandler from rich.markup import escape +from rich.table import Table from ..exceptions import ( AuthenticationError, @@ -25,6 +26,7 @@ SchemaNotFoundError, ServerNotReachableError, ServerNotResponsiveError, + TrackingGroupCleanupError, ValidationError, ) from ..graphql.query_renderer import render_query @@ -67,6 +69,9 @@ def handle_exception(exc: Exception, console: Console, exit_code: int) -> NoRetu if isinstance(exc, GraphQLError): print_graphql_errors(console=console, errors=exc.errors) raise typer.Exit(code=exit_code) + if isinstance(exc, TrackingGroupCleanupError): + print_tracking_group_failures(console=console, failures=exc.failures) + raise typer.Exit(code=exit_code) if isinstance(exc, (SchemaNotFoundError, NodeNotFoundError, ResourceNotDefinedError, GraphQLQueryError)): console.print(f"[red]Error: {exc!s}") raise typer.Exit(code=exit_code) @@ -148,6 +153,17 @@ def print_graphql_errors(console: Console, errors: list) -> None: console.print(f"[red]{escape(str(error))}") +def print_tracking_group_failures(console: Console, failures: dict[str, str]) -> None: + table = Table(title="Unused tracking group members that could not be deleted") + table.add_column("Node ID") + table.add_column("Reason") + for node_id, reason in failures.items(): + table.add_row(escape(node_id), escape(reason)) + + console.print(table) + console.print("[yellow]These nodes remain members of the tracking group and will be retried on the next run.") + + def parse_cli_vars(variables: list[str] | None) -> dict[str, str]: if not variables: return {} diff --git a/infrahub_sdk/exceptions.py b/infrahub_sdk/exceptions.py index 02111b9a..2f35824c 100644 --- a/infrahub_sdk/exceptions.py +++ b/infrahub_sdk/exceptions.py @@ -66,6 +66,24 @@ def __init__(self, errors: list[dict[str, Any]], query: str | None = None, varia super().__init__(self.message) +class TrackingGroupCleanupError(Error): + """Raised when unused members of a tracking group could not be deleted. + + Every unused member is attempted before this is raised, and the ones that failed are + kept in the tracking group so a later run retries them. + """ + + def __init__(self, failures: dict[str, str]) -> None: + self.failures = failures + details = "; ".join(f"{node_id} ({reason})" for node_id, reason in failures.items()) + super().__init__(f"Unable to delete {len(failures)} unused member(s) of the tracking group: {details}") + + def __reduce__(self) -> tuple[type[TrackingGroupCleanupError], tuple[dict[str, str]]]: + # Rebuild from the failures rather than the formatted message, so the exception + # survives the serialization that a task orchestrator applies to a failed run. + return (self.__class__, (self.failures,)) + + class VersionNotSupportedError(Error): """Raised when a feature is used against an Infrahub server version that does not support it.""" diff --git a/infrahub_sdk/query_groups.py b/infrahub_sdk/query_groups.py index 3fcde346..d6b51a92 100644 --- a/infrahub_sdk/query_groups.py +++ b/infrahub_sdk/query_groups.py @@ -1,10 +1,11 @@ from __future__ import annotations from collections.abc import Sequence +from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any from .constants import InfrahubClientMode -from .exceptions import GraphQLError, NodeNotFoundError +from .exceptions import GraphQLError, NodeNotFoundError, TrackingGroupCleanupError from .utils import dict_hash if TYPE_CHECKING: @@ -12,6 +13,30 @@ from .node import InfrahubNode, InfrahubNodeSync, RelatedNodeBase from .schema import MainSchemaTypesAPI +ALREADY_DELETED_ERROR = "Unable to find the node" + + +@dataclass +class ReapResult: + """How far a reap of the unused members got. + + Attributes: + refused: The id of each member the server refused to delete, mapped to its reason. + unattempted: The ids the reap never got to, because something unrelated to any + member stopped it. + error: The failure that stopped the reap, if one did. + + """ + + refused: dict[str, str] = field(default_factory=dict) + unattempted: list[str] = field(default_factory=list) + error: Exception | None = None + + @property + def retained_member_ids(self) -> list[str]: + """The members that must stay in the group so a later run retries them.""" + return list(self.refused) + self.unattempted + class InfrahubGroupContextBase: """Base class for InfrahubGroupContext and InfrahubGroupContextSync.""" @@ -27,6 +52,7 @@ def __init__(self) -> None: self.delete_unused_nodes: bool = False self.group_type: str = "CoreStandardGroup" self.group_params: dict[str, Any] = {} + self.branch: str | None = None def set_properties( self, @@ -51,6 +77,50 @@ def set_properties( self.group_params = group_params or {} self.branch = branch + def _get_members(self) -> list[str]: + """The ids of everything this run tracked.""" + return self.related_group_ids + self.related_node_ids + + def _set_unused_member_ids(self, previous_member_ids: list[str], members: list[str]) -> None: + self.unused_member_ids = list(set(previous_member_ids) - set(members)) + + def _reap_candidates(self) -> list[tuple[str, str]]: + """The (kind, id) of every previous member this run no longer uses.""" + if not self.previous_members or not self.unused_member_ids: + return [] + + unused_member_ids = set(self.unused_member_ids) + candidates: list[tuple[str, str]] = [] + for member in self.previous_members: + if member.id is None or member.typename is None or member.id not in unused_member_ids: + continue + candidates.append((member.typename, member.id)) + + return candidates + + @staticmethod + def _error_messages(exc: GraphQLError) -> list[str]: + """The server-side message of each error, without the query that triggered them.""" + return [ + str(error.get("message", error)) if isinstance(error, dict) else str(error) for error in exc.errors or [] + ] + + @classmethod + def _is_already_deleted(cls, exc: GraphQLError) -> bool: + """Whether the server reported nothing but missing nodes. + + A member removed by the cascade delete of another member is not a failure, but a + response that also carries a genuine refusal must not be tolerated. + """ + messages = cls._error_messages(exc) + return bool(messages) and all(ALREADY_DELETED_ERROR in message for message in messages) + + @classmethod + def _failure_reason(cls, exc: Exception) -> str: + if isinstance(exc, GraphQLError) and (messages := cls._error_messages(exc)): + return "; ".join(messages) + return str(exc) + def _get_params_as_str(self) -> str: """Convert the params in dict format, into a string.""" params_as_str: list[str] = [] @@ -108,17 +178,37 @@ async def get_group(self, store_peers: bool = False) -> InfrahubNode | None: self.previous_members = group._get_relationship_many(name="members").peers return group - async def delete_unused(self) -> None: - if self.previous_members and self.unused_member_ids: - for member in self.previous_members: - if member.id in self.unused_member_ids and member.typename: - try: - await self.client.delete(kind=member.typename, id=member.id) - except GraphQLError as exc: - if not exc.message or "Unable to find the node" not in exc.message: - # If the node already has been deleted, skip the error as it would have been deleted - # by the cascade delete of another node - raise + async def delete_unused(self) -> ReapResult: + """Delete the members that this run no longer uses. + + A refusal by the server is a fact about the member, so it is recorded against it + and the remaining candidates are still attempted. Any other failure is not a fact + about the member being deleted, so it stops the reap and is returned as itself + rather than blamed on every member in turn. + + Nothing is raised here: the caller has to record this run's membership before + reporting the failure, or the nodes it created are left in no group at all. + + Returns: + How far the reap got, and what must stay in the group. + + """ + result = ReapResult() + candidates = self._reap_candidates() + + for position, (kind, member_id) in enumerate(candidates): + try: + await self.client.delete(kind=kind, id=member_id, branch=self.branch) + except GraphQLError as exc: + if self._is_already_deleted(exc): + continue + result.refused[member_id] = self._failure_reason(exc) + except Exception as exc: + result.error = exc + result.unattempted = [candidate_id for _, candidate_id in candidates[position:]] + break + + return result async def add_related_nodes(self, ids: list[str], update_group_context: bool | None = None) -> None: """Add related Nodes IDs to the context. @@ -147,42 +237,51 @@ async def add_related_groups(self, ids: list[str], update_group_context: bool | self.related_group_ids.extend(ids) async def update_group(self) -> None: - """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution.""" - members: list[str] = self.related_group_ids + self.related_node_ids + """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution. - if not members: - return + Raises: + TrackingGroupCleanupError: When the server refused to delete one or more unused members. + Exception: Whatever stopped the reap, when it was stopped by something other than a refusal. - group_name = self._generate_group_name() - schema = await self.client.schema.get(kind=self.group_type) - description = self._generate_group_description(schema=schema) + """ + members = self._get_members() existing_group = None if self.delete_unused_nodes: existing_group = await self.get_group(store_peers=True) + previous_member_ids: list[str] = [] + if existing_group: + previous_member_ids = existing_group._get_relationship_many(name="members").peer_ids + self._set_unused_member_ids(previous_member_ids=previous_member_ids, members=members) + + reap = await self.delete_unused() + + # A run that tracked nothing and has no members left to reconcile must neither + # create an empty group nor re-save one that is already empty. + if not members and not previous_member_ids: + return + + group_name = self._generate_group_name() + schema = await self.client.schema.get(kind=self.group_type) + description = self._generate_group_description(schema=schema) + + # The group is written before any failure is reported: whatever went wrong, the + # nodes this run created are only reachable later once they are recorded here. group = await self.client.create( kind=self.group_type, name=group_name, description=description, - members=members, + members=members + reap.retained_member_ids, branch=self.branch, **self.group_params, ) await group.save(allow_upsert=True, update_group_context=False) - if not existing_group: - return - - # Calculate how many nodes should be deleted - self.unused_member_ids = list( - set(existing_group._get_relationship_many(name="members").peer_ids) - set(members) - ) - - if not self.delete_unused_nodes: - return - - await self.delete_unused() + if reap.error: + raise reap.error + if reap.refused: + raise TrackingGroupCleanupError(failures=reap.refused) # TODO : create anoter "read" group. Could be based of the store items # Need to filters the store items inherited from CoreGroup to add them as children # Need to validate that it's UUIDas "key" if we want to implement other methods to store item @@ -198,7 +297,9 @@ def __init__(self, client: InfrahubClientSync) -> None: def get_group(self, store_peers: bool = False) -> InfrahubNodeSync | None: group_name = self._generate_group_name() try: - group = self.client.get(kind=self.group_type, name__value=group_name, include=["members"]) + group = self.client.get( + kind=self.group_type, name__value=group_name, include=["members"], branch=self.branch + ) except NodeNotFoundError: return None @@ -208,11 +309,37 @@ def get_group(self, store_peers: bool = False) -> InfrahubNodeSync | None: self.previous_members = group._get_relationship_many(name="members").peers return group - def delete_unused(self) -> None: - if self.previous_members and self.unused_member_ids: - for member in self.previous_members: - if member.id in self.unused_member_ids and member.typename: - self.client.delete(kind=member.typename, id=member.id) + def delete_unused(self) -> ReapResult: + """Delete the members that this run no longer uses. + + A refusal by the server is a fact about the member, so it is recorded against it + and the remaining candidates are still attempted. Any other failure is not a fact + about the member being deleted, so it stops the reap and is returned as itself + rather than blamed on every member in turn. + + Nothing is raised here: the caller has to record this run's membership before + reporting the failure, or the nodes it created are left in no group at all. + + Returns: + How far the reap got, and what must stay in the group. + + """ + result = ReapResult() + candidates = self._reap_candidates() + + for position, (kind, member_id) in enumerate(candidates): + try: + self.client.delete(kind=kind, id=member_id, branch=self.branch) + except GraphQLError as exc: + if self._is_already_deleted(exc): + continue + result.refused[member_id] = self._failure_reason(exc) + except Exception as exc: + result.error = exc + result.unattempted = [candidate_id for _, candidate_id in candidates[position:]] + break + + return result def add_related_nodes(self, ids: list[str], update_group_context: bool | None = None) -> None: """Add related Nodes IDs to the context. @@ -241,42 +368,51 @@ def add_related_groups(self, ids: list[str], update_group_context: bool | None = self.related_group_ids.extend(ids) def update_group(self) -> None: - """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution.""" - members: list[str] = self.related_node_ids + self.related_group_ids + """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution. - if not members: - return + Raises: + TrackingGroupCleanupError: When the server refused to delete one or more unused members. + Exception: Whatever stopped the reap, when it was stopped by something other than a refusal. - group_name = self._generate_group_name() - schema = self.client.schema.get(kind=self.group_type) - description = self._generate_group_description(schema=schema) + """ + members = self._get_members() existing_group = None if self.delete_unused_nodes: existing_group = self.get_group(store_peers=True) + previous_member_ids: list[str] = [] + if existing_group: + previous_member_ids = existing_group._get_relationship_many(name="members").peer_ids + self._set_unused_member_ids(previous_member_ids=previous_member_ids, members=members) + + reap = self.delete_unused() + + # A run that tracked nothing and has no members left to reconcile must neither + # create an empty group nor re-save one that is already empty. + if not members and not previous_member_ids: + return + + group_name = self._generate_group_name() + schema = self.client.schema.get(kind=self.group_type) + description = self._generate_group_description(schema=schema) + + # The group is written before any failure is reported: whatever went wrong, the + # nodes this run created are only reachable later once they are recorded here. group = self.client.create( kind=self.group_type, name=group_name, description=description, - members=members, + members=members + reap.retained_member_ids, branch=self.branch, **self.group_params, ) group.save(allow_upsert=True, update_group_context=False) - if not existing_group: - return - - # Calculate how many nodes should be deleted - self.unused_member_ids = list( - set(existing_group._get_relationship_many(name="members").peer_ids) - set(members) - ) - - if not self.delete_unused_nodes: - return - - self.delete_unused() + if reap.error: + raise reap.error + if reap.refused: + raise TrackingGroupCleanupError(failures=reap.refused) # TODO : create anoter "read" group. Could be based of the store items # Need to filters the store items inherited from CoreGroup to add them as children diff --git a/tests/integration/test_tracking_zero_members.py b/tests/integration/test_tracking_zero_members.py new file mode 100644 index 00000000..20690db1 --- /dev/null +++ b/tests/integration/test_tracking_zero_members.py @@ -0,0 +1,353 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from infrahub_sdk.constants import InfrahubClientMode +from infrahub_sdk.exceptions import NodeNotFoundError, TrackingGroupCleanupError +from infrahub_sdk.testing.docker import TestInfrahubDockerClient +from infrahub_sdk.testing.schemas.animal import TESTING_CAT, TESTING_PERSON, SchemaAnimal + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + + from infrahub_sdk import InfrahubClient, InfrahubClientSync + +# Cats are deleted before persons: an animal holds a mandatory relationship to its owner. +TRACKED_KINDS = (TESTING_CAT, TESTING_PERSON, "BuiltinTag", "CoreStandardGroup") + + +async def _delete_tracked_nodes(client: InfrahubClient, prefixes: tuple[str, ...], branch: str) -> None: + """Delete the nodes and tracking groups a test class created, on main and on its branch.""" + for branch_name in ("main", branch): + for kind in TRACKED_KINDS: + for node in await client.filters(kind=kind, branch=branch_name): + if str(node.name.value or "").startswith(prefixes): + await node.delete() + + +class TestTracking(TestInfrahubDockerClient, SchemaAnimal): + BRANCH = "tracking-branch01" + NAME_PREFIXES = ("Tracking", "tracking-", "BranchTracking", "branch-tracking-", "sdk-") + + @pytest.fixture(scope="class") + async def base_dataset(self, client: InfrahubClient, load_schema: None) -> AsyncGenerator[None, None]: + await client.branch.create(branch_name=self.BRANCH) + yield + await _delete_tracked_nodes(client=client, prefixes=self.NAME_PREFIXES, branch=self.BRANCH) + await client.branch.delete(branch_name=self.BRANCH) + + async def test_zero_member_run_prunes_previous_members(self, client: InfrahubClient, base_dataset: None) -> None: + person_name = "TrackingZeroMemberPerson" + tag_name = "tracking-zero-TAG" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + tag = await clt.create(kind="BuiltinTag", name=tag_name) + await tag.save(allow_upsert=True) + person = await clt.create(kind=TESTING_PERSON, name=person_name, tags=[tag]) + await person.save(allow_upsert=True) + + group_name = client.group_context._generate_group_name() + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # A run that saves nothing must still prune everything the previous run tracked. + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + + with pytest.raises(NodeNotFoundError, match=tag_name): + await client.get(kind="BuiltinTag", name__value=tag_name) + with pytest.raises(NodeNotFoundError, match=person_name): + await client.get(kind=TESTING_PERSON, name__value=person_name) + + async def test_zero_member_run_without_existing_group_creates_nothing( + self, client: InfrahubClient, base_dataset: None + ) -> None: + params = {"person_name": "TrackingNeverAnyMembers"} + + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + group_name = client.group_context._generate_group_name() + with pytest.raises(NodeNotFoundError, match=group_name): + await client.get(kind="CoreStandardGroup", name__value=group_name) + + async def test_refused_delete_does_not_abort_remaining_reaps( + self, client: InfrahubClient, base_dataset: None + ) -> None: + person_name = "TrackingRefusedPerson" + doomed_tag_name = "tracking-refused-DOOMED" + keeper_tag_name = "tracking-refused-KEEPER" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = await clt.create(kind=TESTING_PERSON, name=person_name) + await person.save(allow_upsert=True) + doomed_tag = await clt.create(kind="BuiltinTag", name=doomed_tag_name) + await doomed_tag.save(allow_upsert=True) + + group_name = client.group_context._generate_group_name() + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # An animal outside the tracking group makes its owner undeletable, + # because Animal.owner is a mandatory relationship. + cat = await client.create(kind=TESTING_CAT, name="TrackingRefusedCat", breed="Bengal", owner=person) + await cat.save() + + # Second run saves only a new tag, so the person and the first tag both + # become reap candidates. The person's delete is refused by the server. + with pytest.raises(TrackingGroupCleanupError, match="Unable to delete 1 unused member") as exc_info: + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + keeper_tag = await clt.create(kind="BuiltinTag", name=keeper_tag_name) + await keeper_tag.save(allow_upsert=True) + + assert list(exc_info.value.failures) == [person.id] + + # The refused delete must not prevent the other unused member from being reaped. + with pytest.raises(NodeNotFoundError, match=doomed_tag_name): + await client.get(kind="BuiltinTag", name__value=doomed_tag_name) + + # The person survived, and must still be a group member so a later run can retry it. + await client.get(kind=TESTING_PERSON, name__value=person_name) + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert sorted(group.members.peer_ids) == sorted([person.id, keeper_tag.id]) + + async def test_zero_member_run_keeps_undeletable_member(self, client: InfrahubClient, base_dataset: None) -> None: + person_name = "TrackingRetryPerson" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = await clt.create(kind=TESTING_PERSON, name=person_name) + await person.save(allow_upsert=True) + + cat = await client.create(kind=TESTING_CAT, name="TrackingRetryCat", breed="Bengal", owner=person) + await cat.save() + + # A zero-member run now attempts the reap; the person's delete is refused. + with pytest.raises(TrackingGroupCleanupError, match="Unable to delete 1 unused member"): + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + assert client.mode == InfrahubClientMode.DEFAULT + + group_name = client.group_context._generate_group_name() + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert group.members.peer_ids == [person.id] + + # Once the blocking node is gone, the next zero-member run reaps the person. + await client.delete(kind=TESTING_CAT, id=cat.id) + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + with pytest.raises(NodeNotFoundError, match=person_name): + await client.get(kind=TESTING_PERSON, name__value=person_name) + + async def test_zero_member_run_prunes_on_the_tracked_branch( + self, client: InfrahubClient, base_dataset: None + ) -> None: + person_name = "BranchTrackingPerson" + tag_name = "branch-tracking-TAG" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True, branch=self.BRANCH) as clt: + tag = await clt.create(kind="BuiltinTag", name=tag_name, branch=self.BRANCH) + await tag.save(allow_upsert=True) + person = await clt.create(kind=TESTING_PERSON, name=person_name, tags=[tag], branch=self.BRANCH) + await person.save(allow_upsert=True) + + group_name = client.group_context._generate_group_name() + group = await client.get( + kind="CoreStandardGroup", name__value=group_name, include=["members"], branch=self.BRANCH + ) + assert len(group.members.peers) == 2 + + # The group belongs to the branch, not to main. + with pytest.raises(NodeNotFoundError, match=group_name): + await client.get(kind="CoreStandardGroup", name__value=group_name, branch="main") + + # A zero-member run on the branch must delete the branch's nodes. Deleting on the + # client default branch instead would not find them and would silently succeed. + async with client.start_tracking(params=params, delete_unused_nodes=True, branch=self.BRANCH): + pass + + group = await client.get( + kind="CoreStandardGroup", name__value=group_name, include=["members"], branch=self.BRANCH + ) + assert len(group.members.peers) == 0 + + with pytest.raises(NodeNotFoundError, match=tag_name): + await client.get(kind="BuiltinTag", name__value=tag_name, branch=self.BRANCH) + with pytest.raises(NodeNotFoundError, match=person_name): + await client.get(kind=TESTING_PERSON, name__value=person_name, branch=self.BRANCH) + + +class TestTrackingSync(TestInfrahubDockerClient, SchemaAnimal): + BRANCH = "sync-tracking-branch01" + NAME_PREFIXES = ("SyncTracking", "sync-tracking-", "SyncBranchTracking", "sync-branch-tracking-", "sdk-") + + @pytest.fixture(scope="class") + async def base_dataset(self, client: InfrahubClient, load_schema: None) -> AsyncGenerator[None, None]: + await client.branch.create(branch_name=self.BRANCH) + yield + await _delete_tracked_nodes(client=client, prefixes=self.NAME_PREFIXES, branch=self.BRANCH) + await client.branch.delete(branch_name=self.BRANCH) + + def test_zero_member_run_prunes_previous_members(self, client_sync: InfrahubClientSync, base_dataset: None) -> None: + person_name = "SyncTrackingZeroMemberPerson" + tag_name = "sync-tracking-zero-TAG" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt: + tag = clt.create(kind="BuiltinTag", name=tag_name) + tag.save(allow_upsert=True) + person = clt.create(kind=TESTING_PERSON, name=person_name, tags=[tag]) + person.save(allow_upsert=True) + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # A run that saves nothing must still prune everything the previous run tracked. + with client_sync.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + + with pytest.raises(NodeNotFoundError, match=tag_name): + client_sync.get(kind="BuiltinTag", name__value=tag_name) + with pytest.raises(NodeNotFoundError, match=person_name): + client_sync.get(kind=TESTING_PERSON, name__value=person_name) + + def test_zero_member_run_without_existing_group_creates_nothing( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + params = {"person_name": "SyncTrackingNeverAnyMembers"} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True): + pass + + group_name = client_sync.group_context._generate_group_name() + with pytest.raises(NodeNotFoundError, match=group_name): + client_sync.get(kind="CoreStandardGroup", name__value=group_name) + + def test_refused_delete_does_not_abort_remaining_reaps( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + person_name = "SyncTrackingRefusedPerson" + doomed_tag_name = "sync-tracking-refused-DOOMED" + keeper_tag_name = "sync-tracking-refused-KEEPER" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = clt.create(kind=TESTING_PERSON, name=person_name) + person.save(allow_upsert=True) + doomed_tag = clt.create(kind="BuiltinTag", name=doomed_tag_name) + doomed_tag.save(allow_upsert=True) + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # An animal outside the tracking group makes its owner undeletable, + # because Animal.owner is a mandatory relationship. + cat = client_sync.create(kind=TESTING_CAT, name="SyncTrackingRefusedCat", breed="Bengal", owner=person) + cat.save() + + # Second run saves only a new tag, so the person and the first tag both + # become reap candidates. The person's delete is refused by the server. + with ( + pytest.raises(TrackingGroupCleanupError, match="Unable to delete 1 unused member") as exc_info, + client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt, + ): + keeper_tag = clt.create(kind="BuiltinTag", name=keeper_tag_name) + keeper_tag.save(allow_upsert=True) + + assert list(exc_info.value.failures) == [person.id] + + # The refused delete must not prevent the other unused member from being reaped. + with pytest.raises(NodeNotFoundError, match=doomed_tag_name): + client_sync.get(kind="BuiltinTag", name__value=doomed_tag_name) + + # The person survived, and must still be a group member so a later run can retry it. + client_sync.get(kind=TESTING_PERSON, name__value=person_name) + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert sorted(group.members.peer_ids) == sorted([person.id, keeper_tag.id]) + + def test_zero_member_run_keeps_undeletable_member( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + person_name = "SyncTrackingRetryPerson" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = clt.create(kind=TESTING_PERSON, name=person_name) + person.save(allow_upsert=True) + + cat = client_sync.create(kind=TESTING_CAT, name="SyncTrackingRetryCat", breed="Bengal", owner=person) + cat.save() + + # A zero-member run now attempts the reap; the person's delete is refused. + with ( + pytest.raises(TrackingGroupCleanupError, match="Unable to delete 1 unused member"), + client_sync.start_tracking(params=params, delete_unused_nodes=True), + ): + pass + + assert client_sync.mode == InfrahubClientMode.DEFAULT + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert group.members.peer_ids == [person.id] + + # Once the blocking node is gone, the next zero-member run reaps the person. + client_sync.delete(kind=TESTING_CAT, id=cat.id) + with client_sync.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + with pytest.raises(NodeNotFoundError, match=person_name): + client_sync.get(kind=TESTING_PERSON, name__value=person_name) + + def test_zero_member_run_prunes_on_the_tracked_branch( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + person_name = "SyncBranchTrackingPerson" + tag_name = "sync-branch-tracking-TAG" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True, branch=self.BRANCH) as clt: + tag = clt.create(kind="BuiltinTag", name=tag_name, branch=self.BRANCH) + tag.save(allow_upsert=True) + person = clt.create(kind=TESTING_PERSON, name=person_name, tags=[tag], branch=self.BRANCH) + person.save(allow_upsert=True) + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get( + kind="CoreStandardGroup", name__value=group_name, include=["members"], branch=self.BRANCH + ) + assert len(group.members.peers) == 2 + + # The sync group lookup must target the branch. Looking on the client default + # branch would miss this group entirely and skip the cleanup. + with client_sync.start_tracking(params=params, delete_unused_nodes=True, branch=self.BRANCH): + pass + + group = client_sync.get( + kind="CoreStandardGroup", name__value=group_name, include=["members"], branch=self.BRANCH + ) + assert len(group.members.peers) == 0 + + with pytest.raises(NodeNotFoundError, match=tag_name): + client_sync.get(kind="BuiltinTag", name__value=tag_name, branch=self.BRANCH) + with pytest.raises(NodeNotFoundError, match=person_name): + client_sync.get(kind=TESTING_PERSON, name__value=person_name, branch=self.BRANCH) diff --git a/tests/unit/sdk/test_group_context.py b/tests/unit/sdk/test_group_context.py index 7b4de550..f40fea76 100644 --- a/tests/unit/sdk/test_group_context.py +++ b/tests/unit/sdk/test_group_context.py @@ -1,16 +1,43 @@ +from __future__ import annotations + import inspect -from collections.abc import Callable +import pickle # noqa: S403 - round-tripping our own exception, no untrusted data involved +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any +import httpx import pytest -from infrahub_sdk.query_groups import InfrahubGroupContext, InfrahubGroupContextBase, InfrahubGroupContextSync -from infrahub_sdk.schema import NodeSchemaAPI +from infrahub_sdk.exceptions import GraphQLError, ServerNotResponsiveError, TrackingGroupCleanupError +from infrahub_sdk.query_groups import ( + InfrahubGroupContext, + InfrahubGroupContextBase, + InfrahubGroupContextSync, + ReapResult, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + from pytest_httpx import HTTPXMock + + from infrahub_sdk.schema import NodeSchemaAPI + from tests.unit.sdk.conftest import BothClients async_methods = [method for method in dir(InfrahubGroupContext) if not method.startswith("_")] sync_methods = [method for method in dir(InfrahubGroupContextSync) if not method.startswith("_")] client_types = ["standard", "sync"] +GROUP_ID = "gggggggg-gggg-gggg-gggg-gggggggggggg" +TAG_ID = "tttttttt-tttt-tttt-tttt-tttttttttttt" +IDENTIFIER = "unit-tracking" + +MANDATORY_RELATIONSHIP_ERROR = ( + "Cannot delete TestingPerson 'pppp'. It is linked to mandatory relationship owner on node TestingCat 'cccc'" +) +MISSING_NODE_ERROR = "Unable to find the node BuiltinTag/tttt in the database." + async def test_method_sanity() -> None: """Validate that there is at least one public method and that both clients look the same.""" @@ -83,3 +110,168 @@ def test_generate_group_description(std_group_schema: NodeSchemaAPI) -> None: context = InfrahubGroupContextBase() context.set_properties(identifier="MYID", params={"one": "xxxxxxxxxxx", "two": "yyyyyyyyyyy"}) assert context._generate_group_description(schema=std_group_schema) == "one: xxxxxxxxxx..." + + +def test_get_members_combines_groups_and_nodes() -> None: + context = InfrahubGroupContextBase() + context.related_group_ids = ["group1"] + context.related_node_ids = ["node1", "node2"] + assert context._get_members() == ["group1", "node1", "node2"] + + +def test_set_unused_member_ids_diffs_against_the_previous_run() -> None: + context = InfrahubGroupContextBase() + context._set_unused_member_ids(previous_member_ids=["kept", "dropped"], members=["kept", "added"]) + assert context.unused_member_ids == ["dropped"] + + context._set_unused_member_ids(previous_member_ids=["gone1", "gone2"], members=[]) + assert sorted(context.unused_member_ids or []) == ["gone1", "gone2"] + + +@dataclass +class AlreadyDeletedCase: + name: str + errors: list[dict[str, Any]] + expected: bool + + +ALREADY_DELETED_CASES = [ + AlreadyDeletedCase(name="missing-node", errors=[{"message": MISSING_NODE_ERROR}], expected=True), + AlreadyDeletedCase(name="refusal", errors=[{"message": MANDATORY_RELATIONSHIP_ERROR}], expected=False), + AlreadyDeletedCase( + name="missing-node-and-refusal", + errors=[{"message": MISSING_NODE_ERROR}, {"message": MANDATORY_RELATIONSHIP_ERROR}], + expected=False, + ), + AlreadyDeletedCase(name="no-errors", errors=[], expected=False), +] + + +@pytest.mark.parametrize("case", [pytest.param(tc, id=tc.name) for tc in ALREADY_DELETED_CASES]) +def test_is_already_deleted(case: AlreadyDeletedCase) -> None: + """A cascade-deleted member is tolerated, but a refusal alongside it must not be.""" + exc = GraphQLError(errors=case.errors, query="mutation BuiltinTagDelete { BuiltinTagDelete { ok } }") + assert InfrahubGroupContextBase._is_already_deleted(exc) is case.expected + + +def test_failure_reason_omits_the_query() -> None: + query = 'mutation TestingPersonDelete { TestingPersonDelete(data: {id: "pppp"}) { ok } }' + exc = GraphQLError(errors=[{"message": MANDATORY_RELATIONSHIP_ERROR, "path": ["TestingPersonDelete"]}], query=query) + + reason = InfrahubGroupContextBase._failure_reason(exc) + + assert reason == MANDATORY_RELATIONSHIP_ERROR + assert query not in reason + + +def test_failure_reason_of_a_non_graphql_error() -> None: + assert InfrahubGroupContextBase._failure_reason(httpx.ReadTimeout("read timed out")) == "read timed out" + + +def test_tracking_group_cleanup_error_survives_serialization() -> None: + """The exception crosses a task-orchestrator boundary, which serializes failed runs.""" + failures = {TAG_ID: MANDATORY_RELATIONSHIP_ERROR} + + restored = pickle.loads(pickle.dumps(TrackingGroupCleanupError(failures=failures))) # noqa: S301 + + assert restored.failures == failures + assert str(restored) == str(TrackingGroupCleanupError(failures=failures)) + + +def _group_query_response(member_ids: list[str]) -> dict[str, Any]: + return { + "data": { + "CoreStandardGroup": { + "count": 1, + "edges": [ + { + "node": { + "id": GROUP_ID, + "display_label": IDENTIFIER, + "__typename": "CoreStandardGroup", + "name": {"value": IDENTIFIER, "is_default": False, "is_from_profile": False}, + "description": {"value": None, "is_default": True, "is_from_profile": False}, + "members": { + "count": len(member_ids), + "edges": [ + {"node": {"id": member_id, "display_label": member_id, "__typename": "BuiltinTag"}} + for member_id in member_ids + ], + }, + } + } + ], + } + } + } + + +NO_GROUP_RESPONSE: dict[str, Any] = {"data": {"CoreStandardGroup": {"count": 0, "edges": []}}} + + +async def _track_nothing(clients: BothClients, client_type: str, schema: dict) -> None: + """Run a tracking block that saves no node, on whichever client is under test.""" + if client_type == "standard": + clients.standard.schema.set_cache(schema=schema, branch="main") + async with clients.standard.start_tracking(identifier=IDENTIFIER, delete_unused_nodes=True): + pass + return + + clients.sync.schema.set_cache(schema=schema, branch="main") + with clients.sync.start_tracking(identifier=IDENTIFIER, delete_unused_nodes=True): + pass + + +@pytest.mark.parametrize("client_type", client_types) +async def test_zero_member_run_without_group_issues_no_mutation( + clients: BothClients, client_type: str, schema_query_05_data: dict, httpx_mock: HTTPXMock +) -> None: + """A run that tracked nothing and finds no group must not create an empty one.""" + httpx_mock.add_response(method="POST", url="http://mock/graphql/main", json=NO_GROUP_RESPONSE) + + await _track_nothing(clients=clients, client_type=client_type, schema=schema_query_05_data) + + assert len(httpx_mock.get_requests()) == 1 + + +@pytest.mark.parametrize("client_type", client_types) +async def test_zero_member_run_with_empty_group_issues_no_mutation( + clients: BothClients, client_type: str, schema_query_05_data: dict, httpx_mock: HTTPXMock +) -> None: + """Repeated zero-member runs settle on the lookup alone once the group is empty.""" + httpx_mock.add_response(method="POST", url="http://mock/graphql/main", json=_group_query_response(member_ids=[])) + + await _track_nothing(clients=clients, client_type=client_type, schema=schema_query_05_data) + + assert len(httpx_mock.get_requests()) == 1 + + +@pytest.mark.parametrize("client_type", client_types) +async def test_failed_reap_still_records_the_membership( + clients: BothClients, client_type: str, schema_query_05_data: dict, httpx_mock: HTTPXMock +) -> None: + """A transport failure mid-reap must not skip the upsert that keeps the member reachable.""" + httpx_mock.add_response( + method="POST", url="http://mock/graphql/main", json=_group_query_response(member_ids=[TAG_ID]) + ) + httpx_mock.add_exception(httpx.ReadTimeout("read timed out"), method="POST", url="http://mock/graphql/main") + httpx_mock.add_response( + method="POST", + url="http://mock/graphql/main", + json={"data": {"CoreStandardGroupUpsert": {"ok": True, "object": {"id": GROUP_ID}}}}, + ) + + # The failure is reported as itself, not blamed on the member it interrupted. + with pytest.raises(ServerNotResponsiveError, match="Unable to read from"): + await _track_nothing(clients=clients, client_type=client_type, schema=schema_query_05_data) + + requests = httpx_mock.get_requests() + assert len(requests) == 3 + # The member the reap never got through stays in the group so a later run retries it. + assert TAG_ID in requests[-1].read().decode() + + +def test_reap_result_retains_refused_and_unattempted_members() -> None: + result = ReapResult(refused={"refused1": MANDATORY_RELATIONSHIP_ERROR}, unattempted=["unattempted1"]) + assert result.retained_member_ids == ["refused1", "unattempted1"] + assert ReapResult().retained_member_ids == []