Add raw fetch methods; make CLI info resilient to unparseable responses

smarttub/__main__.py

@@ -10,77 +10,85 @@
 from . import SmartTub, SpaLight
 
 
+async def _section(title, fetch):
+    print(f"== {title} ==")
+    try:
+        pprint(await fetch())
+    except Exception as err:
+        print(f"Error: {type(err).__name__}: {err}")
+    print()
+
+
 async def info_command(spas, args):
     for spa in spas:
         print(f"= Spa '{spa.name}' =\n")
-        if args.all or args.status or args.location or args.locks:
-            status = await spa.get_status_full()
+
+        status = None
+        status_error = None
+        if args.all or any(
+            (
+                args.status,
+                args.location,
+                args.pumps,
+                args.lights,
+                args.locks,
+                args.sensors,
+            )
+        ):
+            try:
+                status = await spa.get_status_full_raw()
+            except Exception as err:
+                status_error = f"Error: {type(err).__name__}: {err}"
+
+        def from_status(fetch):
+            async def inner():
+                if status is None:
+                    raise RuntimeError(status_error)
+                return fetch(status)
+
+            return inner
 
         if args.all or args.status:
-            print("== Status ==")
-            status_dict = status.properties.copy()
-            # redact location for privacy
-            location = status_dict.pop("location")
-            pprint(status_dict)
-            print()
+            # location is redacted for privacy unless --location is given
+            await _section(
+                "Status",
+                from_status(lambda s: {k: v for k, v in s.items() if k != "location"}),
+            )
 
         if args.location:
             # not included in --all
-            print(
-                f"Location: {location['latitude']} {location['longitude']} (accuracy: {location['accuracy']})\n"
-            )
+            await _section("Location", from_status(lambda s: s["location"]))
 
         if args.all or args.pumps:
-            print("== Pumps ==")
-            for pump in status.pumps:
-                print(pump)
-            print()
+            await _section("Pumps", from_status(lambda s: s.get("pumps")))
 
         if args.all or args.lights:
-            print("== Lights ==")
-            for light in status.lights:
-                print(light)
-            print()
+            await _section("Lights", from_status(lambda s: s.get("lights")))
 
         if args.all or args.errors:
-            print("== Errors ==")
-            for error in await spa.get_errors():
-                print(error)
-            print()
+            await _section("Errors", spa.get_errors_raw)
 
         if args.all or args.reminders:
-            print("== Reminders ==")
-            for reminder in await spa.get_reminders():
-                print(reminder)
-            print()
+            await _section("Reminders", spa.get_reminders_raw)
 
         if args.all or args.locks:
-            print("== Locks ==")
-            for lock in status.locks.values():
-                print(lock)
-            print()
+            await _section("Locks", from_status(lambda s: s.get("locks")))
 
         if args.all or args.energy:
-            print("== Energy usage ==")
-            energy_usage_day = spa.get_energy_usage(
-                spa.EnergyUsageInterval.DAY,
-                end_date=datetime.date.today(),
-                start_date=datetime.date.today() - datetime.timedelta(days=7),
+            await _section(
+                "Energy usage",
+                lambda: spa.get_energy_usage(
+                    spa.EnergyUsageInterval.DAY,
+                    end_date=datetime.date.today(),
+                    start_date=datetime.date.today() - datetime.timedelta(days=7),
+                ),
             )
-            pprint(await energy_usage_day)
-            print()
 
         if args.all or args.sensors:
-            print("== Sensors ==")
-            for sensor in status.sensors:
-                print(sensor)
-            print()
+            await _section("Sensors", from_status(lambda s: s.get("sensors")))
 
         if args.all or args.debug:
-            debug_status = await spa.get_debug_status()
-            print("== Debug status ==")
-            pprint(debug_status)
-            print()
+            await _section("Debug status", spa.get_debug_status)
 
 
 async def set_command(spas, args):
@@ -203,4 +211,5 @@ async def main(argv):
         await args.func(spas, args)
 
 
-asyncio.run(main(sys.argv[1:]))
+if __name__ == "__main__":
+    asyncio.run(main(sys.argv[1:]))

smarttub/api.py

@@ -254,6 +254,22 @@ async def get_status_full(self) -> "SpaStateFull":
             logger.error(f"Failed to parse fullStatus response: {full_status}")
             raise
 
+    async def get_status_full_raw(self) -> dict:
+        """Retrieve the unparsed fullStatus response.
+
+        Unlike get_status_full(), this does not raise on values the library
+        does not recognize, so it is suitable for diagnostics.
+        """
+        return await self.request("GET", "fullStatus")
+
+    async def get_errors_raw(self) -> list:
+        """Retrieve the unparsed errors, without validating their contents."""
+        return (await self.request("GET", "errors"))["content"]
+
+    async def get_reminders_raw(self) -> list:
+        """Retrieve the unparsed reminders, without validating their contents."""
+        return (await self.request("GET", "reminders"))["reminders"]
+
     async def get_debug_status(self) -> dict:
         return (await self.request("GET", "debugStatus"))["debugStatus"]
 

tests/test_cli.py

@@ -0,0 +1,80 @@
+import argparse
+from unittest.mock import AsyncMock, MagicMock
+
+from smarttub.__main__ import info_command
+
+
+def _args(**overrides):
+    values = dict(
+        all=True,
+        status=False,
+        location=False,
+        pumps=False,
+        lights=False,
+        errors=False,
+        reminders=False,
+        locks=False,
+        sensors=False,
+        energy=False,
+        debug=False,
+    )
+    values.update(overrides)
+    return argparse.Namespace(**values)
+
+
+def _spa():
+    spa = MagicMock()
+    spa.name = "Test Spa"
+    spa.get_status_full_raw = AsyncMock(
+        return_value={
+            "heatMode": "SOME_NEW_MODE",
+            "location": {"latitude": 12.5, "longitude": 34.5},
+            "pumps": [{"id": "P1"}],
+            "lights": [{"zone": 1}],
+            "locks": {"spa": "UNLOCKED"},
+            "sensors": [],
+        }
+    )
+    spa.get_errors_raw = AsyncMock(return_value=[{"code": 7}])
+    spa.get_reminders_raw = AsyncMock(return_value=[{"id": "r1"}])
+    spa.get_energy_usage = AsyncMock(return_value=[])
+    spa.get_debug_status = AsyncMock(return_value={"uptime": 1})
+    return spa
+
+
+async def test_info_all_redacts_location_and_shows_unknown_values(capsys):
+    await info_command([_spa()], _args())
+
+    out = capsys.readouterr().out
+    assert "SOME_NEW_MODE" in out
+    assert "12.5" not in out
+    assert "'code': 7" in out
+
+
+async def test_info_location_flag_shows_location(capsys):
+    await info_command([_spa()], _args(all=False, location=True))
+
+    assert "12.5" in capsys.readouterr().out
+
+
+async def test_info_one_failing_section_does_not_hide_others(capsys):
+    spa = _spa()
+    spa.get_errors_raw.side_effect = KeyError("boom")
+
+    await info_command([spa], _args())
+
+    out = capsys.readouterr().out
+    assert "Error: KeyError: 'boom'" in out
+    assert "'id': 'r1'" in out
+    assert "'uptime': 1" in out
+
+
+async def test_info_status_failure_reported_in_each_dependent_section(capsys):
+    spa = _spa()
+    spa.get_status_full_raw.side_effect = RuntimeError("api down")
+
+    await info_command([spa], _args())
+
+    out = capsys.readouterr().out
+    assert out.count("Error: RuntimeError: api down") >= 5
+    assert "'id': 'r1'" in out

tests/test_spa.py

@@ -693,3 +693,26 @@ async def test_secondary_filtration_cycle(mock_api, spa):
     mock_api.request.assert_called_with(
         "PATCH", f"spas/{spa.id}/config", {"secondaryFiltrationConfig": "FREQUENT"}
     )
+
+
+async def test_get_status_full_raw_tolerates_unknown_values(mock_api, spa):
+    payload = {"heatMode": "SOME_NEW_MODE", "location": {"latitude": 1.0}}
+    mock_api.request.return_value = payload
+
+    with pytest.raises(KeyError):
+        await spa.get_status_full()
+    assert await spa.get_status_full_raw() == payload
+
+
+async def test_get_errors_raw(mock_api, spa):
+    mock_api.request.return_value = {"content": [{"code": 1, "createdAt": None}]}
+
+    assert await spa.get_errors_raw() == [{"code": 1, "createdAt": None}]
+    mock_api.request.assert_called_with("GET", f"spas/{spa.id}/errors", None)
+
+
+async def test_get_reminders_raw(mock_api, spa):
+    mock_api.request.return_value = {"reminders": [{"id": "r1", "lastUpdated": None}]}
+
+    assert await spa.get_reminders_raw() == [{"id": "r1", "lastUpdated": None}]
+    mock_api.request.assert_called_with("GET", f"spas/{spa.id}/reminders", None)