diff --git a/backend/infrahub/core/node/__init__.py b/backend/infrahub/core/node/__init__.py index d36677addfb..d479e8d0c33 100644 --- a/backend/infrahub/core/node/__init__.py +++ b/backend/infrahub/core/node/__init__.py @@ -25,6 +25,7 @@ from infrahub.core.protocols import CoreNumberPool, CoreObjectTemplate from infrahub.core.protocols_base import CoreNode from infrahub.core.query.node import NodeCheckIDQuery, NodeCreateAllQuery, NodeDeleteQuery, NodeUpdateMetadataQuery +from infrahub.core.query.node_agnostic_retirement import RetireNodeAgnosticFieldsQuery from infrahub.core.schema import ( AttributeSchema, NodeSchema, @@ -1250,6 +1251,18 @@ async def delete(self, db: InfrahubDatabase, user_id: str = SYSTEM_USER_ID, at: query = await NodeDeleteQuery.init(db=db, node=self, at=delete_at, user_id=user_id) await query.execute(db=db) + retirement_query = await RetireNodeAgnosticFieldsQuery.init(db=db, node_uuid=self.get_id(), at=delete_at) + await retirement_query.execute(db=db) + retired = retirement_query.get_data() + if retired.edges_closed: + log.debug( + "Retired branch-agnostic fields of a deleted node", + node_id=self.get_id(), + node_kind=self.get_kind(), + edges_closed=retired.edges_closed, + at=delete_at.to_string(), + ) + self._node_changelog = node_changelog async def to_graphql( diff --git a/backend/infrahub/core/query/agnostic_retention.py b/backend/infrahub/core/query/agnostic_retention.py new file mode 100644 index 00000000000..9d623f39edc --- /dev/null +++ b/backend/infrahub/core/query/agnostic_retention.py @@ -0,0 +1,122 @@ +"""The one predicate that decides whether a branch-agnostic field is still retained by any branch. + +A branch-agnostic field keeps its value on edges carrying the global branch name, which every branch +reads. Those edges may only be closed once **NO** branch can still reach a live owner over a live field +edge. Cypher for this is consolidated here to ensure consistency. + +Retention is decided per branch **and** per linked vertex: under that branch's view, the vertex's +existence edge and its edge to the field must both resolve to `active`. Surviving edges are summed +per branch, and only then is the maximum taken across branches. Retention is a disjunction of what +each branch holds live on its own, never a pool the branches contribute to jointly. + +A `:Relationship` needs two qualifying field edges rather than one, because a relationship missing a +peer is not a relationship. A `:Relationship` must also have two distinct active peers to be +considered active. + +The winner lookups are OPTIONAL CALL subqueries and the peers are counted conditionally rather than +filtered so that a field with no live edge on a branch reaches the end of the predicate carrying a +count of zero -- a plain CALL or MATCH would drop its row instead. + +Assumption: every branch forks from the default branch. A branch-of-branch feature would not extend +this logic. +""" + +# Expects `agnostic_candidates` in scope: a list of the candidate `:Attribute` / `:Relationship` +# vertices, plus the `$global_branch_name` and `$at` parameters. Emits one row per candidate that no +# branch retains, with `field` as the only variable in scope. +UNRETAINED_AGNOSTIC_FIELD_PREDICATE = """ +// ---------------------- +// The branches are read once for the whole run and carried as a list. +// ---------------------- +MATCH (branch:Branch) +WHERE branch.name <> $global_branch_name +WITH + agnostic_candidates, + collect({ + name: branch.name, + origin_name: CASE WHEN branch.is_default THEN NULL ELSE branch.origin_branch END, + origin_at: CASE + WHEN branch.is_default THEN NULL + WHEN branch.branched_from < $at THEN branch.branched_from + ELSE $at + END + }) AS branch_windows + +UNWIND agnostic_candidates AS field +WITH DISTINCT field, branch_windows +// ---------------------- +// Quick filter to remove all fields with no active edges on the global branch, covers most fields +// ---------------------- +WHERE EXISTS { + MATCH (field)-[global_edge]-() + WHERE global_edge.branch = $global_branch_name + AND global_edge.status = "active" + AND global_edge.to IS NULL +} +WITH field, branch_windows, CASE WHEN field:Relationship THEN 2 ELSE 1 END AS required_live_peers + +// ---------------------- +// Get all the possible peer :Nodes for each field. Usually 1 for :Attribute and 2 for :Relationship +// ---------------------- +OPTIONAL MATCH (node:Node)-[:HAS_ATTRIBUTE|IS_RELATED]-(field) +WITH DISTINCT field, branch_windows, required_live_peers, node + +UNWIND branch_windows AS branch_window +WITH + field, + required_live_peers, + node, + branch_window.name AS branch_name, + branch_window.origin_name AS origin_name, + branch_window.origin_at AS origin_at + +// ---------------------- +// Resolve the status of this peer's latest edge to the field on this branch. +// ---------------------- +OPTIONAL CALL (field, node, branch_name, origin_name, origin_at) { + MATCH (node)-[field_edge:HAS_ATTRIBUTE|IS_RELATED]-(field) + WHERE (field_edge.branch IN [$global_branch_name, branch_name] + AND field_edge.from <= $at + AND (field_edge.to IS NULL OR field_edge.to > $at)) + OR (field_edge.branch = origin_name + AND field_edge.from <= origin_at + AND (field_edge.to IS NULL OR field_edge.to > origin_at)) + RETURN field_edge.status AS latest_field_edge_status + ORDER BY field_edge.branch_level DESC, field_edge.from DESC, field_edge.status ASC + LIMIT 1 +} + +// ---------------------- +// Resolve whether this peer exists on this branch. +// The global branch stays in the existence match so that an owner which is itself branch-agnostic +// reads as live on every branch and is therefore retained. +// ---------------------- +OPTIONAL CALL (node, branch_name, origin_name, origin_at) { + MATCH (node)-[existence:IS_PART_OF]->(:Root) + WHERE (existence.branch IN [$global_branch_name, branch_name] + AND existence.from <= $at + AND (existence.to IS NULL OR existence.to > $at)) + OR (existence.branch = origin_name + AND existence.from <= origin_at + AND (existence.to IS NULL OR existence.to > origin_at)) + RETURN existence.status AS latest_existence + ORDER BY existence.branch_level DESC, existence.from DESC, existence.status ASC + LIMIT 1 +} + +// ---------------------- +// Peers are counted by uuid. Kind/inheritance migration leaves multiple Node vertices with +// the same uuid for a single entity. +// ---------------------- +WITH + field, + required_live_peers, + branch_name, + count(DISTINCT CASE + WHEN latest_field_edge_status = "active" AND latest_existence = "active" THEN node.uuid + END) AS live_peer_count + +WITH field, required_live_peers, max(live_peer_count) AS most_live_peers +WHERE most_live_peers < required_live_peers +WITH field +""" diff --git a/backend/infrahub/core/query/node_agnostic_retirement.py b/backend/infrahub/core/query/node_agnostic_retirement.py new file mode 100644 index 00000000000..dd62bd90309 --- /dev/null +++ b/backend/infrahub/core/query/node_agnostic_retirement.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from infrahub.core.constants import GLOBAL_BRANCH_NAME +from infrahub.core.query import Query, QueryType +from infrahub.core.query.agnostic_retention import UNRETAINED_AGNOSTIC_FIELD_PREDICATE + +if TYPE_CHECKING: + from infrahub.core.timestamp import Timestamp + from infrahub.database import InfrahubDatabase + + +@dataclass(frozen=True) +class NodeAgnosticRetirementResult: + """What retiring one node's branch-agnostic fields changed.""" + + edges_closed: int + """Global edges given a `to` timestamp. Zero means every field is still retained somewhere.""" + + +_RETIRE_UNRETAINED_FIELDS_OF_NODE = """ +// ----------------- +// MATCH on the branch-agnostic edges we care about to start with. +// ----------------- +MATCH (anchor_node:Node {uuid: $node_uuid})-[anchor:HAS_ATTRIBUTE|IS_RELATED]-(field:Attribute|Relationship) +WHERE anchor.branch = $global_branch_name + AND anchor.status = "active" + AND anchor.from <= $at + AND anchor.to IS NULL +WITH collect(DISTINCT field) AS agnostic_candidates +%(unretained_predicate)s + +MATCH (field)-[e]-() +WHERE e.branch = $global_branch_name + AND e.status = "active" + AND e.from <= $at + AND e.to IS NULL +SET e.to = $at +RETURN count(e) AS edges_closed +""" % {"unretained_predicate": UNRETAINED_AGNOSTIC_FIELD_PREDICATE} + + +class RetireNodeAgnosticFieldsQuery(Query): + """Close the open global edges of one node's branch-agnostic fields that no branch retains. + + Checks if the field is reachable from ANY branch. It is only deleted if it is completely + unreachable. + """ + + name: str = "retire_node_agnostic_fields" + type: QueryType = QueryType.WRITE + + insert_return: bool = False + insert_limit: bool = False + + def __init__(self, node_uuid: str, at: Timestamp, **kwargs: Any) -> None: + self.node_uuid = node_uuid + super().__init__(at=at, **kwargs) + + async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None: # noqa: ARG002 + self.params["global_branch_name"] = GLOBAL_BRANCH_NAME + self.params["node_uuid"] = self.node_uuid + self.params["at"] = self.at.to_string() + + self.add_to_query(_RETIRE_UNRETAINED_FIELDS_OF_NODE) + self.update_return_labels(["edges_closed"]) + + def get_data(self) -> NodeAgnosticRetirementResult: + """Return what the run closed.""" + result = self.get_result() + if result: + return NodeAgnosticRetirementResult(edges_closed=result.get_as_type("edges_closed", int)) + return NodeAgnosticRetirementResult(edges_closed=0) diff --git a/backend/infrahub/git/tasks.py b/backend/infrahub/git/tasks.py index c44205a997a..b3be6e63e55 100644 --- a/backend/infrahub/git/tasks.py +++ b/backend/infrahub/git/tasks.py @@ -1170,8 +1170,9 @@ async def run_check_merge_conflicts(model: CheckRepositoryMergeConflicts) -> Val await check.save() database = await get_database() - async with database.start_session() as db: - await NodeManager.delete(db=db, nodes=list(existing_checks.values())) + for check in existing_checks.values(): + async with database.start_transaction() as dbt: + await NodeManager.delete(db=dbt, nodes=[check]) return validator_conclusion diff --git a/backend/tests/component/core/test_agnostic_retirement.py b/backend/tests/component/core/test_agnostic_retirement.py new file mode 100644 index 00000000000..a1bc2bd16ee --- /dev/null +++ b/backend/tests/component/core/test_agnostic_retirement.py @@ -0,0 +1,429 @@ +"""Test for deleting branch-aware objects with branch-agnostic attributes and relationships. + +Branch-agnostic fields on branch-aware objects should only be deleted when the cannot be accessed +from any branch. + +Every assertion reads the edges directly rather than going through the node manager: the subject is +which edges carry a `to` timestamp and which do not, and a read through the manager would hide the +very states these tests exist to pin down. Where a branch is expected to go on reading the object, the +manager is used as well, because that is the claim being made. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest + +from infrahub.core import registry +from infrahub.core.constants import GLOBAL_BRANCH_NAME, InfrahubKind +from infrahub.core.initialization import create_branch +from infrahub.core.manager import NodeManager +from infrahub.core.node import Node +from infrahub.core.node.resource_manager.number_pool import CoreNumberPool +from infrahub.core.query.node_agnostic_retirement import RetireNodeAgnosticFieldsQuery +from infrahub.core.timestamp import Timestamp +from infrahub.database import InfrahubDatabase, InfrahubDatabaseMode + +if TYPE_CHECKING: + from neo4j import Record + + from infrahub.core.branch import Branch + from infrahub.core.query import QueryType + from infrahub.core.schema.schema_branch import SchemaBranch + +from tests.helpers.agnostic_edges import ( + EdgeState, + attribute_global_edges, + attribute_owning_edges, + edge_summary, + existence_edges, + expected_closed_at, + open_edge_types, + open_edges, + pool_reservation_edges, + relationship_global_edges, + remove_attribute_on_branch, +) +from tests.helpers.schema.agnostic_retirement import ( + AGNOSTIC_RETIREMENT_SCHEMA, + GADGET_KIND, + RELATIONSHIP_IDENTIFIER, + WIDGET_KIND, +) + +SERIAL_POOL_START = 9001 +SERIAL_POOL_END = 9002 + + +class RetirementFailureError(Exception): + """Stands in for whatever the retirement run could fail with.""" + + +class FailingRetirementDatabase(InfrahubDatabase): + """Database that fails the branch-agnostic retirement query and passes every other query through. + + A real database is what makes the claim testable: the deletion's own writes have to reach the + transaction so that the rollback has something to undo. + """ + + @classmethod + def from_db(cls, db: InfrahubDatabase) -> FailingRetirementDatabase: + return cls( + mode=InfrahubDatabaseMode.DRIVER, + driver=db._driver, + db_type=db.db_type, + default_neo4j_runtime=db.default_neo4j_runtime, + queries_names_to_config=db.queries_names_to_config, + ) + + async def execute_query_with_metadata( + self, + query: str, + params: dict[str, Any] | None = None, + name: str = "undefined", + context: dict[str, str] | None = None, + type: QueryType | None = None, + timeout_seconds: float | None = None, + ) -> tuple[list[Record], dict[str, Any]]: + if name == RetireNodeAgnosticFieldsQuery.name: + raise RetirementFailureError("the retirement run could not complete") + return await super().execute_query_with_metadata( + query=query, params=params, name=name, context=context, type=type, timeout_seconds=timeout_seconds + ) + + +async def _create_widget(db: InfrahubDatabase, branch: Branch, name: str, serial: int, **kwargs: Any) -> Node: + widget = await Node.init(db=db, schema=WIDGET_KIND, branch=branch) + await widget.new(db=db, name=name, serial=serial, **kwargs) + await widget.save(db=db) + return widget + + +async def _delete(db: InfrahubDatabase, node_id: str, branch: Branch, at: Timestamp) -> None: + to_delete = await NodeManager.get_one(db=db, id=node_id, branch=branch, raise_on_error=True) + await to_delete.delete(db=db, at=at) + + +class TestAgnosticRetirementOnDelete: + @pytest.fixture(scope="class") + async def default_branch(self, default_branch_scope_class: Branch) -> Branch: + return default_branch_scope_class + + @pytest.fixture(scope="class") + async def agnostic_schema(self, db: InfrahubDatabase, default_branch: Branch) -> None: + registry.schema.register_schema(schema=AGNOSTIC_RETIREMENT_SCHEMA, branch=default_branch.name) + + @pytest.fixture(scope="class") + async def serial_pool( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + register_core_models_schema_scope_class: SchemaBranch, + ) -> CoreNumberPool: + """A pool of two numbers backing the widget's branch-agnostic serial. + + Two rather than one so that an allocation which fails to reuse the freed value still succeeds, + and reports the number it handed out instead of an exhausted pool. + """ + registry.node[InfrahubKind.NUMBERPOOL] = CoreNumberPool + pool = await CoreNumberPool.init(db=db, schema=InfrahubKind.NUMBERPOOL) + await pool.new( + db=db, + name="agnostic-serial-pool", + node=WIDGET_KIND, + node_attribute="serial", + start_range=SERIAL_POOL_START, + end_range=SERIAL_POOL_END, + ) + await pool.save(db=db) + return pool + + async def test_a_field_created_and_deleted_on_the_same_user_branch_is_closed_by_the_delete( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """An object that only ever existed on one branch cannot also exist elsewhere through a merge. + + A branch that has been merged is permanently read-only, so a delete running on a branch proves the + branch was never merged. Nothing else can be holding the field, so the close is unconditional. + """ + branch = await create_branch(db=db, branch_name="creates-and-deletes") + widget = await _create_widget(db=db, branch=branch, name="branch-only", serial=100) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + deleted_at = Timestamp() + await _delete(db=db, node_id=widget.id, branch=branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert {edge.status for edge in after} == {"active"}, "retirement is a time-close, never a status tombstone" + + async def test_a_field_stays_open_while_the_default_branch_still_holds_the_object( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Deleting on a branch says nothing about the object's fate on the branch it forked from.""" + widget = await _create_widget(db=db, branch=default_branch, name="deleted-on-a-branch-only", serial=200) + branch = await create_branch(db=db, branch_name="deletes-its-own-copy") + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + + await _delete(db=db, node_id=widget.id, branch=branch, at=Timestamp()) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ) + still_on_default = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) + assert still_on_default is not None + assert still_on_default.get_attribute(name="serial").value == 200 + + async def test_a_field_is_closed_when_the_default_branch_deletes_the_last_holder( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """No branch forked between the creation and the deletion, so no branch can still read the object.""" + widget = await _create_widget(db=db, branch=default_branch, name="last-holder", serial=300) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + deleted_at = Timestamp() + await _delete(db=db, node_id=widget.id, branch=default_branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert {edge.status for edge in after} == {"active"}, "retirement is a time-close, never a status tombstone" + + async def test_a_field_stays_open_for_a_branch_that_forked_between_creation_and_deletion( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """A branch that forked while the object was live reads it as live, and keeps reading its value.""" + widget = await _create_widget(db=db, branch=default_branch, name="retained-by-a-fork", serial=400) + branch = await create_branch(db=db, branch_name="forked-before-the-delete") + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + await _delete(db=db, node_id=widget.id, branch=default_branch, at=Timestamp()) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ) + assert await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) is None + on_branch = await NodeManager.get_one(db=db, id=widget.id, branch=branch) + assert on_branch is not None + assert on_branch.get_attribute(name="serial").value == 400 + + async def test_a_field_stays_open_until_every_retaining_branch_has_deleted_the_object( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Verify that an object's agnostic fields are only deleted when NO branch can reach them""" + widget = await _create_widget(db=db, branch=default_branch, name="held-by-two", serial=800) + first = await create_branch(db=db, branch_name="first-of-two-holders") + second = await create_branch(db=db, branch_name="second-of-two-holders") + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + await _delete(db=db, node_id=widget.id, branch=default_branch, at=Timestamp()) + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ), "two branches still read the object, so nothing is released" + + await _delete(db=db, node_id=widget.id, branch=first, at=Timestamp()) + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ), "the second branch alone is still enough to retain it" + assert await NodeManager.get_one(db=db, id=widget.id, branch=second) is not None + + last_delete = Timestamp() + await _delete(db=db, node_id=widget.id, branch=second, at=last_delete) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edges(after) == [], "the last holder released it" + assert {edge.to_time for edge in after} == {last_delete.to_string()} + + async def test_a_relationship_is_closed_when_its_peers_are_live_on_different_branches( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Both peers must be live under one branch's view, not one peer each on two branches. + + After deleting each peer on a separate branch, the object is NOT retired. Only when the + relationship is broken on _every_ branch will it be retired. + """ + gadget = await Node.init(db=db, schema=GADGET_KIND, branch=default_branch) + await gadget.new(db=db, name="peer-on-its-own-branch") + await gadget.save(db=db) + widget = await _create_widget(db=db, branch=default_branch, name="split-peers", serial=1100, gadget=gadget) + + keeps_the_gadget = await create_branch(db=db, branch_name="keeps-the-gadget") + keeps_the_widget = await create_branch(db=db, branch_name="keeps-the-widget") + + before = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert [edge.edge_type for edge in open_edges(before)].count("IS_RELATED") == 2 + + # each branch loses one half of the relationship, so neither branch holds both peers + await _delete(db=db, node_id=widget.id, branch=keeps_the_gadget, at=Timestamp()) + await _delete(db=db, node_id=gadget.id, branch=keeps_the_widget, at=Timestamp()) + + assert edge_summary( + await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + ) == edge_summary(before), "the default branch still holds both peers" + + last_delete = Timestamp() + await _delete(db=db, node_id=gadget.id, branch=default_branch, at=last_delete) + + after = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert open_edges(after) == [], ( + "no branch reads both peers as live, so the relationship is released even though each peer " + "survives somewhere" + ) + + async def test_a_field_removed_on_the_only_retaining_branch_is_closed_with_the_object( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Test an attribute removed from the schema on a branch is retired when its object is deleted.""" + widget = await _create_widget(db=db, branch=default_branch, name="field-removed-on-the-fork", serial=500) + branch = await create_branch(db=db, branch_name="removed-the-attribute") + await remove_attribute_on_branch( + db=db, node_id=widget.id, attribute_name="serial", branch=branch, at=Timestamp() + ) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + deleted_at = Timestamp() + await _delete(db=db, node_id=widget.id, branch=default_branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert {edge.status for edge in after} == {"active"}, "retirement is a time-close, never a status tombstone" + + owning_edges = await attribute_owning_edges(db=db, node_id=widget.id, attribute_name="serial") + assert sorted((edge.branch, edge.status, edge.to_time or "") for edge in owning_edges) == [ + (GLOBAL_BRANCH_NAME, "active", deleted_at.to_string()), + (default_branch.name, "deleted", ""), + (branch.name, "deleted", ""), + ], "only the global edge is closed; the branch-scoped tombstones are left exactly as they were" + + still_on_branch = await NodeManager.get_one(db=db, id=widget.id, branch=branch) + assert still_on_branch is not None, "the branch retains the object, which is why only the field was released" + assert still_on_branch.get_attribute(name="serial").value is None, ( + "the branch that dropped the field reads no value for it, which is what made it unretained" + ) + + async def test_a_relationship_is_closed_when_one_of_its_peers_is_deleted( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """A relationship missing a peer is not a relationship, so both of its peer edges are closed.""" + gadget = await Node.init(db=db, schema=GADGET_KIND, branch=default_branch) + await gadget.new(db=db, name="doomed-peer") + await gadget.save(db=db) + widget = await _create_widget(db=db, branch=default_branch, name="surviving-peer", serial=600, gadget=gadget) + + before = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert [edge.edge_type for edge in open_edges(before)].count("IS_RELATED") == 2 + + deleted_at = Timestamp() + await _delete(db=db, node_id=gadget.id, branch=default_branch, at=deleted_at) + + after = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert open_edges(after) == [] + assert {edge.status for edge in after} == {"active"} + assert {edge.to_time for edge in after} == {deleted_at.to_string()} + + async def test_a_retirement_failure_propagates_and_leaves_the_graph_untouched( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """The failure has to reach the caller, because the deletion is what rolls back with it. + + Reporting the failure as a zero commits a deleted object still holding a live branch-agnostic + value, which is the shape retirement exists to prevent and which no later action repairs. + """ + widget = await _create_widget(db=db, branch=default_branch, name="delete-rolls-back", serial=700) + + attribute_before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + existence_before = await existence_edges(db=db, node_id=widget.id) + + failing_db = FailingRetirementDatabase.from_db(db=db) + with pytest.raises(RetirementFailureError, match=r"^the retirement run could not complete$"): + async with failing_db.start_transaction() as dbt: + to_delete = await NodeManager.get_one(db=dbt, id=widget.id, branch=default_branch, raise_on_error=True) + await to_delete.delete(db=dbt, at=Timestamp()) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(attribute_before) + ) + assert sorted((edge.branch, edge.status, edge.to_time or "") for edge in existence_before) == sorted( + (edge.branch, edge.status, edge.to_time or "") for edge in await existence_edges(db=db, node_id=widget.id) + ), "the existence tombstone rolled back with the retirement that failed after it" + still_there = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) + assert still_there is not None + assert still_there.get_attribute(name="serial").value == 700 + + async def test_a_value_freed_by_retirement_is_allocatable_again_from_its_pool( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + serial_pool: CoreNumberPool, + ) -> None: + """Deleting the object holding a pooled value returns that value to the pool.""" + holder = await Node.init(db=db, schema=WIDGET_KIND, branch=default_branch) + await holder.new(db=db, name="holds-a-pooled-serial", serial={"from_pool": {"id": serial_pool.id}}) + await holder.save(db=db) + + assert holder.get_attribute(name="serial").value == SERIAL_POOL_START + assert await serial_pool.get_used(db=db, branch=default_branch) == [SERIAL_POOL_START] + + before = await attribute_global_edges(db=db, node_id=holder.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED", "HAS_SOURCE"} + reserved_before = await pool_reservation_edges(db=db, pool_id=serial_pool.id, identifier=holder.id) + assert reserved_before == [ + EdgeState(edge_type="IS_RESERVED", branch=GLOBAL_BRANCH_NAME, status="active", to_time=None) + ] + + deleted_at = Timestamp() + await _delete(db=db, node_id=holder.id, branch=default_branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=holder.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert await pool_reservation_edges(db=db, pool_id=serial_pool.id, identifier=holder.id) == reserved_before, ( + "the reservation is never cleaned up on delete, and does not need to be" + ) + assert await serial_pool.get_used(db=db, branch=default_branch) == [] + + reallocated = await Node.init(db=db, schema=WIDGET_KIND, branch=default_branch) + await reallocated.new(db=db, name="takes-the-freed-serial", serial={"from_pool": {"id": serial_pool.id}}) + await reallocated.save(db=db) + + assert reallocated.get_attribute(name="serial").value == SERIAL_POOL_START + assert await serial_pool.get_used(db=db, branch=default_branch) == [SERIAL_POOL_START] diff --git a/backend/tests/component/query/__init__.py b/backend/tests/component/query/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/backend/tests/component/query/test_node_agnostic_retirement_query.py b/backend/tests/component/query/test_node_agnostic_retirement_query.py new file mode 100644 index 00000000000..dd3f3d76c9e --- /dev/null +++ b/backend/tests/component/query/test_node_agnostic_retirement_query.py @@ -0,0 +1,453 @@ +"""Graph-shape assertions for the retirement of one node's branch-agnostic fields. + +Every assertion reads the edges directly rather than going through the node manager: the subject is +which edges carry a `to` timestamp and which do not, and a read through the manager would hide the +very states these tests exist to pin down. +""" + +from typing import Any + +import pytest + +from infrahub.core import registry +from infrahub.core.branch import Branch +from infrahub.core.constants import ( + GLOBAL_BRANCH_NAME, + SchemaPathType, +) +from infrahub.core.manager import NodeManager +from infrahub.core.migrations.schema.node_kind_update import ( + NodeKindUpdateMigration, + NodeKindUpdateMigrationQuery01, +) +from infrahub.core.node import Node +from infrahub.core.path import SchemaPath +from infrahub.core.query.node_agnostic_retirement import ( + NodeAgnosticRetirementResult, + RetireNodeAgnosticFieldsQuery, +) +from infrahub.core.timestamp import Timestamp +from infrahub.database import InfrahubDatabase +from tests.helpers.agnostic_edges import ( + attribute_global_edges, + attribute_vertex_count, + edge_summary, + node_vertex_count, + open_active_edges, + open_edges, + relationship_global_edges, + relationship_peer_shape, + tombstone_existence_only, + tombstone_relationship_peer_edge, + values_reachable_over_open_edges, +) +from tests.helpers.schema.agnostic_retirement import ( + AGNOSTIC_RETIREMENT_SCHEMA, + BEACON_KIND, + GADGET_KIND, + RELATIONSHIP_IDENTIFIER, + WIDGET_KIND, +) + + +async def _rename_widget_kind(db: InfrahubDatabase, branch: Branch) -> None: + """Rename the widget kind in the graph, leaving a superseded node vertex under the same uuid. + + The copy shares the original's attribute and relationship vertices, so a field vertex is linked to + both the live copy and the superseded one. + """ + previous_schema = registry.schema.get_node_schema(name=WIDGET_KIND, branch=branch, duplicate=False) + renamed_schema = registry.schema.get_node_schema(name=WIDGET_KIND, branch=branch, duplicate=True) + renamed_schema.name = "RenamedWidget" + + migration = NodeKindUpdateMigration( + previous_node_schema=previous_schema, + new_node_schema=renamed_schema, + schema_path=SchemaPath(path_type=SchemaPathType.ATTRIBUTE, schema_kind=renamed_schema.kind, field_name="name"), + ) + query = await NodeKindUpdateMigrationQuery01.init(db=db, branch=branch, migration=migration) + await query.execute(db=db) + + +async def _retire(db: InfrahubDatabase, node_id: str, at: Timestamp) -> NodeAgnosticRetirementResult: + query = await RetireNodeAgnosticFieldsQuery.init(db=db, node_uuid=node_id, at=at) + await query.execute(db=db) + return query.get_data() + + +async def _create_widget(db: InfrahubDatabase, branch: Branch, name: str, serial: int, **kwargs: Any) -> Node: + widget = await Node.init(db=db, schema=WIDGET_KIND, branch=branch) + await widget.new(db=db, name=name, serial=serial, **kwargs) + await widget.save(db=db) + return widget + + +async def _create_gadget(db: InfrahubDatabase, branch: Branch, name: str) -> Node: + gadget = await Node.init(db=db, schema=GADGET_KIND, branch=branch) + await gadget.new(db=db, name=name) + await gadget.save(db=db) + return gadget + + +class TestRetireNodeAgnosticFields: + @pytest.fixture(scope="class") + async def default_branch(self, default_branch_scope_class: Branch) -> Branch: + return default_branch_scope_class + + @pytest.fixture(scope="class") + async def nodedel_schema(self, db: InfrahubDatabase, default_branch: Branch) -> None: + registry.schema.register_schema(schema=AGNOSTIC_RETIREMENT_SCHEMA, branch=default_branch.name) + + async def test_an_anchor_that_matches_nothing_reports_a_measured_zero( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Ensure the query returns 0 when it does nothing, and that the anchor is what limited it. + + A retirable object is present and left alone, so the zero says "this uuid owns nothing" rather + than "the database held nothing to find". The row itself matters too: the query ends in an + aggregation with no grouping key, and a missing row would leave the outcome unknown. + """ + bystander = await _create_widget(db=db, branch=default_branch, name="not-the-anchor", serial=1000) + await tombstone_existence_only(db=db, node_id=bystander.get_id(), branch=default_branch, at=Timestamp()) + bystander_before = await attribute_global_edges(db=db, node_id=bystander.get_id(), attribute_name="serial") + assert open_active_edges(bystander_before) != [], ( + "the bystander has to be retirable, or the anchor is not what spared it" + ) + + query = await RetireNodeAgnosticFieldsQuery.init(db=db, node_uuid="no-node-carries-this-uuid", at=Timestamp()) + await query.execute(db=db) + + assert query.get_result() is not None + assert query.get_data() == NodeAgnosticRetirementResult(edges_closed=0) + assert edge_summary( + await attribute_global_edges(db=db, node_id=bystander.get_id(), attribute_name="serial") + ) == edge_summary(bystander_before), "the unrelated retirable object is untouched" + + async def test_branch_agnostic_object_is_not_deleted_while_active( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify branch-agnostic object is not improperly deleted by the retirement query.""" + beacon = await Node.init(db=db, schema=BEACON_KIND, branch=default_branch) + await beacon.new(db=db, name="beacon-alive") + await beacon.save(db=db) + + before = await attribute_global_edges(db=db, node_id=beacon.get_id(), attribute_name="name") + assert open_edges(before) != [] + + assert await _retire(db=db, node_id=beacon.get_id(), at=Timestamp()) == NodeAgnosticRetirementResult( + edges_closed=0 + ) + assert edge_summary( + await attribute_global_edges(db=db, node_id=beacon.get_id(), attribute_name="name") + ) == edge_summary(before) + + async def test_an_owner_that_is_itself_branch_agnostic_is_closed_once_by_its_own_deletion( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Ensure deleting a branch-agnostic object leaves nothing for the retirement query to address.""" + beacon = await Node.init(db=db, schema=BEACON_KIND, branch=default_branch) + await beacon.new(db=db, name="beacon-deleted") + await beacon.save(db=db) + beacon_id = beacon.get_id() + + before = await attribute_global_edges(db=db, node_id=beacon_id, attribute_name="name") + open_before = open_edges(before) + assert open_before != [] + + deleted_at = Timestamp() + to_delete = await NodeManager.get_one(db=db, id=beacon_id, branch=default_branch) + await to_delete.delete(db=db, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=beacon_id, attribute_name="name") + + assert {(edge.edge_type, edge.status) for edge in open_edges(after)} == { + ("HAS_ATTRIBUTE", "deleted"), + ("HAS_VALUE", "deleted"), + ("IS_PROTECTED", "deleted"), + } + assert {edge.to_time for edge in after if edge.status == "active"} == {deleted_at.to_string()} + + assert await _retire(db=db, node_id=beacon_id, at=Timestamp()) == NodeAgnosticRetirementResult(edges_closed=0) + + async def test_partially_deleted_object_is_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Object with a closed IS_PART_OF edge and an active agnostic field is retired.""" + widget = await _create_widget(db=db, branch=default_branch, name="orphan-holder", serial=8100) + before = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + open_before = open_edges(before) + assert open_before != [] + + await tombstone_existence_only(db=db, node_id=widget.get_id(), branch=default_branch, at=Timestamp()) + + unchanged = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert edge_summary(unchanged) == edge_summary(before) + + retired_at = Timestamp() + assert await _retire(db=db, node_id=widget.get_id(), at=retired_at) == NodeAgnosticRetirementResult( + edges_closed=len(open_before) + ) + + after = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert open_edges(after) == [] + assert {edge.to_time for edge in after if edge.status == "active"} == {retired_at.to_string()} + + async def test_a_relationship_stays_open_while_both_peers_are_live_on_one_branch( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + gadget = await _create_gadget(db=db, branch=default_branch, name="live-peer") + widget = await _create_widget(db=db, branch=default_branch, name="live-owner", serial=500, gadget=gadget) + + before = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + assert [edge.edge_type for edge in open_edges(before)].count("IS_RELATED") == 2 + + at = Timestamp() + assert await _retire(db=db, node_id=widget.id, at=at) == NodeAgnosticRetirementResult(edges_closed=0) + assert await _retire(db=db, node_id=gadget.id, at=at) == NodeAgnosticRetirementResult(edges_closed=0) + + assert edge_summary( + await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + ) == (edge_summary(before)) + + async def test_a_sourced_and_owned_attribute_closes_every_property_edge_type( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify HAS_OWNER and HAS_SOURCE edges are retired correctly.""" + source = await _create_gadget(db=db, branch=default_branch, name="the-source") + owner = await _create_gadget(db=db, branch=default_branch, name="the-owner") + + widget = await Node.init(db=db, schema=WIDGET_KIND, branch=default_branch) + await widget.new( + db=db, + name="sourced-and-owned", + serial={"value": 2200, "source": source.get_id(), "owner": owner.get_id()}, + ) + await widget.save(db=db) + + before = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert {edge.edge_type for edge in open_edges(before)} == { + "HAS_ATTRIBUTE", + "HAS_VALUE", + "IS_PROTECTED", + "HAS_SOURCE", + "HAS_OWNER", + }, "the attribute is expected to hold all four property edge types plus its owning edge" + + await tombstone_existence_only(db=db, node_id=widget.get_id(), branch=default_branch, at=Timestamp()) + + at = Timestamp() + assert await _retire(db=db, node_id=widget.get_id(), at=at) == NodeAgnosticRetirementResult( + edges_closed=len(open_edges(before)) + ) + + after = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert open_edges(after) == [] + + async def test_a_partially_closed_relationship_is_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify an illegal Relationship with 1 closed IS_RELATED edge is correctly retired""" + widget = await _create_widget(db=db, branch=default_branch, name="tombstoned-peer-edge", serial=2000) + gadget = await _create_gadget(db=db, branch=default_branch, name="tombstoned-peer-edge-gadget") + await widget.get_relationship(name="gadget").update(db=db, data=gadget) + await widget.save(db=db) + assert await relationship_peer_shape(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) == ( + 2, + 2, + ), "the relationship is expected to reach two distinct live peers before one edge is tombstoned" + + await tombstone_relationship_peer_edge( + db=db, + node_id=widget.get_id(), + identifier=RELATIONSHIP_IDENTIFIER, + peer_id=gadget.get_id(), + at=Timestamp(), + ) + before = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + assert edge_summary(before) == [ + ("IS_PROTECTED", "active", ""), + ("IS_RELATED", "active", ""), + ("IS_RELATED", "active", ""), + ("IS_RELATED", "deleted", ""), + ], "the tombstone is a new edge alongside the active one it supersedes, and both are open" + attribute_before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + + at = Timestamp() + assert await _retire(db=db, node_id=widget.id, at=at) == NodeAgnosticRetirementResult(edges_closed=3) + + assert edge_summary( + await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + ) == [ + ("IS_PROTECTED", "active", at.to_string()), + ("IS_RELATED", "active", at.to_string()), + ("IS_RELATED", "active", at.to_string()), + ("IS_RELATED", "deleted", ""), + ], "every active edge is closed and the tombstone is left exactly as it was" + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(attribute_before) + ), "the live node's own attribute is retained, so only the relationship was released" + + async def test_relationship_with_renamed_peer_is_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Relationship linked to a peer with an updated kind needs to still be retired correctly.""" + widget = await _create_widget(db=db, branch=default_branch, name="renamed-peer", serial=3300) + gadget = await _create_gadget(db=db, branch=default_branch, name="renamed-peer-gadget") + await widget.get_relationship(name="gadget").update(db=db, data=gadget) + await widget.save(db=db) + + await _rename_widget_kind(db=db, branch=default_branch) + assert await node_vertex_count(db=db, node_id=widget.id) == 2, ( + "the rename is expected to leave a superseded node vertex sharing the uuid" + ) + + await tombstone_existence_only(db=db, node_id=gadget.get_id(), branch=default_branch, at=Timestamp()) + + before = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + open_active_before = [edge for edge in open_edges(before) if edge.status == "active"] + assert open_active_before != [] + attribute_before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + + at = Timestamp() + assert await _retire(db=db, node_id=widget.id, at=at) == NodeAgnosticRetirementResult( + edges_closed=len(open_active_before) + ) + + after = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + assert [edge for edge in open_edges(after) if edge.status == "active"] == [], ( + "every active edge of the relationship is closed once its second peer is gone" + ) + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(attribute_before) + ), "the surviving `widget` still reads its own attribute, so only the relationship was released" + + async def test_a_renamed_kind_is_not_improperly_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify a kind-migrated object is not improperly retired""" + widget = await _create_widget(db=db, branch=default_branch, name="renamed-kind", serial=1100) + + await _rename_widget_kind(db=db, branch=default_branch) + + assert await node_vertex_count(db=db, node_id=widget.id) == 2, ( + "the rename is expected to leave a superseded node vertex sharing the uuid" + ) + assert await attribute_vertex_count(db=db, node_id=widget.id, attribute_name="serial") == 1, ( + "both node vertices are expected to share one attribute vertex" + ) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert await values_reachable_over_open_edges(db=db, node_id=widget.id, attribute_name="serial") == [1100] + + assert await _retire(db=db, node_id=widget.id, at=Timestamp()) == NodeAgnosticRetirementResult(edges_closed=0) + + assert await values_reachable_over_open_edges(db=db, node_id=widget.id, attribute_name="serial") == [1100] + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ) + + async def test_an_edge_that_begins_after_the_requested_time_is_left_alone( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Closing an edge at a time before it began would invert its interval, so it is not a candidate.""" + widget = await _create_widget(db=db, branch=default_branch, name="edge-from-the-future", serial=4200) + await tombstone_existence_only(db=db, node_id=widget.get_id(), branch=default_branch, at=Timestamp()) + + before = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert open_active_edges(before) != [], "the field has to be retirable, or the bounds are not what spared it" + + # the anchor half: nothing is even a candidate at a time before its owning edge began + assert await _retire( + db=db, node_id=widget.get_id(), at=Timestamp().subtract(hours=1) + ) == NodeAgnosticRetirementResult(edges_closed=0) + assert edge_summary( + await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + ) == edge_summary(before), "no candidate, so nothing is closed" + + # the closure half: a field that IS a candidate, holding one edge that begins later + future = Timestamp().add_delta(hours=1) + await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: "serial"}) + WITH DISTINCT a + MATCH (owner:Node {uuid: $node_id}) + CREATE (a)-[:HAS_SOURCE {branch: $global_branch, branch_level: 1, status: "active", from: $future}]->(owner) + """, + params={ + "node_id": widget.get_id(), + "global_branch": GLOBAL_BRANCH_NAME, + "future": future.to_string(), + }, + ) + with_future = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert len(open_active_edges(with_future)) == len(open_active_edges(before)) + 1 + + at = Timestamp() + assert await _retire(db=db, node_id=widget.get_id(), at=at) == NodeAgnosticRetirementResult( + edges_closed=len(open_active_edges(before)) + ), "every edge that had begun is closed, and the one that had not is not counted" + + after = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + still_open = open_active_edges(after) + assert [edge.edge_type for edge in still_open] == ["HAS_SOURCE"], ( + "the edge that begins later is left open rather than closed before it existed" + ) + assert still_open[0].to_time is None + + async def test_a_second_run_over_an_already_retired_field_closes_nothing( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify that retirement is idempotent.""" + widget = await _create_widget(db=db, branch=default_branch, name="retired-once", serial=2500) + + deleted_at = Timestamp() + to_delete = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch, raise_on_error=True) + await to_delete.delete(db=db, at=deleted_at) + + after_first_run = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after_first_run) == sorted( + (edge_type, "active", deleted_at.to_string()) + for edge_type in ("HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED") + ) + + second_run_at = Timestamp() + assert second_run_at > deleted_at, "the second run's stamp is expected to be distinguishable from the first" + + assert await _retire(db=db, node_id=widget.id, at=second_run_at) == NodeAgnosticRetirementResult(edges_closed=0) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(after_first_run) + ) diff --git a/backend/tests/helpers/agnostic_edges.py b/backend/tests/helpers/agnostic_edges.py new file mode 100644 index 00000000000..06ff0d92147 --- /dev/null +++ b/backend/tests/helpers/agnostic_edges.py @@ -0,0 +1,269 @@ +"""Graph-shape readers for the branch-agnostic retirement tests. + +These read edges directly rather than going through the node manager: the subject of the assertions +is which edges carry a `to` timestamp and which do not, and a read through the manager would hide the +very states the tests exist to pin down. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from infrahub.core.constants import GLOBAL_BRANCH_NAME + +if TYPE_CHECKING: + from infrahub.core.branch import Branch + from infrahub.core.timestamp import Timestamp + from infrahub.database import InfrahubDatabase + + +@dataclass(frozen=True) +class EdgeState: + """One edge as these assertions care about it: what it is, where it sits, and whether it is open.""" + + edge_type: str + branch: str + status: str + to_time: str | None + + @property + def is_open(self) -> bool: + return self.to_time is None + + @property + def is_active(self) -> bool: + return self.status == "active" + + +def open_edges(edges: list[EdgeState]) -> list[EdgeState]: + """Every open edge, one entry each. + + Counting distinct types instead undercounts a vertex holding two edges of one type, which is every + relationship vertex, so an expected closure count is read from this and never from the types. + """ + return [edge for edge in edges if edge.is_open] + + +def open_active_edges(edges: list[EdgeState]) -> list[EdgeState]: + """The edges a retirement run is expected to close: open, and not already a tombstone.""" + return [edge for edge in edges if edge.is_open and edge.is_active] + + +def open_edge_types(edges: list[EdgeState]) -> set[str]: + return {edge.edge_type for edge in open_edges(edges)} + + +def edge_summary(edges: list[EdgeState]) -> list[tuple[str, str, str]]: + """Order-independent view of a vertex's edges, for before/after comparison.""" + return sorted((edge.edge_type, edge.status, edge.to_time or "") for edge in edges) + + +def expected_closed_at(edges: list[EdgeState], at: Timestamp) -> list[tuple[str, str, str]]: + """What `edge_summary` should return once every type present has been closed at `at`.""" + return sorted((edge_type, "active", at.to_string()) for edge_type in {edge.edge_type for edge in edges}) + + +async def attribute_global_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[EdgeState]: + """Every global-branch edge touching the attribute vertex, owning edge included.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WITH DISTINCT a + MATCH (a)-[e]-() + WHERE e.branch = $global_branch + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id, "attribute_name": attribute_name, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def relationship_global_edges(db: InfrahubDatabase, node_id: str, identifier: str) -> list[EdgeState]: + """Every global-branch edge of the relationship vertex reached from this node. + + Reached through the node rather than matched on the identifier alone, so a relationship built by + another test sharing the database is not counted. The traversal is unfiltered on purpose: the edge + it arrives over may itself have been closed by the run under assertion. + """ + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:IS_RELATED]-(r:Relationship {name: $identifier}) + WITH DISTINCT r + MATCH (r)-[e]-() + WHERE e.branch = $global_branch + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id, "identifier": identifier, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def attribute_owning_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[EdgeState]: + """Every owning edge of the attribute vertex, on any branch.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WITH DISTINCT a + MATCH (:Node)-[e:HAS_ATTRIBUTE]->(a) + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id, "attribute_name": attribute_name}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def existence_edges(db: InfrahubDatabase, node_id: str) -> list[EdgeState]: + """Every existence edge of the node, on any branch.""" + results = await db.execute_query( + query=""" + MATCH (n:Node {uuid: $node_id})-[e:IS_PART_OF]->(:Root) + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def relationship_peer_shape(db: InfrahubDatabase, node_id: str, identifier: str) -> tuple[int, int]: + """How many live global `IS_RELATED` edges this node's relationship has, and how many peers.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:IS_RELATED]-(r:Relationship {name: $identifier}) + WITH DISTINCT r + MATCH (r)-[e:IS_RELATED]-(peer:Node) + WHERE e.branch = $global_branch AND e.status = "active" AND e.to IS NULL + RETURN count(e) AS edge_count, count(DISTINCT peer.uuid) AS peer_count + """, + params={"node_id": node_id, "identifier": identifier, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return results[0]["edge_count"], results[0]["peer_count"] + + +async def node_vertex_count(db: InfrahubDatabase, node_id: str) -> int: + """How many `:Node` vertices carry this uuid; more than one after a kind or inheritance change.""" + results = await db.execute_query( + query="MATCH (n:Node {uuid: $node_id}) RETURN count(n) AS vertex_count", + params={"node_id": node_id}, + ) + return results[0]["vertex_count"] + + +async def attribute_vertex_count(db: InfrahubDatabase, node_id: str, attribute_name: str) -> int: + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + RETURN count(DISTINCT a) AS attribute_count + """, + params={"node_id": node_id, "attribute_name": attribute_name}, + ) + return results[0]["attribute_count"] + + +async def values_reachable_over_open_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[Any]: + """The attribute values a node still reads, following only open, active global edges.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[owning:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WHERE owning.branch = $global_branch AND owning.status = "active" AND owning.to IS NULL + MATCH (a)-[value_edge:HAS_VALUE]->(value:AttributeValue) + WHERE value_edge.branch = $global_branch AND value_edge.status = "active" AND value_edge.to IS NULL + RETURN value.value AS value + """, + params={"node_id": node_id, "attribute_name": attribute_name, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return [result["value"] for result in results] + + +async def tombstone_existence_only(db: InfrahubDatabase, node_id: str, branch: Branch, at: Timestamp) -> None: + """Mark the owner deleted while leaving every field edge exactly as it was. + + The ordinary delete tombstones an attribute's edges alongside the existence edge, so both axes flip + together and either one alone would reach the right answer. This builds the state where they + disagree -- an owner no branch reads as live still holding an open, active global value edge -- + which is the orphan shape this feature repairs. + """ + results = await db.execute_query( + query=""" + MATCH (n:Node {uuid: $node_id})-[existing:IS_PART_OF]->(root:Root) + WHERE existing.branch = $branch AND existing.status = "active" AND existing.to IS NULL + SET existing.to = $at + CREATE (n)-[:IS_PART_OF {branch: $branch, branch_level: $branch_level, status: "deleted", from: $at}]->(root) + RETURN count(existing) AS tombstoned + """, + params={ + "node_id": node_id, + "branch": branch.name, + "branch_level": branch.hierarchy_level, + "at": at.to_string(), + }, + ) + assert results[0]["tombstoned"] == 1 + + +async def tombstone_relationship_peer_edge( + db: InfrahubDatabase, node_id: str, identifier: str, peer_id: str, at: Timestamp +) -> None: + """Add a `deleted` global peer edge that supersedes the active one, leaving the active one open. + + A tombstone is a new, more recent edge rather than a rewrite of the one it supersedes -- the graph + never holds a `deleted` edge where the `active` one it replaced has vanished. Both stay open, and + which of them speaks for the branch is decided by the ordering, which is what puts the status under + test here. + """ + results = await db.execute_query( + query=""" + MATCH (peer:Node {uuid: $peer_id})-[active:IS_RELATED]-(r:Relationship {name: $identifier}) + WHERE active.branch = $global_branch AND active.status = "active" AND active.to IS NULL + AND EXISTS { MATCH (:Node {uuid: $node_id})-[:IS_RELATED]-(r) } + CREATE (peer)-[:IS_RELATED {branch: $global_branch, branch_level: active.branch_level, + status: "deleted", from: $at}]->(r) + RETURN count(active) AS tombstoned + """, + params={ + "node_id": node_id, + "identifier": identifier, + "peer_id": peer_id, + "global_branch": GLOBAL_BRANCH_NAME, + "at": at.to_string(), + }, + ) + assert results[0]["tombstoned"] == 1 + + +async def remove_attribute_on_branch( + db: InfrahubDatabase, node_id: str, attribute_name: str, branch: Branch, at: Timestamp +) -> None: + """Mirror the owning edge with a branch-level `deleted` one, as a schema attribute removal does. + + A removal never closes the global edges; it writes a more specific `deleted` edge that wins under + the removing branch's view only, which is what makes the field disappear there while the object + goes on existing everywhere. + """ + await db.execute_query( + query=""" + MATCH (n:Node {uuid: $node_id})-[owning:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WHERE owning.branch = $global_branch AND owning.status = "active" AND owning.to IS NULL + CREATE (n)-[:HAS_ATTRIBUTE {branch: $branch_name, branch_level: $branch_level, status: "deleted", from: $at}]->(a) + """, + params={ + "node_id": node_id, + "attribute_name": attribute_name, + "global_branch": GLOBAL_BRANCH_NAME, + "branch_name": branch.name, + "branch_level": branch.hierarchy_level, + "at": at.to_string(), + }, + ) + + +async def pool_reservation_edges(db: InfrahubDatabase, pool_id: str, identifier: str) -> list[EdgeState]: + """Every reservation edge a pool holds under one identifier, on any branch.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $pool_id})-[e:IS_RESERVED {identifier: $identifier}]->(:AttributeValue) + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"pool_id": pool_id, "identifier": identifier}, + ) + return [EdgeState(**dict(result)) for result in results] diff --git a/backend/tests/helpers/schema/agnostic_retirement.py b/backend/tests/helpers/schema/agnostic_retirement.py new file mode 100644 index 00000000000..9cc86a851f1 --- /dev/null +++ b/backend/tests/helpers/schema/agnostic_retirement.py @@ -0,0 +1,46 @@ +from infrahub.core.constants import BranchSupportType, RelationshipCardinality, RelationshipKind +from infrahub.core.schema import AttributeSchema, NodeSchema, RelationshipSchema, SchemaRoot + +RELATIONSHIP_IDENTIFIER = "agnosticretire_widget__agnosticretire_gadget" + +WIDGET_KIND = "AgnosticretireWidget" +GADGET_KIND = "AgnosticretireGadget" +BEACON_KIND = "AgnosticretireBeacon" + +AGNOSTIC_WIDGET = NodeSchema( + name="Widget", + namespace="Agnosticretire", + branch=BranchSupportType.AWARE, + attributes=[ + AttributeSchema(name="name", kind="Text", unique=True), + AttributeSchema(name="serial", kind="Number", branch=BranchSupportType.AGNOSTIC), + ], + relationships=[ + RelationshipSchema( + name="gadget", + kind=RelationshipKind.GENERIC, + peer=GADGET_KIND, + identifier=RELATIONSHIP_IDENTIFIER, + cardinality=RelationshipCardinality.ONE, + optional=True, + branch=BranchSupportType.AGNOSTIC, + ), + ], +) + +AGNOSTIC_GADGET = NodeSchema( + name="Gadget", + namespace="Agnosticretire", + branch=BranchSupportType.AWARE, + attributes=[AttributeSchema(name="name", kind="Text", unique=True)], +) + +# A kind that is itself branch-agnostic, so its own existence edge lives on the global branch. +AGNOSTIC_BEACON = NodeSchema( + name="Beacon", + namespace="Agnosticretire", + branch=BranchSupportType.AGNOSTIC, + attributes=[AttributeSchema(name="name", kind="Text", unique=True)], +) + +AGNOSTIC_RETIREMENT_SCHEMA = SchemaRoot(nodes=[AGNOSTIC_WIDGET, AGNOSTIC_GADGET, AGNOSTIC_BEACON])