jonas
71803418e5
Shelly device management app with mDNS/subnet discovery, inventory, configuration, and mass operations for Gen1/Gen2+ devices. Includes .gitignore excluding runtime data (device DB, user config), AI conversation history, build artifacts, and common Python/OS patterns.
236 lines
7.4 KiB
Python
236 lines
7.4 KiB
Python
"""Mass config helper filters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from shelly_manager.core.models import ShellyDevice
|
|
from shelly_manager.ui.mass_config_helpers import (
|
|
apply_filters,
|
|
device_to_row,
|
|
firmware_update_category,
|
|
firmware_update_display,
|
|
prepare_table_rows_for_display,
|
|
)
|
|
|
|
|
|
def _dev(name: str, gen: int, online: bool) -> ShellyDevice:
|
|
return ShellyDevice(
|
|
id="AABBCCDDEEFF",
|
|
name=name,
|
|
mac="AA:BB:CC:DD:EE:FF",
|
|
ip="192.168.1.10",
|
|
generation=gen, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1.0",
|
|
online=online,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=["switch"],
|
|
status={},
|
|
settings={"ble": {"enable": True}},
|
|
tags=["t1"],
|
|
)
|
|
|
|
|
|
def test_apply_filters_name() -> None:
|
|
a = _dev("Kitchen", 2, True)
|
|
b = _dev("Garage", 2, True)
|
|
rows = [device_to_row(a), device_to_row(b)]
|
|
out = apply_filters(rows, name_q="kit")
|
|
assert len(out) == 1
|
|
assert out[0]["name"] == "Kitchen"
|
|
|
|
|
|
def test_needs_reboot_column_and_filter() -> None:
|
|
yes = ShellyDevice(
|
|
id="AA",
|
|
name="A",
|
|
mac="AA:BB:CC:DD:EE:01",
|
|
ip="192.168.1.1",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={"sys": {"restart_required": True}},
|
|
settings={},
|
|
)
|
|
no = ShellyDevice(
|
|
id="BB",
|
|
name="B",
|
|
mac="AA:BB:CC:DD:EE:02",
|
|
ip="192.168.1.2",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={"sys": {"restart_required": False}},
|
|
settings={},
|
|
)
|
|
ra, rb = device_to_row(yes), device_to_row(no)
|
|
assert ra["needs_reboot"] == "yes"
|
|
assert rb["needs_reboot"] == "no"
|
|
rows = [ra, rb]
|
|
assert len(apply_filters(rows, needs_reboot="yes")) == 1
|
|
assert apply_filters(rows, needs_reboot="yes")[0]["device_id"] == "AA"
|
|
|
|
|
|
def test_apply_filters_versions_q() -> None:
|
|
a = _dev("Kitchen", 2, True)
|
|
rows = [
|
|
device_to_row(a, version_summaries=["2026-03-19 12:00 — mass:ble_disable"]),
|
|
device_to_row(_dev("Other", 2, True)),
|
|
]
|
|
out = apply_filters(rows, versions_q="mass:ble")
|
|
assert len(out) == 1
|
|
assert "mass:ble" in out[0]["versions"]
|
|
|
|
|
|
def test_firmware_update_column_and_filter() -> None:
|
|
gen1 = _dev("G1", 1, True)
|
|
unchecked = ShellyDevice(
|
|
id="U1",
|
|
name="U",
|
|
mac="AA:BB:CC:DD:EE:03",
|
|
ip="192.168.1.3",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={},
|
|
settings={},
|
|
firmware_check_result=None,
|
|
)
|
|
uptodate = ShellyDevice(
|
|
id="U2",
|
|
name="U2",
|
|
mac="AA:BB:CC:DD:EE:04",
|
|
ip="192.168.1.4",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={},
|
|
settings={},
|
|
firmware_check_result={},
|
|
)
|
|
offer = ShellyDevice(
|
|
id="U3",
|
|
name="U3",
|
|
mac="AA:BB:CC:DD:EE:05",
|
|
ip="192.168.1.5",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={},
|
|
settings={},
|
|
firmware_check_result={"stable": {"version": "2.0.0", "build_id": "abc"}},
|
|
)
|
|
assert firmware_update_display(gen1) == "N/A"
|
|
assert firmware_update_display(unchecked) == "not checked"
|
|
assert firmware_update_display(uptodate) == "up to date"
|
|
assert "stable 2.0.0" in firmware_update_display(offer)
|
|
|
|
rows = [device_to_row(gen1), device_to_row(unchecked), device_to_row(uptodate), device_to_row(offer)]
|
|
assert len(apply_filters(rows, fw_update_filter="N/A")) == 1
|
|
assert len(apply_filters(rows, fw_update_filter="not checked")) == 1
|
|
assert len(apply_filters(rows, fw_update_filter="has_stable")) == 1
|
|
assert len(apply_filters(rows, fw_update_filter="has update")) == 1 # legacy alias → has_stable
|
|
|
|
|
|
def test_apply_filters_tags_mode() -> None:
|
|
tagged = _dev("T", 2, True)
|
|
tagged.tags = ["a"]
|
|
untagged = _dev("U", 2, True)
|
|
untagged.tags = []
|
|
rows = [device_to_row(tagged), device_to_row(untagged)]
|
|
assert len(apply_filters(rows, tags_mode="has_tags")) == 1
|
|
assert apply_filters(rows, tags_mode="has_tags")[0]["name"] == "T"
|
|
assert len(apply_filters(rows, tags_mode="none")) == 1
|
|
|
|
|
|
def test_apply_filters_fw_update_any_of() -> None:
|
|
gen1 = _dev("G1", 1, True)
|
|
auth = _dev("AU", 2, True)
|
|
auth.auth_required = True
|
|
rows = [device_to_row(gen1), device_to_row(auth)]
|
|
out = apply_filters(rows, fw_update_any_of=["N/A", "—"])
|
|
assert len(out) == 2
|
|
|
|
|
|
def test_firmware_offer_uses_sys_available_updates_when_not_checked() -> None:
|
|
"""GetStatus `sys.available_updates` matches device web UI without explicit CheckForUpdate."""
|
|
d = ShellyDevice(
|
|
id="XX",
|
|
name="X",
|
|
mac="AA:BB:CC:DD:EE:FF",
|
|
ip="192.168.1.1",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="SNSW-001P16EU",
|
|
firmware="1.7.1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={
|
|
"sys": {
|
|
"available_updates": {
|
|
"stable": {"version": "1.7.4"},
|
|
}
|
|
}
|
|
},
|
|
settings={},
|
|
firmware_check_result=None,
|
|
)
|
|
assert firmware_update_category(d) == "has_stable"
|
|
assert "1.7.4" in firmware_update_display(d)
|
|
|
|
|
|
def test_firmware_beta_only_category_and_filters() -> None:
|
|
beta_only = ShellyDevice(
|
|
id="B1",
|
|
name="B",
|
|
mac="AA:BB:CC:DD:EE:99",
|
|
ip="192.168.1.99",
|
|
generation=2, # type: ignore[arg-type]
|
|
model="Plus1",
|
|
firmware="1",
|
|
online=True,
|
|
last_seen=datetime.now(UTC),
|
|
capabilities=[],
|
|
status={},
|
|
settings={},
|
|
firmware_check_result={"beta": {"version": "2.1.0-beta", "build_id": "b1"}},
|
|
)
|
|
assert firmware_update_category(beta_only) == "has_beta_only"
|
|
assert "beta" in firmware_update_display(beta_only).lower()
|
|
assert "(beta only)" in firmware_update_display(beta_only)
|
|
row = device_to_row(beta_only)
|
|
assert row["_fw_update_cat"] == "has_beta_only"
|
|
assert len(apply_filters([row], fw_update_filter="has_stable")) == 0
|
|
assert len(apply_filters([row], fw_update_filter="has_beta_only")) == 1
|
|
assert len(apply_filters([row], fw_update_filter="has_any")) == 1
|
|
|
|
|
|
def test_prepare_table_rows_for_display_name_and_ip_links() -> None:
|
|
a = _dev("Kitchen Light", 2, True)
|
|
row = device_to_row(a, version_summaries=["snap-a", "snap-b"])
|
|
assert row["versions"] == "snap-a · snap-b"
|
|
assert "link" not in row
|
|
assert row["ip_http_url"].startswith("http://")
|
|
[prep] = prepare_table_rows_for_display([row])
|
|
assert "_fw_update_cat" not in prep
|
|
assert prep["name"].startswith("/Device?device=AABBCCDDEEFF&label=")
|
|
assert "Kitchen" in prep["name"] or "%20" in prep["name"]
|
|
assert prep["ip"] == row["ip_http_url"]
|
|
assert "ip_http_url" not in prep
|