mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
Show the plug that powers the printer in the card's Power row (#2830)
A printer card has one Power row: a plug name, its draw, and the auto-off
and on/off buttons. Which plug filled it was decided by nothing -- the
endpoint returned the first row the database handed back that was not a
Home Assistant script, from a query with no ORDER BY.
For the reporter that was an enclosure exhaust fan, added before the
outlet their X1C is plugged into. The card showed the fan's name with
'--' for watts, offered to switch the printer off by cutting the fan,
and demoted the metered outlet to the small HA button row. The fan was
marked as not powering the printer and hidden from the card; neither
setting was consulted here, though controls_printer_power has decided
the scheduler's power-on pick since #2629.
Rank the candidates instead: switchable at all, controls_printer_power,
enabled, show_on_printer_card, reports power, lowest id. The first rules
out a script, which can only be run, and an MQTT plug, which the control
endpoint rejects as monitor-only -- and an MQTT plug is exactly the kind
that reports watts, so without it ahead of the power tiebreak the row
could land on a plug whose on/off button answers with an error. The last
is not cosmetic: with no ORDER BY, a plain UPDATE on PostgreSQL can move
a row and silently swap which plug the card calls the printer's power.
None of these excludes a plug. A printer whose only plug is hidden,
disabled or monitor-only still needs its Power row, because that row
holds the on/off button and the HA buttons are drawn inside it.
controls_printer_power sits above show_on_printer_card because the two
only disagree when the plug that really feeds the printer is hidden, and
letting a display preference win there points the power buttons at an
accessory -- the fault #2629 fixed. Power capability is read from the
configuration, not measured: this runs on every card render, and it is
approximate both ways, so it only breaks a tie.
The scripts endpoint shares the same pick and excludes it, so a
switchable main plug is not repeated as a button directly below itself.
A script is left in place: a printer whose only entities are scripts
falls back to showing one in the power row, and taking it out of the
button row too would cost it the one-click run it has always had.
This commit is contained in:
maziggy
@@ -138,6 +138,90 @@ async def create_smart_plug(
|
||||
return plug
|
||||
|
||||
|
||||
def _is_script_plug(plug: SmartPlug) -> bool:
|
||||
"""Whether the plug is a Home Assistant script rather than a switchable device."""
|
||||
return bool(plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script."))
|
||||
|
||||
|
||||
def _can_be_switched(plug: SmartPlug) -> bool:
|
||||
"""Whether ``control_smart_plug`` can actually turn this plug on and off.
|
||||
|
||||
Two kinds cannot, and the card's on/off button is useless on both:
|
||||
|
||||
- A Home Assistant script. It can be run, not switched.
|
||||
- An MQTT plug. Bambuddy subscribes to it and never publishes, so the
|
||||
control endpoint rejects it outright as monitor-only -- and an MQTT plug
|
||||
is exactly the kind that reports watts, so without this it would win the
|
||||
power tiebreak below and take the row off a plug that can be switched.
|
||||
"""
|
||||
return not _is_script_plug(plug) and plug.plug_type != "mqtt"
|
||||
|
||||
|
||||
def _reports_power(plug: SmartPlug) -> bool:
|
||||
"""Whether the plug is configured with somewhere to read watts from (#2830).
|
||||
|
||||
Read from the configuration rather than measured: this runs on every printer
|
||||
card render, and probing each plug would mean an HTTP round trip per plug.
|
||||
So it is approximate in both directions -- an HA plug with no dedicated power
|
||||
sensor may still report watts from the switch entity's own
|
||||
``current_power_w`` attribute, and a Tasmota device without energy metering
|
||||
is counted here as if it had it. Only a live read could tell, and this is
|
||||
used solely to break a tie between plugs that are otherwise equally
|
||||
eligible, so neither miss can decide anything on its own.
|
||||
"""
|
||||
if plug.plug_type == "homeassistant":
|
||||
return bool(plug.ha_power_entity)
|
||||
if plug.plug_type == "mqtt":
|
||||
return bool(plug.mqtt_power_topic or plug.mqtt_topic)
|
||||
if plug.plug_type == "rest":
|
||||
return bool(plug.rest_power_path)
|
||||
return True # Tasmota, whose firmware reports power when the hardware has it
|
||||
|
||||
|
||||
def _main_plug_rank(plug: SmartPlug) -> tuple:
|
||||
"""Sort key for choosing the printer's main power plug, best first (#2830).
|
||||
|
||||
A printer's plugs are not interchangeable. The card's Power row carries the
|
||||
power on/off and auto-off-after-print controls, so it has to land on the plug
|
||||
that actually feeds the printer -- pointing those at an exhaust fan is the
|
||||
same harm #2629 fixed for the scheduler's power-on. Ordered:
|
||||
|
||||
1. It can be switched at all -- see ``_can_be_switched``. The row's buttons
|
||||
are the point of it.
|
||||
2. ``controls_printer_power`` -- the flag that says this plug feeds the
|
||||
printer, as opposed to an accessory that merely follows the print cycle.
|
||||
3. ``enabled`` -- a disabled plug ignores automation, so its auto-off toggle
|
||||
would sit there doing nothing.
|
||||
4. ``show_on_printer_card`` -- ranked, not filtered: excluding hidden plugs
|
||||
outright would strip the Power row, and with it the on/off button, from a
|
||||
printer whose only plug has the flag off. It sorts below the power flag
|
||||
because a display preference must not hand power control to an accessory.
|
||||
5. Reports power, so the row shows watts rather than "--" where there is a
|
||||
choice.
|
||||
6. Lowest id, so the answer never depends on row order. The query had no
|
||||
ORDER BY at all, which on Postgres means a plain UPDATE can move a row and
|
||||
silently swap which plug the card calls the printer's power.
|
||||
"""
|
||||
return (
|
||||
not _can_be_switched(plug),
|
||||
not plug.controls_printer_power,
|
||||
not plug.enabled,
|
||||
not plug.show_on_printer_card,
|
||||
not _reports_power(plug),
|
||||
plug.id,
|
||||
)
|
||||
|
||||
|
||||
def _pick_main_plug(plugs: list[SmartPlug]) -> SmartPlug | None:
|
||||
"""The plug the printer card shows as its power, or None if there are none."""
|
||||
return min(plugs, key=_main_plug_rank, default=None)
|
||||
|
||||
|
||||
async def _plugs_for_printer(db: AsyncSession, printer_id: int) -> list[SmartPlug]:
|
||||
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id).order_by(SmartPlug.id))
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
@router.get("/by-printer/{printer_id}", response_model=SmartPlugResponse | None)
|
||||
async def get_smart_plug_by_printer(
|
||||
printer_id: int,
|
||||
@@ -146,23 +230,11 @@ async def get_smart_plug_by_printer(
|
||||
):
|
||||
"""Get the main smart plug assigned to a printer.
|
||||
|
||||
When multiple plugs are assigned (e.g., a regular plug + script),
|
||||
returns the main (non-script) plug for power control.
|
||||
When several plugs are assigned -- a printer outlet, an enclosure fan, a
|
||||
script -- returns the one that best fits the card's power controls. See
|
||||
``_main_plug_rank`` for the order and why.
|
||||
"""
|
||||
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
|
||||
plugs = result.scalars().all()
|
||||
|
||||
if not plugs:
|
||||
return None
|
||||
|
||||
# If multiple plugs, prefer the non-script one (main power plug)
|
||||
for plug in plugs:
|
||||
is_script = plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
|
||||
if not is_script:
|
||||
return plug
|
||||
|
||||
# All are scripts, return the first one
|
||||
return plugs[0]
|
||||
return _pick_main_plug(await _plugs_for_printer(db, printer_id))
|
||||
|
||||
|
||||
@router.get("/by-printer/{printer_id}/scripts", response_model=list[SmartPlugResponse])
|
||||
@@ -176,13 +248,25 @@ async def get_script_plugs_by_printer(
|
||||
Returns HA entities (switches, scripts, lights, etc.) for the printer that have
|
||||
show_on_printer_card enabled.
|
||||
Used to display action buttons alongside the main power plug.
|
||||
|
||||
A switchable main plug is left out: it is rendered directly above this row
|
||||
with its own on/off button, so listing it here draws the same entity twice
|
||||
(#2830). A script is not, because a printer whose only entities are scripts
|
||||
falls back to showing one of them in the power row -- taking it out of this
|
||||
row too would cost the one-click run it has always had there.
|
||||
"""
|
||||
result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
|
||||
plugs = result.scalars().all()
|
||||
plugs = await _plugs_for_printer(db, printer_id)
|
||||
main_plug = _pick_main_plug(plugs)
|
||||
duplicate_of_power_row = main_plug.id if main_plug and not _is_script_plug(main_plug) else None
|
||||
|
||||
# Filter to HA entities with show_on_printer_card enabled
|
||||
ha_entities = [
|
||||
plug for plug in plugs if plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.show_on_printer_card
|
||||
plug
|
||||
for plug in plugs
|
||||
if plug.plug_type == "homeassistant"
|
||||
and plug.ha_entity_id
|
||||
and plug.show_on_printer_card
|
||||
and plug.id != duplicate_of_power_row
|
||||
]
|
||||
return ha_entities
|
||||
|
||||
|
||||
@@ -0,0 +1,233 @@
|
||||
"""The printer card's Power row and HA row, over the real API (#2830).
|
||||
|
||||
The ranking itself is unit-tested. These drive the two endpoints the card calls,
|
||||
with real rows in the database, because the ranking is only worth anything if
|
||||
the endpoints use it -- and because the second endpoint has to agree with the
|
||||
first about which plug is the main one or the card draws it twice.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
|
||||
|
||||
MAIN = "/api/v1/smart-plugs/by-printer/{}"
|
||||
ENTITIES = "/api/v1/smart-plugs/by-printer/{}/scripts"
|
||||
|
||||
|
||||
async def _ha(smart_plug_factory, printer, entity_id, **kwargs):
|
||||
return await smart_plug_factory(
|
||||
plug_type="homeassistant",
|
||||
ha_entity_id=entity_id,
|
||||
printer_id=printer.id,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
class TestTheReportedStranding:
|
||||
"""An X1C with two Home Assistant plugs: an exhaust fan with no power
|
||||
monitoring, added first, and the outlet the printer is actually plugged
|
||||
into. The card showed the fan, with "--" where its wattage would be."""
|
||||
|
||||
async def test_the_outlet_takes_the_power_row(self, async_client: AsyncClient, printer_factory, smart_plug_factory):
|
||||
printer = await printer_factory()
|
||||
await _ha(
|
||||
smart_plug_factory,
|
||||
printer,
|
||||
"switch.print_farm_exhaust_fan",
|
||||
name="Print Farm Exhaust Fan",
|
||||
controls_printer_power=False,
|
||||
show_on_printer_card=False,
|
||||
)
|
||||
await _ha(
|
||||
smart_plug_factory,
|
||||
printer,
|
||||
"switch.bambu_x1c_outlet",
|
||||
name="Bambu X1C Outlet",
|
||||
ha_power_entity="sensor.bambu_x1c_outlet_power",
|
||||
)
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == "Bambu X1C Outlet"
|
||||
|
||||
async def test_the_outlet_is_not_also_drawn_in_the_ha_row(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
"""It is rendered directly above that row with its own controls."""
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "switch.fan", name="Fan", controls_printer_power=False)
|
||||
await _ha(smart_plug_factory, printer, "switch.outlet", name="Bambu X1C Outlet")
|
||||
|
||||
response = await async_client.get(ENTITIES.format(printer.id))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [p["name"] for p in response.json()] == ["Fan"]
|
||||
|
||||
|
||||
class TestTheHAEntityRow:
|
||||
async def test_scripts_and_lights_still_appear(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "switch.outlet", name="Outlet")
|
||||
await _ha(smart_plug_factory, printer, "script.start", name="Start Script", controls_printer_power=False)
|
||||
await _ha(smart_plug_factory, printer, "light.chamber", name="Chamber Light", controls_printer_power=False)
|
||||
|
||||
response = await async_client.get(ENTITIES.format(printer.id))
|
||||
|
||||
assert sorted(p["name"] for p in response.json()) == ["Chamber Light", "Start Script"]
|
||||
|
||||
async def test_hidden_entities_stay_hidden(self, async_client: AsyncClient, printer_factory, smart_plug_factory):
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "switch.outlet", name="Outlet")
|
||||
await _ha(
|
||||
smart_plug_factory,
|
||||
printer,
|
||||
"light.chamber",
|
||||
name="Chamber Light",
|
||||
controls_printer_power=False,
|
||||
show_on_printer_card=False,
|
||||
)
|
||||
|
||||
response = await async_client.get(ENTITIES.format(printer.id))
|
||||
|
||||
assert response.json() == []
|
||||
|
||||
async def test_a_tasmota_main_plug_leaves_the_row_untouched(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
"""Only HA entities are listed there, so excluding the main plug must
|
||||
not remove anything when the main plug was never in the list."""
|
||||
printer = await printer_factory()
|
||||
await smart_plug_factory(name="Tasmota Outlet", printer_id=printer.id)
|
||||
await _ha(smart_plug_factory, printer, "light.chamber", name="Chamber Light", controls_printer_power=False)
|
||||
|
||||
response = await async_client.get(ENTITIES.format(printer.id))
|
||||
|
||||
assert [p["name"] for p in response.json()] == ["Chamber Light"]
|
||||
|
||||
|
||||
class TestASinglePlugIsNeverDropped:
|
||||
"""Hidden and disabled are ranking criteria, not filters. Excluding those
|
||||
outright would take the Power row -- and with it the on/off button, and the
|
||||
HA row nested inside it -- off a card that has one plug to show."""
|
||||
|
||||
async def test_a_hidden_plug_still_holds_the_row(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "switch.outlet", name="Outlet", show_on_printer_card=False)
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "Outlet"
|
||||
|
||||
async def test_an_accessory_still_holds_the_row(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
printer = await printer_factory()
|
||||
await smart_plug_factory(name="Filter Fan", printer_id=printer.id, controls_printer_power=False)
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "Filter Fan"
|
||||
|
||||
async def test_no_plugs_means_null(self, async_client: AsyncClient, printer_factory):
|
||||
printer = await printer_factory()
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() is None
|
||||
|
||||
|
||||
class TestTheRestOfTheOrder:
|
||||
async def test_an_enabled_plug_beats_a_disabled_one(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
printer = await printer_factory()
|
||||
await smart_plug_factory(name="Disabled", printer_id=printer.id, enabled=False)
|
||||
await smart_plug_factory(name="Live", printer_id=printer.id)
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "Live"
|
||||
|
||||
async def test_a_switch_beats_a_script(self, async_client: AsyncClient, printer_factory, smart_plug_factory):
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "script.start", name="Start Script")
|
||||
await _ha(smart_plug_factory, printer, "switch.outlet", name="Outlet")
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "Outlet"
|
||||
|
||||
async def test_a_printer_with_only_scripts_still_gets_one(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "script.a", name="First Script")
|
||||
await _ha(smart_plug_factory, printer, "script.b", name="Second Script")
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "First Script"
|
||||
|
||||
async def test_a_script_in_the_power_row_keeps_its_button(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
"""The script-only fallback is unchanged from before #2830, including
|
||||
the one-click run in the HA row. Only a switchable main plug is
|
||||
de-duplicated -- a script reached from the power row costs a confirm
|
||||
dialog it never used to need."""
|
||||
printer = await printer_factory()
|
||||
await _ha(smart_plug_factory, printer, "script.a", name="First Script")
|
||||
await _ha(smart_plug_factory, printer, "script.b", name="Second Script")
|
||||
|
||||
response = await async_client.get(ENTITIES.format(printer.id))
|
||||
|
||||
assert sorted(p["name"] for p in response.json()) == ["First Script", "Second Script"]
|
||||
|
||||
async def test_a_switchable_plug_beats_a_monitor_only_mqtt_plug(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
"""An MQTT plug reports watts but cannot be controlled -- the control
|
||||
endpoint rejects it. Put it in the power row and the on/off button
|
||||
answers with an error."""
|
||||
printer = await printer_factory()
|
||||
await smart_plug_factory(
|
||||
name="Monitor", plug_type="mqtt", printer_id=printer.id, mqtt_power_topic="tele/printer/SENSOR"
|
||||
)
|
||||
await _ha(smart_plug_factory, printer, "switch.outlet", name="Outlet")
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "Outlet"
|
||||
|
||||
async def test_a_lone_monitor_only_plug_still_holds_the_row(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
printer = await printer_factory()
|
||||
await smart_plug_factory(
|
||||
name="Monitor", plug_type="mqtt", printer_id=printer.id, mqtt_power_topic="tele/printer/SENSOR"
|
||||
)
|
||||
|
||||
response = await async_client.get(MAIN.format(printer.id))
|
||||
|
||||
assert response.json()["name"] == "Monitor"
|
||||
|
||||
|
||||
class TestOtherPrintersAreUnaffected:
|
||||
async def test_plugs_are_not_borrowed_across_printers(
|
||||
self, async_client: AsyncClient, printer_factory, smart_plug_factory
|
||||
):
|
||||
one = await printer_factory()
|
||||
two = await printer_factory()
|
||||
await smart_plug_factory(name="Plug One", printer_id=one.id)
|
||||
await smart_plug_factory(name="Plug Two", printer_id=two.id)
|
||||
await smart_plug_factory(name="Unlinked")
|
||||
|
||||
assert (await async_client.get(MAIN.format(one.id))).json()["name"] == "Plug One"
|
||||
assert (await async_client.get(MAIN.format(two.id))).json()["name"] == "Plug Two"
|
||||
@@ -0,0 +1,259 @@
|
||||
"""Choosing which of a printer's plugs is "the" power plug (#2830).
|
||||
|
||||
The printer card's Power row carries the power on/off and auto-off-after-print
|
||||
controls, so it has to land on the plug that actually feeds the printer. It used
|
||||
to take the first non-script row the database happened to return, which put an
|
||||
exhaust fan -- flagged as not powering the printer, and explicitly hidden from
|
||||
the card -- in front of the outlet the printer is plugged into.
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.app.api.routes.smart_plugs import (
|
||||
_can_be_switched,
|
||||
_is_script_plug,
|
||||
_main_plug_rank,
|
||||
_pick_main_plug,
|
||||
_reports_power,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
def _plug(plug_id=1, **kwargs):
|
||||
"""A plug with every flag at its model default, overridable per test."""
|
||||
defaults = {
|
||||
"id": plug_id,
|
||||
"name": f"Plug {plug_id}",
|
||||
"plug_type": "tasmota",
|
||||
"ha_entity_id": None,
|
||||
"ha_power_entity": None,
|
||||
"mqtt_topic": None,
|
||||
"mqtt_power_topic": None,
|
||||
"rest_power_path": None,
|
||||
"controls_printer_power": True,
|
||||
"enabled": True,
|
||||
"show_on_printer_card": True,
|
||||
}
|
||||
return SimpleNamespace(**{**defaults, **kwargs})
|
||||
|
||||
|
||||
class TestTheReportedCase:
|
||||
def test_the_printers_outlet_beats_an_exhaust_fan(self):
|
||||
"""The reporter's two Home Assistant plugs on one X1C. The fan was
|
||||
created first, so with no ranking it won on row order alone."""
|
||||
fan = _plug(
|
||||
1,
|
||||
name="Print Farm Exhaust Fan",
|
||||
plug_type="homeassistant",
|
||||
ha_entity_id="switch.print_farm_exhaust_fan",
|
||||
controls_printer_power=False,
|
||||
show_on_printer_card=False,
|
||||
)
|
||||
outlet = _plug(
|
||||
2,
|
||||
name="Bambu X1C Outlet",
|
||||
plug_type="homeassistant",
|
||||
ha_entity_id="switch.bambu_x1c_outlet",
|
||||
ha_power_entity="sensor.bambu_x1c_outlet_power",
|
||||
)
|
||||
|
||||
assert _pick_main_plug([fan, outlet]) is outlet
|
||||
|
||||
def test_and_it_wins_whichever_order_they_arrive_in(self):
|
||||
fan = _plug(1, plug_type="homeassistant", ha_entity_id="switch.fan", controls_printer_power=False)
|
||||
outlet = _plug(2, plug_type="homeassistant", ha_entity_id="switch.outlet")
|
||||
|
||||
assert _pick_main_plug([outlet, fan]) is outlet
|
||||
|
||||
|
||||
class TestTheOrderOfPreference:
|
||||
def test_a_switch_beats_a_script(self):
|
||||
"""A script cannot be switched off, so it can never be the power plug."""
|
||||
script = _plug(1, plug_type="homeassistant", ha_entity_id="script.start_print")
|
||||
switch = _plug(2, plug_type="homeassistant", ha_entity_id="switch.outlet")
|
||||
|
||||
assert _pick_main_plug([script, switch]) is switch
|
||||
|
||||
def test_a_switchable_plug_beats_a_monitor_only_mqtt_plug(self):
|
||||
"""``control_smart_plug`` rejects MQTT plugs as monitor-only, so the
|
||||
row's on/off button would answer with an error. An MQTT plug is also
|
||||
the kind that reports watts, so it would otherwise win on rank 5 --
|
||||
this is the one pair where the tiebreak could have done real harm."""
|
||||
monitor = _plug(1, plug_type="mqtt", mqtt_power_topic="tele/printer/SENSOR")
|
||||
switch = _plug(2, plug_type="homeassistant", ha_entity_id="switch.outlet")
|
||||
|
||||
assert _pick_main_plug([monitor, switch]) is switch
|
||||
|
||||
def test_a_monitor_only_plug_still_holds_the_row_on_its_own(self):
|
||||
"""Rank, not filter: a printer whose only plug is an MQTT monitor keeps
|
||||
the wattage readout it has always had."""
|
||||
monitor = _plug(1, plug_type="mqtt", mqtt_power_topic="tele/printer/SENSOR")
|
||||
|
||||
assert _pick_main_plug([monitor]) is monitor
|
||||
|
||||
def test_a_power_plug_beats_an_accessory(self):
|
||||
accessory = _plug(1, controls_printer_power=False)
|
||||
outlet = _plug(2)
|
||||
|
||||
assert _pick_main_plug([accessory, outlet]) is outlet
|
||||
|
||||
def test_an_enabled_plug_beats_a_disabled_one(self):
|
||||
"""A disabled plug ignores automation -- its auto-off toggle on the card
|
||||
would sit there doing nothing."""
|
||||
disabled = _plug(1, enabled=False)
|
||||
live = _plug(2)
|
||||
|
||||
assert _pick_main_plug([disabled, live]) is live
|
||||
|
||||
def test_a_visible_plug_beats_a_hidden_one(self):
|
||||
hidden = _plug(1, show_on_printer_card=False)
|
||||
visible = _plug(2)
|
||||
|
||||
assert _pick_main_plug([hidden, visible]) is visible
|
||||
|
||||
def test_powering_the_printer_outranks_being_visible(self):
|
||||
"""The two only disagree when someone hides the plug that really feeds
|
||||
the printer. Letting the display flag win would hand the power buttons
|
||||
to an accessory, which is the harm #2629 fixed for the scheduler."""
|
||||
visible_accessory = _plug(1, controls_printer_power=False, show_on_printer_card=True)
|
||||
hidden_outlet = _plug(2, controls_printer_power=True, show_on_printer_card=False)
|
||||
|
||||
assert _pick_main_plug([visible_accessory, hidden_outlet]) is hidden_outlet
|
||||
|
||||
def test_a_plug_that_reports_watts_breaks_a_tie(self):
|
||||
"""Otherwise the row reads "--" while a plug that knows the answer sits
|
||||
one rank below it."""
|
||||
mute = _plug(1, plug_type="homeassistant", ha_entity_id="switch.a")
|
||||
metered = _plug(2, plug_type="homeassistant", ha_entity_id="switch.b", ha_power_entity="sensor.b_power")
|
||||
|
||||
assert _pick_main_plug([mute, metered]) is metered
|
||||
|
||||
def test_reporting_watts_does_not_outrank_powering_the_printer(self):
|
||||
metered_accessory = _plug(1, ha_power_entity="sensor.fan_power", controls_printer_power=False)
|
||||
mute_outlet = _plug(2, plug_type="homeassistant", ha_entity_id="switch.outlet")
|
||||
|
||||
assert _pick_main_plug([metered_accessory, mute_outlet]) is mute_outlet
|
||||
|
||||
|
||||
class TestDeterminism:
|
||||
def test_equal_plugs_resolve_by_id(self):
|
||||
"""The query had no ORDER BY, so on Postgres an unrelated UPDATE could
|
||||
move a row and silently swap which plug the card called the printer's
|
||||
power."""
|
||||
first = _plug(1)
|
||||
second = _plug(2)
|
||||
|
||||
assert _pick_main_plug([second, first]) is first
|
||||
|
||||
def test_the_documented_order_holds_end_to_end(self):
|
||||
"""One plug per criterion, each failing only that one, ranked together.
|
||||
Pairwise tests would pass against an order with the middle two swapped."""
|
||||
ideal = _plug(6)
|
||||
mute = _plug(5, plug_type="homeassistant", ha_entity_id="switch.mute")
|
||||
hidden = _plug(4, show_on_printer_card=False)
|
||||
disabled = _plug(3, enabled=False)
|
||||
accessory = _plug(2, controls_printer_power=False)
|
||||
script = _plug(1, plug_type="homeassistant", ha_entity_id="script.a")
|
||||
|
||||
ranked = sorted([script, accessory, disabled, hidden, mute, ideal], key=_main_plug_rank)
|
||||
|
||||
assert ranked == [ideal, mute, hidden, disabled, accessory, script]
|
||||
|
||||
|
||||
class TestNoPlugs:
|
||||
def test_no_plugs_means_no_main_plug(self):
|
||||
assert _pick_main_plug([]) is None
|
||||
|
||||
def test_all_scripts_still_yields_one(self):
|
||||
"""Pre-existing behaviour: a printer whose only entities are scripts
|
||||
still gets a Power row, because the card nests everything else inside
|
||||
it. Ranking must not turn that into an empty card."""
|
||||
second = _plug(2, plug_type="homeassistant", ha_entity_id="script.b")
|
||||
first = _plug(1, plug_type="homeassistant", ha_entity_id="script.a")
|
||||
|
||||
assert _pick_main_plug([second, first]) is first
|
||||
|
||||
|
||||
class TestSwitchability:
|
||||
@pytest.mark.parametrize(
|
||||
"plug,expected",
|
||||
[
|
||||
(_plug(plug_type="tasmota"), True),
|
||||
(_plug(plug_type="rest"), True),
|
||||
(_plug(plug_type="homeassistant", ha_entity_id="switch.outlet"), True),
|
||||
(_plug(plug_type="homeassistant", ha_entity_id="light.chamber"), True),
|
||||
(_plug(plug_type="homeassistant", ha_entity_id="script.start"), False),
|
||||
(_plug(plug_type="mqtt", mqtt_topic="zigbee2mqtt/plug"), False),
|
||||
],
|
||||
)
|
||||
def test_matches_what_the_control_endpoint_accepts(self, plug, expected):
|
||||
assert _can_be_switched(plug) is expected
|
||||
|
||||
|
||||
class TestUnsetFlags:
|
||||
"""An upgraded database adds these columns by ALTER, so they are nullable
|
||||
there even though a fresh one declares them NOT NULL. Every row is
|
||||
backfilled with the default and nothing writes a null, but a rank that
|
||||
raised on one would take down the whole printers page."""
|
||||
|
||||
def test_a_plug_with_null_flags_ranks_last_rather_than_raising(self):
|
||||
unset = _plug(1, controls_printer_power=None, enabled=None, show_on_printer_card=None)
|
||||
ordinary = _plug(2)
|
||||
|
||||
assert _pick_main_plug([unset, ordinary]) is ordinary
|
||||
|
||||
def test_and_still_holds_the_row_when_it_is_the_only_plug(self):
|
||||
unset = _plug(1, controls_printer_power=None, enabled=None, show_on_printer_card=None)
|
||||
|
||||
assert _pick_main_plug([unset]) is unset
|
||||
|
||||
def test_a_plug_with_no_type_at_all_does_not_raise(self):
|
||||
assert _pick_main_plug([_plug(1, plug_type=None)]) is not None
|
||||
|
||||
|
||||
class TestScriptDetection:
|
||||
@pytest.mark.parametrize(
|
||||
"entity_id,expected",
|
||||
[
|
||||
("script.start_print", True),
|
||||
("switch.outlet", False),
|
||||
("light.chamber", False),
|
||||
(None, False),
|
||||
],
|
||||
)
|
||||
def test_only_ha_script_entities_count(self, entity_id, expected):
|
||||
assert _is_script_plug(_plug(plug_type="homeassistant", ha_entity_id=entity_id)) is expected
|
||||
|
||||
def test_a_tasmota_plug_is_never_a_script(self):
|
||||
assert _is_script_plug(_plug(plug_type="tasmota", ha_entity_id="script.confusing")) is False
|
||||
|
||||
|
||||
class TestPowerCapability:
|
||||
"""Read from configuration, not measured -- this runs on every card render.
|
||||
|
||||
It is a floor: an HA plug with no dedicated sensor may still report watts
|
||||
from the switch entity's own ``current_power_w`` attribute, which only a
|
||||
live read would show. Good enough for a tiebreak, and it never decides
|
||||
anything on its own.
|
||||
"""
|
||||
|
||||
def test_tasmota_reports_power_natively(self):
|
||||
assert _reports_power(_plug(plug_type="tasmota")) is True
|
||||
|
||||
def test_home_assistant_needs_a_power_sensor(self):
|
||||
assert _reports_power(_plug(plug_type="homeassistant", ha_entity_id="switch.a")) is False
|
||||
assert _reports_power(_plug(plug_type="homeassistant", ha_power_entity="sensor.a_power")) is True
|
||||
|
||||
def test_mqtt_accepts_either_the_new_or_the_legacy_topic(self):
|
||||
assert _reports_power(_plug(plug_type="mqtt")) is False
|
||||
assert _reports_power(_plug(plug_type="mqtt", mqtt_power_topic="tele/plug/SENSOR")) is True
|
||||
assert _reports_power(_plug(plug_type="mqtt", mqtt_topic="zigbee2mqtt/plug")) is True
|
||||
|
||||
def test_rest_needs_a_power_path(self):
|
||||
"""The REST service returns no energy at all without one, so a URL on
|
||||
its own proves nothing."""
|
||||
assert _reports_power(_plug(plug_type="rest")) is False
|
||||
assert _reports_power(_plug(plug_type="rest", rest_power_path="apower")) is True
|
||||
Reference in New Issue
Block a user