test_coordinator.py

# fmt: off
# pylint: disable=line-too-long
"""Unit tests for the discovery catalogue coordinator (coordinator.py) - container validation and report collection."""

import inspect
from datetime import datetime

from mock_base import MockBase
from coordinator import CAPABILITY_KEYS, CONTAINER_SPEC, Coordinator, NOT_APPLICABLE_DEFAULTS, Redactor, SCHEMA_VERSION, SECTION_SPEC, VOCAB_CONTAINERS, inverter_definition, inverter_record, validate_report
from config import INVERTER_DEF


def _coordinator():
    """A Coordinator on a bare MockBase."""
    base = MockBase()
    return base, Coordinator(base)


def test_report_keeps_valid_containers():
    """A well-formed inverter record survives validation intact."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{
        "device_id": "givtcp:SN1",
        "inverter_type": "GE",
        "functions": ["solar", "battery"],
        "hardware_ids": {"serial": "SN1"},
        "info": {"firmware": "D0.451"},
        "ratings": {"soc_max": 9.5, "battery_rate_max": 3600},
        "entities": {"soc_kw": {"entity_id": "sensor.predbat_givtcp_0_soc_kw", "domain": "sensor", "access": "r", "unit": "kWh"}},
    }]})
    record = coordinator.reports["givtcp"]["inverters"][0]
    assert record["hardware_ids"] == {"serial": "SN1"}
    assert record["ratings"]["soc_max"] == 9.5
    assert record["functions"] == ["solar", "battery"]
    assert record["entities"]["soc_kw"]["entity_id"] == "sensor.predbat_givtcp_0_soc_kw"
    print("PASS: valid containers preserved")
    return 0


def test_ratings_rejects_non_numeric():
    """A numbers-only container drops a string value rather than emitting it - the structural safety guarantee."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:SN1", "ratings": {"battery_kwh": 9.5, "owner": "Bob Smith, 12 Acacia Ave"}}]})
    ratings = coordinator.reports["givtcp"]["inverters"][0]["ratings"]
    assert ratings == {"battery_kwh": 9.5}, ratings
    print("PASS: non-numeric dropped from ratings")
    return 0


def test_vocabulary_rejects_free_text():
    """Vocabulary lists take short lowercase tokens only, so free text cannot ride in on one."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:SN1", "functions": ["battery", "My Home Battery!", "SOLAR"]}]})
    assert coordinator.reports["givtcp"]["inverters"][0]["functions"] == ["battery"]
    print("PASS: vocabulary constraint enforced")
    return 0


def test_info_rejects_email_shaped_and_long_strings():
    """info takes vendor descriptors, not addresses - an @ or an over-long string is dropped."""
    base, coordinator = _coordinator()
    coordinator.report("ohme", {"chargers": [{"device_id": "ohme:CH1", "info": {"vendor": "Ohme", "login": "bob@example.com", "notes": "x" * 200}}]})
    assert coordinator.reports["ohme"]["chargers"][0]["info"] == {"vendor": "Ohme"}
    print("PASS: info rejects email-shaped and over-long values")
    return 0


def test_credential_named_field_rejected():
    """A field whose NAME trips is_secret_key is refused in any container, however it is nested."""
    base, coordinator = _coordinator()
    coordinator.report("fox", {"inverters": [{"device_id": "fox:SN1", "info": {"model": "H3", "api_key": "abcd1234", "auth_token": "zzz"}}]})
    assert coordinator.reports["fox"]["inverters"][0]["info"] == {"model": "H3"}
    print("PASS: credential-named fields refused")
    return 0


def test_entities_credential_named_key_rejected():
    """An entities descriptor KEYED by a credential-shaped name is refused too - the guard covers every container, not just info."""
    base, coordinator = _coordinator()
    coordinator.report("fox", {"inverters": [{
        "device_id": "fox:SN1",
        "entities": {
            "soc_kw": {"entity_id": "sensor.predbat_fox_0_soc_kw", "domain": "sensor", "access": "r"},
            "api_key": {"entity_id": "sensor.predbat_fox_0_api_key", "domain": "sensor", "access": "r"},
        },
    }]})
    entities = coordinator.reports["fox"]["inverters"][0]["entities"]
    assert "soc_kw" in entities
    assert "api_key" not in entities
    print("PASS: entities key named for a credential refused")
    return 0


def test_entities_free_text_unit_dropped_legitimate_descriptor_intact():
    """A free-text value in a typed descriptor field is dropped, while a fully legitimate descriptor survives with every field intact."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{
        "device_id": "givtcp:SN1",
        "entities": {
            "charge_rate": {
                "entity_id": "number.predbat_givtcp_0_charge_rate",
                "domain": "number", "access": "rw", "unit": "W",
                "device_class": "power", "min": 0, "max": 3600, "step": 100, "precision": 0,
            },
            "soc_kw": {
                "entity_id": "sensor.predbat_givtcp_0_soc_kw",
                "domain": "sensor", "access": "r", "unit": "not a real unit " * 6,
            },
        },
    }]})
    entities = coordinator.reports["givtcp"]["inverters"][0]["entities"]
    assert entities["charge_rate"] == {
        "entity_id": "number.predbat_givtcp_0_charge_rate",
        "domain": "number", "access": "rw", "unit": "W",
        "device_class": "power", "min": 0, "max": 3600, "step": 100, "precision": 0,
    }, entities["charge_rate"]
    assert entities["soc_kw"]["entity_id"] == "sensor.predbat_givtcp_0_soc_kw"
    assert "unit" not in entities["soc_kw"]
    print("PASS: free-text descriptor field dropped, legitimate descriptor intact")
    return 0


def test_entities_options_preserves_realistic_values():
    """options takes bounded strings, not vocabulary tokens - realistic select values like time strings and mixed-case labels survive."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{
        "device_id": "givtcp:SN1",
        "entities": {
            "charge_start_time": {"entity_id": "select.predbat_givtcp_0_charge_start_time", "domain": "select", "access": "rw", "options": ["00:00", "00:30"]},
            "pause_mode": {"entity_id": "select.predbat_givtcp_0_pause_mode", "domain": "select", "access": "rw", "options": ["Disabled", "PauseCharge"]},
        },
    }]})
    entities = coordinator.reports["givtcp"]["inverters"][0]["entities"]
    assert entities["charge_start_time"]["options"] == ["00:00", "00:30"], entities["charge_start_time"]
    assert entities["pause_mode"]["options"] == ["Disabled", "PauseCharge"], entities["pause_mode"]
    print("PASS: options preserves realistic time and label values")
    return 0


def test_coverage_accepts_numbers_booleans_and_vocabulary_lists():
    """coverage accepts numbers, booleans and lists of vocabulary tokens - Solcast's pv10/pv50/pv90 variants must survive whole."""
    base, coordinator = _coordinator()
    coordinator.report("solcast", {"forecasts": [{"device_id": "solcast:site1", "kind": "solar", "coverage": {"horizon_hours": 168, "variants": ["pv10", "pv50"]}}]})
    coverage = coordinator.reports["solcast"]["forecasts"][0]["coverage"]
    assert coverage == {"horizon_hours": 168, "variants": ["pv10", "pv50"]}, coverage
    print("PASS: coverage accepts numbers and vocabulary-token lists")
    return 0


def test_validate_report_never_raises_on_non_dict():
    """A non-dict report (None, a list, ...) never raises - it is treated as empty, per the 'never
    raises' contract. "automatic" is absent, not defaulted to True: an empty/broken report carries
    no evidence the component has any such concept, matching validate_report()'s "only carry
    'automatic' through when the raw report actually provided one" rule."""
    base, coordinator = _coordinator()
    coordinator.report("broken", None)
    assert coordinator.reports["broken"] == {"schema_version": SCHEMA_VERSION}
    coordinator.report("broken2", ["not", "a", "dict"])
    assert coordinator.reports["broken2"] == {"schema_version": SCHEMA_VERSION}
    print("PASS: non-dict report never raises")
    return 0


def test_validate_report_never_raises_on_a_non_list_section():
    """Review round: a section that is not a list of records is dropped whole, never iterated.

    The "never raises" contract has to hold for a malformed SECTION exactly as it already does for
    a malformed report: a scalar ({"inverters": 1}) is not iterable at all, so it raised TypeError
    straight out of report(), and a bare string would have iterated as its characters - one log
    line per character, for a report that was simply the wrong shape.
    """
    base, coordinator = _coordinator()
    coordinator.report("broken", {"inverters": 1, "meters": "octopus:1", "cars": {"device_id": "x"}})
    assert coordinator.reports["broken"] == {"schema_version": SCHEMA_VERSION}, coordinator.reports["broken"]

    # A valid section alongside a malformed one still survives - the drop is per section, not per report
    coordinator.report("mixed", {"inverters": 1, "meters": [{"device_id": "octopus:m1", "direction": "import"}]})
    assert "inverters" not in coordinator.reports["mixed"]
    assert coordinator.reports["mixed"]["meters"][0]["device_id"] == "octopus:m1"
    print("PASS: a non-list section is dropped rather than iterated")
    return 0


def test_automatic_field_absent_when_component_omits_it():
    """A component with no automatic-config concept (Solcast: solar-forecast sourcing is a plain
    apps.yaml choice, never something this catalogue auto-wires) omits "automatic" from its report
    entirely, and neither the cleaned report nor the assembled catalogue's component status may
    default it to True on its behalf - that would claim an auto-config relationship that does not
    exist. A component that DOES report it (every other v1 reporter) keeps whatever value it sent,
    explicit False included."""
    base, coordinator = _coordinator()
    coordinator.report("solar", {"forecasts": [{"device_id": "solcast:site1", "kind": "solar"}]})
    coordinator.report("ohme", {"chargers": [{"device_id": "ohme:CH1"}], "automatic": False})
    assert "automatic" not in coordinator.reports["solar"], coordinator.reports["solar"]
    assert coordinator.reports["ohme"]["automatic"] is False
    components = coordinator.assemble()["components"]
    assert "automatic" not in components["solar"], components["solar"]
    assert components["ohme"]["automatic"] is False
    print("PASS: automatic is absent when a component's report never provided it, and preserved otherwise")
    return 0


def test_unknown_container_dropped():
    """A container this schema does not know is dropped - there is no untyped path into the catalogue."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:SN1", "scratchpad": {"anything": "at all"}}]})
    assert "scratchpad" not in coordinator.reports["givtcp"]["inverters"][0]
    print("PASS: unknown container dropped")
    return 0


def test_record_without_device_id_dropped():
    """device_id is the one genuinely required field; a record without it cannot be referred to, so it is dropped."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"hardware_ids": {"serial": "SN1"}}, {"device_id": "givtcp:SN2"}]})
    records = coordinator.reports["givtcp"]["inverters"]
    assert len(records) == 1 and records[0]["device_id"] == "givtcp:SN2"
    print("PASS: record without device_id dropped")
    return 0


def test_report_is_idempotent_and_versioned():
    """Re-reporting replaces the prior report, and the schema version is stamped."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A"}]})
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A"}, {"device_id": "givtcp:B"}]})
    assert len(coordinator.reports["givtcp"]["inverters"]) == 2
    assert coordinator.reports["givtcp"]["schema_version"] == SCHEMA_VERSION
    print("PASS: report replaces and is versioned")
    return 0


def test_meter_sub_record_validated():
    """A meter's nested tariff record is validated with the same container rules."""
    base, coordinator = _coordinator()
    coordinator.report("octopus", {"meters": [{
        "device_id": "octopus:m1", "direction": "import",
        "account_ids": {"mpan": "1234567890123"},
        "tariff": {"info": {"tariff_code": "E-1R-AGILE-24-10-01-A"}, "flags": ["agile"], "ratings": {"standing_charge_p": 47.5}, "secret_token": "nope"},
    }]})
    tariff = coordinator.reports["octopus"]["meters"][0]["tariff"]
    assert tariff["info"]["tariff_code"] == "E-1R-AGILE-24-10-01-A"
    assert tariff["flags"] == ["agile"]
    assert "secret_token" not in tariff
    print("PASS: nested tariff record validated")
    return 0


def test_credential_guard_fires_inside_sub_record_container():
    """The credential guard is shared code, so it fires identically for a key nested inside a sub-record's container (tariff.info), not just at record level."""
    base, coordinator = _coordinator()
    coordinator.report("octopus", {"meters": [{
        "device_id": "octopus:m1", "direction": "import",
        "tariff": {"info": {"tariff_code": "E-1R-AGILE-24-10-01-A", "api_key": "abcd1234"}},
    }]})
    tariff_info = coordinator.reports["octopus"]["meters"][0]["tariff"]["info"]
    assert tariff_info == {"tariff_code": "E-1R-AGILE-24-10-01-A"}, tariff_info
    print("PASS: credential guard fires inside a sub-record container")
    return 0


class _StubRegistry:
    """Stands in for Components so assemble() can derive a status for every registry entry, and load_salt() can find a stub Storage."""

    def __init__(self, active=(), alive=(), errors=None, all_names=(), components=None):
        """Record which component names are active, alive, failed to load, known at all, and any stub components (e.g. storage) registered by name."""
        self._active = set(active)
        self._alive = set(alive)
        self._errors = errors or {}
        self._all = list(all_names)
        self._components = components or {}

    def get_all(self):
        """Every component name the registry knows."""
        return list(self._all)

    def is_active(self, name):
        """Whether the component was constructed."""
        return name in self._active

    def is_alive(self, name):
        """Whether the component is running and fresh."""
        return name in self._alive

    def load_error(self, name):
        """Why the component failed to construct, or None."""
        return self._errors.get(name)

    def get_component(self, name):
        """The stub component registered under this name, or None."""
        return self._components.get(name)


def test_inverter_record_omits_empty_containers():
    """An empty or unset container is left out of the record rather than written as {} or [].

    Every reporter previously wrote its own `if info: record["info"] = info` ladder. Centralising
    it means a reporter can pass whatever it gathered and let the builder decide, and the shape of
    an inverters record lives in exactly one place.
    """
    record = inverter_record("fox:ABC123", inverter_type="FoxCloud", composition="direct", functions=["solar", "battery"], info={}, ratings={}, entities={}, serials=[])
    assert record == {"device_id": "fox:ABC123", "inverter_type": "FoxCloud", "composition": "direct", "functions": ["solar", "battery"]}, record
    print("PASS: inverter_record omits empty containers")
    return 0


def test_inverter_record_keeps_everything_populated():
    """Every populated field survives, including a falsy-but-real rating like 0 and control=False.

    control is the one top-level field where a falsy value is itself the fact: False means a
    monitor-only inverter. A drop rule "simplified" to `if not value: continue` would silently
    turn every monitor-only inverter into one whose control is unknown, so it is pinned here.
    """
    record = inverter_record(
        "fox:ABC123",
        inverter_type="FoxCloud",
        control=False,
        composition="gateway",
        serials=["S1", "S2"],
        measures_meter="fox:meter:M1",
        functions=["solar"],
        capabilities={"support_charge_freeze": True},
        hardware_ids={"serial": "ABC123"},
        info={"model": "H3-10.0"},
        ratings={"battery_kwh": 10.4, "max_charge_w": 0},
        entities={"charge_rate": {"entity_id": "number.fox_abc123_charge_rate", "domain": "number", "access": "rw"}},
    )
    assert record["serials"] == ["S1", "S2"]
    assert record["measures_meter"] == "fox:meter:M1"
    assert record["capabilities"] == {"support_charge_freeze": True}
    assert record["hardware_ids"] == {"serial": "ABC123"}
    assert record["ratings"]["max_charge_w"] == 0, "a real zero rating is data, not an empty container"
    assert record["entities"]["charge_rate"]["domain"] == "number"
    assert record["control"] is False, "control=False is a fact (monitor-only), not an empty value: {}".format(record)
    print("PASS: inverter_record keeps every populated field")
    return 0


def test_inverter_record_round_trips_through_validation():
    """A record the builder produced survives validate_report() unchanged.

    The builder's whole job is producing something the coordinator will accept. If a field name
    here ever drifts from SECTION_SPEC's structural list, validation silently drops it - so pin
    that the two agree rather than trusting they do.
    """
    record = inverter_record("fox:ABC123", inverter_type="FoxCloud", composition="direct", serials=["S1"], measures_meter="fox:meter:M1", functions=["solar"], hardware_ids={"serial": "ABC123"}, info={"model": "H3"}, ratings={"battery_kwh": 10.4})
    cleaned = validate_report({"inverters": [record]}, "fox", print)["inverters"][0]
    for field in ("device_id", "inverter_type", "composition", "serials", "measures_meter", "functions", "hardware_ids", "info", "ratings"):
        assert field in cleaned, "{} was dropped by validate_report - builder and SECTION_SPEC disagree: {}".format(field, cleaned)
    assert cleaned == record, "validate_report() changed a built record:\n  built:   {}\n  cleaned: {}".format(record, cleaned)
    print("PASS: a built record survives validation with every field intact")
    return 0


def test_inverter_record_parameters_match_the_inverters_schema():
    """inverter_record()'s parameters are exactly the inverters section's fields - no more, no fewer.

    The builder spells the schema out as named parameters so a mistyped field is a TypeError at
    the call site. That only holds while its parameter list and the coordinator's schema agree:
    rename a container in CONTAINER_SPEC, or add a structural field to SECTION_SPEC, and the builder
    would go on emitting the old name, which validate_report() then silently drops from every
    reporter's records at once. This passes today; its job is to fail the day the two drift.
    """
    parameters = set(inspect.signature(inverter_record).parameters)
    assert "device_id" in parameters, "device_id is the record's identity and must stay a parameter: {}".format(sorted(parameters))
    schema = (set(SECTION_SPEC["inverters"]["structural"]) - {"device_id"}) | set(CONTAINER_SPEC) | set(VOCAB_CONTAINERS)
    fields = parameters - {"device_id"}
    assert fields == schema, "inverter_record() and the inverters schema disagree - missing from the builder: {}, not in the schema: {}".format(sorted(schema - fields), sorted(fields - schema))
    print("PASS: inverter_record's parameters are exactly the inverters schema")
    return 0


def test_inverter_record_fields_are_keyword_only():
    """Only device_id may be passed positionally; every other field must be named.

    Fifteen optional parameters in a row are an easy place to slip by one: accepted positionally,
    inverter_record("x:1", "direct", True) would quietly build inverter_type="direct",
    control=True - a wrong record with nothing to say so.
    """
    try:
        inverter_record("gecloud:x", "direct", True)
    except TypeError:
        pass
    else:
        raise AssertionError("a positional second argument must be a TypeError, not silently taken as inverter_type")
    parameters = inspect.signature(inverter_record).parameters
    positional = [name for name, parameter in parameters.items() if parameter.kind is not inspect.Parameter.KEYWORD_ONLY]
    assert positional == ["device_id"], "only device_id may be positional, got {}".format(positional)
    print("PASS: inverter_record's schema fields are keyword-only")
    return 0


def test_inverter_record_copies_the_callers_containers():
    """The record holds its own copies of the containers, never the caller's live objects.

    refresh_discovery() stores the report it filed and compares the next build against it with ==.
    If the record held the caller's own list - a reporter passing serials=self.serials - then
    growing that list would grow the STORED report too, the rebuilt report would compare equal to
    it, and the change would never be filed: a report frozen at its first state, the exact bug
    refresh_discovery() exists to prevent.
    """
    live_serials = ["S1"]
    live_info = {"model": "GIV-GATEWAY"}
    first = inverter_record("gecloud:gateway001", composition="gateway", serials=live_serials, info=live_info)

    live_serials.append("S2")
    live_info["firmware"] = "ARM 1"
    second = inverter_record("gecloud:gateway001", composition="gateway", serials=live_serials, info=live_info)

    assert first != second, "a rebuilt record must differ once the caller's live state has moved on: {}".format(first)
    assert first["serials"] == ["S1"], "the first record's serials changed under it: {}".format(first["serials"])
    assert first["info"] == {"model": "GIV-GATEWAY"}, "the first record's info changed under it: {}".format(first["info"])
    print("PASS: inverter_record copies the caller's containers, so live state cannot freeze a report")
    return 0


def test_inverter_record_normalises_sets_and_tuples():
    """A set or frozenset becomes a sorted list, a tuple becomes a list, and an empty frozenset is omitted.

    validate_report() keeps only a list for a structural or vocabulary field, so anything else
    would be silently dropped from the catalogue. A set is sorted rather than listed as-is: string
    hashing is randomised per process, so list(some_set) comes out in a different order after a
    restart and two dumps of the same hardware would diff for no reason.
    """
    record = inverter_record(
        "gecloud:gateway001",
        serials={"S5", "S3", "S1", "S4", "S2"},
        functions=frozenset({"solar", "battery"}),
        capabilities=frozenset(),
        flags=("monitor_only",),
    )
    assert record["serials"] == ["S1", "S2", "S3", "S4", "S5"], "a set should become a sorted list: {}".format(record.get("serials"))
    assert record["functions"] == ["battery", "solar"], "a frozenset should become a sorted list: {}".format(record.get("functions"))
    assert "capabilities" not in record, "an empty frozenset is empty and must be omitted: {}".format(record)
    assert record["flags"] == ["monitor_only"], "a tuple should become a list: {}".format(record.get("flags"))
    assert "capabilities" not in inverter_record("gecloud:gateway001", capabilities={}), "an empty capabilities dict must be omitted too"
    print("PASS: inverter_record sorts sets, lists tuples and omits an empty frozenset")
    return 0


def test_inverter_record_sorts_a_mixed_type_set_without_raising():
    """A set mixing types that cannot be compared is sorted by string form rather than raising.

    A bare sorted() raises TypeError on {1, "a"}. build_discovery() is called from
    refresh_discovery(), which catches it - so the component survives, but its whole report is
    withheld, on every cycle, because nothing changes the offending data. Sorting by str() makes
    every element comparable, and leaves the order of an all-string set exactly as before.
    """
    record = inverter_record("x:1", serials={1, "a"})
    assert record["serials"] == [1, "a"], "a mixed-type set should sort by string form: {}".format(record.get("serials"))
    print("PASS: inverter_record sorts a mixed-type set by string form instead of raising")
    return 0


def test_inverter_record_drops_an_empty_string():
    """An empty string is unset, like None - but control=False is a real value and is kept."""
    record = inverter_record("fox:ABC123", inverter_type="", measures_meter="", composition="direct", control=False)
    assert "inverter_type" not in record, "an empty inverter_type is unset and must be omitted: {}".format(record)
    assert "measures_meter" not in record, "an empty measures_meter is unset and must be omitted: {}".format(record)
    assert record["composition"] == "direct"
    assert record["control"] is False, "control=False is a bool, not an empty string, and must be kept: {}".format(record)
    print("PASS: inverter_record drops an empty string but keeps control=False")
    return 0


def test_assemble_merges_sections_and_tags_source():
    """Records from several components merge into one list per section, each tagged with its source."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A", "inverter_type": "GE"}]})
    coordinator.report("gecloud", {"inverters": [{"device_id": "gecloud:b", "inverter_type": "GEC"}]})
    catalogue = coordinator.assemble()
    sources = sorted(record["source"] for record in catalogue["inverters"])
    assert sources == ["gecloud", "givtcp"], sources
    assert catalogue["schema_version"] == SCHEMA_VERSION
    assert catalogue["generated"]
    print("PASS: sections merged and tagged with source")
    return 0


def test_catalogue_reassembles_so_a_post_assembly_report_reaches_it():
    """Final review, Ruling R42: catalogue() used to return `self.assembled or self.assemble()`,
    and assemble() was only ever called once, at the startup barrier - so a report filed AFTER
    that point (a retry, a rediscovered device, a changed tariff or vehicle, a later-discovered
    forecast site) could never reach a dump, no matter how many times catalogue() was called
    afterwards. catalogue() must re-assemble on every call so later work is always reflected."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A", "inverter_type": "GE"}]})
    coordinator.assemble()
    first = coordinator.catalogue()
    assert len(first["inverters"]) == 1, first["inverters"]

    # A second inverter is reported AFTER assemble() has already run once - simulating GivTCP's own
    # rediscovery, GE Cloud's retry, or any report arriving after the startup barrier.
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A", "inverter_type": "GE"}, {"device_id": "givtcp:B", "inverter_type": "GE"}]})
    second = coordinator.catalogue()
    assert len(second["inverters"]) == 2, "a report filed after assemble() already ran must still reach a later catalogue() call: {}".format(second["inverters"])

    # catalogue_raw() must behave identically - see its own docstring.
    assert len(coordinator.catalogue_raw()["inverters"]) == 2
    print("PASS: catalogue() re-assembles on every call, so a post-assembly report reaches it")
    return 0


def test_assemble_component_status():
    """A configured component gets a status distinguishing silent from timed out from failed.

    A component the user never configured is absent entirely rather than carrying a status: the
    registry holds 30 entries and a handful can report, so listing them all restated a static list
    from components.py on every dump. Absence from the map is its status.
    """
    base, coordinator = _coordinator()
    base.components = _StubRegistry(active=["givtcp", "octopus", "gecloud"], alive=["givtcp", "octopus"],
                                    errors={"fox": "No module named 'protobuf'"},
                                    all_names=["givtcp", "octopus", "gecloud", "fox", "solis"])
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A"}]})
    components = coordinator.assemble()["components"]
    assert components["givtcp"]["status"] == "ok"
    assert components["octopus"]["status"] == "no_report"      # alive, capable of reporting, silent
    assert components["gecloud"]["status"] == "not_started"    # active but not alive
    assert components["fox"]["status"] == "load_error"
    assert "solis" not in components, "a component that was never configured must not appear at all"
    print("PASS: component statuses derived, unconfigured components omitted")
    return 0


class _SilentComponent:
    """A component with no build_discovery() - working as intended, never going to report."""


class _ReportingComponent:
    """A component that carries a reporter, so silence from it is an anomaly worth showing."""

    def build_discovery(self):
        """Report nothing - this stub exists only to be recognised as reporter-carrying."""
        return None


def test_component_status_omits_components_that_cannot_report():
    """An active component with no build_discovery() is left out; one that has it and stayed silent is kept.

    This is the distinction that makes the map worth reading. Before it, "no_report" covered both
    storage/web/db/ha - which are working perfectly and will never report - and a component whose
    reporter genuinely failed to fire. The first group is the overwhelming majority, so it buried
    the second. Dropping it is what turns a silent reporter back into a visible anomaly.
    """
    base, coordinator = _coordinator()
    base.components = _StubRegistry(
        active=["storage", "givtcp"],
        alive=["storage", "givtcp"],
        all_names=["storage", "givtcp"],
        components={"storage": _SilentComponent(), "givtcp": _ReportingComponent()},
    )
    components = coordinator.assemble()["components"]
    assert "storage" not in components, "a component with no reporter must not be filed as no_report"
    assert components["givtcp"]["status"] == "no_report", "a component that CAN report and did not is the anomaly worth keeping"
    print("PASS: only components capable of reporting are held to account for silence")
    return 0


def test_component_status_reported_at_set_only_for_ok():
    """reported_at carries a UTC timestamp for a component that has actually reported, and stays
    None for every other status - the field was previously never populated anywhere, so a
    published catalogue showed reported_at: null forever even for a component reporting normally."""
    base, coordinator = _coordinator()
    base.components = _StubRegistry(active=["givtcp", "octopus"], alive=["octopus"], all_names=["givtcp", "octopus", "solis"])
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A"}]})
    components = coordinator.assemble()["components"]
    assert components["givtcp"]["status"] == "ok"
    reported_at = components["givtcp"]["reported_at"]
    assert reported_at, "an 'ok' component must carry a reported_at timestamp"
    datetime.fromisoformat(reported_at)  # must parse as a real ISO-8601 timestamp, not just be truthy
    assert components["octopus"]["reported_at"] is None, "a component that has not reported must not have a timestamp"
    assert "solis" not in components, "an unconfigured component is absent, so it has no reported_at at all"
    print("PASS: reported_at is set only for a component with status ok")
    return 0


def test_observations_duplicate_serial():
    """The same serial claimed by two components is recorded, not resolved."""
    base, coordinator = _coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:SN1", "hardware_ids": {"serial": "SA2242"}}]})
    coordinator.report("gecloud", {"inverters": [{"device_id": "gecloud:sn1", "hardware_ids": {"serial": "sa2242"}}]})
    conflicts = coordinator.assemble()["observations"]["conflicts"]
    duplicate = [entry for entry in conflicts if entry["kind"] == "duplicate_serial"]
    assert duplicate and sorted(duplicate[0]["claimed_by"]) == ["gecloud", "givtcp"]
    assert any(entry["kind"] == "multiple_inverter_sources" for entry in conflicts)
    print("PASS: duplicate serial and multiple sources observed")
    return 0


def test_observations_duplicate_serial_sees_a_gateways_fronted_batteries():
    """Review round: a gateway record claims the batteries it fronts, so GE Cloud + GivTCP on the same battery collides.

    GE Cloud in gateway composition reports the GATEWAY's serial in hardware_ids and the fronted
    battery serials in the structural `serials` list, while GivTCP reports each battery's own
    serial in hardware_ids. Grouping only on hardware_ids.serial therefore missed the single most
    likely real-world double-claim - two components controlling the same physical battery - which
    is the observation this catalogue release exists to count.
    """
    base, coordinator = _coordinator()
    coordinator.report("gecloud", {"inverters": [{"device_id": "gecloud:GW1", "hardware_ids": {"serial": "GW1"}, "composition": "gateway", "serials": ["BAT001", "BAT002"]}]})
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:BAT001", "hardware_ids": {"serial": "bat001"}}]})
    conflicts = coordinator.assemble()["observations"]["conflicts"]
    duplicate = [entry for entry in conflicts if entry["kind"] == "duplicate_serial"]
    assert duplicate, "a battery fronted by a gateway and claimed directly by GivTCP is a duplicate serial: {}".format(conflicts)
    assert duplicate[0]["serial"] == "bat001", duplicate
    assert sorted(duplicate[0]["claimed_by"]) == ["gecloud", "givtcp"], duplicate

    # The gateway's own serials must not collide with themselves - one source cannot conflict with itself
    base, coordinator = _coordinator()
    coordinator.report("gecloud", {"inverters": [{"device_id": "gecloud:GW1", "hardware_ids": {"serial": "GW1"}, "composition": "gateway", "serials": ["BAT001", "BAT002"]}]})
    assert not [entry for entry in coordinator.assemble()["observations"]["conflicts"] if entry["kind"] == "duplicate_serial"], "a gateway alone claims nothing twice"
    print("PASS: a gateway's fronted serials take part in duplicate_serial detection")
    return 0


def test_component_load_error_carries_no_free_text():
    """Review round: a load_error status records THAT a component failed, never the exception text.

    components.load_error() is the raw str() of whatever construction raised - unbounded free text
    that could carry a URL, an account value or a credential from a component's own error message.
    Every other value in this document reaches it through a typed container that cannot hold one;
    a dump gets attached to a public issue, so the one field outside that boundary is removed
    rather than trusted to the shape guard. The message stays in the log and on the
    component-status entity, which is where a maintainer already reads it.
    """
    base, coordinator = _coordinator()
    base.components = _StubRegistry(all_names=["fox"], errors={"fox": "auth failed for https://api.example.com/?key=SECRET123"})
    entry = coordinator.assemble()["components"]["fox"]
    assert entry["status"] == "load_error", entry
    assert "error" not in entry, "the raw exception text must not be published into the catalogue: {}".format(entry)
    assert "SECRET123" not in str(coordinator.catalogue()), "no part of an exception message may reach the published catalogue"
    print("PASS: a load_error status carries no free-text exception message")
    return 0


def test_observations_contested_cars_and_meters():
    """A charger component and an intelligent-device component both claiming cars is recorded, as are two import meters."""
    base, coordinator = _coordinator()
    coordinator.report("ohme", {"chargers": [{"device_id": "ohme:CH1"}], "cars": [{"device_id": "ohme:v1"}]})
    coordinator.report("octopus", {"cars": [{"device_id": "octopus:d1"}], "meters": [{"device_id": "octopus:m1", "direction": "import"}]})
    coordinator.report("kraken", {"meters": [{"device_id": "kraken:m1", "direction": "import"}]})
    conflicts = coordinator.assemble()["observations"]["conflicts"]
    kinds = {entry["kind"] for entry in conflicts}
    assert "contested_car_slots" in kinds and "multiple_import_meters" in kinds, kinds
    print("PASS: contested cars and multiple import meters observed")
    return 0


def test_observations_resulting_config():
    """The config discovery did NOT set is recorded alongside it, for comparison."""
    base, coordinator = _coordinator()
    base.args.update({"num_inverters": 2, "num_cars": 1, "inverter_type": ["GE", "GEC"]})
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A"}]})
    resulting = coordinator.assemble()["observations"]["resulting_config"]
    assert resulting["num_inverters"] == 2 and resulting["inverter_type"] == ["GE", "GEC"]
    print("PASS: resulting config recorded")
    return 0


def _redacting_coordinator():
    """A coordinator with a fixed salt, so pseudonym tokens are reproducible inside one test."""
    base, coordinator = _coordinator()
    coordinator.salt = "test-salt-0001"
    return base, coordinator


def test_account_ids_pseudonymised_and_stable():
    """account_ids values never appear in the clear, and the same value maps to the same token throughout."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123", "account": "A-1234ABCD"}}],
            "programmes": [{"device_id": "axle:site1", "kind": "vpp", "meter": "octopus:1234567890123"}],
        },
    )
    catalogue = coordinator.catalogue()
    text = str(catalogue)
    assert "1234567890123" not in text, "raw MPAN must not survive redaction"
    assert "A-1234ABCD" not in text
    meter = catalogue["meters"][0]
    assert meter["account_ids"]["mpan"].startswith("#")
    # the cross-link still resolves to the same meter
    assert catalogue["programmes"][0]["meter"] == meter["device_id"]
    print("PASS: account ids pseudonymised, cross-link preserved")
    return 0


def test_account_id_noted_in_two_forms_across_records_resolves_to_one_token():
    """Final review, Ruling R41: _note() used to derive the token from whichever textual form it
    was handed and overwrite any mapping an earlier call had already made for a shared variant, so
    the SAME real-world identifier noted twice in different forms produced two different,
    order-dependent tokens - defeating the stability a maintainer relies on to correlate records
    within one dump. Reproduces the review's own two verified cases in one test: an MPAN reported
    as a string by one component and as a float by another, and an account id reported in two
    case/separator forms by two components - each pair must resolve to exactly one token."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123"}}]})
    coordinator.report("kraken", {"meters": [{"device_id": "kraken:1234567890123", "direction": "import", "account_ids": {"mpan": 1234567890123.0}}]})
    coordinator.report("ohme", {"chargers": [{"device_id": "ohme:CH1", "account_ids": {"account": "AB-12CD34"}}]})
    coordinator.report("solis", {"inverters": [{"device_id": "solis:INV1", "account_ids": {"account": "ab_12cd34"}}]})

    catalogue = coordinator.catalogue()

    mpan_tokens = {record["source"]: record["account_ids"]["mpan"] for record in catalogue["meters"]}
    assert mpan_tokens["octopus"] == mpan_tokens["kraken"], "a string mpan and its float echo must resolve to one token: {}".format(mpan_tokens)

    account_tokens = {record["source"]: record["account_ids"]["account"] for record in catalogue["chargers"] + catalogue["inverters"] if "account_ids" in record}
    assert account_tokens["ohme"] == account_tokens["solis"], "'AB-12CD34' and 'ab_12cd34' noted in different records must resolve to one token: {}".format(account_tokens)
    print("PASS: an identifier noted in two different textual forms across two records resolves to exactly one token")
    return 0


def test_note_returns_same_token_when_called_twice_with_different_variants():
    """Redactor._note() unit-level: calling it a second time with a DIFFERENT textual form of an
    already-noted value must return the SAME token, not mint a fresh one - the fix builds the full
    variant set before minting anything and reuses whatever token any variant already maps to.
    Covers both axes _note() composes: numeric (string vs float) and identifier (case/separator)."""
    redactor = Redactor("test-salt-0001")
    token_a = redactor._note("1234567890123", substring=True)
    token_b = redactor._note(1234567890123.0, substring=True)
    assert token_a == token_b, "a string and its float echo must mint/reuse the same token: {} != {}".format(token_a, token_b)

    token_c = redactor._note("AB-12CD34", substring=True)
    token_d = redactor._note("ab_12cd34", substring=True)
    assert token_c == token_d, "case/separator variants must mint/reuse the same token: {} != {}".format(token_c, token_d)

    # And the reverse order - the variant noted FIRST here was noted SECOND above - to prove this
    # is genuinely order-independent, not merely "the second call wins" restated.
    redactor2 = Redactor("test-salt-0001")
    token_e = redactor2._note(1234567890123.0, substring=True)
    token_f = redactor2._note("1234567890123", substring=True)
    assert token_e == token_f, "order must not matter: {} != {}".format(token_e, token_f)
    print("PASS: _note() reuses one token across different textual forms of an identifier, regardless of call order")
    return 0


def test_cross_link_resolves_without_coincidental_substring():
    """The device_id/meter cross-link resolves even when device_id does not literally embed the account
    identifier as a substring - proving the link is via noting the owning record's device_id, not a
    coincidence of the previous test's device_id happening to spell out the raw MPAN."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [{"device_id": "octopus:m1", "direction": "import", "account_ids": {"mpan": "1234567890123"}}],
            "programmes": [{"device_id": "axle:site1", "kind": "vpp", "meter": "octopus:m1"}],
        },
    )
    catalogue = coordinator.catalogue()
    meter = catalogue["meters"][0]
    assert meter["device_id"] != "octopus:m1", "device_id of a record with account_ids must itself be pseudonymised"
    assert catalogue["programmes"][0]["meter"] == meter["device_id"]
    print("PASS: cross-link resolves without a coincidental substring match")
    return 0


def test_measures_meter_cross_link_resolves():
    """The measures_meter cross-link (inverter -> meter) resolves the same way as meter (programme -> meter)."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "givtcp",
        {
            "meters": [{"device_id": "givtcp:m1", "direction": "import", "account_ids": {"mpan": "1234567890123"}}],
            "inverters": [{"device_id": "givtcp:inv1", "measures_meter": "givtcp:m1"}],
        },
    )
    catalogue = coordinator.catalogue()
    meter = catalogue["meters"][0]
    assert meter["device_id"] != "givtcp:m1", "this test is only load-bearing if device_id actually changed - it must not stay equal by both sides going unredacted"
    assert catalogue["inverters"][0]["measures_meter"] == meter["device_id"]
    print("PASS: measures_meter cross-link resolves")
    return 0


def test_pseudonym_differs_across_salts():
    """The same value under a different installation salt produces a different token."""
    base_a, coordinator_a = _redacting_coordinator()
    base_b, coordinator_b = _coordinator()
    coordinator_b.salt = "a-different-salt"
    report = {"meters": [{"device_id": "octopus:m", "direction": "import", "account_ids": {"mpan": "1234567890123"}}]}
    coordinator_a.report("octopus", report)
    coordinator_b.report("octopus", report)
    token_a = coordinator_a.catalogue()["meters"][0]["account_ids"]["mpan"]
    token_b = coordinator_b.catalogue()["meters"][0]["account_ids"]["mpan"]
    assert token_a != token_b
    print("PASS: tokens differ across installations")
    return 0


def test_pseudonym_substituted_inside_entity_ids():
    """A vendor that embeds an identifier in an entity name does not republish what the field just hid."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "solar",
        {
            "forecasts": [
                {
                    "device_id": "solcast:abcdef123456",
                    "kind": "solar",
                    "account_ids": {"site_id": "abcdef123456"},
                    "entities": {"pv_forecast_today": {"entity_id": "sensor.predbat_solcast_abcdef123456_today", "domain": "sensor", "access": "r"}},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    assert "abcdef123456" not in str(catalogue), catalogue["forecasts"][0]["entities"]
    entity = catalogue["forecasts"][0]["entities"]["pv_forecast_today"]
    assert entity["domain"] == "sensor" and entity["access"] == "r", "legitimate neighbouring descriptor fields are untouched"
    print("PASS: pseudonymised value substituted inside entity ids")
    return 0


def test_serials_and_tariff_codes_stay_clear():
    """Hardware identity and product codes stay readable - they are what makes a bug report diagnosable."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:SA2242G123", "hardware_ids": {"serial": "SA2242G123"}, "info": {"firmware": "D0.451"}}]})
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:m", "direction": "import", "tariff": {"info": {"tariff_code": "E-1R-AGILE-24-10-01-A"}}}]})
    text = str(coordinator.catalogue())
    assert "SA2242G123" in text and "D0.451" in text and "E-1R-AGILE-24-10-01-A" in text
    print("PASS: serials, firmware and tariff codes kept clear")
    return 0


def test_misfiled_identifier_caught_by_shape_guard():
    """An MPAN is a number, so ratings accepts it - the shape guard pseudonymises it anyway and logs."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:m", "direction": "import", "ratings": {"standing_charge_p": 47.5, "supply_number": 1234567890123}}]})
    catalogue = coordinator.catalogue()
    assert "1234567890123" not in str(catalogue)
    assert catalogue["meters"][0]["ratings"]["standing_charge_p"] == 47.5, "a real measurement is untouched"
    print("PASS: misfiled identifier caught by the shape guard")
    return 0


def test_misfiled_identifier_inside_entity_descriptor_caught():
    """An MPAN misfiled into a nested entity descriptor field (e.g. its "max") is still caught: entities
    values are descriptor dicts, not scalars, so the shape guard has to recurse into them rather than
    stringify the whole descriptor - the exact gap Correction 2 exists to close."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {
                    "device_id": "octopus:m",
                    "direction": "import",
                    "entities": {"supply_number": {"entity_id": "sensor.predbat_octopus_m_supply_number", "domain": "sensor", "access": "r", "max": 1234567890123}},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    assert "1234567890123" not in str(catalogue)
    entity = catalogue["meters"][0]["entities"]["supply_number"]
    assert entity["entity_id"] == "sensor.predbat_octopus_m_supply_number", "the descriptor itself survives, only the misfiled value is pseudonymised"
    print("PASS: identifier misfiled inside a nested entity descriptor field caught")
    return 0


def test_misfiled_email_inside_entity_id_caught():
    """entity_id is kept verbatim by validation (unlike every other descriptor field, which is cleaned by
    type), so it is the one realistic path an email-shaped value can reach a clear container - info and
    every other string field already refuse "@" at validation. The shape guard is what stands between a
    buggy component's entity_id and a public dump."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("ohme", {"chargers": [{"device_id": "ohme:CH1", "info": {"vendor": "Ohme"}, "entities": {"login": {"entity_id": "sensor.bob@example.com", "domain": "sensor", "access": "r"}}}]})
    assert "bob@example.com" in str(coordinator.catalogue_raw()), "sanity check: the email really does reach the raw catalogue unguarded"
    catalogue = coordinator.catalogue()
    assert "bob@example.com" not in str(catalogue)
    assert catalogue["chargers"][0]["info"]["vendor"] == "Ohme", "a legitimate neighbouring field is untouched"
    print("PASS: email-shaped entity_id caught by the shape guard")
    return 0


def test_misfiled_identifier_inside_vocabulary_list_caught():
    """A vocabulary token pattern allows digits, so an MPAN-shaped value misfiled into a flags/functions
    list is still caught, even though vocabulary lists are validated and walked differently from the
    dict-shaped clear containers."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:m", "direction": "import", "tariff": {"flags": ["agile", "1234567890123"]}}]})
    catalogue = coordinator.catalogue()
    assert "1234567890123" not in str(catalogue)
    assert "agile" in catalogue["meters"][0]["tariff"]["flags"]
    print("PASS: misfiled identifier inside a vocabulary list caught")
    return 0


def test_catalogue_raw_is_unredacted():
    """catalogue_raw is the in-process view and keeps originals, so the redacted path is demonstrably doing work."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:m", "direction": "import", "account_ids": {"mpan": "1234567890123"}}]})
    assert "1234567890123" in str(coordinator.catalogue_raw())
    print("PASS: raw catalogue retains originals")
    return 0


# --- Review round 2: root cause A - the shape guard's digit detection was too narrow ---


def test_misfiled_float_identifier_caught():
    """str(1234567890123.0) is '1234567890123.0', which an anchored ^\\d{10,}$ pattern rejects even
    though _clean_number accepts floats happily - any component that passes an API number through
    float() must not be able to launder an identifier that way."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:m", "direction": "import", "ratings": {"standing_charge_p": 47.5, "supply_number": 1234567890123.0}}]})
    catalogue = coordinator.catalogue()
    assert "1234567890123" not in str(catalogue)
    assert catalogue["meters"][0]["ratings"]["standing_charge_p"] == 47.5
    print("PASS: a misfiled identifier reported as a float is still caught")
    return 0


def test_misfiled_identifier_embedded_or_separated_caught():
    """An identifier does not have to be a bare digit string to leak: embedded in prose ("MPAN
    1234567890123"), broken up by formatting separators ("1234-5678-90123"), or written as a phone
    number ("+447700900123") - only whitespace was stripped before, so all three previously
    survived."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {"device_id": "octopus:m1", "direction": "import", "info": {"supply": "MPAN 1234567890123"}},
                {"device_id": "octopus:m2", "direction": "import", "info": {"ref": "1234-5678-90123"}},
                {"device_id": "octopus:m3", "direction": "import", "info": {"contact": "+447700900123"}},
            ]
        },
    )
    catalogue = coordinator.catalogue()
    text = str(catalogue)
    assert "1234567890123" not in text
    assert "1234-5678-90123" not in text and "5678" not in text
    assert "447700900123" not in text
    assert catalogue["meters"][1]["device_id"] == "octopus:m2", "an ordinary device_id is a completely different shape and is untouched"
    print("PASS: an embedded, separator-formatted or phone-shaped identifier is caught")
    return 0


def test_misfiled_identifier_grouped_by_underscore_or_comma_caught():
    """Adversarial pass: an identifier grouped with "_" (a Python-style numeric literal) or ","
    (thousands separators) reads identically to a human as the same identifier, but was found to
    survive redaction untouched when only whitespace/hyphen/slash/dot were stripped before the
    digit-run search - fixed by widening the stripped separator set rather than narrowing this
    test to only the separators already handled."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {"device_id": "octopus:m1", "direction": "import", "info": {"ref": "1_234_567_890_123"}, "ratings": {"standing_charge_p": 47.5}},
                {"device_id": "octopus:m2", "direction": "import", "info": {"ref": "1,234,567,890,123"}},
            ]
        },
    )
    catalogue = coordinator.catalogue()
    text = str(catalogue)
    assert "1_234_567_890_123" not in text and "1,234,567,890,123" not in text
    assert "1234567890123" not in text
    assert catalogue["meters"][0]["ratings"]["standing_charge_p"] == 47.5, "a real measurement is untouched"
    print("PASS: an underscore- or comma-grouped identifier is caught")
    return 0


def test_location_shaped_key_pseudonymised_regardless_of_value_shape():
    """latitude/longitude cannot be recognised as a pair from one value, so the guard checks the
    KEY NAME instead - this sidesteps the pair problem entirely and cannot false-positive on an
    ordinary rating like battery_kwh: 9.5."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("solar", {"forecasts": [{"device_id": "solcast:site1", "kind": "solar", "ratings": {"latitude": 51.5074, "longitude": -0.1278, "horizon_hours": 168}}]})
    catalogue = coordinator.catalogue()
    ratings = catalogue["forecasts"][0]["ratings"]
    assert str(ratings["latitude"]).startswith("#") and ratings["latitude"] != 51.5074
    assert str(ratings["longitude"]).startswith("#") and ratings["longitude"] != -0.1278
    assert ratings["horizon_hours"] == 168, "an ordinary rating under an unrelated key name is untouched"
    print("PASS: a location-named key is pseudonymised regardless of its value's shape")
    return 0


# --- Review round 3: NEW-1 - the shape guard, widened for round 2, over-corrected on ordinary floats ---


def test_legitimate_long_float_survives_unchanged_and_still_numeric():
    """NEW-1 (Critical): stripping "." and searching the whole repr for a 10+ digit run means any
    float whose repr carries that many digits after the decimal point gets pseudonymised - and a
    numbers-only container silently starts handing consumers a str. An efficiency, a percentage or
    a unit conversion routinely produces exactly such a float. The guard must judge a NUMERIC
    value on its INTEGER PART's digit count, not its full stringified repr. Also asserts a genuine
    identifier on the SAME record WAS redacted, so this is not a false pass against a no-op
    redact() - the same defect fixed in the timestamp test, applied here too."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "givtcp",
        {
            "inverters": [
                {
                    "device_id": "givtcp:inv1",
                    "account_ids": {"owner_ref": "1234567890123"},
                    "ratings": {"third": 1 / 3, "long_division": 700 / 11, "precise": 17 / 6, "float_sum": 0.1 + 0.2, "big_but_legit": 123456789.0},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    inverter = catalogue["inverters"][0]
    ratings = inverter["ratings"]
    for key, expected in (("third", 1 / 3), ("long_division", 700 / 11), ("precise", 17 / 6), ("float_sum", 0.1 + 0.2), ("big_but_legit", 123456789.0)):
        assert ratings[key] == expected, (key, ratings[key])
        assert isinstance(ratings[key], float), "a numbers-only container must still hand back a number, not a str"
    assert inverter["account_ids"]["owner_ref"] != "1234567890123" and inverter["account_ids"]["owner_ref"].startswith("#"), "redaction must still be doing real work on the same record"
    print("PASS: a legitimate long float survives unchanged and stays numeric, while real redaction still happens")
    return 0


# --- Review round 3: NEW-4 - hardware_ids over-pseudonymised real vendor serials ---


def test_hardware_ids_only_flags_all_digit_values_not_prefixed_serials():
    """NEW-4 (Important): a letter-prefixed serial with a long digit tail is the industry-standard
    shape ("HV2160123456", "1102B1234567890", "SVT1234567890") and hardware_ids exists precisely
    so a serial survives in the clear - over-hiding it is a real design regression, not a safe
    default. Inside hardware_ids only, the digit-run guard must fire solely when the value is
    nothing BUT digits; a bare all-digit string there is still genuinely suspicious."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "givtcp",
        {
            "inverters": [
                {
                    "device_id": "givtcp:inv1",
                    "hardware_ids": {"serial": "HV2160123456", "battery_serial": "1102B1234567890", "wifi_serial": "SVT1234567890", "mpan_misfiled": "1234567890123"},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    hardware_ids = catalogue["inverters"][0]["hardware_ids"]
    assert hardware_ids["serial"] == "HV2160123456"
    assert hardware_ids["battery_serial"] == "1102B1234567890"
    assert hardware_ids["wifi_serial"] == "SVT1234567890"
    assert hardware_ids["mpan_misfiled"] != "1234567890123" and str(hardware_ids["mpan_misfiled"]).startswith("#"), "a BARE all-digit value in hardware_ids is still caught"
    print("PASS: prefixed vendor serials stay clear in hardware_ids; a bare all-digit value is still caught")
    return 0


# --- Review round 2: root cause B - the substitution pass was too narrow and too broad ---


def test_pseudonymised_value_substituted_inside_entity_id_value():
    """A site id embedded in an entity_id VALUE is still hidden wherever it appears - substring
    replacement is still full-strength for values, only dict KEYS are exact-match-only (see
    NEW-3/test_account_ids_value_does_not_corrupt_structural_or_descriptor_keys below, which is
    what a version of this test that also embedded the id in the descriptor's KEY would now have
    to accept as a residual, deliberate trade-off)."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "solar",
        {
            "forecasts": [
                {
                    "device_id": "solcast:site1",
                    "kind": "solar",
                    "account_ids": {"site_id": "abcdef123456"},
                    "entities": {"pv_today": {"entity_id": "sensor.pv_abcdef123456_today", "domain": "sensor", "access": "r"}},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    entity = catalogue["forecasts"][0]["entities"]["pv_today"]
    assert "abcdef123456" not in entity["entity_id"], entity
    assert entity["domain"] == "sensor" and entity["access"] == "r", "a legitimate neighbouring descriptor field is untouched"
    print("PASS: pseudonymised value substituted inside an entity_id value")
    return 0


def test_account_ids_value_does_not_corrupt_structural_or_descriptor_keys():
    """NEW-3: an account_ids value can be as ordinary as a short English word - nothing forces a
    component to report something identifier-shaped there - and substring-rewriting dict keys
    turned exactly that coincidence into an entire published section vanishing: with
    account_ids: {"acct": "charge"}, the top-level "chargers" key was corrupted and the entities
    key "charge_rate" lost half its name. Keys are now rewritten by exact match only, never
    substring, so a section name and a component-chosen entity name both survive intact."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "ohme",
        {
            "chargers": [
                {
                    "device_id": "ohme:CH1",
                    "account_ids": {"acct": "charge"},
                    "entities": {"charge_rate": {"entity_id": "number.predbat_ohme_0_charge_rate", "domain": "number", "access": "rw"}},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    assert "chargers" in catalogue, list(catalogue.keys())
    assert "charge_rate" in catalogue["chargers"][0]["entities"], list(catalogue["chargers"][0]["entities"].keys())
    assert catalogue["chargers"][0]["account_ids"]["acct"] != "charge", "the value itself is still pseudonymised"
    print("PASS: an account_ids value does not corrupt structural or descriptor keys via substring")
    return 0


def test_account_ids_value_equal_to_a_structural_name_does_not_delete_it():
    """Final review (carried from Task 7): an account_ids value equal to a structural name did not
    even need to be a substring coincidence (the test above) - EXACT equality alone was enough,
    since _substitute_key rewrote any key equal to a noted original with nothing excluding
    Predbat's own vocabulary from that match. account_ids: {"acct": "chargers"} deleted the entire
    top-level "chargers" section; account_ids: {"acct": "account_ids"} deleted the account_ids
    container out from under its own record. PROTECTED_KEY_NAMES closes this for every SECTION_SPEC
    and CONTAINER_SPEC key name plus "components"/"observations"."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "ohme",
        {"chargers": [{"device_id": "ohme:CH1", "account_ids": {"acct": "chargers"}, "info": {"model": "Home Pro"}}]},
    )
    coordinator.report(
        "solis",
        {"inverters": [{"device_id": "solis:INV1", "account_ids": {"acct": "account_ids"}, "info": {"model": "S6"}}]},
    )
    catalogue = coordinator.catalogue()

    assert "chargers" in catalogue and len(catalogue["chargers"]) == 1, "the top-level chargers section must survive: {}".format(list(catalogue.keys()))
    assert catalogue["chargers"][0]["info"]["model"] == "Home Pro"
    assert catalogue["chargers"][0]["account_ids"]["acct"] != "chargers", "the value itself is still pseudonymised"

    inverter = catalogue["inverters"][0]
    assert "account_ids" in inverter, "the account_ids container must survive a value equal to its own container name: {}".format(inverter)
    assert inverter["account_ids"]["acct"] != "account_ids", "the value itself is still pseudonymised"
    print("PASS: an account_ids value equal to a section or container name does not delete it")
    return 0


def test_int_identifier_echoed_outside_guarded_container_is_substituted():
    """A non-string scalar (an int) that exactly repeats an account identifier elsewhere in the
    catalogue is rewritten too - substitution previously only ever looked at strings, so an
    identifier echoed as a bare int survived untouched."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {"meters": [{"device_id": "octopus:m", "direction": "import", "account_ids": {"customer": "123456"}, "ratings": {"customer_ref": 123456}}]},
    )
    catalogue = coordinator.catalogue()
    meter = catalogue["meters"][0]
    assert meter["ratings"]["customer_ref"] == meter["account_ids"]["customer"]
    assert meter["ratings"]["customer_ref"] != 123456
    print("PASS: an int identifier echoed outside its guarded container is substituted")
    return 0


def test_numeric_identifier_echo_substituted_regardless_of_int_float_or_string_form():
    """NEW-2 (Important): the int-echo fix above compares str(node), so it missed the FLOAT echo -
    account_ids: {"customer": "123456"} alongside ratings: {"customer_ref": 123456.0} published
    123456.0 in the clear right next to the tokenised original, for any identifier of <=8 digits
    (9+ get caught by the shape guard directly). The same gap works in reverse: a float noted in
    account_ids must be caught when echoed as a plain string elsewhere too."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {"device_id": "octopus:m1", "direction": "import", "account_ids": {"customer": "123456"}, "ratings": {"customer_ref": 123456.0}},
                {"device_id": "octopus:m2", "direction": "import", "account_ids": {"customer": 654321.0}, "info": {"note": "654321"}},
            ]
        },
    )
    catalogue = coordinator.catalogue()
    m1, m2 = catalogue["meters"]
    assert m1["ratings"]["customer_ref"] == m1["account_ids"]["customer"] and m1["ratings"]["customer_ref"] != 123456.0
    assert m2["info"]["note"] == m2["account_ids"]["customer"] and m2["info"]["note"] != "654321"
    print("PASS: a numeric identifier is substituted the same way whether echoed as a string, int or float")
    return 0


def test_structural_scalar_shape_guard_catches_bare_identifier_device_id():
    """A meter reported with a raw MPAN AS its device_id, and no account_ids container at all,
    still gets caught - structural fields sit outside CONTAINER_SPEC, but that must not exempt
    them from the same shape guard clear-container values get."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("octopus", {"meters": [{"device_id": "1234567890123", "direction": "import"}]})
    catalogue = coordinator.catalogue()
    assert "1234567890123" not in str(catalogue)
    print("PASS: a bare identifier reported directly as a structural field is still caught")
    return 0


def test_device_id_without_identity_data_never_corrupts_unrelated_text():
    """A device_id with no account_ids alongside it carries no identity-linking data at all, so it
    is never noted for cross-linking at all (not whole-string, not substring) - an ordinary word
    used as one (a charger literally named "charger") cannot corrupt unrelated text that happens
    to contain the same word, however common it is. (See NEW-5 below for the complementary case:
    a device_id that IS identity-derived, because its own record carries account_ids, is deliberately
    substring-eligible, since its identity comes from a real identifier and under-redacting it
    would make its token trivially correlatable.) Also asserts a genuine identity-derived device_id
    ELSEWHERE in the same catalogue WAS redacted, so this is not a false pass against a no-op
    redact() - the same defect fixed in the timestamp test, applied here too."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "ohme",
        {
            "chargers": [{"device_id": "charger", "info": {"model": "charger v2"}}],
            "cars": [{"device_id": "ohme:v1", "entities": {"status": {"entity_id": "sensor.ohme_charger_status", "domain": "sensor", "access": "r"}}}],
        },
    )
    coordinator.report("octopus", {"meters": [{"device_id": "octopus:m", "direction": "import", "account_ids": {"mpan": "1234567890123"}}]})
    catalogue = coordinator.catalogue()
    charger = catalogue["chargers"][0]
    assert charger["device_id"] == "charger", "no account_ids on this record, so nothing marks device_id as identity-derived"
    assert charger["info"]["model"] == "charger v2", "an unrelated string containing the word must not be corrupted"
    assert catalogue["cars"][0]["entities"]["status"]["entity_id"] == "sensor.ohme_charger_status", "an unrelated entity_id containing the word must not be corrupted either"
    assert catalogue["meters"][0]["device_id"] != "octopus:m", "redaction must still be doing real work elsewhere in the same catalogue"
    print("PASS: a device_id with no account_ids alongside it never corrupts unrelated text, while real redaction still happens elsewhere")
    return 0


def test_identity_derived_device_id_substituted_wherever_it_is_echoed():
    """NEW-5 (Important): a device_id whose record carries account_ids is identity-derived, but the
    shape guard does not independently catch every such id ("A-AAAA1111" is Octopus-account-shaped
    but has no 10+ digit run) - it was tokenised in place but left in the clear wherever it was
    merely ECHOED elsewhere in the same record (an entity_id, a free-text note), making its token
    trivially correlatable back to the raw text sitting right next to it. An identity-derived
    device_id must be substring-matched too, not just whole-string."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {
                    "device_id": "A-AAAA1111",
                    "direction": "import",
                    "account_ids": {"mpan": "1234567890123"},
                    "entities": {"pv": {"entity_id": "sensor.octopus_A-AAAA1111_import", "domain": "sensor", "access": "r"}},
                    "info": {"note": "linked to A-AAAA1111"},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    text = str(catalogue)
    assert "A-AAAA1111" not in text, text
    meter = catalogue["meters"][0]
    assert meter["entities"]["pv"]["entity_id"].startswith("sensor.octopus_#") and meter["entities"]["pv"]["entity_id"].endswith("_import")
    print("PASS: an identity-derived device_id is substituted wherever it is echoed, not just where it was noted")
    return 0


# --- Review round 4: ITEM 1 - the identity-derived precondition only checked the record's own top-level keys ---


def test_pseudonym_container_nested_in_sub_record_still_marks_device_id_identity_derived():
    """ITEM 1 (the substantive finding of this round): _walk's device_id precondition tested
    `any(name in node for name in PSEUDONYM_CONTAINERS)` against the SAME dict only, so a record
    whose account_ids sits inside a sub-record (tariff, which _validate_record structurally
    accepts even though v1's own Octopus reporter puts account_ids at meter level) never had its
    device_id noted at all - neither whole-string nor substring. The device_id itself, a
    free-text echo of it, and a cross-link pointing at it all published raw. Fixed by recursing
    the precondition (_has_pseudonym_container) into every nested dict/list, not just node's own
    keys."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {
                    "device_id": "A-AAAA1111",
                    "direction": "import",
                    "tariff": {"account_ids": {"contract": "1234567890123"}},
                    "info": {"note": "for A-AAAA1111"},
                }
            ],
            "programmes": [{"device_id": "axle:site1", "kind": "vpp", "meter": "A-AAAA1111"}],
        },
    )
    catalogue = coordinator.catalogue()
    meter = catalogue["meters"][0]
    assert meter["device_id"] != "A-AAAA1111", "device_id must be identity-derived even when account_ids sits inside a sub-record, not just a direct sibling"
    assert "A-AAAA1111" not in meter["info"]["note"], meter["info"]["note"]
    assert catalogue["programmes"][0]["meter"] == meter["device_id"], "the cross-link must still resolve"
    print("PASS: a pseudonym container nested in a sub-record still marks device_id identity-derived")
    return 0


def test_substitution_does_not_touch_the_catalogue_timestamp():
    """generated is the catalogue's own timestamp, stamped by assemble() itself rather than any
    component, so a short account identifier that coincidentally matches digits inside it must not
    corrupt it. Constructed directly against Redactor so the collision is deterministic rather than
    depending on the real clock. Also asserts the account id WAS actually redacted elsewhere - a
    no-op redact() would pass the timestamp assertion for the wrong reason."""
    redactor = Redactor("test-salt-0001")
    catalogue = {
        "schema_version": SCHEMA_VERSION,
        "generated": "2026-09-11T12:34:56.123456+00:00",
        "meters": [{"source": "octopus", "device_id": "octopus:m", "direction": "import", "account_ids": {"customer": "123456"}}],
    }
    redacted = redactor.redact(catalogue)
    assert redacted["generated"] == "2026-09-11T12:34:56.123456+00:00"
    assert redacted["meters"][0]["account_ids"]["customer"] != "123456" and redacted["meters"][0]["account_ids"]["customer"].startswith("#"), "redact() must still be doing real work, not a no-op"
    print("PASS: a coincidental digit collision does not corrupt the catalogue's own timestamp, while real redaction still happens")
    return 0


def test_shorter_original_does_not_fragment_a_longer_one():
    """When one noted original is a substring of a longer one, and NEITHER is independently caught
    by the shape guard (both are plain alphabetic strings here, not digit runs), the longer
    original must still be substituted whole rather than leaving fragments of it exposed around a
    shorter token spliced into the middle. The previous version of this test used two identifiers
    that both tripped the shape guard directly (both had an embedded 13-digit run), so it passed
    even with the sort removed and insertion order used instead - this repro is the one that
    actually exercises the substitution ordering fix and nothing else.

    Detection-power note: production behaviour (redact()'s `sorted(self.substring_ok, key=len,
    reverse=True)`) is fully deterministic regardless of set-iteration order, since the four
    originals below have four DISTINCT lengths - a stable sort on an all-distinct key has only one
    possible output. What is NOT fully deterministic is whether *reverting* that sort (falling
    back to raw `set` iteration) gets caught by this test: CPython's per-string hashing is
    PYTHONHASHSEED-randomised, so an unsorted mutant's output depends on the salt-of-the-run, not
    on anything this test controls. A 2-original chain (as this test used to be) is "accidentally"
    passed by an unsorted mutant whenever the longer original happens to iterate first - a
    coin-flip, empirically caught in only about half of sampled hash seeds. Chaining FOUR nested
    originals instead of two cuts that accidental-pass probability roughly to 1-in-4 (only the
    longest happening to iterate first defeats detection, regardless of the other three's
    relative order) - a cheap, meaningful improvement, but still not a deterministic guarantee.
    """
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [
                {
                    "device_id": "octopus:m",
                    "direction": "import",
                    "account_ids": {"a": "abcdef", "b": "xxabcdefyy", "c": "wwxxabcdefyyzz", "d": "vvwwxxabcdefyyzzuu"},
                    "info": {"note": "ref vvwwxxabcdefyyzzuu end"},
                }
            ]
        },
    )
    catalogue = coordinator.catalogue()
    note = catalogue["meters"][0]["info"]["note"]
    assert "abcdef" not in note, note
    assert note.startswith("ref #") and note.endswith(" end"), note
    print("PASS: the longest original is substituted before any shorter one that is its substring")
    return 0


# --- Review round 2: adversarial pass - a raw identifier used as a dict key, not a value ---


def test_declared_serials_stay_readable_wherever_they_appear():
    """A hardware serial is not secret, so an all-digit one stays readable - and so does what is built from it.

    Solis, Deye and Sunsynk serials are nothing but digits (these are real shapes: 16, 10 and 10
    digits), which the misfiled-identifier guard used to treat as a possible MPAN. That hid the
    serial, the device_id built from it and the duplicate_serial observation naming it, and logged
    a Warn per inverter on every catalogue read. A serial a record declares - hardware_ids.serial
    or an entry in serials - is now clear everywhere it appears as a whole token.

    Deye also reports its station id in account_ids, which used to mark the record's device_id as
    identity-derived and replace it wholesale. That rule is for a device_id built from an account
    identifier (Octopus's "octopus:{mpan}"); "deye:{serial}" is built from the serial, contains no
    account identifier, and stays readable while the station id itself is still pseudonymised.
    """
    base, coordinator = _redacting_coordinator()
    messages = []
    coordinator.log = messages.append
    coordinator.report("solis", {"inverters": [{"device_id": "solis:1031260253072197", "hardware_ids": {"serial": "1031260253072197"}}]})
    coordinator.report("sunsynk", {"inverters": [{"device_id": "sunsynk:2405116013", "hardware_ids": {"serial": "2405116013"}}]})
    coordinator.report("deye", {"inverters": [{"device_id": "deye:2306178123", "hardware_ids": {"serial": "2306178123"}, "account_ids": {"station_id": 61234567}}]})
    # A gateway fronting all-digit battery serials, one of which Sunsynk also claims
    coordinator.report("gecloud", {"inverters": [{"device_id": "gecloud:GW2242G123", "composition": "gateway", "hardware_ids": {"serial": "GW2242G123"}, "serials": ["2405116013", "7700112233"]}]})

    catalogue = coordinator.catalogue()

    by_id = {record["device_id"]: record for record in catalogue["inverters"]}
    assert set(by_id) == {"solis:1031260253072197", "sunsynk:2405116013", "deye:2306178123", "gecloud:GW2242G123"}, sorted(by_id)
    assert by_id["solis:1031260253072197"]["hardware_ids"] == {"serial": "1031260253072197"}
    assert by_id["sunsynk:2405116013"]["hardware_ids"] == {"serial": "2405116013"}
    assert by_id["deye:2306178123"]["hardware_ids"] == {"serial": "2306178123"}
    assert by_id["gecloud:GW2242G123"]["serials"] == ["2405116013", "7700112233"]
    station = by_id["deye:2306178123"]["account_ids"]["station_id"]
    assert station != 61234567 and str(station).startswith("#"), "the station id is an account identifier and is still pseudonymised"
    duplicates = [entry for entry in catalogue["observations"]["conflicts"] if entry.get("kind") == "duplicate_serial"]
    assert [entry["serial"] for entry in duplicates] == ["2405116013"], duplicates
    assert not [message for message in messages if "looks like an identifier" in message], messages
    print("PASS: declared serials, and the device_ids and observations built from them, stay readable")
    return 0


def test_declared_serials_do_not_shelter_a_misfiled_identifier():
    """Keeping serials readable must not let anything else through.

    A serial is only ignored where it appears as a whole token: a short serial that happens to sit
    inside a longer number must not break that number's digit run and let an MPAN through. A value
    under any other hardware_ids key is still guarded as before, and a device_id that is built from
    an account identifier is still replaced wholesale even when the record also declares a serial.
    """
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "solis",
        {
            "inverters": [
                {
                    "device_id": "solis:inv1",
                    "hardware_ids": {"serial": "123456", "mpan_misfiled": "1234567890123"},
                    "info": {"note": "MPAN 1234567890123"},
                },
                {"device_id": "solis:SERIAL01:ACCT123456", "hardware_ids": {"serial": "SERIAL01"}, "account_ids": {"account": "ACCT123456"}},
            ]
        },
    )

    catalogue = coordinator.catalogue()

    text = str(catalogue)
    assert "1234567890123" not in text, "a serial inside a longer number must not unhide it"
    first = catalogue["inverters"][0]
    assert first["hardware_ids"]["serial"] == "123456"
    assert str(first["hardware_ids"]["mpan_misfiled"]).startswith("#"), "another hardware_ids key is still guarded"
    assert "ACCT123456" not in text, "the account identifier is still hidden"
    assert str(catalogue["inverters"][1]["device_id"]).startswith("#"), "a device_id built from an account identifier is still replaced wholesale"
    assert catalogue["inverters"][1]["hardware_ids"]["serial"] == "SERIAL01"
    print("PASS: declared serials shelter nothing but themselves")
    return 0


def test_serial_derived_device_id_needs_whole_tokens():
    """A device_id is only treated as built from its serial on whole-token matches, both ways.

    The serial must stand as a whole token in the device_id - merely sitting inside a longer token
    is coincidence, not construction - and any account_ids value that stands as a whole token in it
    marks the device_id as identity-derived however short that value is, since a short account id
    is not otherwise caught by the substring pass (MIN_SUBSTITUTE).
    """
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "deye",
        {
            "inverters": [
                # the serial only sits inside "inv98765432", so nothing says the device_id is built from it
                {"device_id": "deye:inv98765432", "hardware_ids": {"serial": "98765"}, "account_ids": {"account": "XY12"}},
                # built from the serial, but a short station id stands in it as a whole token too
                {"device_id": "deye:2306178123:ST42", "hardware_ids": {"serial": "2306178123"}, "account_ids": {"station_id": "ST42"}},
            ]
        },
    )

    catalogue = coordinator.catalogue()

    first, second = catalogue["inverters"]
    assert str(first["device_id"]).startswith("#"), "a serial inside a longer token does not make the device_id serial-derived: {}".format(first["device_id"])
    assert str(second["device_id"]).startswith("#"), "a short account id in the device_id makes it identity-derived: {}".format(second["device_id"])
    assert "ST42" not in str(catalogue), "the station id must not survive"
    assert second["hardware_ids"]["serial"] == "2306178123", "the serial itself stays readable"
    print("PASS: serial-derived device_ids need whole-token matches, both ways")
    return 0


def test_misfiled_identifier_used_as_a_container_key_caught():
    """Adversarial: a component keys hardware_ids by the serial itself instead of naming the field.
    Before this fix only VALUES were shape-guarded and only NOTED originals were substituted into
    keys, so a raw identifier that a component used AS a dict key - never a value anywhere -
    reached the published catalogue untouched by either pass."""
    base, coordinator = _redacting_coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:inv1", "hardware_ids": {"1234567890123": "primary", "model": "H3"}}]})
    catalogue = coordinator.catalogue()
    assert "1234567890123" not in str(catalogue)
    assert catalogue["inverters"][0]["hardware_ids"]["model"] == "H3", "a legitimate neighbouring key is untouched"
    print("PASS: an identifier used as a clear-container key is caught")
    return 0


def test_guard_key_does_not_log_the_raw_value():
    """Final review: _guard_key logged the raw key text it had just decided to pseudonymise -
    Predbat's log file is attached to the same public issues a debug dump is, so writing the
    unredacted form into a log line republishes exactly what this guard exists to hide. Must match
    _guard_scalar, which already logs only the container label, never the value it hides."""
    messages = []
    redactor = Redactor("test-salt-0001", log=messages.append)
    redactor._guard_key("hardware_ids", "1234567890123")
    redactor._flush_warnings()
    assert any("hardware_ids" in message for message in messages), "expected a warning naming the container: {}".format(messages)
    assert not any("1234567890123" in message for message in messages), "the raw identifier must never reach the log: {}".format(messages)
    print("PASS: _guard_key logs only the container, never the raw key text it is hiding")
    return 0


def test_identity_derived_values_do_not_warn():
    """Octopus reports each meter as "octopus:{mpan}" with the MPAN in account_ids, and a programme
    cross-links to it. That is the redaction working as designed - every one of those is still
    pseudonymised - so it must not log a warning on every debug dump, whichever record is walked first."""
    base, coordinator = _redacting_coordinator()
    messages = []
    coordinator.log = messages.append
    coordinator.report(
        "octopus",
        {
            "programmes": [{"device_id": "axle:site1", "kind": "vpp", "meter": "octopus:1234567890123"}],
            "meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123", "station": 61234567890}}],
        },
    )
    coordinator.report("solis", {"inverters": [{"device_id": "solis:x1", "ratings": {"echo": 61234567890}}]})
    catalogue = coordinator.catalogue()
    text = str(catalogue)
    assert "1234567890123" not in text and "61234567890" not in text, text
    assert not [message for message in messages if "identifier" in message], messages
    print("PASS: an identifier redacted where it belongs, and its echoes, are not warned about")
    return 0


def test_misfiled_identifier_warns_once_per_process():
    """A genuine misfiling warns, but once - not again on every later debug dump or publish - and an
    account identifier alongside it does not hide an unrelated identifier sharing the same value."""
    base, coordinator = _redacting_coordinator()
    messages = []
    coordinator.log = messages.append
    coordinator.report(
        "octopus",
        {"meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123"}, "info": {"note": "1234567890123 and 9876543210987"}}]},
    )
    for _ in range(3):
        catalogue = coordinator.catalogue()
        assert "9876543210987" not in str(catalogue)
    warnings = [message for message in messages if "identifier" in message]
    assert warnings == ["Warn: Coordinator: info.note looks like an identifier in a clear container - pseudonymised"], warnings
    print("PASS: a genuine misfiling warns exactly once across repeated redactions")
    return 0


def test_short_or_unrelated_identities_do_not_hide_misfiled_values():
    """Suppression is for a value wholly explained by an account identifier: a short identifier
    must not split a longer number's digit run, a device_id that embeds no account identifier is
    not identity-derived, and a numeric identifier is judged by its integer part."""
    messages = []
    redactor = Redactor("test-salt-0001", log=messages.append)
    redactor.redact(
        {
            "meters": [
                {"device_id": "octopus:9876543210987", "direction": "import", "account_ids": {"site": "1"}, "info": {"long": "1213141516171"}},
                {"device_id": "octopus:5", "direction": "import", "account_ids": {"site": "1"}, "info": {"big": 1213141516171}},
            ]
        }
    )
    warnings = sorted(message for message in messages if "identifier" in message)
    assert any("info.long" in message for message in warnings), warnings
    assert any("info.big" in message for message in warnings), warnings
    assert any("device_id" in message for message in warnings), warnings
    print("PASS: short or unrelated identities do not hide genuinely misfiled values")
    return 0


def test_warning_is_once_per_record_and_ignores_joined_digit_runs():
    """A second record misfiling the same field is a new leak and warns too (once per record, not
    once per field); and removing an explained identifier must not join the digit runs either side
    of it into one that looks like an identifier."""
    base, coordinator = _redacting_coordinator()
    messages = []
    coordinator.log = messages.append

    def meter(device_id, mpan, note):
        """One Octopus-shaped meter record with a note in its clear info container."""
        return {"device_id": device_id, "direction": "import", "account_ids": {"mpan": mpan}, "info": {"note": note}}

    coordinator.report(
        "octopus",
        {"meters": [meter("octopus:1111111111111", "1111111111111", "9876543210987"), meter("octopus:2222222222222", "2222222222222", "9876543210988"), meter("octopus:3333333333333", "3333333333333", "ref 12345 3333333333333 67890")]},
    )
    for _ in range(2):
        coordinator.catalogue()
    warnings = [message for message in messages if "info.note" in message]
    assert len(warnings) == 2, warnings
    print("PASS: a warning is once per record, and an explained identifier does not join digit runs")
    return 0


# --- Review round 3: Task 7 review - a case/separator-transformed echo of a pseudonymised value ---


def test_identifier_variants_registered_for_case_and_separator_transforms():
    """Redactor._note() registers case-folded and separator-swapped variants of a noted original,
    all resolving to the SAME token, and marks each substring-eligible when the original was -
    this is the general mechanism a component's get_entity_name()-style helper (Octopus's own
    included, and Solcast's, which embeds a site id in an entity name the same way) relies on: such
    a helper commonly lower-cases an identifier and swaps "-" for "_" when folding it into an
    entity id, producing a string that is neither equal to, nor contains as a literal substring,
    the noted original."""
    redactor = Redactor("test-salt-0001")
    token = redactor._note("A-1234ABCD", substring=True)
    for variant in ("A-1234ABCD", "a-1234abcd", "A_1234ABCD", "a_1234abcd"):
        assert redactor.originals.get(variant) == token, "{} should map to the same token".format(variant)
        assert variant in redactor.substring_ok, "{} should be substring-eligible, matching the original's own substring=True".format(variant)
    print("PASS: _note registers case-folded and separator-swapped variants, all mapping to the same token")
    return 0


def test_identifier_variants_fold_up_as_well_as_down():
    """Final review: _identifier_variants() previously folded DOWN only (lower-casing), so an
    UPPER-cased echo of a lower-cased CANONICAL identifier survived - the reverse of the case every
    other test here exercises, where the noted original happens to already be upper/mixed case.
    Noting a lower-case original (an entity_id-shaped id, say) must still catch an upper-cased echo
    of it elsewhere, exactly as noting an upper-case original already caught a lower-cased echo."""
    redactor = Redactor("test-salt-0001")
    token = redactor._note("a-1234abcd", substring=True)
    for variant in ("a-1234abcd", "A-1234ABCD", "a_1234abcd", "A_1234ABCD"):
        assert redactor.originals.get(variant) == token, "{} should map to the same token".format(variant)
        assert variant in redactor.substring_ok, "{} should be substring-eligible, matching the original's own substring=True".format(variant)
    print("PASS: _note registers fold-UP variants too, not just fold-down, from a lower-case original")
    return 0


def test_pseudonymised_value_hidden_when_case_folded_and_separator_swapped_in_entity_id():
    """Task 7 review, Finding 1: a vendor helper shaped like Octopus's get_entity_name() lower-cases
    an identifier and swaps "-" for "_" when building an entity id, producing a DIFFERENT string
    from the noted original - "A-1234ABCD" becomes "a_1234abcd" - which neither the exact-match nor
    the (pre-fix, verbatim-only) substring pass would ever match, so it reached the redacted
    catalogue in the clear right next to its own pseudonymised original. MPAN is included alongside
    to confirm the fix does not disturb its own, independent protection - the all-digit shape guard,
    unaffected by this change."""
    base, coordinator = _redacting_coordinator()
    coordinator.report(
        "octopus",
        {
            "meters": [{"device_id": "octopus:1234567890123", "direction": "import", "account_ids": {"mpan": "1234567890123", "account": "A-1234ABCD"}}],
            "cars": [
                {
                    "device_id": "octopus:dev1",
                    "entities": {"octopus_intelligent_slot": {"entity_id": "binary_sensor.predbat_octopus_a_1234abcd_intelligent_dispatch_1", "domain": "binary_sensor", "access": "r"}},
                }
            ],
        },
    )
    catalogue = coordinator.catalogue()
    catalogue_text = str(catalogue)
    assert "A-1234ABCD" not in catalogue_text, "raw account id must not survive"
    assert "a_1234abcd" not in catalogue_text, "the case-folded, separator-swapped form must not survive either"
    assert "1234567890123" not in catalogue_text, "raw MPAN must still be hidden"
    entity = catalogue["cars"][0]["entities"]["octopus_intelligent_slot"]
    assert entity["domain"] == "binary_sensor" and entity["access"] == "r", "a legitimate neighbouring descriptor field is untouched"
    print("PASS: a case-folded, separator-swapped echo of a pseudonymised value is hidden, and MPAN protection is unaffected")
    return 0


# --- Review round 2: test gaps ---


class _StubStorage:
    """A minimal async Storage stand-in, recording save() calls and replaying them from load()."""

    def __init__(self):
        """Start with nothing stored."""
        self.saved = {}

    async def load(self, module, filename):
        """Return the previously saved payload for this module/filename, or None."""
        return self.saved.get((module, filename))

    async def save(self, module, filename, data, format="yaml", expiry=None, indent=None):
        """Record a payload under its module/filename key."""
        self.saved[(module, filename)] = data


def test_load_salt_fallback_without_storage():
    """Without a Storage component (MockBase.components is None), load_salt() falls back to a
    fresh per-process salt rather than running unsalted or reusing a fixed value - previously
    every test set coordinator.salt directly, so this branch had zero coverage."""
    base1, coordinator1 = _coordinator()
    salt1 = coordinator1.load_salt()
    assert len(salt1) == 32 and all(c in "0123456789abcdef" for c in salt1)
    base2, coordinator2 = _coordinator()
    salt2 = coordinator2.load_salt()
    assert salt2 != salt1, "two installations must not somehow end up with the same fallback salt"
    print("PASS: load_salt falls back to a fresh random salt without Storage")
    return 0


def test_load_salt_round_trips_through_storage():
    """With a Storage component available, load_salt() persists the salt and a second coordinator
    sharing that Storage loads the SAME salt rather than minting a new one - the other half of
    load_salt()'s contract that setting coordinator.salt directly in every other test never
    exercised."""
    base1, coordinator1 = _coordinator()
    storage = _StubStorage()
    base1.components = _StubRegistry(components={"storage": storage})
    salt1 = coordinator1.load_salt()

    base2, coordinator2 = _coordinator()
    base2.components = _StubRegistry(components={"storage": storage})
    salt2 = coordinator2.load_salt()

    assert salt1 == salt2, "a second coordinator must load the persisted salt, not mint a new one"
    assert ("coordinator", "salt") in storage.saved
    print("PASS: load_salt persists through Storage and a fresh coordinator reuses it")
    return 0


def test_report_discovery_helper():
    """A component reports through the base-class helper, and does nothing harmful when there is no coordinator."""
    from component_base import ComponentBase

    class _FakeComponent(ComponentBase):
        """Minimal component used to exercise report_discovery."""

        def initialize(self, **kwargs):
            """Nothing to set up."""
            pass

        async def run(self, seconds, first):
            """Never called here."""
            return True

    base, coordinator = _coordinator()
    component = _FakeComponent(base)
    component.component_name = "givtcp"
    component.report_discovery({"inverters": [{"device_id": "givtcp:A"}]})  # no coordinator yet - must not raise
    assert coordinator.reports == {}

    class _Registry:
        """Registry stub exposing the coordinator, as Components does."""

        def __init__(self):
            """Hold the coordinator."""
            self.coordinator = coordinator

        def get_all(self):
            """One known component."""
            return ["givtcp"]

        def is_active(self, name):
            """Active."""
            return True

        def is_alive(self, name):
            """Alive."""
            return True

        def load_error(self, name):
            """No error."""
            return None

    base.components = _Registry()
    component.report_discovery({"inverters": [{"device_id": "givtcp:A"}]})
    assert coordinator.reports["givtcp"]["inverters"][0]["device_id"] == "givtcp:A"
    print("PASS: report_discovery routes to the coordinator and no-ops without one")
    return 0


def test_publish_writes_sensor():
    """publish() writes a SUMMARY to the entity, not the full catalogue.

    Two inverters sharing a serial gives the test a real observations.conflicts entry to check
    threads through, and a hardware_ids container on each record gives it real per-section
    payload that must NOT leak into the entity's attributes - a test that only checked presence
    would not notice a regression back to the full attributes.update(catalogue) dump.
    """
    base, coordinator = _redacting_coordinator()
    coordinator.report("givtcp", {"inverters": [{"device_id": "givtcp:A", "inverter_type": "GE", "hardware_ids": {"serial": "SA2242G123"}}]})
    coordinator.report("gecloud", {"inverters": [{"device_id": "gecloud:b", "inverter_type": "GEC", "hardware_ids": {"serial": "SA2242G123"}}]})
    coordinator.assemble()
    coordinator.publish()
    entity = base.entities.get("sensor.predbat_discovery")
    assert entity, base.entities
    attributes = entity["attributes"]

    # Summary half: present and correct
    assert attributes["friendly_name"] == "Predbat discovery"
    assert attributes["icon"] == "mdi:sitemap"
    assert attributes["schema_version"] == SCHEMA_VERSION
    assert attributes["generated"]
    assert attributes["counts"]["inverters"] == 2, attributes["counts"]
    assert attributes["components"]["givtcp"]["status"] == "ok", attributes["components"]
    conflicts = attributes["observations"]["conflicts"]
    assert any(entry["kind"] == "duplicate_serial" for entry in conflicts), conflicts
    assert entity["state"] == sum(attributes["counts"].values())

    # NOT-present half: no section record list anywhere in the attributes
    for section in SECTION_SPEC:
        assert section not in attributes, "{} record list must not reach the entity - it bloats the HA recorder".format(section)

    print("PASS: discovery sensor carries a summary (counts, components, conflicts), never the full catalogue")
    return 0


def _one_inverter(**fields):
    """Validate a single inverter record carrying the given fields and return what survives."""
    record = {"device_id": "test:SN1", "inverter_type": "SunsynkCloud"}
    record.update(fields)
    logs = []
    cleaned = validate_report({"inverters": [record]}, "test", logs.append)
    return (cleaned.get("inverters") or [{}])[0], logs


def test_schema_version_is_two():
    """The document shape changed (dict capabilities, renamed ratings, required access), so the version moved."""
    assert SCHEMA_VERSION == 2
    print("PASS: schema version 2")
    return 0


def test_capabilities_dict_keeps_known_bool_keys():
    """A capabilities dict of the seven behaviour keys survives validation intact."""
    capabilities = {key: True for key in CAPABILITY_KEYS}
    capabilities["can_span_midnight"] = False
    record, _ = _one_inverter(capabilities=capabilities)
    assert record["capabilities"] == capabilities, record
    print("PASS: capabilities dict preserved")
    return 0


def test_capabilities_drops_unknown_key_and_non_bool():
    """An unknown key, or a value that is not a bool, is dropped and logged; the rest survives."""
    record, logs = _one_inverter(capabilities={"support_charge_freeze": True, "has_reserve_soc": True, "can_span_midnight": 1})
    assert record["capabilities"] == {"support_charge_freeze": True}, record
    assert any("has_reserve_soc" in line for line in logs), logs
    assert any("can_span_midnight" in line for line in logs), logs
    print("PASS: capabilities rejects unknown keys and non-bools")
    return 0


def test_capabilities_list_is_dropped():
    """Schema 2: capabilities is a dict of the seven behaviour keys, so a token list no longer validates."""
    record, _ = _one_inverter(capabilities=["schedule", "target_soc"])
    assert "capabilities" not in record, record
    base, coordinator = _coordinator()
    coordinator.report("test", {"inverters": [{"device_id": "test:SN1", "capabilities": ["schedule"]}]})
    assert "capabilities" not in coordinator.catalogue()["inverters"][0]
    print("PASS: capability token list dropped")
    return 0


def test_descriptor_requires_access():
    """An entities descriptor without access rw or r is dropped."""
    record, _ = _one_inverter(entities={
        "soc_percent": {"entity_id": "sensor.a", "access": "r"},
        "charge_limit": {"entity_id": "number.b"},
        "reserve": {"entity_id": "number.c", "access": "write"},
    })
    assert set(record["entities"]) == {"soc_percent"}, record
    print("PASS: descriptor needs access")
    return 0


def test_descriptor_needs_exactly_one_of_entity_id_and_value():
    """entity_id and value are alternatives: both, or neither, drops the descriptor."""
    record, _ = _one_inverter(entities={
        "pv_power": {"value": 0, "access": "r"},
        "grid_power": {"entity_id": "sensor.g", "value": 0, "access": "r"},
        "load_power": {"access": "r"},
    })
    assert record["entities"] == {"pv_power": {"value": 0, "access": "r"}}, record
    print("PASS: exactly one of entity_id/value")
    return 0


def test_descriptor_value_rejects_free_text():
    """A string value must pass the info-string guard: no "@", no over-long text."""
    record, _ = _one_inverter(entities={
        "battery_power_invert_note": {"value": "True", "access": "r"},
        "email": {"value": "someone@example.com", "access": "r"},
        "blob": {"value": "x" * 200, "access": "r"},
    })
    assert set(record["entities"]) == {"battery_power_invert_note"}, record
    print("PASS: descriptor value rejects free text")
    return 0


def test_descriptor_invert_must_be_bool():
    """invert is kept when it is a bool and dropped (descriptor kept) when it is not."""
    record, _ = _one_inverter(entities={
        "grid_power": {"entity_id": "sensor.g", "access": "r", "invert": True},
        "battery_power": {"entity_id": "sensor.b", "access": "r", "invert": "yes"},
    })
    assert record["entities"]["grid_power"]["invert"] is True, record
    assert "invert" not in record["entities"]["battery_power"], record
    print("PASS: invert must be bool")
    return 0


def test_descriptor_step_percent_of_capacity_must_be_a_percentage():
    """step_percent_of_capacity is kept when it is a percentage above 0 and up to 100, dropped otherwise (#5324)."""
    record, _ = _one_inverter(entities={
        "charge_rate": {"entity_id": "number.c", "access": "rw", "step_percent_of_capacity": 1},
        "discharge_rate": {"entity_id": "number.d", "access": "rw", "step_percent_of_capacity": True},
        "charge_limit": {"entity_id": "number.l", "access": "rw", "step_percent_of_capacity": 0},
        "reserve": {"entity_id": "number.r", "access": "rw", "step_percent_of_capacity": 150},
    })
    assert record["entities"]["charge_rate"]["step_percent_of_capacity"] == 1, record
    for setting in ("discharge_rate", "charge_limit", "reserve"):
        assert "step_percent_of_capacity" not in record["entities"][setting], record
    print("PASS: step_percent_of_capacity must be a percentage")
    return 0


def test_new_containers_survive_redaction_unchanged():
    """Bool capabilities, a strftime format and a string value stand-in pass the redactor untouched."""
    base, coordinator = _coordinator()
    coordinator.report("test", {"inverters": [{
        "device_id": "test:SN1",
        "capabilities": {"support_charge_freeze": True},
        "entities": {
            "inverter_time": {"entity_id": "sensor.t", "access": "r", "format": "%Y-%m-%d %H:%M:%S"},
            "pv_power": {"value": "0", "access": "r"},
        },
    }]})
    record = coordinator.catalogue()["inverters"][0]
    assert record["capabilities"] == {"support_charge_freeze": True}, record
    assert record["entities"]["inverter_time"]["format"] == "%Y-%m-%d %H:%M:%S", record
    assert record["entities"]["pv_power"]["value"] == "0", record
    print("PASS: new containers survive redaction")
    return 0


def _full_record():
    """A record carrying every source inverter_definition() reads, shaped like a cloud inverter."""
    return {
        "device_id": "test:SN1",
        "inverter_type": "SunsynkCloud",
        "capabilities": {key: key != "can_span_midnight" for key in CAPABILITY_KEYS},
        "entities": {
            "soc_percent": {"entity_id": "sensor.soc", "access": "r", "unit": "%"},
            "charge_rate": {"entity_id": "number.rate", "access": "rw", "unit": "W"},
            "charge_start_time": {"entity_id": "select.start", "access": "rw", "domain": "select", "format": "HH:MM:SS"},
            "charge_limit": {"entity_id": "number.limit", "access": "rw"},
            "reserve": {"entity_id": "number.reserve", "access": "rw"},
            "scheduled_charge_enable": {"entity_id": "switch.c", "access": "rw"},
            "scheduled_discharge_enable": {"entity_id": "switch.d", "access": "rw"},
            "schedule_write_button": {"entity_id": "switch.w", "access": "rw"},
        },
    }


def test_inverter_definition_builds_every_field_without_a_base():
    """A full record yields every derived field and no gaps."""
    definition, gaps, not_applicable = inverter_definition(_full_record(), 2)
    assert gaps == [], gaps
    assert not_applicable == ["clock_time_format", "current_dp"], not_applicable
    assert definition["support_charge_freeze"] is True and definition["can_span_midnight"] is False
    assert definition["has_charge_enable_time"] and definition["has_discharge_enable_time"]
    assert definition["has_target_soc"] and definition["has_reserve_soc"]
    assert not definition["has_idle_time"] and not definition["has_timed_pause"]
    assert not definition["has_ge_inverter_mode"] and not definition["has_ge_eco_toggle"] and not definition["has_mqtt_api"]
    assert definition["charge_time_entity_is_option"] is True and definition["charge_time_format"] == "HH:MM:SS"
    assert definition["clock_time_format"] == NOT_APPLICABLE_DEFAULTS["clock_time_format"]
    assert definition["soc_units"] == "%" and definition["output_charge_control"] == "power"
    assert definition["time_button_press"] is True and definition["num_load_entities"] == 1
    assert definition["write_and_poll_sleep"] == 2 and definition["name"] == "SunsynkCloud"
    print("PASS: definition built without a base")
    return 0


def test_inverter_definition_presence_needs_a_real_rw_entity():
    """An r entry, or a value stand-in, does not make a presence flag True."""
    record = _full_record()
    record["entities"]["reserve"] = {"entity_id": "number.reserve", "access": "r"}
    record["entities"]["charge_limit"] = {"value": 100, "access": "rw"}
    record["entities"]["pause_mode"] = {"entity_id": "select.pause", "access": "rw"}
    definition, _, _ = inverter_definition(record, 2)
    assert definition["has_reserve_soc"] is False and definition["has_target_soc"] is False
    assert definition["has_timed_pause"] is True
    print("PASS: presence needs rw entity")
    return 0


def test_inverter_definition_rate_step_from_charge_rate():
    """rate_step_percent_of_capacity comes from charge_rate's step_percent_of_capacity; absent, base's value stands, else 0 (#5324)."""
    record = _full_record()
    definition, gaps, _ = inverter_definition(record, 2)
    assert definition["rate_step_percent_of_capacity"] == 0 and "rate_step_percent_of_capacity" not in gaps, definition
    definition, _, _ = inverter_definition(record, 2, base={"rate_step_percent_of_capacity": 1})
    assert definition["rate_step_percent_of_capacity"] == 1, definition
    record["entities"]["charge_rate"]["step_percent_of_capacity"] = 1
    definition, _, _ = inverter_definition(record, 2)
    assert definition["rate_step_percent_of_capacity"] == 1, definition
    del record["entities"]["charge_rate"]
    definition, _, _ = inverter_definition(record, 2)
    assert definition["rate_step_percent_of_capacity"] == 0, definition
    print("PASS: rate step from charge_rate")
    return 0


def test_inverter_definition_ge_mode_flags_follow_inverter_mode_domain():
    """inverter_mode as a select means a GE inverter-mode select; as a switch, GE Cloud's eco toggle."""
    record = _full_record()
    record["entities"]["inverter_mode"] = {"entity_id": "select.mode", "access": "rw", "domain": "select"}
    definition, _, _ = inverter_definition(record, 2)
    assert definition["has_ge_inverter_mode"] is True and definition["has_ge_eco_toggle"] is False
    record["entities"]["inverter_mode"] = {"entity_id": "switch.eco", "access": "rw", "domain": "switch"}
    definition, _, _ = inverter_definition(record, 2)
    assert definition["has_ge_inverter_mode"] is False and definition["has_ge_eco_toggle"] is True
    print("PASS: GE mode flags from inverter_mode domain")
    return 0


def test_inverter_definition_charge_rate_units():
    """Amps mean current control with decimal places from the step; no charge_rate means none, unless charge_rate_percent is bound."""
    record = _full_record()
    record["entities"]["charge_rate"] = {"entity_id": "number.amps", "access": "rw", "unit": "A", "step": 0.1}
    definition, gaps, not_applicable = inverter_definition(record, 2)
    assert definition["output_charge_control"] == "current" and definition["current_dp"] == 1, definition
    assert "current_dp" not in not_applicable and gaps == []
    record["entities"]["charge_rate"]["step"] = 1
    assert inverter_definition(record, 2)[0]["current_dp"] == 0
    del record["entities"]["charge_rate"]
    definition, _, _ = inverter_definition(record, 2)
    assert definition["output_charge_control"] == "none"
    record["entities"]["charge_rate_percent"] = {"entity_id": "number.pct", "access": "rw", "unit": "%"}
    definition, gaps, _ = inverter_definition(record, 2)
    assert definition["output_charge_control"] == "power" and "output_charge_control" not in gaps, "a percentage-rate model is still power control"
    del record["entities"]["charge_rate_percent"]
    record["entities"]["charge_rate"] = {"entity_id": "number.rate", "access": "rw"}
    _, gaps, _ = inverter_definition(record, 2)
    assert "output_charge_control" in gaps, gaps
    print("PASS: charge_rate unit handling")
    return 0


def test_inverter_definition_gaps_and_not_applicable():
    """A bound entity missing its format is a gap; an unbound source is not applicable; a missing capability is a gap."""
    record = _full_record()
    del record["entities"]["charge_start_time"]["format"]
    del record["capabilities"]["support_feedin_first"]
    record["entities"]["inverter_time"] = {"entity_id": "sensor.t", "access": "r", "format": "%H:%M:%S"}
    definition, gaps, not_applicable = inverter_definition(record, 2)
    assert "charge_time_format" in gaps and "support_feedin_first" in gaps, gaps
    assert "charge_time_format" not in definition and "support_feedin_first" not in definition
    assert definition["clock_time_format"] == "%H:%M:%S" and "clock_time_format" not in not_applicable
    del record["entities"]["charge_start_time"]
    _, gaps, not_applicable = inverter_definition(record, 2)
    assert "charge_time_format" in not_applicable and "charge_time_format" not in gaps
    print("PASS: gaps vs not applicable")
    return 0


def test_inverter_definition_counts_load_entities():
    """num_load_entities is 1 plus the consecutive load_power_N entities bound."""
    record = _full_record()
    record["entities"]["load_power"] = {"entity_id": "sensor.l0", "access": "r"}
    record["entities"]["load_power_1"] = {"entity_id": "sensor.l1", "access": "r"}
    record["entities"]["load_power_3"] = {"entity_id": "sensor.l3", "access": "r"}
    assert inverter_definition(record, 2)[0]["num_load_entities"] == 2
    print("PASS: load entity count")
    return 0


def test_inverter_definition_base_fills_gaps_and_is_never_mutated():
    """With the live INVERTER_DEF row as base, a gap keeps the base value and the row is untouched."""
    row = INVERTER_DEF["SunsynkCloud"]
    before = dict(row)
    record = _full_record()
    record["capabilities"] = {}
    definition, gaps, _ = inverter_definition(record, 2, base=row)
    assert set(CAPABILITY_KEYS) <= set(gaps)
    assert definition["support_charge_freeze"] == row["support_charge_freeze"]
    definition["support_charge_freeze"] = "changed"
    assert INVERTER_DEF["SunsynkCloud"] == before, "inverter_definition() must never mutate INVERTER_DEF"
    print("PASS: base fills gaps, never mutated")
    return 0


def test_coordinator_all(my_predbat=None):
    """Run every coordinator test, returning the number of failures."""
    failures = 0
    failures += test_report_keeps_valid_containers()
    failures += test_ratings_rejects_non_numeric()
    failures += test_vocabulary_rejects_free_text()
    failures += test_info_rejects_email_shaped_and_long_strings()
    failures += test_credential_named_field_rejected()
    failures += test_entities_credential_named_key_rejected()
    failures += test_entities_free_text_unit_dropped_legitimate_descriptor_intact()
    failures += test_entities_options_preserves_realistic_values()
    failures += test_coverage_accepts_numbers_booleans_and_vocabulary_lists()
    failures += test_validate_report_never_raises_on_non_dict()
    failures += test_validate_report_never_raises_on_a_non_list_section()
    failures += test_automatic_field_absent_when_component_omits_it()
    failures += test_unknown_container_dropped()
    failures += test_record_without_device_id_dropped()
    failures += test_report_is_idempotent_and_versioned()
    failures += test_meter_sub_record_validated()
    failures += test_credential_guard_fires_inside_sub_record_container()
    failures += test_assemble_merges_sections_and_tags_source()
    failures += test_catalogue_reassembles_so_a_post_assembly_report_reaches_it()
    failures += test_assemble_component_status()
    failures += test_inverter_record_omits_empty_containers()
    failures += test_inverter_record_keeps_everything_populated()
    failures += test_inverter_record_round_trips_through_validation()
    failures += test_inverter_record_parameters_match_the_inverters_schema()
    failures += test_inverter_record_fields_are_keyword_only()
    failures += test_inverter_record_copies_the_callers_containers()
    failures += test_inverter_record_normalises_sets_and_tuples()
    failures += test_inverter_record_sorts_a_mixed_type_set_without_raising()
    failures += test_inverter_record_drops_an_empty_string()
    failures += test_component_status_omits_components_that_cannot_report()
    failures += test_component_status_reported_at_set_only_for_ok()
    failures += test_observations_duplicate_serial()
    failures += test_observations_duplicate_serial_sees_a_gateways_fronted_batteries()
    failures += test_component_load_error_carries_no_free_text()
    failures += test_observations_contested_cars_and_meters()
    failures += test_observations_resulting_config()
    failures += test_account_ids_pseudonymised_and_stable()
    failures += test_account_id_noted_in_two_forms_across_records_resolves_to_one_token()
    failures += test_note_returns_same_token_when_called_twice_with_different_variants()
    failures += test_cross_link_resolves_without_coincidental_substring()
    failures += test_measures_meter_cross_link_resolves()
    failures += test_pseudonym_differs_across_salts()
    failures += test_pseudonym_substituted_inside_entity_ids()
    failures += test_serials_and_tariff_codes_stay_clear()
    failures += test_misfiled_identifier_caught_by_shape_guard()
    failures += test_misfiled_identifier_inside_entity_descriptor_caught()
    failures += test_misfiled_email_inside_entity_id_caught()
    failures += test_misfiled_identifier_inside_vocabulary_list_caught()
    failures += test_catalogue_raw_is_unredacted()
    failures += test_misfiled_float_identifier_caught()
    failures += test_misfiled_identifier_embedded_or_separated_caught()
    failures += test_misfiled_identifier_grouped_by_underscore_or_comma_caught()
    failures += test_location_shaped_key_pseudonymised_regardless_of_value_shape()
    failures += test_legitimate_long_float_survives_unchanged_and_still_numeric()
    failures += test_hardware_ids_only_flags_all_digit_values_not_prefixed_serials()
    failures += test_declared_serials_stay_readable_wherever_they_appear()
    failures += test_declared_serials_do_not_shelter_a_misfiled_identifier()
    failures += test_serial_derived_device_id_needs_whole_tokens()
    failures += test_pseudonymised_value_substituted_inside_entity_id_value()
    failures += test_account_ids_value_does_not_corrupt_structural_or_descriptor_keys()
    failures += test_account_ids_value_equal_to_a_structural_name_does_not_delete_it()
    failures += test_int_identifier_echoed_outside_guarded_container_is_substituted()
    failures += test_numeric_identifier_echo_substituted_regardless_of_int_float_or_string_form()
    failures += test_structural_scalar_shape_guard_catches_bare_identifier_device_id()
    failures += test_device_id_without_identity_data_never_corrupts_unrelated_text()
    failures += test_identity_derived_device_id_substituted_wherever_it_is_echoed()
    failures += test_pseudonym_container_nested_in_sub_record_still_marks_device_id_identity_derived()
    failures += test_substitution_does_not_touch_the_catalogue_timestamp()
    failures += test_shorter_original_does_not_fragment_a_longer_one()
    failures += test_misfiled_identifier_used_as_a_container_key_caught()
    failures += test_guard_key_does_not_log_the_raw_value()
    failures += test_identity_derived_values_do_not_warn()
    failures += test_misfiled_identifier_warns_once_per_process()
    failures += test_short_or_unrelated_identities_do_not_hide_misfiled_values()
    failures += test_warning_is_once_per_record_and_ignores_joined_digit_runs()
    failures += test_identifier_variants_registered_for_case_and_separator_transforms()
    failures += test_identifier_variants_fold_up_as_well_as_down()
    failures += test_pseudonymised_value_hidden_when_case_folded_and_separator_swapped_in_entity_id()
    failures += test_load_salt_fallback_without_storage()
    failures += test_load_salt_round_trips_through_storage()
    failures += test_report_discovery_helper()
    failures += test_publish_writes_sensor()
    failures += test_schema_version_is_two()
    failures += test_capabilities_dict_keeps_known_bool_keys()
    failures += test_capabilities_drops_unknown_key_and_non_bool()
    failures += test_capabilities_list_is_dropped()
    failures += test_descriptor_requires_access()
    failures += test_descriptor_needs_exactly_one_of_entity_id_and_value()
    failures += test_descriptor_value_rejects_free_text()
    failures += test_descriptor_invert_must_be_bool()
    failures += test_descriptor_step_percent_of_capacity_must_be_a_percentage()
    failures += test_new_containers_survive_redaction_unchanged()
    failures += test_inverter_definition_builds_every_field_without_a_base()
    failures += test_inverter_definition_presence_needs_a_real_rw_entity()
    failures += test_inverter_definition_ge_mode_flags_follow_inverter_mode_domain()
    failures += test_inverter_definition_rate_step_from_charge_rate()
    failures += test_inverter_definition_charge_rate_units()
    failures += test_inverter_definition_gaps_and_not_applicable()
    failures += test_inverter_definition_counts_load_entities()
    failures += test_inverter_definition_base_fills_gaps_and_is_never_mutated()
    return failures