130 lines
3.4 KiB
Python
130 lines
3.4 KiB
Python
# Copyright (c) 2026 Umber
|
|
#
|
|
# Licensed under the Apache License, Version 2.0; see LICENSE.
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from fiwicontrol.power import AcronamePower, MonsoonPower, MonsoonReading, Power
|
|
|
|
|
|
class _FakePort:
|
|
def __init__(self) -> None:
|
|
self.enabled: bool | None = None
|
|
|
|
def setEnabled(self, enable: bool) -> int:
|
|
self.enabled = bool(enable)
|
|
return 0
|
|
|
|
|
|
class _FakeHub:
|
|
def __init__(self, n: int = 4) -> None:
|
|
self.port = [_FakePort() for _ in range(n)]
|
|
|
|
|
|
class _FakeModule:
|
|
def __init__(self, n: int = 4) -> None:
|
|
self.hub = _FakeHub(n)
|
|
|
|
|
|
async def _acroname_power_port_on_off() -> None:
|
|
mod = _FakeModule(4)
|
|
ap = AcronamePower(mod, default_port=1)
|
|
await ap.port_on()
|
|
assert mod.hub.port[1].enabled is True
|
|
await ap.port_off()
|
|
assert mod.hub.port[1].enabled is False
|
|
await ap.port_on(2)
|
|
assert mod.hub.port[2].enabled is True
|
|
|
|
|
|
def test_acroname_power_port_on_off() -> None:
|
|
asyncio.run(_acroname_power_port_on_off())
|
|
|
|
|
|
async def _acroname_power_set_port_validates_range() -> None:
|
|
ap = AcronamePower(_FakeModule(2))
|
|
with pytest.raises(ValueError):
|
|
await ap.set_port(5, True)
|
|
|
|
|
|
def test_acroname_power_set_port_validates_range() -> None:
|
|
asyncio.run(_acroname_power_set_port_validates_range())
|
|
|
|
|
|
async def _monsoon_power_custom_sampler() -> None:
|
|
async def sample() -> MonsoonReading:
|
|
return MonsoonReading(volts=5.0, amps=0.4, watts=2.0)
|
|
|
|
mp = MonsoonPower(sample)
|
|
assert await mp.voltage() == 5.0
|
|
assert await mp.current() == 0.4
|
|
assert await mp.watts() == 2.0
|
|
|
|
|
|
def test_monsoon_power_custom_sampler() -> None:
|
|
asyncio.run(_monsoon_power_custom_sampler())
|
|
|
|
|
|
def test_monsoon_reading_json_line_parses_short_keys() -> None:
|
|
from fiwicontrol.power.monsoon import _read_monsoon_json_line
|
|
|
|
class _FakeSerial:
|
|
def __init__(self, *a, **k) -> None:
|
|
pass
|
|
|
|
def __enter__(self) -> _FakeSerial:
|
|
return self
|
|
|
|
def __exit__(self, *a) -> None:
|
|
pass
|
|
|
|
def readline(self) -> bytes:
|
|
return b'{"v":3.3,"a":0.1}\n'
|
|
|
|
with mock.patch("serial.Serial", _FakeSerial):
|
|
r = _read_monsoon_json_line("/dev/fake", 115200, 1.0)
|
|
assert r.volts == pytest.approx(3.3)
|
|
assert r.amps == pytest.approx(0.1)
|
|
assert r.watts == pytest.approx(0.33)
|
|
|
|
|
|
async def _power_facade_delegates() -> None:
|
|
mod = _FakeModule(2)
|
|
ap = AcronamePower(mod, default_port=0)
|
|
|
|
async def sample() -> MonsoonReading:
|
|
return MonsoonReading(12.0, 0.5, 6.0)
|
|
|
|
mp = MonsoonPower(sample)
|
|
p = Power(acroname=ap, monsoon=mp)
|
|
await p.on()
|
|
assert mod.hub.port[0].enabled is True
|
|
await p.off()
|
|
assert mod.hub.port[0].enabled is False
|
|
assert await p.voltage() == 12.0
|
|
assert await p.current() == 0.5
|
|
assert await p.watts() == 6.0
|
|
s = await p.sample()
|
|
assert s.volts == 12.0
|
|
|
|
|
|
def test_power_facade_delegates() -> None:
|
|
asyncio.run(_power_facade_delegates())
|
|
|
|
|
|
async def _power_facade_missing_backend_errors() -> None:
|
|
p = Power(acroname=None, monsoon=None)
|
|
with pytest.raises(RuntimeError, match="Acroname"):
|
|
await p.on()
|
|
with pytest.raises(RuntimeError, match="Monsoon"):
|
|
await p.voltage()
|
|
|
|
|
|
def test_power_facade_missing_backend_errors() -> None:
|
|
asyncio.run(_power_facade_missing_backend_errors())
|