From 7ab68c84ecdd7cfda105e8ee5a0f8bc0d7b0ab99 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 30 Jul 2026 10:35:30 +0100 Subject: [PATCH] feat(lattice): add provider fallback foundation --- apps/predbat/lattice_provider_fallback.py | 728 ++++++++++ .../tests/test_lattice_provider_fallback.py | 1175 +++++++++++++++++ 2 files changed, 1903 insertions(+) create mode 100644 apps/predbat/lattice_provider_fallback.py create mode 100644 apps/predbat/tests/test_lattice_provider_fallback.py diff --git a/apps/predbat/lattice_provider_fallback.py b/apps/predbat/lattice_provider_fallback.py new file mode 100644 index 000000000..d7898a8d4 --- /dev/null +++ b/apps/predbat/lattice_provider_fallback.py @@ -0,0 +1,728 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Lattice provider-path fallback boundary +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure, default-off provider-path selection for future Lattice control. + +The current production control path does not import this module. It defines a +small fail-closed boundary for a later integration tranche: + +* targets are resolved only through provider-qualified aliases; +* gateway-local paths always rank before vendor-cloud paths; +* per-target path health has an explicit freshness TTL and retry cooldown; +* executors return APPLIED, NOT_APPLIED, or UNKNOWN without truthy coercion; +* fallback is possible only after a definite, fallback-safe NOT_APPLIED. + +There are no network, filesystem, MQTT, Home Assistant, inverter, or component +registration dependencies here. Disabled dispatch returns before consulting +the clock, aliases, health ledger, path declarations, or executors. +""" + +# cspell:ignore codepoint cooldown noncharacter + +import asyncio +import inspect +import math +import re +import time +from dataclasses import dataclass, field +from enum import Enum +from typing import Callable, Optional + + +MAX_IDENTITY_BYTES = 128 +MAX_DETAIL_BYTES = 256 +_IDENTIFIER = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9._:/-]{0,126}[A-Za-z0-9])?$") +_OBSERVED_AT_UNSET = object() + + +class ProviderFallbackValidationError(ValueError): + """Raised when declarations cannot produce an unambiguous safe router.""" + + +class ProviderExecutionState(Enum): + """What an executor can honestly prove about one attempted write.""" + + APPLIED = "APPLIED" + NOT_APPLIED = "NOT_APPLIED" + UNKNOWN = "UNKNOWN" + + +class ProviderPathKind(Enum): + """The only access-path classes supported by the first selector.""" + + GATEWAY_LOCAL = "gw-local" + VENDOR_CLOUD = "vendor-cloud" + + +class ProviderPathHealthPhase(Enum): + """Selection-relevant projection of one path's last observation.""" + + UNOBSERVED = "unobserved" + HEALTHY = "healthy" + PROBE = "probe" + COOLING_FALLBACK_SAFE = "cooling_fallback_safe" + COOLING_BLOCKED = "cooling_blocked" + + +@dataclass(frozen=True, order=True) +class QualifiedTargetAlias: + """One provider-scoped alias; unqualified lookup is intentionally absent.""" + + provider_id: str + alias: str + + def __post_init__(self): + """Reject ambiguous, unbounded, or non-canonical identity text.""" + _require_identifier(self.provider_id, "provider_id", lowercase=True) + _require_identifier(self.alias, "target alias") + + +@dataclass(frozen=True, order=True) +class ProviderPathIdentity: + """Stable identity for health and execution observations.""" + + target_id: str + provider_id: str + path_id: str + + def __post_init__(self): + """Validate every component before it can become a ledger key.""" + _require_identifier(self.target_id, "target_id") + _require_identifier(self.provider_id, "provider_id", lowercase=True) + _require_identifier(self.path_id, "path_id") + + +@dataclass(frozen=True) +class ProviderPathHealthPolicy: + """Freshness and retry bounds for one target-specific path.""" + + ttl_seconds: float = 30.0 + cooldown_seconds: float = 60.0 + + def __post_init__(self): + """Require finite positive durations and reject bool-as-number.""" + _require_duration(self.ttl_seconds, "health ttl_seconds") + _require_duration(self.cooldown_seconds, "health cooldown_seconds") + + +@dataclass(frozen=True) +class ProviderExecutionResult: + """Strict executor result whose fallback claim is independently explicit.""" + + state: ProviderExecutionState + fallback_safe: bool = False + verified: bool = False + applied_value: object = None + detail: str = "" + + def __post_init__(self): + """Forbid combinations that overstate what the executor knows.""" + if not isinstance(self.state, ProviderExecutionState): + raise ProviderFallbackValidationError( + "execution state must be ProviderExecutionState", + ) + if type(self.fallback_safe) is not bool: + raise ProviderFallbackValidationError( + "fallback_safe must be a boolean", + ) + if type(self.verified) is not bool: + raise ProviderFallbackValidationError("verified must be a boolean") + _require_detail(self.detail) + + if self.state is ProviderExecutionState.APPLIED: + if not self.verified: + raise ProviderFallbackValidationError( + "APPLIED must be verified", + ) + if self.fallback_safe: + raise ProviderFallbackValidationError( + "APPLIED cannot be fallback-safe", + ) + return + + if self.applied_value is not None: + raise ProviderFallbackValidationError( + "{} cannot carry applied_value".format(self.state.value), + ) + if self.verified: + raise ProviderFallbackValidationError( + "{} cannot be verified".format(self.state.value), + ) + if self.state is ProviderExecutionState.UNKNOWN and self.fallback_safe: + raise ProviderFallbackValidationError( + "UNKNOWN can never be fallback-safe", + ) + + @classmethod + def applied(cls, applied_value=None, detail=""): + """Build a verified APPLIED result.""" + return cls( + ProviderExecutionState.APPLIED, + verified=True, + applied_value=applied_value, + detail=detail, + ) + + @classmethod + def not_applied(cls, fallback_safe=False, detail=""): + """Build a definite NOT_APPLIED result with an explicit safety claim.""" + return cls( + ProviderExecutionState.NOT_APPLIED, + fallback_safe=fallback_safe, + detail=detail, + ) + + @classmethod + def unknown(cls, detail=""): + """Build an UNKNOWN result that can never authorize fallback.""" + return cls(ProviderExecutionState.UNKNOWN, detail=detail) + + +@dataclass(frozen=True) +class ProviderPath: + """One pure declaration joining a logical target to an injected executor.""" + + target_id: str + provider_id: str + path_id: str + kind: ProviderPathKind + executor: Callable + aliases: tuple = () + health_policy: ProviderPathHealthPolicy = field( + default_factory=ProviderPathHealthPolicy, + ) + + def __post_init__(self): + """Validate path identity, callable seam, aliases, and health policy.""" + ProviderPathIdentity( + self.target_id, + self.provider_id, + self.path_id, + ) + if not isinstance(self.kind, ProviderPathKind): + raise ProviderFallbackValidationError( + "path kind must be ProviderPathKind", + ) + if not callable(self.executor): + raise ProviderFallbackValidationError( + "path executor must be callable", + ) + if type(self.aliases) is not tuple: + raise ProviderFallbackValidationError("path aliases must be a tuple") + seen = set() + for alias in self.aliases: + qualified = QualifiedTargetAlias(self.provider_id, alias) + if qualified in seen: + raise ProviderFallbackValidationError( + "duplicate alias {} for provider {}".format( + alias, + self.provider_id, + ), + ) + seen.add(qualified) + if not isinstance(self.health_policy, ProviderPathHealthPolicy): + raise ProviderFallbackValidationError( + "health_policy must be ProviderPathHealthPolicy", + ) + + @property + def identity(self): + """Return the stable per-target path identity.""" + return ProviderPathIdentity( + self.target_id, + self.provider_id, + self.path_id, + ) + + @property + def qualified_aliases(self): + """Return explicit aliases plus the provider-qualified target id.""" + aliases = [QualifiedTargetAlias(self.provider_id, self.target_id)] + aliases.extend(QualifiedTargetAlias(self.provider_id, alias) for alias in self.aliases) + return tuple(aliases) + + +@dataclass(frozen=True) +class ProviderPathHealthRecord: + """Last accepted result for exactly one target/provider/path identity.""" + + identity: ProviderPathIdentity + result: ProviderExecutionResult + observed_at: float + + def __post_init__(self): + """Validate a self-contained immutable health observation.""" + if not isinstance(self.identity, ProviderPathIdentity): + raise ProviderFallbackValidationError( + "health identity must be ProviderPathIdentity", + ) + if not isinstance(self.result, ProviderExecutionResult): + raise ProviderFallbackValidationError( + "health result must be ProviderExecutionResult", + ) + _require_timestamp(self.observed_at, "health observed_at") + + +@dataclass(frozen=True) +class ProviderPathAttempt: + """One bounded, provider-qualified attempt in dispatch order.""" + + identity: ProviderPathIdentity + kind: ProviderPathKind + result: ProviderExecutionResult + + def __post_init__(self): + """Validate retained attempt evidence.""" + if not isinstance(self.identity, ProviderPathIdentity): + raise ProviderFallbackValidationError( + "attempt identity must be ProviderPathIdentity", + ) + if not isinstance(self.kind, ProviderPathKind): + raise ProviderFallbackValidationError( + "attempt kind must be ProviderPathKind", + ) + if not isinstance(self.result, ProviderExecutionResult): + raise ProviderFallbackValidationError( + "attempt result must be ProviderExecutionResult", + ) + + +@dataclass(frozen=True) +class ProviderFallbackResult: + """Final router decision with the complete ordered attempt evidence.""" + + state: ProviderExecutionState + fallback_safe: bool + attempts: tuple = () + selected_path: Optional[ProviderPathIdentity] = None + disabled: bool = False + detail: str = "" + + def __post_init__(self): + """Keep aggregate truth no stronger than its terminal evidence.""" + if not isinstance(self.state, ProviderExecutionState): + raise ProviderFallbackValidationError( + "fallback result state must be ProviderExecutionState", + ) + if type(self.fallback_safe) is not bool: + raise ProviderFallbackValidationError( + "fallback result fallback_safe must be a boolean", + ) + if type(self.attempts) is not tuple: + raise ProviderFallbackValidationError( + "fallback result attempts must be a tuple", + ) + if any(not isinstance(attempt, ProviderPathAttempt) for attempt in self.attempts): + raise ProviderFallbackValidationError( + "fallback result attempts are invalid", + ) + if self.selected_path is not None and not isinstance(self.selected_path, ProviderPathIdentity): + raise ProviderFallbackValidationError( + "selected_path must be ProviderPathIdentity", + ) + if type(self.disabled) is not bool: + raise ProviderFallbackValidationError( + "disabled must be a boolean", + ) + _require_detail(self.detail) + if self.state is not ProviderExecutionState.NOT_APPLIED and self.fallback_safe: + raise ProviderFallbackValidationError( + "only NOT_APPLIED can be fallback-safe", + ) + if self.disabled and (self.state is not ProviderExecutionState.UNKNOWN or self.fallback_safe or self.attempts or self.selected_path is not None): + raise ProviderFallbackValidationError( + "disabled result must fail closed without attempts", + ) + + +class ProviderPathHealthLedger: + """Process-local health ledger with monotonic observation protection.""" + + def __init__(self, clock=None): + """Create an empty ledger around an injectable monotonic clock.""" + self._clock = clock or time.monotonic + if not callable(self._clock): + raise ProviderFallbackValidationError("clock must be callable") + self._records = {} + + def now(self): + """Read and validate the injected clock.""" + value = self._clock() + _require_timestamp(value, "clock value") + return float(value) + + def observe(self, identity, result, observed_at=_OBSERVED_AT_UNSET): + """Accept a new or idempotent observation; reject time regressions.""" + if not isinstance(identity, ProviderPathIdentity): + raise ProviderFallbackValidationError( + "health identity must be ProviderPathIdentity", + ) + if not isinstance(result, ProviderExecutionResult): + raise ProviderFallbackValidationError( + "health result must be ProviderExecutionResult", + ) + timestamp = self.now() if observed_at is _OBSERVED_AT_UNSET else observed_at + _require_timestamp(timestamp, "health observed_at") + timestamp = float(timestamp) + previous = self._records.get(identity) + if previous is not None: + if timestamp < previous.observed_at: + raise ProviderFallbackValidationError( + "health observation time regressed", + ) + if timestamp == previous.observed_at and result != previous.result: + raise ProviderFallbackValidationError( + "health observation reused timestamp with different result", + ) + record = ProviderPathHealthRecord(identity, result, timestamp) + self._records[identity] = record + return record + + def record(self, identity): + """Return the immutable current observation for one exact identity.""" + if not isinstance(identity, ProviderPathIdentity): + raise ProviderFallbackValidationError( + "health identity must be ProviderPathIdentity", + ) + return self._records.get(identity) + + def phase(self, path, now=None): + """Project health into a deterministic selection decision.""" + if not isinstance(path, ProviderPath): + raise ProviderFallbackValidationError( + "health path must be ProviderPath", + ) + timestamp = self.now() if now is None else now + _require_timestamp(timestamp, "health decision time") + timestamp = float(timestamp) + record = self._records.get(path.identity) + if record is None: + return ProviderPathHealthPhase.UNOBSERVED + if timestamp < record.observed_at: + raise ProviderFallbackValidationError( + "health decision time precedes observation", + ) + + age = timestamp - record.observed_at + if record.result.state is ProviderExecutionState.APPLIED: + if age <= path.health_policy.ttl_seconds: + return ProviderPathHealthPhase.HEALTHY + return ProviderPathHealthPhase.PROBE + if age >= path.health_policy.cooldown_seconds: + return ProviderPathHealthPhase.PROBE + if record.result.state is ProviderExecutionState.NOT_APPLIED and record.result.fallback_safe: + return ProviderPathHealthPhase.COOLING_FALLBACK_SAFE + return ProviderPathHealthPhase.COOLING_BLOCKED + + +class LatticeProviderFallbackRouter: + """Default-off serial selector that never falls through UNKNOWN.""" + + def __init__(self, paths, enabled=False, health_ledger=None): + """Validate an immutable routing table without registering it anywhere.""" + if type(enabled) is not bool: + raise ProviderFallbackValidationError( + "router enabled gate must be a boolean", + ) + if type(paths) is not tuple: + raise ProviderFallbackValidationError("paths must be a tuple") + if any(not isinstance(path, ProviderPath) for path in paths): + raise ProviderFallbackValidationError( + "every path must be ProviderPath", + ) + if health_ledger is None: + health_ledger = ProviderPathHealthLedger() + if not isinstance(health_ledger, ProviderPathHealthLedger): + raise ProviderFallbackValidationError( + "health_ledger must be ProviderPathHealthLedger", + ) + + identities = set() + aliases = {} + by_target = {} + for path in paths: + identity = path.identity + if identity in identities: + raise ProviderFallbackValidationError( + "provider path identity collision: {}".format(identity), + ) + identities.add(identity) + for alias in path.qualified_aliases: + previous_target = aliases.get(alias) + if previous_target is not None and previous_target != path.target_id: + raise ProviderFallbackValidationError( + "provider-qualified alias collision: {}".format(alias), + ) + aliases[alias] = path.target_id + by_target.setdefault(path.target_id, []).append(path) + + self._enabled = enabled + self._health = health_ledger + self._aliases = aliases + self._by_target = {target_id: tuple(sorted(target_paths, key=_path_rank)) for target_id, target_paths in by_target.items()} + self._lock = None + self._lock_loop = None + + @property + def enabled(self): + """Expose the immutable explicit gate.""" + return self._enabled + + @property + def health_ledger(self): + """Expose health observations for detached diagnostics and tests.""" + return self._health + + def ranked_paths(self, qualified_alias): + """Resolve one qualified alias and return deterministic path order.""" + if not isinstance(qualified_alias, QualifiedTargetAlias): + raise ProviderFallbackValidationError( + "target lookup must be QualifiedTargetAlias", + ) + target_id = self._aliases.get(qualified_alias) + if target_id is None: + return () + return self._by_target.get(target_id, ()) + + async def dispatch(self, qualified_alias, request): + """Execute ranked paths, falling through only explicit safe rejection.""" + if not self._enabled: + return ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=False, + disabled=True, + detail="provider fallback router is disabled", + ) + + try: + lock = self._dispatch_lock() + except Exception as exc: + return ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=False, + detail=_safe_detail( + "dispatch lock failed: {}".format(type(exc).__name__), + ), + ) + + async with lock: + try: + paths = self.ranked_paths(qualified_alias) + decision_time = self._health.now() + except Exception as exc: + return ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=False, + detail=_safe_detail( + "routing preflight failed: {}".format( + type(exc).__name__, + ), + ), + ) + if not paths: + return ProviderFallbackResult( + ProviderExecutionState.NOT_APPLIED, + fallback_safe=True, + detail="no declared provider path", + ) + + attempts = [] + skipped_safe = False + for path in paths: + try: + phase = self._health.phase(path, now=decision_time) + except Exception as exc: + return ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=False, + attempts=tuple(attempts), + selected_path=path.identity, + detail=_safe_detail( + "health decision failed: {}".format( + type(exc).__name__, + ), + ), + ) + if phase is ProviderPathHealthPhase.COOLING_FALLBACK_SAFE: + skipped_safe = True + continue + if phase is ProviderPathHealthPhase.COOLING_BLOCKED: + return ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=False, + attempts=tuple(attempts), + selected_path=path.identity, + detail="higher-ranked path is cooling after unsafe result", + ) + + result = await _execute(path, request) + try: + self._health.observe(path.identity, result) + except Exception as exc: + attempt = ProviderPathAttempt( + path.identity, + path.kind, + ProviderExecutionResult.unknown( + "health persistence failed", + ), + ) + attempts.append(attempt) + return ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=False, + attempts=tuple(attempts), + selected_path=path.identity, + detail=_safe_detail( + "health persistence failed: {}".format( + type(exc).__name__, + ), + ), + ) + attempts.append( + ProviderPathAttempt(path.identity, path.kind, result), + ) + if result.state is ProviderExecutionState.APPLIED: + return ProviderFallbackResult( + ProviderExecutionState.APPLIED, + fallback_safe=False, + attempts=tuple(attempts), + selected_path=path.identity, + detail=result.detail, + ) + if result.state is ProviderExecutionState.NOT_APPLIED and result.fallback_safe: + skipped_safe = True + continue + return ProviderFallbackResult( + result.state, + fallback_safe=False, + attempts=tuple(attempts), + selected_path=path.identity, + detail=result.detail, + ) + + return ProviderFallbackResult( + ProviderExecutionState.NOT_APPLIED, + fallback_safe=skipped_safe, + attempts=tuple(attempts), + detail="all provider paths definitely declined", + ) + + def _dispatch_lock(self): + """Lazily bind the serial lock to the active event loop.""" + loop = asyncio.get_running_loop() + if self._lock_loop is loop: + return self._lock + if self._lock is not None and self._lock.locked(): + raise ProviderFallbackValidationError( + "router cannot move an active dispatch across event loops", + ) + self._lock = asyncio.Lock() + self._lock_loop = loop + return self._lock + + +async def _execute(path, request): + """Invoke one executor and convert exceptions or invalid truth to UNKNOWN.""" + try: + result = path.executor(request) + if inspect.isawaitable(result): + result = await result + except BaseException as exc: + if isinstance( + exc, + (asyncio.CancelledError, KeyboardInterrupt, SystemExit), + ): + raise + return ProviderExecutionResult.unknown( + _safe_detail("executor raised {}".format(type(exc).__name__)), + ) + if not isinstance(result, ProviderExecutionResult): + return ProviderExecutionResult.unknown( + "executor returned invalid result", + ) + return result + + +def _path_rank(path): + """Rank local before cloud, then use stable provider/path identities.""" + kind_rank = { + ProviderPathKind.GATEWAY_LOCAL: 0, + ProviderPathKind.VENDOR_CLOUD: 1, + } + return ( + kind_rank[path.kind], + path.provider_id.encode("ascii"), + path.path_id.encode("ascii"), + ) + + +def _require_identifier(value, name, lowercase=False): + """Require bounded canonical ASCII identity text.""" + if not isinstance(value, str): + raise ProviderFallbackValidationError( + "{} must be a string".format(name), + ) + try: + encoded = value.encode("ascii", "strict") + except UnicodeEncodeError as exc: + raise ProviderFallbackValidationError( + "{} must be canonical ASCII".format(name), + ) from exc + if not encoded or len(encoded) > MAX_IDENTITY_BYTES or _IDENTIFIER.fullmatch(value) is None: + raise ProviderFallbackValidationError( + "{} must be a bounded canonical identifier".format(name), + ) + if lowercase and value != value.lower(): + raise ProviderFallbackValidationError( + "{} must be lowercase".format(name), + ) + + +def _require_duration(value, name): + """Require a finite strictly-positive duration.""" + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0: + raise ProviderFallbackValidationError( + "{} must be a finite positive number".format(name), + ) + + +def _require_timestamp(value, name): + """Require a finite non-negative monotonic timestamp.""" + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0: + raise ProviderFallbackValidationError( + "{} must be a finite non-negative number".format(name), + ) + + +def _require_detail(value): + """Require bounded printable diagnostic text.""" + if not isinstance(value, str): + raise ProviderFallbackValidationError("detail must be a string") + try: + encoded = value.encode("utf-8", "strict") + except UnicodeEncodeError as exc: + raise ProviderFallbackValidationError( + "detail must be valid UTF-8", + ) from exc + if len(encoded) > MAX_DETAIL_BYTES: + raise ProviderFallbackValidationError("detail is too long") + if any(ord(character) < 0x20 or ord(character) == 0x7F or _is_noncharacter(ord(character)) for character in value): + raise ProviderFallbackValidationError( + "detail contains unsafe characters", + ) + + +def _safe_detail(value): + """Return a bounded printable ASCII diagnostic without raw exception text.""" + text = "".join(character if 0x20 <= ord(character) < 0x7F else "?" for character in str(value)) + encoded = text.encode("ascii") + if len(encoded) <= MAX_DETAIL_BYTES: + return text + return encoded[:MAX_DETAIL_BYTES].decode("ascii") + + +def _is_noncharacter(codepoint): + """Return whether one Unicode scalar is reserved as a noncharacter.""" + return 0xFDD0 <= codepoint <= 0xFDEF or codepoint & 0xFFFF in (0xFFFE, 0xFFFF) diff --git a/apps/predbat/tests/test_lattice_provider_fallback.py b/apps/predbat/tests/test_lattice_provider_fallback.py new file mode 100644 index 000000000..3c275f0d9 --- /dev/null +++ b/apps/predbat/tests/test_lattice_provider_fallback.py @@ -0,0 +1,1175 @@ +"""Tests for the detached Lattice provider-path fallback boundary.""" + +# cspell:ignore codepoint codepoints cooldown fffe noncanonical noncharacter overclaim + +import asyncio +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_provider_fallback import ( + LatticeProviderFallbackRouter, + ProviderExecutionResult, + ProviderExecutionState, + ProviderFallbackResult, + ProviderFallbackValidationError, + ProviderPath, + ProviderPathHealthLedger, + ProviderPathHealthPhase, + ProviderPathHealthPolicy, + ProviderPathIdentity, + ProviderPathKind, + QualifiedTargetAlias, +) + + +class ExplodingValue: + """Object that proves the disabled path performs no accidental inspection.""" + + def __getattribute__(self, name): + """Raise on every attempted attribute read.""" + raise AssertionError("disabled request was inspected: {}".format(name)) + + def __str__(self): + """Raise on accidental text coercion.""" + raise AssertionError("disabled request was coerced") + + +class Clock: + """Controllable monotonic clock that records reads.""" + + def __init__(self, value=100.0): + self.value = value + self.calls = 0 + + def __call__(self): + self.calls += 1 + return self.value + + +class Executor: + """Scripted sync or async executor with exact call evidence.""" + + def __init__(self, results, asynchronous=False): + self.results = list(results) + self.asynchronous = asynchronous + self.calls = [] + + def __call__(self, request): + self.calls.append(request) + result = self.results.pop(0) + if isinstance(result, BaseException): + raise result + if not self.asynchronous: + return result + + async def respond(): + return result + + return respond() + + +def path( + provider_id, + path_id, + kind, + executor, + target_id="battery-main", + aliases=(), + ttl=30.0, + cooldown=60.0, +): + """Build one concise valid provider path.""" + return ProviderPath( + target_id=target_id, + provider_id=provider_id, + path_id=path_id, + kind=kind, + executor=executor, + aliases=aliases, + health_policy=ProviderPathHealthPolicy(ttl, cooldown), + ) + + +class TestStrictModels(unittest.TestCase): + """Exhaustively reject outcome combinations that could overclaim safety.""" + + def test_valid_execution_result_truth_table(self): + cases = ( + ( + ProviderExecutionResult.applied(2750), + ProviderExecutionState.APPLIED, + False, + True, + ), + ( + ProviderExecutionResult.not_applied(False), + ProviderExecutionState.NOT_APPLIED, + False, + False, + ), + ( + ProviderExecutionResult.not_applied(True), + ProviderExecutionState.NOT_APPLIED, + True, + False, + ), + ( + ProviderExecutionResult.unknown(), + ProviderExecutionState.UNKNOWN, + False, + False, + ), + ) + for result, state, fallback_safe, verified in cases: + with self.subTest(state=state, fallback_safe=fallback_safe): + self.assertIs(result.state, state) + self.assertIs(result.fallback_safe, fallback_safe) + self.assertIs(result.verified, verified) + + def test_invalid_execution_result_truth_table(self): + for state in ProviderExecutionState: + for fallback_safe in (False, True): + for verified in (False, True): + for has_value in (False, True): + valid = ( + (state is ProviderExecutionState.APPLIED and not fallback_safe and verified) + or (state is ProviderExecutionState.NOT_APPLIED and not verified and not has_value) + or (state is ProviderExecutionState.UNKNOWN and not fallback_safe and not verified and not has_value) + ) + if valid: + continue + with self.subTest( + state=state, + fallback_safe=fallback_safe, + verified=verified, + has_value=has_value, + ): + with self.assertRaises( + ProviderFallbackValidationError, + ): + ProviderExecutionResult( + state, + fallback_safe=fallback_safe, + verified=verified, + applied_value=1 if has_value else None, + ) + + def test_non_enum_and_non_boolean_fields_are_rejected(self): + invalid = ( + lambda: ProviderExecutionResult("APPLIED"), + lambda: ProviderExecutionResult( + ProviderExecutionState.UNKNOWN, + fallback_safe=1, + ), + lambda: ProviderExecutionResult( + ProviderExecutionState.UNKNOWN, + verified=1, + ), + ) + for factory in invalid: + with self.subTest(factory=factory): + with self.assertRaises(ProviderFallbackValidationError): + factory() + + def test_detail_rejects_every_ascii_control(self): + for codepoint in tuple(range(0x20)) + (0x7F,): + with self.subTest(codepoint=codepoint): + with self.assertRaises(ProviderFallbackValidationError): + ProviderExecutionResult.unknown(chr(codepoint)) + + def test_detail_rejects_every_unicode_noncharacter(self): + codepoints = list(range(0xFDD0, 0xFDF0)) + codepoints.extend(codepoint for plane in range(17) for codepoint in (plane * 0x10000 + 0xFFFE, plane * 0x10000 + 0xFFFF)) + for codepoint in codepoints: + with self.subTest(codepoint=hex(codepoint)): + with self.assertRaises(ProviderFallbackValidationError): + ProviderExecutionResult.unknown(chr(codepoint)) + + def test_detail_is_bounded_by_utf8_bytes(self): + self.assertEqual( + ProviderExecutionResult.unknown("£" * 128).detail, + "£" * 128, + ) + with self.assertRaises(ProviderFallbackValidationError): + ProviderExecutionResult.unknown("£" * 129) + + def test_aggregate_result_cannot_overclaim(self): + invalid = ( + lambda: ProviderFallbackResult( + ProviderExecutionState.APPLIED, + True, + ), + lambda: ProviderFallbackResult( + ProviderExecutionState.UNKNOWN, + True, + ), + lambda: ProviderFallbackResult( + ProviderExecutionState.NOT_APPLIED, + False, + disabled=True, + ), + ) + for factory in invalid: + with self.subTest(factory=factory): + with self.assertRaises(ProviderFallbackValidationError): + factory() + + +class TestIdentityAndDeclarations(unittest.TestCase): + """Prove provider qualification, canonical identity, and collision handling.""" + + def test_provider_qualified_aliases_can_repeat_across_providers(self): + gateway = path( + "predbat-gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + Executor([]), + aliases=("SERIAL-1",), + ) + cloud = path( + "givenergy-cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + Executor([]), + aliases=("SERIAL-1",), + ) + router = LatticeProviderFallbackRouter((gateway, cloud)) + self.assertEqual( + router.ranked_paths( + QualifiedTargetAlias("predbat-gateway", "SERIAL-1"), + ), + (gateway, cloud), + ) + self.assertEqual( + router.ranked_paths( + QualifiedTargetAlias("givenergy-cloud", "SERIAL-1"), + ), + (gateway, cloud), + ) + + def test_alias_collision_within_provider_fails_closed(self): + first = path( + "vendor-cloud", + "one", + ProviderPathKind.VENDOR_CLOUD, + Executor([]), + target_id="battery-one", + aliases=("SERIAL-1",), + ) + second = path( + "vendor-cloud", + "two", + ProviderPathKind.VENDOR_CLOUD, + Executor([]), + target_id="battery-two", + aliases=("SERIAL-1",), + ) + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "alias collision", + ): + LatticeProviderFallbackRouter((first, second)) + + def test_implicit_target_alias_collision_fails_closed(self): + first = path( + "vendor-cloud", + "one", + ProviderPathKind.VENDOR_CLOUD, + Executor([]), + target_id="battery-one", + aliases=("battery-two",), + ) + second = path( + "vendor-cloud", + "two", + ProviderPathKind.VENDOR_CLOUD, + Executor([]), + target_id="battery-two", + ) + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "alias collision", + ): + LatticeProviderFallbackRouter((first, second)) + + def test_duplicate_path_identity_fails_even_if_declarations_match(self): + executor = Executor([]) + declaration = path( + "vendor-cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + executor, + ) + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "path identity collision", + ): + LatticeProviderFallbackRouter((declaration, declaration)) + + def test_duplicate_alias_inside_one_declaration_fails(self): + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "duplicate alias", + ): + path( + "vendor-cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + Executor([]), + aliases=("SERIAL-1", "SERIAL-1"), + ) + + def test_unqualified_lookup_is_rejected(self): + router = LatticeProviderFallbackRouter(()) + for value in ("battery-main", ("vendor", "battery-main"), None): + with self.subTest(value=value): + with self.assertRaises(ProviderFallbackValidationError): + router.ranked_paths(value) + + def test_identity_rejects_noncanonical_values(self): + values = ( + "", + " leading", + "trailing ", + "-leading", + "trailing-", + "contains space", + "unicode-£", + "a" * 129, + "line\nbreak", + ) + for value in values: + with self.subTest(value=repr(value)): + with self.assertRaises(ProviderFallbackValidationError): + QualifiedTargetAlias("provider", value) + for value in ("UPPER", "Provider", "provider-£"): + with self.subTest(provider=value): + with self.assertRaises(ProviderFallbackValidationError): + QualifiedTargetAlias(value, "target") + + def test_identifier_boundary_is_exact(self): + valid = "a" * 128 + self.assertEqual( + QualifiedTargetAlias("provider", valid).alias, + valid, + ) + with self.assertRaises(ProviderFallbackValidationError): + QualifiedTargetAlias("provider", "a" * 129) + + def test_invalid_path_shapes_fail_at_construction(self): + valid_executor = Executor([]) + invalid = ( + lambda: ProviderPath( + "target", + "provider", + "path", + "gw-local", + valid_executor, + ), + lambda: ProviderPath( + "target", + "provider", + "path", + ProviderPathKind.GATEWAY_LOCAL, + None, + ), + lambda: ProviderPath( + "target", + "provider", + "path", + ProviderPathKind.GATEWAY_LOCAL, + valid_executor, + aliases=[], + ), + lambda: LatticeProviderFallbackRouter([], enabled=False), + lambda: LatticeProviderFallbackRouter((), enabled=1), + ) + for factory in invalid: + with self.subTest(factory=factory): + with self.assertRaises(ProviderFallbackValidationError): + factory() + + +class TestHealthLedger(unittest.TestCase): + """Exercise TTL, cooldown, identity isolation, and monotonicity.""" + + def setUp(self): + self.clock = Clock() + self.ledger = ProviderPathHealthLedger(self.clock) + self.executor = Executor([]) + self.path = path( + "predbat-gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + self.executor, + ttl=10.0, + cooldown=20.0, + ) + + def test_unobserved_then_applied_ttl_then_probe(self): + self.assertIs( + self.ledger.phase(self.path), + ProviderPathHealthPhase.UNOBSERVED, + ) + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.applied(), + ) + self.clock.value = 110.0 + self.assertIs( + self.ledger.phase(self.path), + ProviderPathHealthPhase.HEALTHY, + ) + self.clock.value = 110.0001 + self.assertIs( + self.ledger.phase(self.path), + ProviderPathHealthPhase.PROBE, + ) + + def test_safe_not_applied_cools_then_probes(self): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.not_applied(fallback_safe=True), + ) + self.clock.value = 119.999 + self.assertIs( + self.ledger.phase(self.path), + ProviderPathHealthPhase.COOLING_FALLBACK_SAFE, + ) + self.clock.value = 120.0 + self.assertIs( + self.ledger.phase(self.path), + ProviderPathHealthPhase.PROBE, + ) + + def test_unknown_and_unsafe_not_applied_cool_blocked(self): + for result in ( + ProviderExecutionResult.unknown(), + ProviderExecutionResult.not_applied(fallback_safe=False), + ): + with self.subTest(result=result): + ledger = ProviderPathHealthLedger(Clock()) + ledger.observe(self.path.identity, result) + self.assertIs( + ledger.phase(self.path, now=119.999), + ProviderPathHealthPhase.COOLING_BLOCKED, + ) + self.assertIs( + ledger.phase(self.path, now=120.0), + ProviderPathHealthPhase.PROBE, + ) + + def test_health_is_isolated_by_target_provider_and_path(self): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.unknown(), + ) + variants = ( + path( + "predbat-gateway", + "other", + ProviderPathKind.GATEWAY_LOCAL, + self.executor, + ), + path( + "other-provider", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + self.executor, + ), + path( + "predbat-gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + self.executor, + target_id="other-target", + ), + ) + for variant in variants: + with self.subTest(identity=variant.identity): + self.assertIs( + self.ledger.phase(variant), + ProviderPathHealthPhase.UNOBSERVED, + ) + + def test_time_regression_and_same_time_conflict_fail_closed(self): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.applied(), + observed_at=100.0, + ) + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "time regressed", + ): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.unknown(), + observed_at=99.0, + ) + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "reused timestamp", + ): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.unknown(), + observed_at=100.0, + ) + repeated = self.ledger.observe( + self.path.identity, + ProviderExecutionResult.applied(), + observed_at=100.0, + ) + self.assertEqual(repeated, self.ledger.record(self.path.identity)) + + def test_decision_before_observation_fails_closed(self): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.applied(), + observed_at=100.0, + ) + with self.assertRaisesRegex( + ProviderFallbackValidationError, + "precedes observation", + ): + self.ledger.phase(self.path, now=99.0) + + def test_invalid_time_and_policy_values_are_rejected(self): + values = ( + True, + False, + -1, + 0, + float("inf"), + float("-inf"), + float("nan"), + "1", + None, + ) + for value in values: + with self.subTest(policy=value): + with self.assertRaises(ProviderFallbackValidationError): + ProviderPathHealthPolicy(value, 1) + with self.assertRaises(ProviderFallbackValidationError): + ProviderPathHealthPolicy(1, value) + for value in ( + True, + -1, + float("inf"), + float("-inf"), + float("nan"), + "1", + None, + ): + with self.subTest(timestamp=value): + with self.assertRaises(ProviderFallbackValidationError): + self.ledger.observe( + self.path.identity, + ProviderExecutionResult.unknown(), + observed_at=value, + ) + + +class TestRouting(unittest.TestCase): + """Prove deterministic rank, strict fallthrough, and zero-effect disable.""" + + def test_disabled_returns_before_all_caller_objects_and_clock(self): + clock = Clock() + ledger = ProviderPathHealthLedger(clock) + executor = Executor( + [ProviderExecutionResult.applied()], + ) + router = LatticeProviderFallbackRouter( + ( + path( + "predbat-gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + ), + ), + health_ledger=ledger, + ) + result = asyncio.run( + router.dispatch(ExplodingValue(), ExplodingValue()), + ) + self.assertTrue(result.disabled) + self.assertIs(result.state, ProviderExecutionState.UNKNOWN) + self.assertFalse(result.fallback_safe) + self.assertEqual(result.attempts, ()) + self.assertEqual(clock.calls, 0) + self.assertEqual(executor.calls, []) + + def test_gateway_local_always_ranks_before_vendor_cloud(self): + calls = [] + + def executor(name, result): + def execute(_request): + calls.append(name) + return result + + return execute + + declarations = ( + path( + "z-vendor", + "z-rest", + ProviderPathKind.VENDOR_CLOUD, + executor( + "z-cloud", + ProviderExecutionResult.applied(), + ), + aliases=("SERIAL-1",), + ), + path( + "z-gateway", + "z-mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor( + "z-local", + ProviderExecutionResult.not_applied(True), + ), + ), + path( + "a-vendor", + "a-rest", + ProviderPathKind.VENDOR_CLOUD, + executor( + "a-cloud", + ProviderExecutionResult.applied(), + ), + ), + path( + "a-gateway", + "a-mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor( + "a-local", + ProviderExecutionResult.not_applied(True), + ), + ), + ) + router = LatticeProviderFallbackRouter( + declarations, + enabled=True, + ) + ranked = router.ranked_paths( + QualifiedTargetAlias("z-vendor", "SERIAL-1"), + ) + self.assertEqual( + tuple((item.kind, item.provider_id, item.path_id) for item in ranked), + ( + (ProviderPathKind.GATEWAY_LOCAL, "a-gateway", "a-mqtt"), + (ProviderPathKind.GATEWAY_LOCAL, "z-gateway", "z-mqtt"), + (ProviderPathKind.VENDOR_CLOUD, "a-vendor", "a-rest"), + (ProviderPathKind.VENDOR_CLOUD, "z-vendor", "z-rest"), + ), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("z-vendor", "SERIAL-1"), + {"watts": 2500}, + ), + ) + self.assertIs(result.state, ProviderExecutionState.APPLIED) + self.assertEqual(calls, ["a-local", "z-local", "a-cloud"]) + + def test_each_terminal_state_controls_fallback_exactly(self): + first_results = ( + ProviderExecutionResult.applied(), + ProviderExecutionResult.not_applied(False), + ProviderExecutionResult.not_applied(True), + ProviderExecutionResult.unknown(), + ) + for first_result in first_results: + with self.subTest(first_result=first_result): + local = Executor([first_result]) + cloud = Executor([ProviderExecutionResult.applied()]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + local, + aliases=("target-alias",), + ), + path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud, + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger( + Clock(), + ), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias( + "gateway", + "target-alias", + ), + object(), + ), + ) + should_fallback = first_result.state is ProviderExecutionState.NOT_APPLIED and first_result.fallback_safe + self.assertEqual(len(cloud.calls), int(should_fallback)) + self.assertEqual( + result.state, + (ProviderExecutionState.APPLIED if should_fallback else first_result.state), + ) + + def test_unknown_never_falls_through_to_cloud(self): + local = Executor([ProviderExecutionResult.unknown()]) + cloud = Executor([ProviderExecutionResult.applied()]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + local, + aliases=("battery",), + ), + path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud, + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger(Clock()), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs(result.state, ProviderExecutionState.UNKNOWN) + self.assertFalse(result.fallback_safe) + self.assertEqual(len(result.attempts), 1) + self.assertEqual(cloud.calls, []) + + def test_exception_and_invalid_return_become_unknown_without_fallback(self): + invalid_returns = ( + True, + False, + "NOT_APPLIED", + None, + {"state": "NOT_APPLIED", "fallback_safe": True}, + RuntimeError("secret message must not escape"), + ) + for invalid_return in invalid_returns: + with self.subTest(invalid_return=repr(invalid_return)): + local = Executor([invalid_return], asynchronous=True) + cloud = Executor([ProviderExecutionResult.applied()]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + local, + aliases=("battery",), + ), + path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud, + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger( + Clock(), + ), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs( + result.state, + ProviderExecutionState.UNKNOWN, + ) + self.assertFalse(result.fallback_safe) + self.assertEqual(cloud.calls, []) + self.assertNotIn("secret", result.detail) + + def test_safe_cooling_path_skips_to_cloud(self): + clock = Clock() + ledger = ProviderPathHealthLedger(clock) + local = path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + Executor([]), + aliases=("battery",), + cooldown=60, + ) + cloud_executor = Executor([ProviderExecutionResult.applied()]) + cloud = path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud_executor, + ) + ledger.observe( + local.identity, + ProviderExecutionResult.not_applied(True), + ) + router = LatticeProviderFallbackRouter( + (local, cloud), + enabled=True, + health_ledger=ledger, + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs(result.state, ProviderExecutionState.APPLIED) + self.assertEqual(cloud_executor.calls.__len__(), 1) + self.assertEqual(len(result.attempts), 1) + + def test_unknown_cooling_path_blocks_cloud(self): + clock = Clock() + ledger = ProviderPathHealthLedger(clock) + local = path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + Executor([]), + aliases=("battery",), + cooldown=60, + ) + cloud_executor = Executor([ProviderExecutionResult.applied()]) + cloud = path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud_executor, + ) + ledger.observe(local.identity, ProviderExecutionResult.unknown()) + router = LatticeProviderFallbackRouter( + (local, cloud), + enabled=True, + health_ledger=ledger, + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs(result.state, ProviderExecutionState.UNKNOWN) + self.assertEqual(result.selected_path, local.identity) + self.assertEqual(result.attempts, ()) + self.assertEqual(cloud_executor.calls, []) + + def test_cooldown_expiry_retries_higher_ranked_path(self): + clock = Clock() + ledger = ProviderPathHealthLedger(clock) + local_executor = Executor([ProviderExecutionResult.applied()]) + local = path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + local_executor, + aliases=("battery",), + cooldown=60, + ) + cloud_executor = Executor([ProviderExecutionResult.applied()]) + cloud = path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud_executor, + ) + ledger.observe(local.identity, ProviderExecutionResult.unknown()) + clock.value = 160.0 + router = LatticeProviderFallbackRouter( + (cloud, local), + enabled=True, + health_ledger=ledger, + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs(result.state, ProviderExecutionState.APPLIED) + self.assertEqual(len(local_executor.calls), 1) + self.assertEqual(cloud_executor.calls, []) + + def test_all_definite_safe_declines_are_fallback_safe(self): + executors = ( + Executor([ProviderExecutionResult.not_applied(True)]), + Executor([ProviderExecutionResult.not_applied(True)]), + ) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executors[0], + aliases=("battery",), + ), + path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + executors[1], + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger(Clock()), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs( + result.state, + ProviderExecutionState.NOT_APPLIED, + ) + self.assertTrue(result.fallback_safe) + self.assertEqual(len(result.attempts), 2) + + def test_unknown_alias_is_definite_no_path_without_executor_effects(self): + executor = Executor([ProviderExecutionResult.applied()]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger(Clock()), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "not-declared"), + object(), + ), + ) + self.assertIs( + result.state, + ProviderExecutionState.NOT_APPLIED, + ) + self.assertTrue(result.fallback_safe) + self.assertEqual(executor.calls, []) + + def test_health_observation_is_recorded_for_each_attempt(self): + clock = Clock() + ledger = ProviderPathHealthLedger(clock) + executor = Executor([ProviderExecutionResult.applied(1234)]) + declaration = path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + aliases=("battery",), + ) + router = LatticeProviderFallbackRouter( + (declaration,), + enabled=True, + health_ledger=ledger, + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertEqual( + ledger.record(declaration.identity).result, + result.attempts[0].result, + ) + + def test_health_clock_failure_is_unknown_before_executor(self): + def broken_clock(): + raise RuntimeError("clock secret") + + executor = Executor([ProviderExecutionResult.applied()]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + aliases=("battery",), + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger(broken_clock), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs(result.state, ProviderExecutionState.UNKNOWN) + self.assertEqual(executor.calls, []) + self.assertNotIn("secret", result.detail) + + def test_health_write_failure_is_unknown_and_stops_fallback(self): + class FailingLedger(ProviderPathHealthLedger): + def observe(self, identity, result, observed_at=None): + raise RuntimeError("storage secret") + + local = Executor([ProviderExecutionResult.not_applied(True)]) + cloud = Executor([ProviderExecutionResult.applied()]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + local, + aliases=("battery",), + ), + path( + "cloud", + "rest", + ProviderPathKind.VENDOR_CLOUD, + cloud, + ), + ), + enabled=True, + health_ledger=FailingLedger(Clock()), + ) + result = asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIs(result.state, ProviderExecutionState.UNKNOWN) + self.assertEqual(cloud.calls, []) + self.assertNotIn("secret", result.detail) + + def test_keyboard_interrupt_and_system_exit_are_not_swallowed(self): + for exception in (KeyboardInterrupt(), SystemExit()): + with self.subTest(exception=type(exception).__name__): + executor = Executor([exception]) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + aliases=("battery",), + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger( + Clock(), + ), + ) + with self.assertRaises(type(exception)): + asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + + def test_cancellation_is_not_swallowed_before_or_during_await(self): + def cancelled_synchronously(_request): + raise asyncio.CancelledError() + + async def cancelled_while_awaiting(_request): + await asyncio.sleep(0) + raise asyncio.CancelledError() + + for executor in ( + cancelled_synchronously, + cancelled_while_awaiting, + ): + with self.subTest(executor=executor.__name__): + ledger = ProviderPathHealthLedger(Clock()) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + aliases=("battery",), + ), + ), + enabled=True, + health_ledger=ledger, + ) + with self.assertRaises(asyncio.CancelledError): + asyncio.run( + router.dispatch( + QualifiedTargetAlias("gateway", "battery"), + object(), + ), + ) + self.assertIsNone( + ledger.record( + ProviderPathIdentity( + "battery-main", + "gateway", + "mqtt", + ), + ), + ) + + def test_router_can_be_reused_across_sequential_event_loops(self): + executor = Executor( + [ + ProviderExecutionResult.applied(), + ProviderExecutionResult.applied(), + ], + ) + router = LatticeProviderFallbackRouter( + ( + path( + "gateway", + "mqtt", + ProviderPathKind.GATEWAY_LOCAL, + executor, + aliases=("battery",), + ), + ), + enabled=True, + health_ledger=ProviderPathHealthLedger(Clock()), + ) + alias = QualifiedTargetAlias("gateway", "battery") + + first = asyncio.run(router.dispatch(alias, object())) + second = asyncio.run(router.dispatch(alias, object())) + + self.assertIs(first.state, ProviderExecutionState.APPLIED) + self.assertIs(second.state, ProviderExecutionState.APPLIED) + self.assertEqual(len(executor.calls), 2) + + +if __name__ == "__main__": + unittest.main()