admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
SAFETY-112
goodwe-addon/goodwe_controller/tools/rehearse_roster.py
T

1217 lines
56 KiB
Python
Raw Permalink Blame History

"""ARCH-02: rehearse the flat -> roster switch against a COPY of real state.
python3 tools/rehearse_roster.py --state <path> \
--before <flat.yaml> --after <roster.yaml>
WHAT THIS IS FOR, in one sentence: on the day inverter B is commissioned the
add-on stops reading the flat single-inverter options and starts reading a
declared `inverters:` roster, and two things happen at that restart that are
ONE-WAY and INVISIBLE UNTIL AFTERWARDS - the store migration renames the keys
that hold a safety latch and a month of maintenance history, and the entity ids
Home Assistant composes change shape.
THE STORE. /data/state.json on the live box was written by a flat-config
controller. A latch that migrates to a name nothing reads is a
SAFETY GUARD THAT COMES UP DISARMED on an install whose state
file says it is protected (store.py's own header, and SAFETY-02:
"a latch that a restart clears is not a latch"). MAINT-05 had to
fix exactly this shape once already.
THE ENTITIES. Home Assistant keeps the entity_id it composed at FIRST
discovery. An id that stops being announced is not deleted - it
is ORPHANED: history stays behind on the dead id, and every
automation whose trigger names it SILENTLY STOPS FIRING. An
automation with a vanished trigger entity does not error. It
never runs again.
⚠️ THIS TOOL RUNS THE REAL `Controller`, NOT A MODEL OF IT. `snapshot()` builds
`app.main.Controller` over the real `app.store.Store`, so the migration under
test is the shipped one and the readers that are asked "can you still find your
latch" are the shipped readers. A rehearsal that re-implemented the migration
would be comparing a structure to itself, which is `MUTATION.md` habit 4 and
this repository's most-repeated defect.
⚠️ IT NEVER WRITES TO THE FILE IT IS GIVEN. `--state` is opened once, copied to
a scratch file made with `mkstemp()`, and every `Store.save()` in the rehearsal
lands on the copy. The scratch file is unique per invocation, which is habit 18.
⚠️ IT DOES NOT TOUCH THE LIVE BOX AND CANNOT. There is no SSH, no Supervisor
call and no HTTP in this file. Fetching the state file is an OWNER STEP and it
is written down in DOCS.md under "Before you commission inverter B:
tools/rehearse_roster.py (ARCH-02)".
EXIT CODES
0 the transition is clean: nothing vanished, nothing was stranded under a
name no machine reads, no entity was orphaned, no history was lost.
1 ⭐ AT LEAST ONE OF THOSE HAPPENED. This is the code that must be
reachable: `test_rehearse.py` mutates the shipped migration nine
different ways and requires this exit code for each.
2 the rehearsal could not be performed (bad file, unparseable config, a
state or options file of the wrong shape, the entity-id model failed
its own anchor). ⚠️ NOT A PASS.
3 no findings, but --no-live-entities: the orphan check was not graded
against the install. ⚠️ NOT A PASS.
"""
import argparse
import json
import os
import shutil
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.inverter import UNIT_A, UNIT_STORE_KEYS, Inverter, roster_names # noqa: E402
from app.main import Controller # noqa: E402
from app import mqtt # noqa: E402
from app.p1 import is_enabled # noqa: E402
from app.store import (DEFAULTS, MAINT_SCHEMA_KEY, MAINT_UNIT_KEYS, # noqa: E402
SCHEMA_KEY, Store, unit_key)
from predeploy_check import option_keys, shape_problem # noqa: E402
EXIT_OK, EXIT_FINDING, EXIT_COULD_NOT_RUN, EXIT_DEGRADED = 0, 1, 2, 3
ADDON = Path(__file__).resolve().parent.parent
class Refused(Exception):
"""The rehearsal could not be performed. Exit 2, never exit 0."""
# --------------------------------------------------------------------------- #
# 1. THE CONFIGS
# --------------------------------------------------------------------------- #
# The live box's options blob is JSON (`/data/options.json`); the card's command
# line spells the two configs as `.yaml` because that is what an operator edits.
# Both are accepted. ⚠️ THE YAML READER REFUSES WHAT IT DOES NOT UNDERSTAND
# RATHER THAN GUESSING: a config silently mis-parsed here would produce a
# rehearsal of a configuration nobody is going to deploy, which is worse than no
# rehearsal at all. PyYAML is used when it is importable; the add-on's check
# image has no `yaml` module, so the subset below is what actually runs in CI.
def _scalar(text: str):
raw = text.strip()
if raw.startswith("#"):
raise Refused("comment where a value was expected")
if (raw.startswith('"') and raw.endswith('"') and len(raw) >= 2) or \
(raw.startswith("'") and raw.endswith("'") and len(raw) >= 2):
return raw[1:-1]
low = raw.lower()
if low in ("true", "yes", "on"):
return True
if low in ("false", "no", "off"):
return False
if low in ("null", "~", ""):
return None
if raw == "[]":
return []
if raw == "{}":
return {}
try:
return int(raw)
except ValueError:
pass
try:
return float(raw)
except ValueError:
pass
if raw.startswith("[") or raw.startswith("{"):
raise Refused(f"inline collection {raw!r} is not supported by the "
"built-in reader - install PyYAML or use JSON")
return raw
def _mini_yaml(text: str) -> dict:
"""The subset an add-on options file is written in, and nothing else.
Block mappings, and lists of block mappings (`inverters:`). Two indent
levels of nesting, which is all `config.yaml`'s option surface has. Every
other construct raises `Refused`.
"""
root: dict = {}
cur_list = None
cur_entry = None
list_indent = None
for lineno, line in enumerate(text.splitlines(), 1):
if not line.strip() or line.lstrip().startswith("#"):
continue
indent = len(line) - len(line.lstrip(" "))
if "\t" in line[:indent]:
raise Refused(f"line {lineno}: tab in indentation")
body = line.strip()
try:
if body.startswith("- "):
if cur_list is None:
raise Refused(f"line {lineno}: list item outside a key")
if list_indent is None:
list_indent = indent
if indent != list_indent:
raise Refused(f"line {lineno}: ragged list indentation")
cur_entry = {}
cur_list.append(cur_entry)
item = body[2:].strip()
if item:
k, _, v = item.partition(":")
if not _:
raise Refused(f"line {lineno}: bare list scalar")
cur_entry[k.strip()] = _scalar(v)
continue
if ":" not in body:
raise Refused(f"line {lineno}: {body!r} is not `key: value`")
key, _, value = body.partition(":")
key, value = key.strip(), value.strip()
if cur_entry is not None and list_indent is not None \
and indent > list_indent:
cur_entry[key] = _scalar(value)
continue
cur_list = cur_entry = None
list_indent = None
if value == "":
cur_list = root[key] = []
continue
root[key] = _scalar(value)
except Refused:
raise
# A key written `foo:` with nothing under it is an EMPTY list here, which is
# exactly what `inverters: []` means to `roster_entries()`.
return root
def load_options(path: str) -> dict:
text = Path(path).read_text(encoding="utf-8")
try:
obj = json.loads(text)
except ValueError:
try:
import yaml # noqa: PLC0415
except ImportError:
obj = _mini_yaml(text)
else:
obj = yaml.safe_load(text)
# SAFETY-62: a wrapped export is a dict too, so assert the keys are options.
bad = shape_problem(obj, option_keys(str(ADDON)).__contains__, path)
if bad:
raise Refused(bad)
return obj
_UNIT_PREFIX = unit_key("\0", "").split("\0")[0]
# "auto" is the Automatic toggle, written by app/main.py outside DEFAULTS.
_STORE_KEYS = {*DEFAULTS, *MAINT_UNIT_KEYS, SCHEMA_KEY, MAINT_SCHEMA_KEY, "auto"}
def load_state(path: str) -> dict:
"""Refuse a state file `Store.load()` would silently start fresh on (SAFETY-62)."""
try:
obj = json.loads(Path(path).read_text(encoding="utf-8"))
except ValueError as err:
raise Refused(f"{path}: not a readable JSON state file ({err}); "
"Store.load() would start fresh and grade an empty store")
bad = shape_problem(
obj, lambda k: k in _STORE_KEYS or str(k).startswith(_UNIT_PREFIX),
path, allow_empty=True)
if bad:
raise Refused(bad)
return obj
# --------------------------------------------------------------------------- #
# 2. THE STORE, AND THE READ LOG THAT MAKES "NOTHING READS THIS" MEASURABLE
# --------------------------------------------------------------------------- #
class _ReadLog(dict):
"""`Store.data`, with every value-read remembered.
⚠️ THIS IS THE INSTRUMENT THE WHOLE SAFETY-44 QUESTION TURNS ON. "A key
nothing reads" cannot be established by reasoning about which name the
migration was handed; it is established by asking the SHIPPED READERS to
restore themselves and recording which keys they actually touched. A key in
the file that no reader touched is a key whose value is not information.
`__contains__` is deliberately NOT recorded: `_move_flat_keys()` tests
`target in data` to decide whether a namespaced value already exists, and
that is a write-side decision, not a machine reading its latch.
"""
def __init__(self, *a, **kw):
super().__init__(*a, **kw)
self.reads = set()
def get(self, key, default=None):
self.reads.add(key)
return super().get(key, default)
def __getitem__(self, key):
self.reads.add(key)
return super().__getitem__(key)
class RehearsalStore(Store):
"""The SHIPPED `Store`, pointed at a scratch copy of the operator's file.
⚠️ THE REAL CLASS, NOT A STAND-IN. `load()`, `save()`, `get_time()` and
`set_time()` are the ones the add-on runs, so the ISO-8601 round trip that
the module docstring says once took the controller down for hours is the
round trip under rehearsal. A `FakeStore` here would rehearse the fake.
"""
def __init__(self, path: str):
super().__init__(path)
self.data = _ReadLog(self.data)
self.data.reads.clear()
def _scratch_copy(state_path: str, tmpdir: str, tag: str) -> str:
"""A per-invocation copy. ⚠️ The operator's file is never opened for write."""
dest = os.path.join(tmpdir, f"state-{tag}.json")
if state_path is None:
Path(dest).write_text("{}", encoding="utf-8")
else:
shutil.copyfile(state_path, dest)
return dest
# --------------------------------------------------------------------------- #
# 3. THE ENTITY IDS
# --------------------------------------------------------------------------- #
# ⚠️ THE SLUGIFIER IS DELIBERATELY NOT `mqtt._slug`. Importing the module's own
# spelling would make the model and the subject move together, so a mutation in
# the composition would change both sides equally and this tool would report
# "no orphans" about a rename that orphaned everything - habit 4, written in one
# import. `test_mqtt.py` part 6 makes the same move for the same reason.
import re # noqa: E402
def compose(device_name, entity_name) -> str:
"""The entity_id Home Assistant composes, per `app/mqtt.py`'s own note."""
s = re.sub(r"[^a-z0-9]+", "_", f"{device_name} {entity_name}".lower())
return "sensor." + s.strip("_")
# The ONE id verified against the live system (2026-08-26, recorded in
# `app/mqtt.py`) and subscribed to by the ESP32 ingest guard. If the model above
# cannot reproduce it, the model is wrong and every orphan claim below it is
# worthless - so the tool REFUSES rather than reporting.
ANCHOR_ID = "sensor.goodwe_p1_sample_age"
class Row:
"""One announced entity, in the three terms Home Assistant keys on.
⚠️ THE UNIQUE_ID AND THE DEVICE IDENTIFIER ARE NOT DECORATION, AND LEAVING
THEM OUT OF THE COMPARISON WAS THIS TOOL'S SECOND HABIT-4. HA's MQTT
integration keys its ENTITY registry on `unique_id` and its DEVICE registry
on the device identifier. Change either one and HA creates a NEW entity
beside the old - while the COMPOSED entity_id is byte-identical, so a
comparison over composed ids alone reports a clean transition about a change
that orphans the whole install. Measured: renaming `mqtt.BASE` and changing
`DEVICE["identifiers"]` both came back exit 0 before this class existed.
"""
__slots__ = ("entity_id", "unique_id", "device_id")
def __init__(self, entity_id, unique_id, device_id):
self.entity_id = entity_id
self.unique_id = unique_id
self.device_id = device_id
def __repr__(self):
return (f"{self.entity_id} (unique_id {self.unique_id}, device "
f"{self.device_id})")
# ⚠️ WHAT THE REFERENCE STILL DOES NOT CARRY: THE DISCOVERY TOPIC.
#
# The three columns are `entity_id unique_id device`, and the payload is graded
# against all three. The TOPIC is not. Measured by review:
#
# homeassistant/sensor/... -> ha/sensor/... exit 0, NO FINDINGS
# sensor -> binary_sensor on the unit rows exit 0, NO FINDINGS
#
# In the first, HA never receives any discovery at all and every announced
# entity dies - reported clean. `discovery_payloads()` already returns the
# topic; `announced_rows` discards it, so a fourth column closes it.
#
# ⚠️ RECORDED AND NOT FIXED, deliberately, with the reason: the prefix is HA's
# fixed contract rather than a name this project chooses, so it is not the kind
# of value a tidy-up rename reaches - which is exactly the risk `mqtt.py`'s own
# header warns about for `unique_id`. That makes it lower-likelihood than the
# N1 gap it resembles, not lower-consequence. If the topic ever becomes
# configurable, this stops being a note and becomes the same defect.
def announced_rows(opts: dict) -> "dict[str, Row]":
"""`{entity_id: Row}` for every row this add-on announces on `opts`.
⚠️ `discovery_payloads()` IS THE SHIPPED ANNOUNCER and `roster_names()` is
the shipped naming rule, so the roster shape decision is not restated here.
"""
payloads = mqtt.discovery_payloads(
units=roster_names(opts),
omit=() if is_enabled(opts) else ("p1_age",))
out = {}
for _topic, cfg in payloads:
eid = compose(cfg["device"]["name"], cfg["name"])
out[eid] = Row(eid, cfg["unique_id"],
(cfg["device"].get("identifiers") or [""])[0])
return out
# The device name and entity name that composed it, recorded as LITERALS.
# ⚠️ LITERALS, NOT `mqtt.DEVICE["name"]`. Anchoring on the shipped dict would
# make this check pass for any renaming of the device - the model and the
# subject would move together and the anchor would be decoration. These two
# strings are what `release/1.0` had on the day the id was verified.
ANCHOR_DEVICE_NAME = "GoodWe"
ANCHOR_ENTITY_NAME = "P1 sample age"
def _check_anchor() -> None:
"""Is the MODEL right? Nothing about the shipped code is asked here."""
got = compose(ANCHOR_DEVICE_NAME, ANCHOR_ENTITY_NAME)
if got != ANCHOR_ID:
raise Refused(
f"the entity-id model composes {got!r} where app/mqtt.py records "
f"{ANCHOR_ID!r} as verified against the live system on 2026-08-26. "
"The model is wrong; no orphan claim from it can be believed.")
# Shipped beside this file, and LOADED ON EVERY RUN. See `load_reference()`.
DEFAULT_REFERENCE = (Path(__file__).resolve().parent / "rehearsal"
/ "announced-reference.txt")
def load_reference(path) -> "dict[str, Row]":
"""What this build is SUPPOSED to announce: `<entity_id> <unique_id> <dev>`.
⚠️ EXTERNAL DATA, AND THE TOOL IS BLIND WITHOUT IT - which is why it is
loaded unconditionally and a missing one is EXIT 2 rather than a quieter
run. The before/after diff is composed twice by the SAME build, so a change
to the composition itself moves both sides equally and the diff reports
nothing. Only a reference that does not move with the code can see it.
⚠️ THREE COLUMNS ARE REQUIRED AND A SHORT LINE IS REFUSED. A two-column
reference would silently stop grading device identifiers, which is exactly
the blind spot this file exists to close - and it would do it without
saying so, which is the failure mode the whole card is about.
"""
rows = {}
for lineno, line in enumerate(
Path(path).read_text(encoding="utf-8").splitlines(), 1):
line = line.split("#", 1)[0].strip()
if not line:
continue
parts = line.split()
if len(parts) != 3:
raise Refused(
f"{path} line {lineno}: expected `<entity_id> <unique_id> "
f"<device identifier>`, got {len(parts)} field(s). A short "
f"line would silently stop grading what HA keys on.")
rows[parts[0]] = Row(*parts)
if not rows:
raise Refused(f"{path}: no rows in it")
return rows
def load_entity_list(path: str) -> set:
"""The entity ids that EXIST ON THE INSTALL, one per line, `#` comments.
This is the owner's export from Home Assistant, and it answers a different
question from `load_reference()`: that one says what the BUILD announces,
this one says what the INSTALL has. An id the install has and the new shape
does not announce is an orphan whatever the build's own history says.
"""
out = set()
for line in Path(path).read_text(encoding="utf-8").splitlines():
line = line.split("#", 1)[0].strip()
if line:
out.add(line.split()[0])
if not out:
raise Refused(f"{path}: no entity ids in it")
return out
# --------------------------------------------------------------------------- #
# 4. ONE CONTROLLER START, OBSERVED
# --------------------------------------------------------------------------- #
class Hass:
"""No I/O. `Controller.__init__` performs none; this only has to exist."""
async def number(self, entity, invert=False):
return None
async def limits(self, entity):
return (None, None, None)
async def set_number(self, entity, value):
return False
async def call(self, domain, service, data):
return False
class Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
class Snapshot:
"""What one `Controller` start over one state file looks like."""
def __init__(self, label, opts, store, controller):
self.label = label
self.opts = opts
self.store = store
self.units = [inv.name for inv in controller.inverters]
self.data = dict(store.data)
self.read = set(store.data.reads)
self.entities = announced_rows(opts)
self.latched = {inv.name: bool(inv.sign_wd.tripped)
for inv in controller.inverters}
self.latch_reason = {inv.name: inv.sign_wd.reason
for inv in controller.inverters}
self.flat_tripped = bool(store.data.get("sign_tripped", False))
self.history = {}
for m in controller.maints:
self.history[m.name] = {k: m.store.get(k) for k in MAINT_UNIT_KEYS}
self.store_migration = controller.store_migration
self.maint_migration = controller.maint_migration
self.machine_read = machine_reads(controller, store)
# Keys no machine reads, and keys that disappeared, whose value is the
# default `Store.load()` would have supplied. Reported, never a
# finding: see `diff()`.
self.stranded_at_default = []
self.vanished_at_default = []
self.vanished_but_migrated = []
@property
def unit_keys(self):
return sorted(k for k in self.data if k.startswith("unit/"))
def machine_reads(controller, store) -> set:
"""Every store key the CONSTRUCTED machines read, measured not transcribed.
⚠️ THE PROBE IS THE SHIPPED CONSTRUCTOR, RUN AGAIN. `Inverter.__init__` is
the code that restores a unit's SAFETY-02 and SAFETY-42 latches, so building
one more `Inverter` per roster entry against the post-migration store
replays EXACTLY the reads a restart performs - including any key this file
has never heard of. Writing out the key names here instead would be a second
transcription of `UNIT_STORE_KEYS`, and the copy that drifts is always the
one that reads (store.py, `unit_key()`).
The maintenance half iterates `MAINT_UNIT_KEYS`, which is the same tuple
`migrate_maint_keys()` moves and `maintenance.UnitHistory` indexes - so a
key renamed there cannot silently stop being probed here either.
"""
store.data.reads.clear()
for inv in controller.inverters:
Inverter(inv.name, store=store)
for m in controller.maints:
for key in MAINT_UNIT_KEYS:
# ⚠️ `get()` AND NOT ALSO `get_time()`. Both route through
# `store.data.get(self.key(name))`, so either one records the read
# this probe exists to record - but `get_time()` on `phase`, which
# holds "idle" and not a timestamp, makes `Store.get_time()` log
# "unparseable timestamp" at WARNING. A probe that manufactures
# alarming log lines about a healthy store teaches an operator to
# ignore this tool's stderr, which is where the controller's own
# real warnings appear.
m.store.get(key)
seen = set(store.data.reads)
store.data.reads.clear()
return seen
def snapshot(label: str, opts: dict, state_path: str, tmpdir: str) -> Snapshot:
store = RehearsalStore(_scratch_copy(state_path, tmpdir, label))
controller = Controller(dict(opts), Hass(), store, Pub(), stop=None)
return Snapshot(label, opts, store, controller)
def snapshot_over(label: str, opts: dict, prior: Snapshot) -> Snapshot:
"""A second start over the store the previous start LEFT BEHIND.
⚠️ THE SAME FILE, NOT A SECOND COPY OF THE ORIGINAL. This is the sequence a
real box performs: the add-on restarts against `/data/state.json` as its own
previous run wrote it.
"""
store = RehearsalStore(prior.store.path)
controller = Controller(dict(opts), Hass(), store, Pub(), stop=None)
return Snapshot(label, opts, store, controller)
# --------------------------------------------------------------------------- #
# 5. THE DIFF
# --------------------------------------------------------------------------- #
class Finding:
def __init__(self, kind, text):
self.kind = kind
self.text = text
def __str__(self):
return f"{self.kind}: {self.text}"
def diff(before: Snapshot, after: Snapshot) -> list:
"""Everything that changed, and which of those changes is a defect."""
findings = []
# -- store keys ---------------------------------------------------------
# ⚠️ THE SAME DEFAULT/NON-DEFAULT RULE THE STRANDED CHECK USES, AND FOR THE
# SAME REASON - `MigrationReport.lost`'s, in store.py: a dropped
# `sign_tripped = False` has lost the DEFAULT, which is the statement
# "absent" already makes, so nothing is gone; a dropped `True` has lost a
# LATCH. Applying it to STRANDED and not here was an inconsistency with a
# cost: a store whose flat start wrote the two SAFETY-02 defaults under one
# name and whose roster start writes them under another reported two
# VANISHED findings about a pair of defaults, and CONTROL-C went red on a
# build that was CORRECT.
# ⚠️ `claimed` MAKES THE MOVE INJECTIVE. Without it N vanished keys can all
# claim the SAME destination - and `DEFAULTS["sign_tripped"] is False`, so
# `True` is the ONLY non-default value a latch can hold, which means they
# collide by CONSTRUCTION rather than by coincidence. Two keys cannot both
# have become one key; the second claimant has not moved anywhere.
claimed = set()
for key in sorted(set(before.data) - set(after.data)):
value = before.data[key]
bare = key.rsplit("/", 1)[-1]
if value in (None, DEFAULTS.get(bare)):
after.vanished_at_default.append(f"{key} = {value!r}")
continue
# ⚠️ AND A KEY THAT MOVED TO A NAMESPACE A MACHINE READS HAS NOT
# VANISHED, WHICH IS THE SAME CORRECTION ONE PARAGRAPH UP, ONE CARD
# LATER. SAFETY-44's fix made `migrate_unit_keys()` write under the
# ROSTER's first name instead of the hardcoded `inverter-a`, so on a
# renamed roster the before-snapshot holds `unit/inverter-a/
# sign_tripped = True` and the after-snapshot holds
# `unit/garage/sign_tripped = True`. Measured on that build: two
# VANISHED findings for a LATCH THAT IS SITTING ON THE SAME PAGE OF THE
# REPORT, three lines above them, under the name the roster reads.
#
# ⚠️ THE TEST IS NOT "the name appears somewhere". It is: the SAME bare
# key, at the SAME value, under a key in `machine_read` - the set built
# by driving a real start, not by pattern-matching a namespace.
#
# ⚠️ THIS USED TO CITE "M3 (delete the flat rollback view)" AS WHAT
# BOUNDS IT, AND THAT CITATION WAS WRONG IN BOTH HALVES. M3 drops
# `sign_reason` from `LEGACY_UNIT_KEYS` and expects `LATCH KEY NOT UNDER
# A ROSTER NAME`; the delete-the-rollback-view mutant is M9, and every
# key M9 vanishes is FLAT, so it can never satisfy the `unit/` gate
# here. Neither mutation exercises this branch. The sentence "what keeps
# this from being a softening of the check" was not supported by the
# thing it named - and a citation is the worst possible carrier for that,
# because it reads as having been checked.
#
# ⚠️ AND THE REPLACEMENT CITATION WENT STALE THE SAME WAY. It read "the
# branch is now bounded by M12" through a card that DELETED M12 and a
# card that restored it on a different leg - eight lines below the
# paragraph condemning exactly this. Every clause of this comprehension
# is now graded by a row in test_rehearse.py part 5e, one case per
# clause, and each clause-removal must flip exactly its own row. That
# is a property of the section, not a pointer at a named check.
# ⚠️ AND THE DESTINATION MUST BE ONE **THIS TRANSITION MADE**
# (`before.data.get(k) != value`) AND NOT ALREADY SPOKEN FOR
# (`k not in claimed`). Without the first, a roster unit that was
# already latched in its own right is a free suppression slot that
# nobody moved into.
moved_to = sorted(
k for k in after.machine_read
if k.rsplit("/", 1)[-1] == bare and after.data.get(k) == value
and before.data.get(k) != value and k not in claimed)
# ⚠️ AND THE VANISHED KEY'S OWN UNIT MUST BE **OFF THE ROSTER**, WHICH
# THE FIRST VERSION OF THIS RULE DID NOT REQUIRE AND WHICH IS WHAT MADE
# IT A SOFTENING. The match was bare-key-plus-value over the whole of
# `machine_read`, with no identity - and `DEFAULTS["sign_tripped"] is
# False`, so the ONLY non-default value a latch can hold is `True`.
# Any surviving latched roster unit therefore suppressed any OTHER
# unit's vanished latch. Measured: a build that DROPS
# `unit/shed/sign_*` while `unit/garage/sign_*` survives, both latched
# by one site event so the reasons match too, reported
#
# none in LEG SAME-STEP: nothing vanished, nothing stranded ...
# NO FINDINGS in the rehearsed transitions
#
# exit 0, on a roster unit's SAFETY-02 latch that had been deleted.
#
# The legitimate case is a RENAME: the key left `inverter-a`, which is
# not on this roster, and arrived under `garage`, which is. A key whose
# own unit is still on the roster has not been renamed - it has been
# LOST - and no amount of the same value elsewhere makes that untrue.
vanished_unit = key.split("/")[1] if key.startswith("unit/") else None
if vanished_unit is not None and vanished_unit not in after.units \
and moved_to:
claimed.add(moved_to[0])
after.vanished_but_migrated.append(
f"{key} = {value!r} -> {moved_to[0]}")
continue
findings.append(Finding(
"VANISHED STORE KEY",
f"{key} = {value!r} was in the store before the "
f"transition and is not there afterwards"))
# ⭐ THE ONE SAFETY-44 IS ABOUT. A key that is present, correct, verified by
# the migration's own read-back - and that no machine on the new roster will
# ever look under. `MigrationReport.misdirected` says the same thing about
# the keys the migration itself wrote; this says it about the WHOLE FILE,
# including keys written by a version that is no longer installed.
#
# ⚠️ ONLY WHERE THE VALUE IS NOT THE DEFAULT, WHICH IS `MigrationReport`'s
# OWN DISTINCTION AND NOT A SOFTENING OF THIS ONE. A stranded
# `sign_tripped = False` has lost the DEFAULT - the same statement "absent"
# already makes - so nothing is gone; a stranded `sign_tripped = True` has
# lost A LATCH. Reporting the first would make this tool red on a FRESH
# INSTALL, whose migration writes both defaults under `inverter-a` before
# any roster exists, and an alarm that fires on the harmless case is an
# alarm an operator learns to dismiss before the real one arrives - which is
# the argument `MigrationReport.lost` already carries in store.py.
for key in sorted(after.data):
if not key.startswith("unit/") or key in after.machine_read:
continue
value = after.data[key]
bare = key.rsplit("/", 1)[-1]
if value in (None, DEFAULTS.get(bare)):
after.stranded_at_default.append(key)
continue
findings.append(Finding(
"STRANDED STORE KEY",
f"{key} = {value!r} is in the store and NO machine on "
f"the roster {after.units!r} read it during a full start. Its "
f"value is not information."))
# -- the SAFETY-02 latch, by name, because SAFETY-44 asks for it by name --
for key in ("sign_tripped", "sign_reason"):
findings.extend(_latch_finding(key, before, after))
if before.flat_tripped and not any(after.latched.values()):
findings.append(Finding(
"LATCH CAME UP DISARMED",
"the state file carries a LATCHED SAFETY-02 sign-disagreement "
"failsafe under the pre-DEV-02 flat key `sign_tripped`, and NOT "
f"ONE inverter on the roster {after.units!r} came up tripped. "
"SAFETY-02: a latch that a restart clears is not a latch."))
# -- maintenance history -------------------------------------------------
findings.extend(_history_findings(before, after))
# -- entity ids ----------------------------------------------------------
orphans = sorted(set(before.entities) - set(after.entities))
for eid in orphans:
findings.append(Finding(
"ORPHANED ENTITY",
f"{eid} (unique_id {before.entities[eid]}) is announced before the "
"transition and NOT afterwards. Home Assistant does not delete it: "
"its history stops, and every automation whose trigger names it "
"silently stops firing - without an error."))
return findings
def _latch_finding(key, before, after):
"""Did the two SAFETY-44 keys land somewhere a machine reads?
⚠️ THE FIRST ROSTER ENTRY, AND ONLY THE FIRST. The flat pair belongs to
whichever inverter the single-inverter install was driving, which is
`inverters[0]` - that is what `migrate_maint_keys()` targets for the history
and what `Controller._grade_store_migration()` re-arms. Asking whether the
flat latch reached inverter B's namespace would report a finding on a
CORRECT migration, and an instrument that is red on the right answer is
worth nothing when it is red on the wrong one.
"""
out = []
if not after.units or key not in UNIT_STORE_KEYS:
return out
unit = after.units[0]
target = unit_key(unit, key)
was = before.data.get(key)
if target not in after.data and was not in (None, DEFAULTS.get(key)):
out.append(Finding(
"LATCH KEY NOT UNDER A ROSTER NAME",
f"the flat `{key}` = {was!r} did not reach {target}, which is the "
f"key `Inverter({unit!r})` restores from"))
elif (target in after.data and target not in after.machine_read
and after.data[target] not in (None, DEFAULTS.get(key))):
out.append(Finding(
"LATCH KEY NOT READ BY ANY MACHINE",
f"{target} = {after.data[target]!r} exists but no machine on the "
f"roster {after.units!r} read it during a full start"))
return out
# ⚠️ THE REFERENCE IS GRADED AT A FIXED, CANONICAL SHAPE AND NOT AT THE
# OPERATOR'S CONFIG. The reference records what `release/1.0` announces at
# `meter_source: firmware`, roster of one. Grading it against the BEFORE config
# instead asks two questions at once and gets a wrong answer to both: a site
# with `meter_source: off` legitimately omits `p1_age` (see `discovery_payloads`
# and `is_enabled`), so the tool reported "the reference records
# sensor.goodwe_p1_sample_age and this build does not announce it" about a
# perfectly correct build. The question here is ONLY "has this build's
# composition drifted", which is a property of the CODE, so it is asked of the
# code at the shape the reference was taken at.
REFERENCE_PROBE_OPTS = {"meter_source": "firmware"}
def reference_findings(reference: "dict[str, Row]") -> list:
"""Does this BUILD still announce what the reference records?
⚠️ RUN UNCONDITIONALLY, AND OVER ALL THREE TERMS HA KEYS ON. The
before/after diff cannot see a change to the composition, because both of
its sides are composed by the same build. This is the only check in the
tool whose reference does not move with the code, so it is the only one
that can - and it is graded on the entity_id, the `unique_id` AND the
device identifier, because a change to either of the latter two orphans an
install while leaving every composed entity_id byte-identical.
"""
out = []
built = announced_rows(REFERENCE_PROBE_OPTS)
for eid in sorted(set(built) - set(reference)):
out.append(Finding(
"COMPOSITION DRIFT",
f"this build announces {eid} and the reference does not record "
f"it. The composed entity_id has changed: Home Assistant keeps the "
f"id it composed at FIRST discovery, so what this creates is a NEW "
f"entity beside one that stops updating."))
for eid in sorted(set(reference) - set(built)):
out.append(Finding(
"COMPOSITION DRIFT",
f"the reference records {eid} and this build does NOT announce it "
f"at the shape the reference was taken at "
f"({REFERENCE_PROBE_OPTS})."))
for eid in sorted(set(reference) & set(built)):
want, got = reference[eid], built[eid]
if want.unique_id != got.unique_id:
out.append(Finding(
"UNIQUE_ID DRIFT",
f"{eid} is announced with unique_id {got.unique_id!r} where "
f"the reference records {want.unique_id!r}. ⚠️ THE COMPOSED "
f"ENTITY_ID IS UNCHANGED, WHICH IS WHY THIS NEEDS ITS OWN "
f"CHECK: HA's MQTT entity registry keys on `unique_id`, so a "
f"changed one is a NEW entity - and discovery is retained with "
f"no removal payload, so the old config stays on the broker "
f"and the old entity stays in the registry, named by every "
f"automation that referenced it and never updating again."))
if want.device_id != got.device_id:
out.append(Finding(
"DEVICE IDENTIFIER DRIFT",
f"{eid} is announced on device {got.device_id!r} where the "
f"reference records {want.device_id!r}. HA's device registry "
f"keys on the identifier, so this is a NEW device carrying new "
f"entities, beside the old device and its dead ones."))
return out
def install_findings(installed: set, before: Snapshot,
after: Snapshot) -> list:
"""Grade the transition against the ids the INSTALL actually has."""
out = []
owned = installed & set(before.entities)
for eid in sorted(owned - set(after.entities)):
out.append(Finding(
"ORPHANED ENTITY (against the install's own entity list)",
f"{eid} exists on the install and is announced in the BEFORE "
"shape, and the AFTER shape does not announce it."))
return out
def report_entity_list(installed: set, before: Snapshot,
after: Snapshot) -> None:
announced_before = set(before.entities)
foreign = sorted(installed - announced_before)
print()
print(" -- AGAINST THE INSTALL'S OWN ENTITY LIST -------------------------")
print(f" {len(installed)} ids on the install, "
f"{len(installed & announced_before)} of them announced by this "
f"add-on in the BEFORE shape")
print(f" {len(foreign)} on the install that this add-on announces in "
"NEITHER shape - the roster switch cannot orphan these, they belong "
"to the")
print(" ESPHome firmware, to template sensors or to helpers:")
for eid in foreign[:60]:
print(f" {eid}")
if len(foreign) > 60:
print(f" ... and {len(foreign) - 60} more")
def _history_findings(before, after):
"""MAINT-05 moved this once. A second move must not drop it."""
out = []
if not before.history or not after.history:
return out
was = list(before.history.values())[0]
now = list(after.history.values())[0]
inheritor = list(after.history)[0]
for key in ("phase", "last_completed"):
old, new = was.get(key), now.get(key)
if old in (None, DEFAULTS.get(key)):
continue
if old != new:
out.append(Finding(
"MAINTENANCE HISTORY LOST",
f"`{key}` read {old!r} through the first pack's own history "
f"before the transition and reads {new!r} through "
f"{inheritor!r}'s afterwards"))
return out
# --------------------------------------------------------------------------- #
# 6. THE REPORT
# --------------------------------------------------------------------------- #
RETREAT_WINDOW_COMPLETION = "2026-09-20T18:00:00+00:00"
def _recalibrate_while_retreated(retreated: Snapshot) -> str:
"""A maintenance cycle finishes while the install is rolled back to flat.
⚠️ WRITTEN THROUGH THE SHIPPED WRITER. `Maintenance.store` is a
`maintenance.UnitHistory`, which is the one object that decides which key a
pack's history lands under and whether the flat rollback view is mirrored.
A raw `store.set("last_completed", ...)` here would be this file's own
opinion about where a completion goes, and the round trip would then be
grading that opinion instead of the add-on.
"""
store = RehearsalStore(retreated.store.path)
controller = Controller(dict(retreated.opts), Hass(), store, Pub(), stop=None)
controller.maints[0].store.set("last_completed", RETREAT_WINDOW_COMPLETION)
return RETREAT_WINDOW_COMPLETION
def _round_trip_findings(completed: str, readvanced: Snapshot) -> list:
"""Can the re-advanced roster still see work done during the retreat?"""
out = []
if not readvanced.history:
return out
inheritor = list(readvanced.history)[0]
seen = readvanced.history[inheritor].get("last_completed")
if seen != completed:
out.append(Finding(
"RETREAT-WINDOW WORK DISCARDED",
f"a maintenance cycle completed at {completed} while the install "
f"was rolled back to the flat config, and after re-advancing to "
f"the roster {inheritor!r} reads `last_completed` = {seen!r}. The "
"pack will be recalibrated again on a schedule that does not know "
"it already was."))
return out
def report_transition(leg_id, name, before, after, findings) -> None:
"""One leg of the rehearsal.
⚠️ `LEG <id>` IS A MACHINE-PARSEABLE MARKER AND IT IS LOAD-BEARING. Every
finding is printed twice - once here and once in the closing summary - so
COUNTING a finding's text across the whole report cannot tell ONE red leg
from TWO. `test_rehearse.py` part 5 was written that way and reported `ok`
on a build whose failure was a single leg. The pin now splits the report on
these markers and asks each leg separately.
"""
print()
print("=" * 78)
print(f" LEG {leg_id}")
print(f" {name}")
print("=" * 78)
print(f" roster before : {before.units}")
print(f" roster after : {after.units}")
print(f" DEV-02 migration : {after.store_migration.explain()}")
print(f" MAINT-05 migration: {after.maint_migration.explain()}")
print()
print(" -- STORE KEYS ----------------------------------------------------")
keys = sorted(set(before.data) | set(after.data))
width = max([len(k) for k in keys] + [10])
print(f" {'key'.ljust(width)} {'before':<26} {'after':<26} status")
for key in keys:
inb, ina = key in before.data, key in after.data
b = repr(before.data[key]) if inb else "-"
a = repr(after.data[key]) if ina else "-"
if inb and not ina:
status = "VANISHED"
elif ina and not inb:
status = "appeared"
elif before.data[key] != after.data[key]:
status = "changed"
else:
status = ""
if key.startswith("unit/") and ina and key not in after.machine_read:
status = (status + " " if status else "") + "STRANDED-nothing-reads-it"
print(f" {key.ljust(width)} {b[:26]:<26} {a[:26]:<26} {status}")
print()
print(" keys the AFTER roster's machines actually read, measured by "
"replaying")
print(" the shipped constructors against the post-transition store:")
for key in sorted(after.machine_read):
mark = " " if key in after.data else " (absent, defaulted) "
print(f" {key}{mark}")
if after.vanished_but_migrated:
print()
print(" keys that MOVED to the namespace this roster reads - not "
"lost, and shown")
print(" because a rename that relocates a LATCH is the operator's "
"business:")
for text in after.vanished_but_migrated:
print(f" migrated {text}")
if after.stranded_at_default or after.vanished_at_default:
print()
print(" keys whose value IS the default `Store.load()` supplies, so "
"nothing is lost")
print(" by stranding or dropping them - reported rather than "
"hidden, not findings:")
for key in after.stranded_at_default:
print(f" stranded {key} = {after.data[key]!r}")
for text in after.vanished_at_default:
print(f" vanished {text}")
print()
print(" -- MAINTENANCE HISTORY (MAINT-05: a second move must not drop it) -")
for label, hist in (("before", before.history), ("after", after.history)):
for unit, vals in hist.items():
print(f" {label:<7} {unit:<12} "
+ " ".join(f"{k}={vals.get(k)!r}" for k in MAINT_UNIT_KEYS))
print()
print(" -- SAFETY-02 LATCH (SAFETY-44 asks about these two keys by name) -")
print(f" flat `sign_tripped` in the file : {before.flat_tripped}")
print(f" flat `sign_reason` in the file : "
f"{before.data.get('sign_reason')!r}")
print(f" before: which units came up latched : "
f"{[u for u, v in before.latched.items() if v] or 'none'}")
print(f" after : which units came up latched : "
f"{[u for u, v in after.latched.items() if v] or 'none'}")
for unit in after.units:
target = unit_key(unit, "sign_tripped")
print(f" {target} = {after.data.get(target, '<absent>')!r} "
f"(read by a machine: {target in after.machine_read})")
print()
print(" -- ENTITY IDS ----------------------------------------------------")
allids = sorted(set(before.entities) | set(after.entities))
print(f" {'entity_id':<44} {'unique_id':<30} {'device':<26} state")
for eid in allids:
row = after.entities.get(eid) or before.entities[eid]
inb, ina = eid in before.entities, eid in after.entities
mark = "both" if (inb and ina) else ("ORPHANED" if inb else "new")
print(f" {eid:<44} {row.unique_id:<30} {row.device_id:<26} {mark}")
print(f" {len(before.entities)} announced before, "
f"{len(after.entities)} after")
print()
print(f" -- FINDINGS in LEG {leg_id} "
+ "-" * max(0, 50 - len(str(leg_id))))
if not findings:
print(f" none in LEG {leg_id}: nothing vanished, nothing stranded, "
"nothing orphaned, no history lost")
for f in findings:
print(f" ⚠️ {f}")
# ⚠️ AN EXPLICIT CLOSE, AND IT IS NOT TIDINESS. Without it a reader that
# splits on the opening markers alone gives every line after the LAST one
# to that leg - including the closing summary, which repeats every finding
# from EVERY leg. The last leg then reads red whenever any leg is red, and
# a per-leg check built on that is the same defect it was written to fix,
# one level down. Measured: with only the opening marker, a build whose
# SAME-STEP leg was clean still reported SAME-STEP disarmed.
print(f" END LEG {leg_id}")
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--state", required=True,
help="path to a COPY of /data/state.json. Never written to.")
ap.add_argument("--before", required=True,
help="the options in force now (flat single-inverter)")
ap.add_argument("--after", required=True,
help="the options being switched to (declared roster)")
ap.add_argument("--reverse", action="store_true",
help="also rehearse the RETREAT: roster -> flat")
ap.add_argument("--round-trip", action="store_true",
help="also rehearse RETREAT then RE-ADVANCE, with a "
"maintenance cycle completing while retreated. ⚠️ A "
"retreat is only a retreat if you can come back from "
"it, so the round trip is the leg that decides "
"whether the reverse direction is supported.")
ap.add_argument("--reference", default=str(DEFAULT_REFERENCE),
help="what this build is SUPPOSED to announce: "
"`<entity_id> <unique_id> <device identifier>` per "
"line. Loaded on EVERY run; a missing or malformed "
"one is exit 2, because it is the only reference in "
"this tool that does not move with the code.")
ap.add_argument("--live-entities", default=None,
help="a file of the entity ids the INSTALL actually has, "
"one per line - the owner's export from Home "
"Assistant. Orphans are graded against it.")
ap.add_argument("--no-live-entities", action="store_true",
help="⚠️ RUN WITHOUT THE INSTALL'S ENTITY LIST, ACCEPTING "
"THAT THE ORPHAN CHECK IS NOT GRADED AGAINST "
"REALITY. Required to omit --live-entities; the "
"report then carries a standing banner and the exit "
"line says so, because a forgotten flag must never "
"look like a clean bill of health.")
ap.add_argument("--paths", default="both", choices=("staged", "same-step",
"both"),
help="staged: the box runs the BEFORE config once, then "
"restarts on AFTER. same-step: the roster is declared "
"in the SAME step as the upgrade, so the BEFORE "
"controller never runs against this store. "
"Default: both.")
args = ap.parse_args(argv)
# ⚠️ THE ASYMMETRY THAT USED TO BE HERE WAS BACKWARDS AND IT IS THE WHOLE
# REASON THIS BLOCK EXISTS. A `--live-entities` file that could not be READ
# refused with exit 2; the flag simply not being PASSED produced "NO
# FINDINGS" and exit 0, with nothing in the report to say half the orphan
# check had not run - measured, `grep -ci` for the words returned 0. The
# owner runs this ONCE, at commissioning, under time pressure. A forgotten
# flag must not hand back a clean bill of health for the half of the
# transition that orphans entities. So omitting it is now a REFUSAL, and
# accepting the weaker run has to be said out loud.
try:
if not args.live_entities and not args.no_live_entities:
raise Refused(
"no --live-entities. The orphan check is graded against the "
"ids the INSTALL has, and without that file it is not graded "
"against reality at all. Export it (DOCS.md, 'Before you "
"commission inverter B'), or pass --no-live-entities to "
"accept the weaker run - which is then stated in the report "
"and on the exit line rather than left to be assumed.")
if args.live_entities and args.no_live_entities:
raise Refused("--live-entities and --no-live-entities are "
"contradictory; pass one.")
_check_anchor()
reference = load_reference(args.reference)
before_opts = load_options(args.before)
after_opts = load_options(args.after)
installed = (load_entity_list(args.live_entities)
if args.live_entities else None)
if not Path(args.state).is_file():
raise Refused(f"{args.state}: not a file")
load_state(args.state)
except (OSError, Refused) as err:
print(f"REFUSED: {err}", file=sys.stderr)
return EXIT_COULD_NOT_RUN
degraded = installed is None
print(__doc__.split("EXIT CODES")[0].rstrip())
print()
print(f" state file (copied, never written): {args.state}")
print(f" before config: {args.before}")
print(f" after config: {args.after}")
print(f" reference : {args.reference} ({len(reference)} rows, graded "
f"on entity_id + unique_id + device identifier)")
if degraded:
print()
print(" " + "!" * 74)
print(" !! DEGRADED RUN: --no-live-entities. The ORPHAN CHECK IS NOT "
"GRADED AGAINST")
print(" !! THE INSTALL. This run can still see the composition "
"drifting away from")
print(" !! the reference, and it can see the store migration - but it "
"cannot tell")
print(" !! you that an entity the install HAS stops being announced, "
"because it")
print(" !! has not been told what the install has. ⚠️ A CLEAN RESULT "
"BELOW IS NOT A")
print(" !! CLEAN BILL OF HEALTH FOR THE ENTITIES.")
print(" " + "!" * 74)
else:
print(f" install list : {args.live_entities} ({len(installed)} ids)")
findings = []
tmpdir = tempfile.mkdtemp(prefix="rehearse-roster-")
try:
if args.paths in ("staged", "both"):
b = snapshot("staged-before", before_opts, args.state, tmpdir)
a = snapshot_over("staged-after", after_opts, b)
f = diff(b, a) + reference_findings(reference)
if installed is not None:
f += install_findings(installed, b, a)
report_transition(
"STAGED",
"the box runs the flat config, then restarts on the roster",
b, a, f)
if installed is not None:
report_entity_list(installed, b, a)
findings += f
if args.paths in ("same-step", "both"):
# ⚠️ SAFETY-44's EXACT PATH. A store that ran pre-DEV-02 flat,
# upgraded straight to 0.7.0+, AND declared a multi-inverter roster
# in the same step - so nothing ever ran the flat config under a
# version that knew about namespaces.
b2 = snapshot("samestep-before", before_opts, args.state, tmpdir)
a2 = snapshot("samestep-after", after_opts, args.state, tmpdir)
f2 = diff(b2, a2) + reference_findings(reference)
if installed is not None:
f2 += install_findings(installed, b2, a2)
report_transition(
"SAME-STEP",
"SAFETY-44's path: a pre-DEV-02 flat store meets a declared "
"roster on its FIRST start under this version",
b2, a2, f2)
findings += f2
if args.reverse or args.round_trip:
r1 = snapshot("retreat-before", after_opts, args.state, tmpdir)
r2 = snapshot_over("retreat-after", before_opts, r1)
fr = diff(r1, r2)
if installed is not None:
fr += install_findings(installed, r1, r2)
report_transition(
"RETREAT",
"roster -> flat, which is the back-out if commissioning B "
"goes wrong", r1, r2, fr)
findings += fr
if args.round_trip:
# ⚠️ THE RETREAT WINDOW HAS TO CONTAIN WORK, OR THIS LEG IS HABIT 4.
# A round trip over a store that nothing changed in between compares
# a structure to itself and reports "clean" whatever the code does.
# So a maintenance cycle COMPLETES while the install is retreated -
# written through `maintenance.UnitHistory`, which is the shipped
# writer, not through a raw `store.set()` - and the question this
# leg asks is whether the pack can still see it after re-advancing.
q1 = snapshot("roundtrip-roster", after_opts, args.state, tmpdir)
q2 = snapshot_over("roundtrip-retreated", before_opts, q1)
worked = _recalibrate_while_retreated(q2)
q3 = snapshot_over("roundtrip-readvanced", after_opts, q2)
fq = diff(q2, q3) + _round_trip_findings(worked, q3)
report_transition(
"ROUND-TRIP",
f"roster -> flat -> roster, with a maintenance cycle "
f"completing at {worked!r} while retreated",
q2, q3, fq)
findings += fq
except Refused as err:
print(f"REFUSED: {err}", file=sys.stderr)
return EXIT_COULD_NOT_RUN
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
print()
print("=" * 78)
if findings:
print(f" {len(findings)} FINDING(S) - this transition is not clean")
for f in findings:
print(f" ⚠️ {f}")
else:
print(" NO FINDINGS in the rehearsed transitions")
# ⚠️ ON THE EXIT LINE, NOT ONLY IN THE HEADER. A report is skimmed from the
# bottom.
if degraded:
print(" ⚠️ DEGRADED RUN (--no-live-entities): the orphan check was "
"NOT graded against")
print(" the install's own entity list. This result does not cover "
"orphaned entities.")
print("=" * 78)
if findings:
return EXIT_FINDING
return EXIT_DEGRADED if degraded else EXIT_OK
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink