diff --git a/plugins/module_utils/nd_output.py b/plugins/module_utils/nd_output.py index 934a36a24..d38f6e71d 100644 --- a/plugins/module_utils/nd_output.py +++ b/plugins/module_utils/nd_output.py @@ -2,11 +2,13 @@ # GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) -from __future__ import absolute_import, division, print_function +from __future__ import absolute_import, annotations, division, print_function from typing import Any, Dict, List, Optional, Union -from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import NDConfigCollection +from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import ( + NDConfigCollection, +) from ansible_collections.cisco.nd.plugins.module_utils.rest.results import Results @@ -14,27 +16,54 @@ class NDOutput: def __init__(self, output_level: str): self._output_level: str = output_level self._changed: bool = False + self._changed_explicit: Optional[bool] = None self._before: Union[NDConfigCollection, List] = [] self._after: Union[NDConfigCollection, List] = [] + self._after_available: bool = True self._diff: Union[NDConfigCollection, List] = [] + self._diff_available: bool = True self._proposed: Union[NDConfigCollection, List] = [] self._logs: List = [] self._extra: Dict[str, Any] = {} def format(self, **kwargs) -> Dict[str, Any]: - if isinstance(self._before, NDConfigCollection) and isinstance(self._after, NDConfigCollection) and self._before.get_diff_collection(self._after): + if ( + self._changed_explicit is None + and self._after_available + and isinstance(self._before, NDConfigCollection) + and isinstance(self._after, NDConfigCollection) + and self._before.get_diff_collection(self._after) + ): self._changed = True output = { "output_level": self._output_level, "changed": self._changed, - "after": self._after.to_ansible_config() if isinstance(self._after, NDConfigCollection) else self._after, - "before": self._before.to_ansible_config() if isinstance(self._before, NDConfigCollection) else self._before, - "diff": self._diff.to_ansible_config() if isinstance(self._diff, NDConfigCollection) else self._diff, + "before": ( + self._before.to_ansible_config() + if isinstance(self._before, NDConfigCollection) + else self._before + ), } + if self._after_available: + output["after"] = ( + self._after.to_ansible_config() + if isinstance(self._after, NDConfigCollection) + else self._after + ) + if self._diff_available: + output["diff"] = ( + self._diff.to_ansible_config() + if isinstance(self._diff, NDConfigCollection) + else self._diff + ) if self._output_level in ("debug", "info"): - output["proposed"] = self._proposed.to_ansible_config() if isinstance(self._proposed, NDConfigCollection) else self._proposed + output["proposed"] = ( + self._proposed.to_ansible_config() + if isinstance(self._proposed, NDConfigCollection) + else self._proposed + ) if self._output_level == "debug": output["logs"] = self._logs @@ -45,7 +74,62 @@ def format(self, **kwargs) -> Dict[str, Any]: return output - def format_with_verbosity(self, verbosity: int, results: Optional[Results] = None, **kwargs) -> Dict[str, Any]: + def set_changed(self, changed: bool) -> None: + """Set changed from mutation evidence rather than collection comparison.""" + self._changed = bool(changed) + self._changed_explicit = bool(changed) + + def set_after_state( + self, + after: NDConfigCollection, + *, + status: str, + verification_performed: bool = False, + reconciliation_complete: bool = True, + ) -> None: + """Expose a known planned, confirmed, or observed after-state.""" + if not isinstance(after, NDConfigCollection): + raise TypeError("after must be an NDConfigCollection") + self._after = after + self._after_available = True + self._diff_available = True + self._extra.update( + after_status=status, + diff_status=status, + reconciliation_required=not reconciliation_complete, + reconciliation_complete=reconciliation_complete, + verification_performed=verification_performed, + ) + self._extra.pop("affected_identifiers", None) + self._extra.pop("may_have_changed", None) + self._extra.pop("verification_error", None) + + def mark_after_unknown( + self, + *, + affected_identifiers: List[Any], + may_have_changed: bool, + verification_performed: bool = False, + verification_error: str | None = None, + ) -> None: + """Omit after and diff when final controller state is not provable.""" + self._after_available = False + self._diff_available = False + self._extra.update( + after_status="unknown", + diff_status="unknown", + reconciliation_required=True, + reconciliation_complete=False, + verification_performed=verification_performed, + affected_identifiers=affected_identifiers, + may_have_changed=bool(may_have_changed), + ) + if verification_error is not None: + self._extra["verification_error"] = verification_error + + def format_with_verbosity( + self, verbosity: int, results: Optional[Results] = None, **kwargs + ) -> Dict[str, Any]: """ Build output dict filtered by CLI verbosity level. @@ -68,7 +152,7 @@ def format_with_verbosity(self, verbosity: int, results: Optional[Results] = Non final = results.final_result # Merge changed/failed from Results (API-level) with NDOutput (config-level). - if final.get("changed"): + if final.get("changed") and self._changed_explicit is None: output["changed"] = True if final.get("failed"): output["failed"] = True @@ -105,6 +189,7 @@ def assign( ) -> None: if isinstance(after, NDConfigCollection): self._after = after + self._after_available = True if isinstance(before, NDConfigCollection): self._before = before if isinstance(diff, NDConfigCollection): diff --git a/plugins/module_utils/nd_state_machine.py b/plugins/module_utils/nd_state_machine.py index e9adfc59e..3aaf42e6e 100644 --- a/plugins/module_utils/nd_state_machine.py +++ b/plugins/module_utils/nd_state_machine.py @@ -5,16 +5,40 @@ from __future__ import absolute_import, annotations, division, print_function +import time from typing import Any, Callable from ansible.module_utils.basic import AnsibleModule -from ansible_collections.cisco.nd.plugins.module_utils.common.exceptions import NDStateMachineError +from ansible_collections.cisco.nd.plugins.module_utils.common.exceptions import ( + NDStateMachineError, +) from ansible_collections.cisco.nd.plugins.module_utils.models.base import NDBaseModel -from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import NDConfigCollection +from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import ( + NDConfigCollection, +) from ansible_collections.cisco.nd.plugins.module_utils.nd_output import NDOutput -from ansible_collections.cisco.nd.plugins.module_utils.orchestrators.base import NDBaseOrchestrator -from ansible_collections.cisco.nd.plugins.module_utils.orchestrators.types import ResponseType -from ansible_collections.cisco.nd.plugins.module_utils.rest.response_handler_nd import ResponseHandler +from ansible_collections.cisco.nd.plugins.module_utils.nd_state_plan import ( + NDStatePlan, + NDStatePlanner, +) +from ansible_collections.cisco.nd.plugins.module_utils.nd_state_reconciliation import ( + DeferredMutation, + MutationCheckpoint, + MutationEffect, + MutationJournal, + MutationOperation, + MutationOutcome, +) +from ansible_collections.cisco.nd.plugins.module_utils.orchestrators.base import ( + NDBaseOrchestrator, +) +from ansible_collections.cisco.nd.plugins.module_utils.orchestrators.types import ( + FinalizationContext, + ResponseType, +) +from ansible_collections.cisco.nd.plugins.module_utils.rest.response_handler_nd import ( + ResponseHandler, +) from ansible_collections.cisco.nd.plugins.module_utils.rest.rest_send import RestSend from ansible_collections.cisco.nd.plugins.module_utils.rest.results import Results from ansible_collections.cisco.nd.plugins.module_utils.rest.sender_nd import Sender @@ -25,7 +49,11 @@ class NDStateMachine: Generic State Machine for Nexus Dashboard (Bulk Support). """ - def __init__(self, module: AnsibleModule, model_orchestrator: type[NDBaseOrchestrator] | NDBaseOrchestrator): + def __init__( + self, + module: AnsibleModule, + model_orchestrator: type[NDBaseOrchestrator] | NDBaseOrchestrator, + ): """ Initialize the ND State Machine. """ @@ -48,13 +76,19 @@ def __init__(self, module: AnsibleModule, model_orchestrator: type[NDBaseOrchest # Configuration # Accept either an orchestrator instance or a class. - if isinstance(model_orchestrator, type) and issubclass(model_orchestrator, NDBaseOrchestrator): - self.model_orchestrator = model_orchestrator(rest_send=self.rest_send, results=self.results) + if isinstance(model_orchestrator, type) and issubclass( + model_orchestrator, NDBaseOrchestrator + ): + self.model_orchestrator = model_orchestrator( + rest_send=self.rest_send, results=self.results + ) elif isinstance(model_orchestrator, NDBaseOrchestrator): self.model_orchestrator = model_orchestrator self.model_orchestrator.results = self.results else: - raise NDStateMachineError(f"model_orchestrator must be an NDBaseOrchestrator class or instance. Got: {type(model_orchestrator)}") + raise NDStateMachineError( + f"model_orchestrator must be an NDBaseOrchestrator class or instance. Got: {type(model_orchestrator)}" + ) self.model_class = self.model_orchestrator.model_class self.state = self.module.params["state"] @@ -64,14 +98,25 @@ def __init__(self, module: AnsibleModule, model_orchestrator: type[NDBaseOrchest self.ignore_errors = self.module.params.get("ignore_errors", False) self.supports_bulk_create = self.model_orchestrator.supports_bulk_create self.supports_bulk_delete = self.model_orchestrator.supports_bulk_delete + self.journal = MutationJournal() + self.plan: NDStatePlan | None = None + self.observed: NDConfigCollection | None = None + self._finalized = False + self._verify_settings = self._verification_settings() # Initialize collections try: response_data = self.model_orchestrator.query_all() # State of configuration objects in ND before change execution - self.before = NDConfigCollection.from_api_response(response_data=response_data, model_class=self.model_class) - # State of current configuration objects in ND during change execution - self.existing = self.before.copy() + self.before = NDConfigCollection.from_api_response( + response_data=response_data, model_class=self.model_class + ) + # Planned and confirmed state must never alias each other. The + # legacy ``existing`` name remains a compatibility alias for the + # evidence-backed confirmed collection. + self.planned = self.before.copy() + self.confirmed = self.before.copy() + self.existing = self.confirmed # Ongoing collection of configuration objects that were changed self.sent = NDConfigCollection(model_class=self.model_class) # Collection of configuration objects given by user. @@ -79,181 +124,411 @@ def __init__(self, module: AnsibleModule, model_orchestrator: type[NDBaseOrchest # state-aware validation (e.g. require certain fields for write states while accepting # identifier-only items for ``deleted``). Models that do not read the context ignore it. self.proposed = NDConfigCollection.from_ansible_config( - data=self.module.params.get("config", []), model_class=self.model_class, context={"state": self.state} + data=self.module.params.get("config", []), + model_class=self.model_class, + context={"state": self.state}, ) - self.output.assign(after=self.existing, before=self.before, proposed=self.proposed) + self.output.assign(before=self.before, proposed=self.proposed) + self.output.set_after_state(self.confirmed, status="confirmed") except Exception as e: raise NDStateMachineError(f"Initialization failed: {str(e)}") from e # State Management (core function) def manage_state(self) -> None: - """ - Manage state according to desired configuration. - """ - if self.state in ["merged", "replaced", "overridden"]: - proposed_items = list(self.proposed) + """Plan first, then apply only controller effects supported by evidence.""" + if self.state not in {"merged", "replaced", "overridden", "deleted"}: + raise NDStateMachineError(f"Invalid state: {self.state}") - # Policy-required-on-create guard (issue #350) runs FIRST: it is local-only (self.existing is - # already in memory), so it fails before the API-backed capability preflight below and before - # _manage_create_update_state mutates self.existing, which NDOutput aliases as `after`. Create - # subset = proposed items not present in the existing inventory -- the same key-membership - # criterion get_diff_config uses to classify "new" (PR #362 review). - items_to_create = [item for item in proposed_items if self.existing.get(item.get_identifier_value()) is None] - - # Normalize preflight failures to NDStateMachineError (PR #362 review, gmicol). Both preflight - # hooks raise a bare RuntimeError (base_interface.preflight_create / the capability preflight), - # but nd_interface_svi and nd_interface_subinterface_managed/_unmanaged catch only - # NDStateMachineError at their entrypoint. Without this wrap a policy-less (or capability) preflight - # failure in those modules escapes as an unhandled RuntimeError, bypassing fail_json and losing the - # structured before/after/changed output the guard exists to provide. This wrap deliberately does - # NOT route through _execute_operation: both preflights must run in check mode too, and - # _execute_operation skips execution during a dry-run. + if self.state in {"merged", "replaced", "overridden"}: + proposed_items = list(self.proposed) + items_to_create = [ + item + for item in proposed_items + if self.before.get(item.get_identifier_value()) is None + ] try: self.model_orchestrator.preflight_create(items_to_create) - - # Capability preflight runs here -- before _manage_create_update_state, whose mutations are - # skipped in check mode -- so dry-runs surface incapable switches (PR #275 / issue #273). self.model_orchestrator.preflight(proposed_items) except NDStateMachineError: raise except Exception as e: raise NDStateMachineError(f"Preflight failed: {e}") from e - self._manage_create_update_state() + try: + self.plan = NDStatePlanner.plan( + state=self.state, + before=self.before, + proposed=self.proposed, + ignore_errors=self.ignore_errors, + ) + except Exception as e: + raise NDStateMachineError(f"Planning failed: {e}") from e - if self.state == "overridden": - self._manage_override_deletions() + self.planned = self.plan.after + if self.check_mode: + self._preview_plan() + return - elif self.state == "deleted": - # Capability preflight intentionally NOT run for deletes: removing configuration does not - # depend on a switch's capability to host the interface type (PR #275 scope decision). - self._manage_delete_state() + self._execute_plan() + self._update_output_from_journal() + + def _preview_plan(self) -> None: + """Expose the prospective plan without executing controller writes.""" + if self.plan is None: + raise NDStateMachineError("State plan is not available") + for item in (*self.plan.updates, *self.plan.creates, *self.plan.deletes): + self._add_sent(item) + self.output.set_changed(self.plan.changed) + self.output.set_after_state(self.planned, status="planned") + + def _effect( + self, operation: MutationOperation, item: NDBaseModel + ) -> MutationEffect: + """Build a resource transition from one planned operation model.""" + identifier = item.get_identifier_value() + before = self.before.get(identifier) + after = None if operation is MutationOperation.DELETE else item + return MutationEffect( + operation=operation, identifier=identifier, before=before, after=after + ) + + def _execute_plan(self) -> None: + """Execute the plan in the state machine's established operation order.""" + if self.plan is None: + raise NDStateMachineError("State plan is not available") + + execution_steps: list[ + tuple[ + MutationCheckpoint, + Callable[..., ResponseType | DeferredMutation], + tuple[Any, ...], + str, + ] + ] = [] + + for item in self.plan.updates: + checkpoint = self.journal.open( + phase="update", effects=(self._effect(MutationOperation.UPDATE, item),) + ) + execution_steps.append( + ( + checkpoint, + self.model_orchestrator.update, + (item,), + f"Failed to update {item.get_identifier_value()}", + ) + ) + creates = list(self.plan.creates) + if creates and self.supports_bulk_create: + checkpoint = self.journal.open( + phase="create", + effects=( + self._effect(MutationOperation.CREATE, item) for item in creates + ), + ) + execution_steps.append( + ( + checkpoint, + self.model_orchestrator.create_bulk, + (creates,), + "Failed to create in bulk", + ) + ) else: - raise NDStateMachineError(f"Invalid state: {self.state}") + for item in creates: + checkpoint = self.journal.open( + phase="create", + effects=(self._effect(MutationOperation.CREATE, item),), + ) + execution_steps.append( + ( + checkpoint, + self.model_orchestrator.create, + (item,), + f"Failed to create {item.get_identifier_value()}", + ) + ) + + deletes = list(self.plan.deletes) + if deletes and self.supports_bulk_delete: + checkpoint = self.journal.open( + phase="delete", + effects=( + self._effect(MutationOperation.DELETE, item) for item in deletes + ), + ) + execution_steps.append( + ( + checkpoint, + self.model_orchestrator.delete_bulk, + (deletes,), + "Failed to delete in bulk", + ) + ) + else: + for item in deletes: + checkpoint = self.journal.open( + phase="delete", + effects=(self._effect(MutationOperation.DELETE, item),), + ) + execution_steps.append( + ( + checkpoint, + self.model_orchestrator.delete, + (item,), + f"Failed to delete {item.get_identifier_value()}", + ) + ) + + for checkpoint, operation, args, error_msg_prefix in execution_steps: + self._execute_operation( + checkpoint, + operation, + *args, + error_msg_prefix=error_msg_prefix, + ) + + @staticmethod + def _is_write_verb(verb: str) -> bool: + return verb.upper() in {"POST", "PUT", "PATCH", "DELETE"} + + @staticmethod + def _proves_no_change(call) -> bool: + """Require explicit endpoint certainty; aggregate changed=False is insufficient.""" + return call.failed and call.result.get("outcome_certainty") == "no_change" + + def _classify_operation( + self, + *, + result_sequence: int, + attempt_sequence: int, + raised: bool, + returned: ResponseType | DeferredMutation, + ) -> tuple[MutationOutcome, bool, bool, tuple[int, ...]]: + """Classify one logical operation from request-attempt and response evidence.""" + calls = self.results.calls_since(result_sequence) + attempts = self.results.attempts_since(attempt_sequence) + write_calls = tuple(call for call in calls if self._is_write_verb(call.verb)) + write_attempts = tuple( + attempt for attempt in attempts if self._is_write_verb(attempt.verb) + ) + incomplete_attempt = any(not attempt.completed for attempt in write_attempts) + failed_calls = tuple( + call + for call in write_calls + if call.failed or call.result.get("success") is False + ) + successful_calls = tuple( + call + for call in write_calls + if not call.failed and call.result.get("success") is True + ) + changed = bool(successful_calls) or any( + call.changed or call.result.get("changed") is True for call in write_calls + ) + call_sequences = tuple(call.sequence_number for call in calls) + + if isinstance(returned, DeferredMutation) and not raised: + return MutationOutcome.QUEUED, False, False, call_sequences + + if incomplete_attempt: + return MutationOutcome.UNKNOWN, changed, not changed, call_sequences + + if failed_calls: + no_change_is_proven = not successful_calls and all( + self._proves_no_change(call) for call in failed_calls + ) + if no_change_is_proven: + return MutationOutcome.FAILED, False, False, call_sequences + return MutationOutcome.UNKNOWN, changed, not changed, call_sequences + + if raised: + if write_calls or write_attempts: + return MutationOutcome.UNKNOWN, changed, not changed, call_sequences + return MutationOutcome.FAILED, False, False, call_sequences + + return MutationOutcome.SUCCEEDED, True, False, call_sequences def _execute_operation( self, - operation: Callable[..., ResponseType], + checkpoint: MutationCheckpoint, + operation: Callable[..., ResponseType | DeferredMutation], *args: Any, error_msg_prefix: str = "Operation failed", **kwargs: Any, - ) -> ResponseType | None: - """Execute an API operation with standardized error handling.""" - try: - if not self.check_mode: - return operation(*args, **kwargs) - return None - except Exception as e: - error_msg = f"{error_msg_prefix}: {e}" - if not self.ignore_errors: - raise NDStateMachineError(error_msg) from e - return None - - def _manage_create_update_state(self) -> None: - """ - Handle merged/replaced/overridden states. - """ - items_to_create: list[NDBaseModel] = [] - items_to_update: list[NDBaseModel] = [] + ) -> MutationOutcome: + """Execute, classify, journal, and reconcile one logical mutation.""" + result_sequence = self.results.task_sequence_number + attempt_sequence = self.results.api_attempt_sequence_number + returned: ResponseType | DeferredMutation = None + caught: Exception | None = None - for proposed_item in self.proposed: - identifier = None - try: - # Extract identifier - identifier = proposed_item.get_identifier_value() - # Determine diff status - # For merged state, only compare fields explicitly provided by - # the user so that Pydantic default values do not trigger false - # diffs or overwrite existing configuration. - exclude_unset = self.state == "merged" - diff_status = self.existing.get_diff_config(proposed_item, exclude_unset=exclude_unset) - - # No changes needed - if diff_status == "no_diff": - continue - - # Prepare final config based on state - if self.state == "merged": - # Merge with existing - final_item = self.existing.merge(proposed_item) - else: - # Replace or creates - if diff_status == "changed": - self.existing.replace(proposed_item) - else: - self.existing.add(proposed_item) - final_item = proposed_item - - # Categorize by operation type - if diff_status == "changed": - items_to_update.append(final_item) - elif diff_status == "new": - items_to_create.append(final_item) - - except Exception as e: - if identifier: - error_msg = f"Failed to process {identifier}: {e}" - else: - error_msg = f"Failed to process: {e}" - if not self.ignore_errors: - raise NDStateMachineError(error_msg) from e - - # The policy-required-on-create guard (issue #350) runs in manage_state, before the capability - # preflight and before this method mutates self.existing (PR #362 review). - - # Execute updates (always individual) - for item in items_to_update: - self._execute_operation(self.model_orchestrator.update, item, error_msg_prefix=f"Failed to update {item.get_identifier_value()}") - - # Execute creates (bulk or individual) - if items_to_create: - if self.supports_bulk_create: - self._execute_operation(self.model_orchestrator.create_bulk, items_to_create, error_msg_prefix="Failed to create in bulk") + try: + returned = operation(*args, **kwargs) + except Exception as e: # outcome must be journaled before propagation + caught = e + + outcome, changed, may_have_changed, call_sequences = self._classify_operation( + result_sequence=result_sequence, + attempt_sequence=attempt_sequence, + raised=caught is not None, + returned=returned, + ) + error_msg = f"{error_msg_prefix}: {caught}" if caught is not None else None + if error_msg is None and outcome in { + MutationOutcome.FAILED, + MutationOutcome.UNKNOWN, + }: + error_msg = f"{error_msg_prefix}: controller outcome is {outcome.value}" + checkpoint.resolve( + outcome, + changed=changed, + may_have_changed=may_have_changed, + error=error_msg, + api_call_sequences=call_sequences, + ) + + if outcome is MutationOutcome.SUCCEEDED: + self._apply_confirmed_effects(checkpoint.effects) + self._update_output_from_journal() + + if caught is not None and not self.ignore_errors: + raise NDStateMachineError(error_msg or error_msg_prefix) from caught + if ( + outcome in {MutationOutcome.FAILED, MutationOutcome.UNKNOWN} + and not self.ignore_errors + ): + raise NDStateMachineError(error_msg or error_msg_prefix) + return outcome + + def _apply_confirmed_effects(self, effects: tuple[MutationEffect, ...]) -> None: + """Apply proven transitions immediately and populate sent from them only.""" + for effect in effects: + if effect.operation is MutationOperation.DELETE: + self.confirmed.delete(effect.identifier) + if effect.before is not None: + self._add_sent(effect.before) + continue + + if effect.after is None: + raise NDStateMachineError( + f"Missing final model for {effect.operation.value} {effect.identifier}" + ) + if self.confirmed.get(effect.identifier) is None: + self.confirmed.add(effect.after) else: - for item in items_to_create: - self._execute_operation(self.model_orchestrator.create, item, error_msg_prefix=f"Failed to create {item.get_identifier_value()}") - - # Mark as sent only after successful API operations - successfully_sent = items_to_update + items_to_create - if successfully_sent: - self.sent.add_many(successfully_sent) - - # Log operation - self.output.assign(after=self.existing) - - def _manage_override_deletions(self) -> None: - """ - Delete items not in proposed config (for overridden state). - """ - diff_identifiers = self.before.get_diff_identifiers(self.proposed) - items_to_delete = [existing_item for identifier in diff_identifiers if (existing_item := self.existing.get(identifier)) is not None] - self._delete_items(items_to_delete) - - def _manage_delete_state(self) -> None: - """Handle deleted state.""" - items_to_delete = [ - existing_item for proposed_item in self.proposed if (existing_item := self.existing.get(proposed_item.get_identifier_value())) is not None - ] - self._delete_items(items_to_delete) - - def _delete_items(self, items: list[NDBaseModel]) -> None: - """Delete a list of items individually or in bulk.""" - if not items: - return - - # Execute deletes (bulk or individual) - if self.supports_bulk_delete: - self._execute_operation(self.model_orchestrator.delete_bulk, items, error_msg_prefix="Failed to delete in bulk") + self.confirmed.replace(effect.after) + self._add_sent(effect.after) + + def _add_sent(self, item: NDBaseModel) -> None: + """Upsert one confirmed or check-mode-preview item into sent.""" + identifier = item.get_identifier_value() + if self.sent.get(identifier) is None: + self.sent.add(item) else: - for item in items: - self._execute_operation(self.model_orchestrator.delete, item, error_msg_prefix=f"Failed to delete {item.get_identifier_value()}") - - # Batch remove from collection (single index rebuild) - keys_to_delete = [item.get_identifier_value() for item in items] - self.existing.delete_many(keys_to_delete) + self.sent.replace(item) + + def _update_output_from_journal(self) -> None: + """Keep output truthful after every checkpoint, including failures.""" + self.output.set_changed(self.journal.changed) + if self.journal.has_unknown: + self.output.mark_after_unknown( + affected_identifiers=list(self.journal.unknown_identifiers), + may_have_changed=self.journal.may_have_changed, + ) + return + self.output.set_after_state(self.confirmed, status="confirmed") + + @staticmethod + def _positive_int_setting(settings: dict[str, Any], name: str, default: int) -> int: + value = settings.get(name, default) + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise NDStateMachineError(f"verify.{name} must be a positive integer") + return value + + def _verification_settings(self) -> dict[str, int] | None: + """Validate the opt-in unknown-state readback settings before writes.""" + raw_verify = self.module.params.get("verify") + if raw_verify in (None, False): + return None + if raw_verify is True: + raw_verify = {} + if not isinstance(raw_verify, dict): + raise NDStateMachineError("verify must be a boolean or dictionary") + enabled = raw_verify.get("enabled", True) + if not isinstance(enabled, bool): + raise NDStateMachineError("verify.enabled must be a boolean") + if not enabled: + return None + delay = raw_verify.get("delay", 1) + if isinstance(delay, bool) or not isinstance(delay, int) or delay < 0: + raise NDStateMachineError("verify.delay must be a non-negative integer") + return { + "retries": self._positive_int_setting(raw_verify, "retries", 5), + "timeout": self._positive_int_setting(raw_verify, "timeout", 10), + "delay": delay, + } + + def finalize(self, *, primary_error: Exception | None = None) -> None: + """Resolve output once at the outer workflow boundary.""" + if self._finalized: + return + if self.check_mode: + self.output.set_changed(bool(self.plan and self.plan.changed)) + self.output.set_after_state(self.planned, status="planned") + self._finalized = True + return + if not self.journal.has_unknown: + self._update_output_from_journal() + self._finalized = True + return + if self._verify_settings is None: + self._update_output_from_journal() + self._finalized = True + return - # Log deletion - self.output.assign(after=self.existing) + context = FinalizationContext( + state=self.state, + affected_identifiers=self.journal.unknown_identifiers, + confirmed_identifiers=tuple(self.confirmed.keys()), + ) + settings = self._verify_settings + rest_send = self.model_orchestrator.rest_send + original_timeout = rest_send.timeout + errors: list[str] = [] + try: + rest_send.timeout = settings["timeout"] + for attempt in range(1, settings["retries"] + 1): + try: + response_data = self.model_orchestrator.query_final_state(context) + self.observed = NDConfigCollection.from_api_response( + response_data=response_data, model_class=self.model_class + ) + self.existing = self.confirmed + self.output.set_changed(self.journal.changed) + self.output.set_after_state( + self.observed, status="observed", verification_performed=True + ) + self._finalized = True + return + except Exception as e: + errors.append(str(e)) + if attempt < settings["retries"] and settings["delay"]: + time.sleep(settings["delay"]) + finally: + rest_send.timeout = original_timeout + + verification_error = f"Final-state verification failed after {settings['retries']} attempts: {'; '.join(errors)}" + self.output.set_changed(self.journal.changed) + self.output.mark_after_unknown( + affected_identifiers=list(self.journal.unknown_identifiers), + may_have_changed=self.journal.may_have_changed, + verification_performed=True, + verification_error=verification_error, + ) + self._finalized = True + if primary_error is None: + raise NDStateMachineError(verification_error) diff --git a/plugins/module_utils/nd_state_plan.py b/plugins/module_utils/nd_state_plan.py new file mode 100644 index 000000000..bc5e41470 --- /dev/null +++ b/plugins/module_utils/nd_state_plan.py @@ -0,0 +1,211 @@ +# Copyright: (c) 2026, Mike Wiebe (@mikewiebe) mwiebe@cisco.com + +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) + +"""Pure configuration planning shared by state machines and aggregate workflows.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Literal, Sequence + +from ansible_collections.cisco.nd.plugins.module_utils.models.base import NDBaseModel +from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import ( + NDConfigCollection, +) + +SupportedState = Literal["merged", "replaced", "overridden", "deleted"] +SUPPORTED_STATES: frozenset[str] = frozenset( + {"merged", "replaced", "overridden", "deleted"} +) + + +@dataclass(frozen=True, init=False) +class NDStatePlan: + """Immutable public snapshot of one resource family's planned operations. + + ``NDConfigCollection`` and ``NDBaseModel`` are mutable. The plan therefore + owns private deep copies and returns defensive copies from every public + collection or operation property. A caller can safely pass an operation + model to an orchestrator that enriches or otherwise mutates it without + changing the stored plan or its planned ``after`` state. + """ + + state: str + _before: NDConfigCollection = field(repr=False) + _proposed: NDConfigCollection = field(repr=False) + _after: NDConfigCollection = field(repr=False) + _creates: tuple[NDBaseModel, ...] = field(repr=False) + _updates: tuple[NDBaseModel, ...] = field(repr=False) + _deletes: tuple[NDBaseModel, ...] = field(repr=False) + _errors: tuple[str, ...] = field(repr=False) + + def __init__( + self, + *, + state: str, + before: NDConfigCollection, + proposed: NDConfigCollection, + after: NDConfigCollection, + creates: Sequence[NDBaseModel] = (), + updates: Sequence[NDBaseModel] = (), + deletes: Sequence[NDBaseModel] = (), + errors: Sequence[str] = (), + ) -> None: + object.__setattr__(self, "state", state) + object.__setattr__(self, "_before", before.copy()) + object.__setattr__(self, "_proposed", proposed.copy()) + object.__setattr__(self, "_after", after.copy()) + object.__setattr__(self, "_creates", tuple(deepcopy(tuple(creates)))) + object.__setattr__(self, "_updates", tuple(deepcopy(tuple(updates)))) + object.__setattr__(self, "_deletes", tuple(deepcopy(tuple(deletes)))) + object.__setattr__(self, "_errors", tuple(errors)) + + @property + def before(self) -> NDConfigCollection: + """Return an isolated copy of the initial-state snapshot.""" + return self._before.copy() + + @property + def proposed(self) -> NDConfigCollection: + """Return an isolated copy of the proposed-state snapshot.""" + return self._proposed.copy() + + @property + def after(self) -> NDConfigCollection: + """Return an isolated copy of the planned after-state snapshot.""" + return self._after.copy() + + @property + def creates(self) -> tuple[NDBaseModel, ...]: + """Return isolated models planned for creation.""" + return tuple(deepcopy(self._creates)) + + @property + def updates(self) -> tuple[NDBaseModel, ...]: + """Return isolated models planned for update.""" + return tuple(deepcopy(self._updates)) + + @property + def deletes(self) -> tuple[NDBaseModel, ...]: + """Return isolated models planned for deletion.""" + return tuple(deepcopy(self._deletes)) + + @property + def errors(self) -> tuple[str, ...]: + """Return planning errors suppressed by ``ignore_errors``.""" + return self._errors + + @property + def changed(self) -> bool: + """Return whether the plan contains any mutation.""" + return bool(self._creates or self._updates or self._deletes) + + @property + def mutation_count(self) -> int: + """Return the total number of planned create, update, and delete operations.""" + return len(self._creates) + len(self._updates) + len(self._deletes) + + +class NDStatePlanner: + """Calculate state operations without invoking an orchestrator mutation method.""" + + @staticmethod + def _error_message(identifier, exc: Exception) -> str: + """Return a consistent per-item planning error.""" + if identifier is None: + return f"Failed to process: {exc}" + return f"Failed to process {identifier}: {exc}" + + @classmethod + def plan( + cls, + *, + state: str, + before: NDConfigCollection, + proposed: NDConfigCollection, + ignore_errors: bool = False, + ) -> NDStatePlan: + """Return create/update/delete operations and their prospective state.""" + if state not in SUPPORTED_STATES: + raise ValueError(f"Invalid state: {state}") + + after = before.copy() + creates: list[NDBaseModel] = [] + updates: list[NDBaseModel] = [] + deletes: list[NDBaseModel] = [] + errors: list[str] = [] + + if state in {"merged", "replaced", "overridden"}: + for proposed_item in proposed: + identifier = None + try: + # Plan each item against a disposable collection. If an + # item-specific merge raises after partially mutating its + # receiver, ``ignore_errors`` cannot leak that partial state + # into the final plan. + candidate_after = after.copy() + working_item = deepcopy(proposed_item) + identifier = working_item.get_identifier_value() + diff_status = candidate_after.get_diff_config( + working_item, exclude_unset=state == "merged" + ) + if diff_status == "no_diff": + continue + + if state == "merged": + final_item = candidate_after.merge(working_item) + else: + if diff_status == "changed": + candidate_after.replace(working_item) + else: + candidate_after.add(working_item) + final_item = working_item + + after = candidate_after + if diff_status == "changed": + updates.append(final_item) + elif diff_status == "new": + creates.append(final_item) + except Exception as exc: + message = cls._error_message(identifier, exc) + if not ignore_errors: + raise ValueError(message) from exc + errors.append(message) + + if state == "overridden": + proposed_identifiers = set(proposed.keys()) + deletes = [ + item + for item in after + if item.get_identifier_value() not in proposed_identifiers + ] + after.delete_many([item.get_identifier_value() for item in deletes]) + + elif state == "deleted": + for proposed_item in proposed: + identifier = None + try: + identifier = proposed_item.get_identifier_value() + existing_item = after.get(identifier) + if existing_item is None: + continue + deletes.append(existing_item) + after.delete(identifier) + except Exception as exc: + message = cls._error_message(identifier, exc) + if not ignore_errors: + raise ValueError(message) from exc + errors.append(message) + + return NDStatePlan( + state=state, + before=before, + proposed=proposed, + after=after, + creates=creates, + updates=updates, + deletes=deletes, + errors=errors, + ) diff --git a/plugins/module_utils/nd_state_reconciliation.py b/plugins/module_utils/nd_state_reconciliation.py new file mode 100644 index 000000000..81edb11de --- /dev/null +++ b/plugins/module_utils/nd_state_reconciliation.py @@ -0,0 +1,145 @@ +# Copyright: (c) 2026, Cisco Systems, Inc. + +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) + +"""Mutation outcomes used to build evidence-backed ND state-machine output.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Iterable + +from ansible_collections.cisco.nd.plugins.module_utils.models.base import NDBaseModel + + +class MutationOperation(str, Enum): + """Controller configuration operations represented by a planned effect.""" + + CREATE = "create" + UPDATE = "update" + DELETE = "delete" + + +class MutationOutcome(str, Enum): + """Reconciliation outcome for one logical mutation checkpoint.""" + + NOT_ATTEMPTED = "not_attempted" + SUCCEEDED = "succeeded" + FAILED = "failed" + UNKNOWN = "unknown" + QUEUED = "queued" + SKIPPED = "skipped" + + +@dataclass(frozen=True) +class MutationEffect: + """One planned resource-state transition owned by a checkpoint.""" + + operation: MutationOperation + identifier: Any + before: NDBaseModel | None + after: NDBaseModel | None + + def __post_init__(self) -> None: + object.__setattr__(self, "identifier", deepcopy(self.identifier)) + object.__setattr__(self, "before", deepcopy(self.before)) + object.__setattr__(self, "after", deepcopy(self.after)) + + +@dataclass(frozen=True) +class DeferredMutation: + """Return receipt for a mutation queued for a later controller write.""" + + phase: str + + +@dataclass +class MutationCheckpoint: + """Execution and response evidence for one logical controller mutation.""" + + sequence_number: int + phase: str + effects: tuple[MutationEffect, ...] + outcome: MutationOutcome = MutationOutcome.NOT_ATTEMPTED + changed: bool = False + may_have_changed: bool = False + error: str | None = None + api_call_sequences: tuple[int, ...] = () + + @property + def affected_identifiers(self) -> tuple[Any, ...]: + """Return resource identifiers covered by this checkpoint.""" + return tuple(deepcopy(effect.identifier) for effect in self.effects) + + def resolve( + self, + outcome: MutationOutcome, + *, + changed: bool = False, + may_have_changed: bool = False, + error: str | None = None, + api_call_sequences: Iterable[int] = (), + ) -> None: + """Record the checkpoint's final outcome exactly once.""" + if self.outcome is not MutationOutcome.NOT_ATTEMPTED: + raise ValueError( + f"Checkpoint {self.sequence_number} is already resolved as {self.outcome.value}" + ) + self.outcome = outcome + self.changed = bool(changed) + self.may_have_changed = bool(may_have_changed) + self.error = error + self.api_call_sequences = tuple(api_call_sequences) + + +@dataclass +class MutationJournal: + """Ordered logical checkpoints for one state-machine execution.""" + + checkpoints: list[MutationCheckpoint] = field(default_factory=list) + + def open( + self, *, phase: str, effects: Iterable[MutationEffect] + ) -> MutationCheckpoint: + """Append and return a not-yet-attempted checkpoint.""" + checkpoint = MutationCheckpoint( + sequence_number=len(self.checkpoints) + 1, + phase=phase, + effects=tuple(effects), + ) + if not checkpoint.effects: + raise ValueError("A mutation checkpoint requires at least one effect") + self.checkpoints.append(checkpoint) + return checkpoint + + @property + def changed(self) -> bool: + """Return whether any checkpoint proves a controller change.""" + return any(checkpoint.changed for checkpoint in self.checkpoints) + + @property + def may_have_changed(self) -> bool: + """Return whether uncertain delivery may have changed controller state.""" + return any(checkpoint.may_have_changed for checkpoint in self.checkpoints) + + @property + def has_unknown(self) -> bool: + """Return whether any affected scope cannot be reconciled exactly.""" + return any( + checkpoint.outcome is MutationOutcome.UNKNOWN + for checkpoint in self.checkpoints + ) + + @property + def unknown_identifiers(self) -> tuple[Any, ...]: + """Return de-duplicated identifiers from unknown checkpoints in order.""" + identifiers: list[Any] = [] + for checkpoint in self.checkpoints: + if checkpoint.outcome is not MutationOutcome.UNKNOWN: + continue + for identifier in checkpoint.affected_identifiers: + if identifier not in identifiers: + identifiers.append(identifier) + return tuple(identifiers) diff --git a/plugins/module_utils/orchestrators/base.py b/plugins/module_utils/orchestrators/base.py index 3ed88ae69..05919ba72 100644 --- a/plugins/module_utils/orchestrators/base.py +++ b/plugins/module_utils/orchestrators/base.py @@ -12,7 +12,7 @@ from ansible_collections.cisco.nd.plugins.module_utils.endpoints.base import NDEndpointBaseModel from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum, OperationType from ansible_collections.cisco.nd.plugins.module_utils.models.base import NDBaseModel -from ansible_collections.cisco.nd.plugins.module_utils.orchestrators.types import ResponseType +from ansible_collections.cisco.nd.plugins.module_utils.orchestrators.types import FinalizationContext, ResponseType from ansible_collections.cisco.nd.plugins.module_utils.rest.rest_send import RestSend from ansible_collections.cisco.nd.plugins.module_utils.rest.results import Results @@ -107,11 +107,17 @@ def _request( self.rest_send.verb = verb if data is not None: self.rest_send.payload = data + + attempt_sequence = None + if self.results is not None: + attempt_sequence = self.results.begin_api_call(path, verb) self.rest_send.commit() # Register with Results before success/error checks so that # both successful and failed calls are captured for troubleshooting. self._register_api_call(path, verb, operation_type, self.rest_send.committed_payload) + if self.results is not None and attempt_sequence is not None: + self.results.complete_api_call(attempt_sequence) # Check not_found_ok before success because ResponseHandler treats # GET 404 as success=True (found=False). Without this early return, @@ -194,6 +200,15 @@ def query_all(self, model_instance: ModelType | None = None, **kwargs) -> Respon except Exception as e: raise Exception(f"Query all failed: {e}") from e + def invalidate_query_cache(self) -> None: + """Invalidate cached query data before a forced final-state readback.""" + return + + def query_final_state(self, context: FinalizationContext) -> ResponseType: + """Return a fresh collection-shaped response for final reconciliation.""" + self.invalidate_query_cache() + return self.query_all() + def prepare_config_data(self, raw_config): """Hook for subclasses to backfill or normalize raw user config before the proposed collection is built. Returns the list unchanged by default.""" return raw_config diff --git a/plugins/module_utils/orchestrators/base_interface.py b/plugins/module_utils/orchestrators/base_interface.py index 56254d2e2..a1a7ea338 100644 --- a/plugins/module_utils/orchestrators/base_interface.py +++ b/plugins/module_utils/orchestrators/base_interface.py @@ -147,6 +147,10 @@ def _switch_interfaces(self, switch_id: str) -> dict[str, dict]: self._switch_interfaces_cache[switch_id] = {iface["interfaceName"].lower(): iface for iface in interfaces if iface.get("interfaceName")} return self._switch_interfaces_cache[switch_id] + def invalidate_query_cache(self) -> None: + """Discard the initialization inventory before forced reconciliation.""" + self._switch_interfaces_cache.clear() + def _switches_to_query(self) -> dict[str, str]: """ # Summary diff --git a/plugins/module_utils/orchestrators/manage_policy_group.py b/plugins/module_utils/orchestrators/manage_policy_group.py index 1d101dd24..c51282b1e 100644 --- a/plugins/module_utils/orchestrators/manage_policy_group.py +++ b/plugins/module_utils/orchestrators/manage_policy_group.py @@ -258,6 +258,10 @@ def _invalidate_cache(self) -> None: self._raw_cache = None self._policy_summary_cache = None + def invalidate_query_cache(self) -> None: + """Drop policy-group caches before forced final-state reconciliation.""" + self._invalidate_cache() + @staticmethod def _is_policy_group_summary(row: dict) -> bool: """Return True when a policySummary row represents a policy group.""" diff --git a/plugins/module_utils/orchestrators/types.py b/plugins/module_utils/orchestrators/types.py index 415526c79..7cff025d5 100644 --- a/plugins/module_utils/orchestrators/types.py +++ b/plugins/module_utils/orchestrators/types.py @@ -2,8 +2,18 @@ # GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) -from __future__ import absolute_import, division, print_function +from __future__ import absolute_import, annotations, division, print_function +from dataclasses import dataclass from typing import Any, Union, List, Dict ResponseType = Union[List[Dict[str, Any]], Dict[str, Any], None] + + +@dataclass(frozen=True) +class FinalizationContext: + """Confirmed and unresolved scope available to a final-state query.""" + + state: str + affected_identifiers: tuple[Any, ...] + confirmed_identifiers: tuple[Any, ...] diff --git a/plugins/module_utils/rest/results.py b/plugins/module_utils/rest/results.py index 3007d2c8c..62695710b 100644 --- a/plugins/module_utils/rest/results.py +++ b/plugins/module_utils/rest/results.py @@ -17,6 +17,7 @@ import copy import logging +from dataclasses import dataclass, replace from typing import Any, Optional from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import ( @@ -29,6 +30,16 @@ from ansible_collections.cisco.nd.plugins.module_utils.enums import HttpVerbEnum, OperationType +@dataclass(frozen=True) +class ApiCallAttempt: + """Request attempt recorded before controller delivery is known.""" + + sequence_number: int + path: str + verb: str + completed: bool = False + + class ApiCallResult(BaseModel): """ # Summary @@ -386,6 +397,12 @@ def __init__(self) -> None: # Task sequence tracking self.task_sequence_number: int = 0 + # Request attempts are recorded before RestSend.commit(). A request can + # therefore be classified as delivery-unknown even when commit raises + # before a normal ApiCallResult can be registered. + self.api_attempt_sequence_number: int = 0 + self._api_attempts: list[ApiCallAttempt] = [] + # Registered tasks (immutable after registration) self._tasks: list[ApiCallResult] = [] @@ -412,6 +429,39 @@ def _increment_task_sequence_number(self) -> None: msg = f"self.task_sequence_number: {self.task_sequence_number}" self.log.debug(msg) + def begin_api_call(self, path: str, verb: HttpVerbEnum) -> int: + """Record an API attempt before the request is sent and return its id.""" + if not isinstance(path, str): + raise TypeError(f"path must be a string. Got {type(path).__name__}.") + if not isinstance(verb, HttpVerbEnum): + raise TypeError(f"verb must be an HttpVerbEnum. Got {type(verb).__name__}.") + + self.api_attempt_sequence_number += 1 + self._api_attempts.append( + ApiCallAttempt( + sequence_number=self.api_attempt_sequence_number, + path=path, + verb=verb.value, + ) + ) + return self.api_attempt_sequence_number + + def complete_api_call(self, sequence_number: int) -> None: + """Mark an attempt complete after its ApiCallResult is registered.""" + for index, attempt in enumerate(self._api_attempts): + if attempt.sequence_number == sequence_number: + self._api_attempts[index] = replace(attempt, completed=True) + return + raise ValueError(f"Unknown API attempt sequence number: {sequence_number}") + + def attempts_since(self, sequence_number: int) -> tuple[ApiCallAttempt, ...]: + """Return attempts recorded after ``sequence_number``.""" + return tuple(attempt for attempt in self._api_attempts if attempt.sequence_number > sequence_number) + + def calls_since(self, sequence_number: int) -> tuple[ApiCallResult, ...]: + """Return defensive copies of completed calls after ``sequence_number``.""" + return tuple(copy.deepcopy(task) for task in self._tasks if task.sequence_number > sequence_number) + def _determine_if_changed(self) -> bool: """ # Summary @@ -553,6 +603,9 @@ def register_api_call(self) -> None: # Register the task self._tasks.append(task_data) + # A later API call must invalidate any previously-built aggregation. + self._final_result = None + # Reset current task for next task self._current = PendingApiCall() diff --git a/tests/unit/module_utils/test_nd_state_machine_reconciliation.py b/tests/unit/module_utils/test_nd_state_machine_reconciliation.py new file mode 100644 index 000000000..64ee5195a --- /dev/null +++ b/tests/unit/module_utils/test_nd_state_machine_reconciliation.py @@ -0,0 +1,348 @@ +# -*- coding: utf-8 -*- + +# Copyright: (c) 2026, Cisco Systems, Inc. + +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) + +"""Failure-path tests for evidence-backed NDStateMachine after-state.""" + +from __future__ import annotations + +from typing import Any, ClassVar, Literal + +import pytest +from ansible_collections.cisco.nd.plugins.module_utils.common.exceptions import ( + NDStateMachineError, +) +from ansible_collections.cisco.nd.plugins.module_utils.enums import ( + HttpVerbEnum, + OperationType, +) +from ansible_collections.cisco.nd.plugins.module_utils.models.base import NDBaseModel +from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import ( + NDConfigCollection, +) +from ansible_collections.cisco.nd.plugins.module_utils.nd_output import NDOutput +from ansible_collections.cisco.nd.plugins.module_utils.nd_state_machine import ( + NDStateMachine, +) +from ansible_collections.cisco.nd.plugins.module_utils.nd_state_reconciliation import ( + DeferredMutation, + MutationJournal, + MutationOutcome, +) +from ansible_collections.cisco.nd.plugins.module_utils.rest.results import Results + + +class _Model(NDBaseModel): + identifiers: ClassVar[list[str] | None] = ["name"] + identifier_strategy: ClassVar[ + Literal["single", "composite", "hierarchical", "singleton"] | None + ] = "single" + + name: str + value: str | None = None + + +class _Module: + def __init__( + self, *, state: str, check_mode: bool, ignore_errors: bool, verify: Any = None + ) -> None: + self.check_mode = check_mode + self.params = { + "state": state, + "config": [], + "output_level": "normal", + "ignore_errors": ignore_errors, + "verify": verify, + } + + +class _RestSend: + timeout = 30 + + +class _Orchestrator: + model_class = _Model + supports_bulk_create = False + supports_bulk_delete = False + + def __init__( + self, results: Results, behaviors: dict[str, str] | None = None + ) -> None: + self.results = results + self.behaviors = behaviors or {} + self.calls: list[str] = [] + self.final_responses: list[Any] = [] + self.rest_send = _RestSend() + + def preflight_create(self, _items) -> None: + return + + def preflight(self, _items) -> None: + return + + def _register( + self, *, success: bool, changed: bool, certainty: str | None = None + ) -> None: + attempt = self.results.begin_api_call("/api/v1/items", HttpVerbEnum.PUT) + self.results.action = OperationType.UPDATE.value + self.results.operation_type = OperationType.UPDATE + self.results.path_current = "/api/v1/items" + self.results.verb_current = HttpVerbEnum.PUT + self.results.response_current = {"RETURN_CODE": 200 if success else 207} + result = {"success": success, "changed": changed} + if certainty is not None: + result["outcome_certainty"] = certainty + self.results.result_current = result + self.results.diff_current = {} + self.results.register_api_call() + self.results.complete_api_call(attempt) + + def _mutate(self, item: _Model): + self.calls.append(item.name) + behavior = self.behaviors.get(item.name, "success") + if behavior == "success": + self._register(success=True, changed=True) + return {} + if behavior == "local_failure": + raise RuntimeError("local validation failed") + if behavior == "no_change": + self._register(success=False, changed=False, certainty="no_change") + raise RuntimeError("controller rejected request") + if behavior == "mixed": + self._register(success=False, changed=True) + raise RuntimeError("mixed multi-status failure") + if behavior == "timeout": + self.results.begin_api_call("/api/v1/items", HttpVerbEnum.PUT) + raise RuntimeError("response lost") + if behavior == "partial_multi_call": + self._register(success=True, changed=True) + self._register(success=False, changed=False) + raise RuntimeError("second request failed") + if behavior == "deferred": + return DeferredMutation(phase="remove") + raise AssertionError(f"Unknown behavior: {behavior}") + + create = _mutate + update = _mutate + delete = _mutate + + def create_bulk(self, items: list[_Model]): + self.calls.extend(item.name for item in items) + behavior = self.behaviors.get("bulk", "success") + if behavior == "mixed": + self._register(success=False, changed=True) + raise RuntimeError("mixed multi-status failure") + self._register(success=True, changed=True) + return {} + + def delete_bulk(self, items: list[_Model]): + return self.create_bulk(items) + + def query_final_state(self, _context): + response = self.final_responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + +def _collection(items: list[_Model]) -> NDConfigCollection: + return NDConfigCollection(model_class=_Model, items=items) + + +def _state_machine( + *, + state: str = "replaced", + before: list[_Model] | None = None, + proposed: list[_Model] | None = None, + behaviors: dict[str, str] | None = None, + ignore_errors: bool = False, + check_mode: bool = False, + verify: Any = None, + bulk_create: bool = False, +) -> NDStateMachine: + sm = object.__new__(NDStateMachine) + sm.module = _Module( + state=state, check_mode=check_mode, ignore_errors=ignore_errors, verify=verify + ) + sm.state = state + sm.check_mode = check_mode + sm.ignore_errors = ignore_errors + sm.model_class = _Model + sm.results = Results() + sm.results.state = state + sm.results.check_mode = check_mode + sm.model_orchestrator = _Orchestrator(sm.results, behaviors) + sm.model_orchestrator.supports_bulk_create = bulk_create + sm.supports_bulk_create = bulk_create + sm.supports_bulk_delete = False + sm.before = _collection(before or []) + sm.planned = sm.before.copy() + sm.confirmed = sm.before.copy() + sm.existing = sm.confirmed + sm.proposed = _collection(proposed or []) + sm.sent = _collection([]) + sm.output = NDOutput("normal") + sm.output.assign(before=sm.before, proposed=sm.proposed) + sm.output.set_after_state(sm.confirmed, status="confirmed") + sm.journal = MutationJournal() + sm.plan = None + sm.observed = None + sm._finalized = False + sm._verify_settings = sm._verification_settings() + return sm + + +def _values(output: dict) -> dict[str, str | None]: + return {item["name"]: item.get("value") for item in output["after"]} + + +def test_earlier_success_is_confirmed_when_later_update_fails() -> None: + sm = _state_machine( + before=[ + _Model(name="a", value="old"), + _Model(name="b", value="old"), + _Model(name="c", value="old"), + ], + proposed=[ + _Model(name="a", value="new"), + _Model(name="b", value="new"), + _Model(name="c", value="new"), + ], + behaviors={"b": "local_failure"}, + ) + + with pytest.raises(NDStateMachineError, match="Failed to update b"): + sm.manage_state() + + output = sm.output.format() + assert sm.model_orchestrator.calls == ["a", "b"] + assert _values(output) == {"a": "new", "b": "old", "c": "old"} + assert output["after_status"] == "confirmed" + assert output["changed"] is True + assert [item.name for item in sm.sent] == ["a"] + assert [checkpoint.outcome for checkpoint in sm.journal.checkpoints] == [ + MutationOutcome.SUCCEEDED, + MutationOutcome.FAILED, + MutationOutcome.NOT_ATTEMPTED, + ] + + +def test_unknown_with_internal_continuation_keeps_later_evidence_but_omits_after() -> ( + None +): + sm = _state_machine( + before=[_Model(name="a", value="old"), _Model(name="b", value="old")], + proposed=[_Model(name="a", value="new"), _Model(name="b", value="new")], + behaviors={"a": "timeout"}, + ignore_errors=True, + ) + + sm.manage_state() + + output = sm.output.format() + assert sm.model_orchestrator.calls == ["a", "b"] + assert "after" not in output + assert "diff" not in output + assert output["after_status"] == "unknown" + assert output["may_have_changed"] is True + assert output["changed"] is True # later b success is still proven + assert [item.name for item in sm.sent] == ["b"] + + +def test_mixed_unkeyed_bulk_response_is_unknown() -> None: + sm = _state_machine( + state="merged", + proposed=[_Model(name="a", value="new"), _Model(name="b", value="new")], + behaviors={"bulk": "mixed"}, + bulk_create=True, + ) + + with pytest.raises(NDStateMachineError, match="mixed multi-status failure"): + sm.manage_state() + + output = sm.output.format() + assert output["changed"] is True + assert output["after_status"] == "unknown" + assert set(output["affected_identifiers"]) == {"a", "b"} + assert "after" not in output + + +def test_opt_in_finalization_replaces_unknown_with_observed_state() -> None: + sm = _state_machine( + before=[_Model(name="a", value="old")], + proposed=[_Model(name="a", value="new")], + behaviors={"a": "timeout"}, + verify={"enabled": True, "retries": 1, "timeout": 5, "delay": 0}, + ) + sm.model_orchestrator.final_responses = [[{"name": "a", "value": "controller"}]] + + with pytest.raises(NDStateMachineError) as exc_info: + sm.manage_state() + sm.finalize(primary_error=exc_info.value) + + output = sm.output.format() + assert _values(output) == {"a": "controller"} + assert output["after_status"] == "observed" + assert output["verification_performed"] is True + + +def test_known_success_does_not_requery_even_when_verify_is_enabled() -> None: + sm = _state_machine( + before=[_Model(name="a", value="old")], + proposed=[_Model(name="a", value="new")], + verify={"enabled": True, "retries": 1, "timeout": 5, "delay": 0}, + ) + + sm.manage_state() + sm.finalize() + + assert sm.model_orchestrator.final_responses == [] + assert sm.output.format()["after_status"] == "confirmed" + + +def test_multi_request_partial_failure_is_unknown_without_semantic_subcheckpoints() -> ( + None +): + sm = _state_machine( + before=[_Model(name="a", value="old")], + proposed=[_Model(name="a", value="new")], + behaviors={"a": "partial_multi_call"}, + ) + + with pytest.raises(NDStateMachineError): + sm.manage_state() + + assert sm.journal.checkpoints[0].outcome is MutationOutcome.UNKNOWN + assert sm.output.format()["after_status"] == "unknown" + + +def test_deferred_receipt_is_not_promoted_to_confirmed_or_sent() -> None: + sm = _state_machine( + before=[_Model(name="a", value="old")], + proposed=[_Model(name="a", value="new")], + behaviors={"a": "deferred"}, + ) + + sm.manage_state() + + assert sm.journal.checkpoints[0].outcome is MutationOutcome.QUEUED + assert _values(sm.output.format()) == {"a": "old"} + assert len(sm.sent) == 0 + + +def test_check_mode_returns_planned_without_controller_calls() -> None: + sm = _state_machine( + before=[_Model(name="a", value="old")], + proposed=[_Model(name="a", value="new")], + check_mode=True, + ) + + sm.manage_state() + + output = sm.output.format() + assert sm.model_orchestrator.calls == [] + assert _values(output) == {"a": "new"} + assert output["after_status"] == "planned" + assert output["changed"] is True diff --git a/tests/unit/module_utils/test_nd_state_plan.py b/tests/unit/module_utils/test_nd_state_plan.py new file mode 100644 index 000000000..4858ea79a --- /dev/null +++ b/tests/unit/module_utils/test_nd_state_plan.py @@ -0,0 +1,203 @@ +# -*- coding: utf-8 -*- + +# Copyright: (c) 2026, Mike Wiebe (@mikewiebe) mwiebe@cisco.com + +# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) + +"""Unit tests for the mutation-free ND state planning boundary.""" + +from __future__ import annotations + +from dataclasses import FrozenInstanceError +from typing import ClassVar, Literal + +import pytest +from ansible_collections.cisco.nd.plugins.module_utils.models.base import NDBaseModel +from ansible_collections.cisco.nd.plugins.module_utils.models.interfaces.loopback_interface import ( + LoopbackInterfaceModel, +) +from ansible_collections.cisco.nd.plugins.module_utils.nd_config_collection import ( + NDConfigCollection, +) +from ansible_collections.cisco.nd.plugins.module_utils.nd_state_plan import ( + NDStatePlanner, +) + + +def _loopback( + name: str, *, ip: str | None = None, description: str | None = None +) -> dict: + policy = { + key: value + for key, value in {"ip": ip, "description": description}.items() + if value is not None + } + return { + "switch_ip": "192.0.2.1", + "interface_name": name, + "config_data": {"network_os": {"policy": policy}}, + } + + +def _collection(config: list[dict], state: str = "merged") -> NDConfigCollection: + return NDConfigCollection.from_ansible_config( + data=config, model_class=LoopbackInterfaceModel, context={"state": state} + ) + + +def _description(collection: NDConfigCollection) -> str | None: + return collection.to_ansible_config()[0]["config_data"]["network_os"]["policy"].get( + "description" + ) + + +def _set_description(model: LoopbackInterfaceModel, value: str) -> None: + model.config_data.network_os.policy.description = value + + +def test_merged_plan_preserves_fields_without_mutating_inputs() -> None: + """Merged planning emits the final merged item and leaves both inputs unchanged.""" + before = _collection( + [_loopback("loopback10", ip="192.0.2.10/32", description="old")] + ) + proposed = _collection([_loopback("LOOPBACK10", description="new")]) + original_before = before.to_ansible_config() + original_proposed = proposed.to_ansible_config() + + plan = NDStatePlanner.plan(state="merged", before=before, proposed=proposed) + + assert len(plan.updates) == 1 + assert not plan.creates + assert not plan.deletes + policy = plan.after.to_ansible_config()[0]["config_data"]["network_os"]["policy"] + assert policy["ip"] == "192.0.2.10" + assert policy["description"] == "new" + assert before.to_ansible_config() == original_before + assert proposed.to_ansible_config() == original_proposed + + +def test_overridden_plan_calculates_update_create_delete_and_after_state() -> None: + """Override planning is complete, deterministic, and mutation-free.""" + before = _collection( + [ + _loopback("loopback10", ip="192.0.2.10/32"), + _loopback("loopback20", ip="192.0.2.20/32"), + ], + state="overridden", + ) + proposed = _collection( + [ + _loopback("loopback10", ip="192.0.2.110/32"), + _loopback("loopback30", ip="192.0.2.30/32"), + ], + state="overridden", + ) + + plan = NDStatePlanner.plan(state="overridden", before=before, proposed=proposed) + + assert [item.interface_name for item in plan.updates] == ["loopback10"] + assert [item.interface_name for item in plan.creates] == ["loopback30"] + assert [item.interface_name for item in plan.deletes] == ["loopback20"] + assert set(plan.after.keys()) == { + ("192.0.2.1", "loopback10"), + ("192.0.2.1", "loopback30"), + } + assert plan.changed is True + assert plan.mutation_count == 3 + assert list(before.keys()) == [ + ("192.0.2.1", "loopback10"), + ("192.0.2.1", "loopback20"), + ] + + +def test_deleted_plan_only_selects_identifiers_that_exist() -> None: + """Delete planning ignores absent identifiers and preserves the input collection.""" + before = _collection([_loopback("loopback10", ip="192.0.2.10/32")]) + proposed = _collection( + [ + {"switch_ip": "192.0.2.1", "interface_name": "loopback10"}, + {"switch_ip": "192.0.2.1", "interface_name": "loopback99"}, + ], + state="deleted", + ) + + plan = NDStatePlanner.plan(state="deleted", before=before, proposed=proposed) + + assert [item.interface_name for item in plan.deletes] == ["loopback10"] + assert len(plan.after) == 0 + assert len(before) == 1 + + +def test_plan_public_values_are_defensive_snapshots() -> None: + """Mutating inputs or returned plan values cannot change the stored plan.""" + key = ("192.0.2.1", "loopback10") + before = _collection([_loopback("loopback10", description="old")]) + proposed = _collection([_loopback("loopback10", description="new")]) + plan = NDStatePlanner.plan(state="merged", before=before, proposed=proposed) + + _set_description(before.get(key), "mutated-before") + _set_description(proposed.get(key), "mutated-proposed") + assert _description(plan.before) == "old" + assert _description(plan.proposed) == "new" + assert _description(plan.after) == "new" + + returned_after = plan.after + returned_after.delete(key) + assert len(plan.after) == 1 + + returned_update = plan.updates[0] + _set_description(returned_update, "mutated-operation") + assert _description(plan.after) == "new" + assert plan.updates[0].config_data.network_os.policy.description == "new" + + with pytest.raises(FrozenInstanceError): + plan.state = "deleted" + + +class _ExplodingMergeModel(NDBaseModel): + """Model whose merge mutates its receiver before raising.""" + + identifiers: ClassVar[list[str] | None] = ["name"] + identifier_strategy: ClassVar[ + Literal["single", "composite", "hierarchical", "singleton"] | None + ] = "single" + + name: str + value: str + + def merge(self, _other: "NDBaseModel") -> "NDBaseModel": + self.value = "partial" + raise RuntimeError("merge failed") + + +def test_ignored_planning_error_does_not_leak_partial_item_mutation() -> None: + """A suppressed item error leaves the last valid planned state untouched.""" + before = NDConfigCollection( + model_class=_ExplodingMergeModel, + items=[_ExplodingMergeModel(name="one", value="before")], + ) + proposed = NDConfigCollection( + model_class=_ExplodingMergeModel, + items=[_ExplodingMergeModel(name="one", value="after")], + ) + + plan = NDStatePlanner.plan( + state="merged", before=before, proposed=proposed, ignore_errors=True + ) + + assert plan.after.get("one").value == "before" + assert plan.changed is False + assert plan.mutation_count == 0 + assert plan.errors == ("Failed to process one: merge failed",) + + +def test_invalid_state_fails_without_changing_inputs() -> None: + """Unsupported state values cannot reach execution.""" + before = _collection([]) + proposed = _collection([]) + + with pytest.raises(ValueError, match="Invalid state"): + NDStatePlanner.plan(state="gathered", before=before, proposed=proposed) + + assert len(before) == 0 + assert len(proposed) == 0 diff --git a/tests/unit/module_utils/test_results.py b/tests/unit/module_utils/test_results.py index a7d3609f1..822cf04a6 100644 --- a/tests/unit/module_utils/test_results.py +++ b/tests/unit/module_utils/test_results.py @@ -300,6 +300,54 @@ def test_defaults_when_not_set(self): assert task.verbosity_level == 3 +class TestApiCallAttemptTracking: + """Tests the pre-delivery request-attempt evidence used by reconciliation.""" + + def test_attempt_is_visible_before_completion(self): + results = Results() + + sequence = results.begin_api_call("/api/v1/test", HttpVerbEnum.PUT) + + assert sequence == 1 + assert results.api_attempt_sequence_number == 1 + assert results.attempts_since(0)[0].completed is False + assert results.attempts_since(0)[0].verb == "PUT" + + def test_attempt_is_completed_after_registration(self): + results = Results() + sequence = results.begin_api_call("/api/v1/test", HttpVerbEnum.DELETE) + + _register_task(results, path="/api/v1/test", verb=HttpVerbEnum.DELETE) + results.complete_api_call(sequence) + + assert results.attempts_since(0)[0].completed is True + assert results.calls_since(0)[0].sequence_number == 1 + + def test_calls_since_returns_defensive_copies(self): + results = Results() + _register_task(results, path="/api/v1/one") + _register_task(results, path="/api/v1/two") + + returned = results.calls_since(1) + returned[0].result["changed"] = False + + assert len(returned) == 1 + assert results.results[1]["changed"] is True + + def test_new_registration_invalidates_cached_final_result(self): + results = Results() + _register_task(results, path="/api/v1/one") + results.build_final_result() + assert len(results.final_result["path"]) == 1 + + _register_task(results, path="/api/v1/two") + + with pytest.raises(ValueError, match="build_final_result"): + _ = results.final_result + results.build_final_result() + assert results.final_result["path"] == ["/api/v1/one", "/api/v1/two"] + + # ============================================================================= # Test: aggregate properties (path, verb, payload, verbosity_level) # =============================================================================