107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
"""Offline tests for the business operations (transport stubbed)."""
|
|
|
|
import unittest
|
|
|
|
from franklinwh_agate_client.device_local import LocalDeviceClient
|
|
from franklinwh_agate_client.protocol import CommunicationCmd
|
|
|
|
|
|
def _client(handler):
|
|
client = LocalDeviceClient("SN", host="127.0.0.1", timeout=0.1)
|
|
calls = []
|
|
|
|
def fake_send_rpc(cmd, equip_no=None, operate=None, data=None,
|
|
envelope=None):
|
|
calls.append((cmd, equip_no, data))
|
|
return handler(cmd, equip_no, data)
|
|
|
|
client.send_rpc = fake_send_rpc
|
|
return client, calls
|
|
|
|
|
|
class TestBatterySocOp(unittest.TestCase):
|
|
def test_apower_win(self):
|
|
client, calls = _client(
|
|
lambda cmd, eq, data: {"dataArea": {"currentBatterySoc": 78}})
|
|
out = client.get_battery_soc()
|
|
self.assertEqual(out["value"], 78)
|
|
self.assertEqual(out["source"], "currentBatterySoc")
|
|
|
|
def test_bms_pack_fallback(self):
|
|
def handler(cmd, eq, data):
|
|
if cmd == CommunicationCmd.DEVICE_APOWER_INFO:
|
|
return {"dataArea": {"result": 1, "reason": -1}}
|
|
if (data or {}).get("id") == 1:
|
|
return {"dataArea": {"opt": 0, "result": 0, "id": 1,
|
|
"batSoc": 64, "batSoh": 99,
|
|
"batTotalVolt": 512}}
|
|
return {"dataArea": {"opt": 0, "result": 0, "id": 0,
|
|
"batSoc": 0, "batSoh": 0,
|
|
"batTotalVolt": 0}}
|
|
|
|
client, calls = _client(handler)
|
|
out = client.get_battery_soc()
|
|
self.assertEqual(out["value"], 64)
|
|
self.assertEqual(out["source"], "batSoc")
|
|
# discovery + apower + bms id=1 (empty id=0 never needed)
|
|
ids = [d.get("id") if isinstance(d, dict) else None
|
|
for _, _, d in calls]
|
|
self.assertIn(1, ids)
|
|
|
|
def test_empty_slots_skipped(self):
|
|
client, _ = _client(
|
|
lambda cmd, eq, data: {"dataArea": {"opt": 0, "result": 0,
|
|
"id": 0, "batSoc": 0,
|
|
"batSoh": 0,
|
|
"batTotalVolt": 0}})
|
|
out = client.get_battery_soc()
|
|
self.assertIsNone(out["value"])
|
|
|
|
def test_transport_error_propagates(self):
|
|
client, _ = _client(None)
|
|
|
|
def boom(cmd, equip_no=None, operate=None, data=None,
|
|
envelope=None):
|
|
raise ConnectionError("down")
|
|
|
|
client.send_rpc = boom
|
|
with self.assertRaises(ConnectionError):
|
|
client.get_battery_soc()
|
|
|
|
|
|
class TestWattageOp(unittest.TestCase):
|
|
def test_detail_win(self):
|
|
client, _ = _client(
|
|
lambda cmd, eq, data: {"dataArea": {"apowerCount": 2,
|
|
"totalPower": 1872.5}})
|
|
out = client.get_household_wattage()
|
|
self.assertEqual(out["value"], 1872.5)
|
|
self.assertEqual(out["source"], "totalPower")
|
|
|
|
def test_ems_math_fallback(self):
|
|
def handler(cmd, eq, data):
|
|
if cmd == CommunicationCmd.EMS:
|
|
return {"dataArea": {"gridVol1": 120.0, "gridVol2": 120.0,
|
|
"loadCurr1": 5.0, "loadCurr2": 5.0}}
|
|
return {"dataArea": {"result": 1, "reason": -1}}
|
|
|
|
client, _ = _client(handler)
|
|
out = client.get_household_wattage()
|
|
self.assertAlmostEqual(out["value"], 1200.0)
|
|
self.assertEqual(out["source"], "ems(loadCurr*V)")
|
|
|
|
def test_inventory(self):
|
|
client, _ = _client(
|
|
lambda cmd, eq, data: {
|
|
"cmdType": 1102, "equipNo": "IBG",
|
|
"dataArea": {"opt": 0, "result": 0, "FHP_SN": ["A", "B"],
|
|
"BMS_SN": ["C"], "protocolVer": "V1"}})
|
|
out = client.get_device_inventory()
|
|
self.assertEqual(out["ibg_sn"], "IBG")
|
|
self.assertEqual(out["fhp_sn"], ["A", "B"])
|
|
self.assertEqual(out["protocol_ver"], "V1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|