Skip to content
14 changes: 13 additions & 1 deletion custom_components/zaptec/zaptec/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -269,19 +269,31 @@ async def build(self) -> None:
redact = self.zaptec.redact

self.chargers = []
if not hierarchy.get("Circuits"):
_LOGGER.warning("Installation %s contains no circuits.", self.qual_id)
return

for circuit in hierarchy["Circuits"]:
ctid = circuit["Id"]
redact.add_uid(ctid, "Circuit")
_LOGGER.debug(" Circuit %s", redact(ctid))

if not circuit.get("Chargers"):
_LOGGER.warning(
"Circuit %s of installation %s contains no chargers.",
redact(ctid),
self.qual_id,
)
continue

for charger_item in circuit["Chargers"]:
chgid = charger_item["Id"]
redact.add_uid(chgid, "Charger")

# Inject additional attributes
charger_item["InstallationId"] = self.id
charger_item["CircuitId"] = ctid
charger_item["CircuitName"] = circuit["Name"]
charger_item["CircuitName"] = circuit.get("Name")
charger_item["CircuitMaxCurrent"] = circuit["MaxCurrent"]

# Add or update the charger
Expand Down
33 changes: 17 additions & 16 deletions custom_components/zaptec/zaptec/validate.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,27 +16,27 @@ class Installation(BaseModel):

model_config = ConfigDict(extra="allow")
Id: str
Active: bool
CurrentUserRoles: int
InstallationType: int
NetworkType: int
Active: bool | None = None
CurrentUserRoles: int | None = None
InstallationType: int | None = None
NetworkType: int | None = None


class Installations(BaseModel):
"""Pydantic model for a list of Zaptec installations."""

model_config = ConfigDict(extra="allow")
Data: list[Installation]
Pages: int
Pages: int | None = None


class Charger(BaseModel):
"""Pydantic model for a Zaptec charger."""

model_config = ConfigDict(extra="allow")
Id: str
Name: str
Active: bool
Name: str | None = None
Active: bool | None = None
DeviceType: int


Expand All @@ -61,17 +61,18 @@ class Circuit(BaseModel):

model_config = ConfigDict(extra="allow")
Id: str
Name: str
Chargers: list[Charger]
Name: str | None = None
MaxCurrent: float
Chargers: list[Charger] | None = None


class Hierarchy(BaseModel):
"""Pydantic model for the hierarchy of Zaptec objects in an installation."""

model_config = ConfigDict(extra="allow")
Id: str
Name: str
NetworkType: int
Name: str | None = None
NetworkType: int | None = None
Circuits: list[Circuit]


Expand All @@ -80,11 +81,11 @@ class ChargerFirmware(BaseModel):

model_config = ConfigDict(extra="allow")
ChargerId: str
DeviceType: int
IsOnline: bool
CurrentVersion: str
AvailableVersion: str
IsUpToDate: bool
DeviceType: int | None = None
IsOnline: bool | None = None
CurrentVersion: str | None = None
AvailableVersion: str | None = None
IsUpToDate: bool | None = None


class InstallationConnectionDetails(BaseModel):
Expand Down
68 changes: 68 additions & 0 deletions tests/zaptec/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -968,6 +968,74 @@ def test_zaptec_collections_and_accessors() -> None:
assert set(zap) == {"i1"}


@pytest.mark.asyncio
async def test_build_hierarchy_handles_null_circuit_chargers() -> None:
"""A null Circuit.Chargers (nullable per the API docs) yields no chargers, not a TypeError."""

hierarchy_payload = {
"Id": "abcdef01-2345-6789-abcd-ef0123456789",
"Circuits": [
{
"Id": "11111111-1111-1111-1111-111111111111",
"MaxCurrent": 32.0,
"Chargers": None,
},
],
}
zap, _ = _make_zaptec([FakeResponse(HTTPStatus.OK, json_data=hierarchy_payload)])
inst = Installation({"Id": "abcdef01-2345-6789-abcd-ef0123456789"}, zap)
zap.register(inst.id, inst)

await inst.build()

assert inst.chargers == []


@pytest.mark.asyncio
async def test_build_hierarchy_charger_with_device_type_survives_full_build() -> None:
"""A hierarchy-only charger survives Zaptec.build()'s chg["DeviceType"] subscript.

A charger seen only via the hierarchy is never re-merged with the /chargers
list, so its DeviceType comes solely from the hierarchy stub -- which the
Charger model now requires. Drives the full build() to prove no KeyError.
"""
inst_id = "abcdef01-2345-6789-abcd-ef0123456789"
charger_id = "12345678-90ab-cdef-1234567890ab"

outcomes = [
FakeResponse(HTTPStatus.OK, json_data={}), # constants
FakeResponse( # installation list
HTTPStatus.OK, json_data={"Pages": 1, "Data": [{"Id": inst_id}]}
),
FakeResponse( # installation/{id}/hierarchy
HTTPStatus.OK,
json_data={
"Id": inst_id,
"Circuits": [
{
"Id": "11111111-1111-1111-1111-111111111111",
"MaxCurrent": 32.0,
"Chargers": [{"Id": charger_id, "DeviceType": 4}],
},
],
},
),
FakeResponse( # chargers list
HTTPStatus.OK,
json_data={"Pages": 1, "Data": [{"Id": charger_id, "DeviceType": 4}]},
),
]
zap, _ = _make_zaptec(outcomes)

await zap.build()

assert zap.is_built
charger: Charger = zap[charger_id]
# str(val) fallback: the constants schema has no DeviceType entry here. The
# point is only that build() populated it at all (no KeyError on subscript).
assert charger["DeviceType"] == "4"


@pytest.mark.asyncio
async def test_zaptec_async_context_manager_closes_internal_client() -> None:
"""Entering/exiting the context manager works with an internally-created client."""
Expand Down
160 changes: 158 additions & 2 deletions tests/zaptec/test_validate.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,9 +90,21 @@ def test_installation_validation() -> None:
with pytest.raises(ValidationError):
validate(invalid_installation_list, installation_list_url)

# check that an installation missing NetworkType fails validation
# Users without the Owner/Service role get a reduced installation object
# missing Active/CurrentUserRoles/InstallationType/NetworkType (see #357).
# api.py only ever indexes Id directly, so this must still validate.
limited_installation = {"Id": valid_installation["Id"]}
validate(limited_installation, single_installation_url)

limited_installation_list = {
"Pages": 1,
"Data": [limited_installation],
}
validate(limited_installation_list, installation_list_url)

# Id is required: Zaptec.build() indexes inst_item["Id"] directly.
invalid_installation = valid_installation.copy()
invalid_installation.pop("NetworkType")
invalid_installation.pop("Id")
with pytest.raises(ValidationError):
validate(invalid_installation, single_installation_url)

Expand All @@ -104,6 +116,150 @@ def test_installation_validation() -> None:
validate(invalid_installation_list2, installation_list_url)


def test_charger_validation() -> None:
"""Check validation of /chargers and /chargers/{id} responses."""

chargers_list_url = "chargers"
single_charger_url = "chargers/12345678-90ab-cdef-1234567890ab"

valid_charger = {
"Id": "12345678-90ab-cdef-1234567890ab",
"Name": "Garage",
"Active": True,
"DeviceType": 4,
}
validate(valid_charger, single_charger_url)
validate({"Pages": 1, "Data": [valid_charger]}, chargers_list_url)

# Users without the Owner role get a reduced charger object missing
# Name/Active; only Id and DeviceType are consumed directly by api.py.
limited_charger = {"Id": valid_charger["Id"], "DeviceType": 4}
validate(limited_charger, single_charger_url)
validate({"Pages": 1, "Data": [limited_charger]}, chargers_list_url)

# DeviceType is required: Zaptec.build() indexes chg["DeviceType"] on
# every registered charger once merged from the /chargers list.
missing_device_type = {"Id": valid_charger["Id"]}
with pytest.raises(ValidationError):
validate(missing_device_type, single_charger_url)

# Id is required: Zaptec.build() indexes charger_item["Id"] directly.
missing_id = {"DeviceType": 4}
with pytest.raises(ValidationError):
validate(missing_id, single_charger_url)


def test_hierarchy_validation() -> None:
"""Check validation of installation/{id}/hierarchy responses."""

hierarchy_url = "installation/abcdef01-2345-6789-abcd-ef0123456789/hierarchy"

valid_hierarchy = {
"Id": "abcdef01-2345-6789-abcd-ef0123456789",
"Name": "Main hierarchy",
"NetworkType": 2,
"Circuits": [
{
"Id": "11111111-1111-1111-1111-111111111111",
"Name": "Circuit 1",
"MaxCurrent": 32.0,
"Chargers": [
{"Id": "12345678-90ab-cdef-1234567890ab", "DeviceType": 4},
],
},
],
}
validate(valid_hierarchy, hierarchy_url)

hierarchy_id = "abcdef01-2345-6789-abcd-ef0123456789"

# Name/NetworkType on the hierarchy itself aren't read by api.py, and per
# the Zaptec API docs a circuit's Name/Chargers may be null -- all of this
# must still validate. The hierarchy Id, however, is always present.
minimal_hierarchy = {
"Id": hierarchy_id,
"Circuits": [
{
"Id": "11111111-1111-1111-1111-111111111111",
"MaxCurrent": 32.0,
"Chargers": None,
},
],
}
validate(minimal_hierarchy, hierarchy_url)

# The hierarchy Id is required: the response always carries it.
missing_hierarchy_id = {
"Circuits": [
{"Id": "11111111-1111-1111-1111-111111111111", "MaxCurrent": 32.0},
],
}
with pytest.raises(ValidationError):
validate(missing_hierarchy_id, hierarchy_url)

# MaxCurrent is required: Installation.build() indexes
# circuit["MaxCurrent"] directly with no validation coverage today --
# exactly the class of bug #359 asks to close.
missing_max_current = {
"Id": hierarchy_id,
"Circuits": [{"Id": "11111111-1111-1111-1111-111111111111"}],
}
with pytest.raises(ValidationError):
validate(missing_max_current, hierarchy_url)

# A circuit's Id is required: Installation.build() indexes circuit["Id"].
missing_circuit_id = {
"Id": hierarchy_id,
"Circuits": [{"MaxCurrent": 32.0}],
}
with pytest.raises(ValidationError):
validate(missing_circuit_id, hierarchy_url)

# DeviceType is required: Zaptec.build() hard-subscripts chg["DeviceType"],
# including hierarchy-only chargers never re-merged with the /chargers list.
missing_device_type_in_hierarchy = {
"Id": hierarchy_id,
"Circuits": [
{
"Id": "11111111-1111-1111-1111-111111111111",
"MaxCurrent": 32.0,
"Chargers": [{"Id": "12345678-90ab-cdef-1234567890ab"}],
},
],
}
with pytest.raises(ValidationError):
validate(missing_device_type_in_hierarchy, hierarchy_url)


def test_charger_firmware_validation() -> None:
"""Check validation of chargerFirmware/installation/{id} responses."""

firmware_url = "chargerFirmware/installation/abcdef01-2345-6789-abcd-ef0123456789"

valid_firmware = [
{
"ChargerId": "12345678-90ab-cdef-1234567890ab",
"DeviceType": 4,
"IsOnline": True,
"CurrentVersion": "1.2.3",
"AvailableVersion": "1.2.4",
"IsUpToDate": False,
},
]
validate(valid_firmware, firmware_url)

# A charger not yet initialized reports only ChargerId; poll_firmware_info()
# treats the rest as optional. All fields except ChargerId are nullable, so
# validation must not reject this before that defensive code runs.
uninitialized_firmware = [{"ChargerId": "12345678-90ab-cdef-1234567890ab"}]
validate(uninitialized_firmware, firmware_url)

# ChargerId is required: poll_firmware_info() indexes fm["ChargerId"] directly.
missing_charger_id = [{"DeviceType": 4}]
with pytest.raises(ValidationError):
validate(missing_charger_id, firmware_url)


def test_missing_and_skipped_validation() -> None:
"""Check that unknown urls and urls setup with None as the Validation model pass."""

Expand Down
Loading