diff --git a/backend/app/api/routes/smart_plugs.py b/backend/app/api/routes/smart_plugs.py index 031674966..58a2dafbb 100644 --- a/backend/app/api/routes/smart_plugs.py +++ b/backend/app/api/routes/smart_plugs.py @@ -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 diff --git a/backend/tests/integration/test_main_plug_pick_2830.py b/backend/tests/integration/test_main_plug_pick_2830.py new file mode 100644 index 000000000..27360bd6c --- /dev/null +++ b/backend/tests/integration/test_main_plug_pick_2830.py @@ -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" diff --git a/backend/tests/unit/test_main_plug_rank_2830.py b/backend/tests/unit/test_main_plug_rank_2830.py new file mode 100644 index 000000000..09143017b --- /dev/null +++ b/backend/tests/unit/test_main_plug_rank_2830.py @@ -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