Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 19 additions & 5 deletions iometer/status.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Device status for IOmeter bridge and core"""

import json
from dataclasses import dataclass
from dataclasses import dataclass, field


@dataclass
Expand Down Expand Up @@ -38,15 +38,25 @@ class Device:
class Meter:
"""Represents the meter device."""

number: str
number: str | None


class NullMeter(Meter):
"""Null Object for Meter to avoid None-attribute errors."""

def __init__(self) -> None:
super().__init__(number=None)

def __bool__(self) -> bool:
return False


@dataclass
class Status:
"""Top level class representing the complete device status"""

meter: Meter
device: Device
meter: Meter = field(default_factory=NullMeter)
typename: str = "iometer.status.v1"

@classmethod
Expand Down Expand Up @@ -76,7 +86,10 @@ def from_json(cls, json_str: str) -> "Status":
# Create device
device = Device(bridge=bridge, id=data["device"]["id"], core=core)

meter = Meter(number=data["meter"]["number"])
# Create meter (use Null Object if missing)
meter = (
Meter(number=data["meter"]["number"]) if data.get("meter") else NullMeter()
)

# Create full status
return cls(meter=meter, device=device)
Expand All @@ -86,7 +99,8 @@ def to_json(self) -> str:
return json.dumps(
{
"__typename": self.typename,
"meter": {"number": self.meter.number},
# If meter is a NullMeter, serialize as null
"meter": {"number": self.meter.number} if self.meter else None,
"device": {
"bridge": {
"rssi": self.device.bridge.rssi,
Expand Down
58 changes: 52 additions & 6 deletions tests/test.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,14 @@
from iometer.client import IOmeterClient
from iometer.exceptions import IOmeterConnectionError, IOmeterTimeoutError
from iometer.reading import Reading
from iometer.status import Status
from iometer.status import NullMeter, Status

HOST = "192.168.1.100"


@pytest.fixture(name="reading_json")
def reading_json_fixture():
"""Sample reading JSON response."""
"""Fixture reading response."""
return {
"__typename": "iometer.reading.v1",
"meter": {
Expand Down Expand Up @@ -57,7 +57,7 @@ def status_json_fixture():

@pytest.fixture(name="status_wired_json")
def status_wired_json_fixture():
""" "Fixture status response"""
""" "Fixture status response with wired power"""
return {
"__typename": "iometer.status.v1",
"meter": {
Expand All @@ -79,8 +79,8 @@ def status_wired_json_fixture():


@pytest.fixture(name="status_detached_json")
def status_deatached_json_fixture():
""" "Fixture status response"""
def status_detached_json_fixture():
""" "Fixture status response with detached core"""
return {
"__typename": "iometer.status.v1",
"meter": {
Expand All @@ -102,7 +102,7 @@ def status_deatached_json_fixture():

@pytest.fixture(name="status_disconnected_json")
def status_disconnected_json_fixture():
""" "Fixture status response"""
""" "Fixture status response with disconnected core"""
return {
"__typename": "iometer.status.v1",
"meter": {
Expand All @@ -116,6 +116,26 @@ def status_disconnected_json_fixture():
}


@pytest.fixture(name="status_no_meter_json")
def status_no_meter_json_fixture():
""" "Fixture status response"""
return {
"__typename": "iometer.status.v1",
"device": {
"bridge": {"rssi": -30, "version": "build-65"},
"id": "658c2b34-2017-45f2-a12b-731235f8bb97",
"core": {
"connectionStatus": "connected",
"rssi": -30,
"version": "build-58",
"powerStatus": "battery",
"batteryLevel": 100,
"attachmentStatus": "attached",
},
},
}


@pytest.fixture(name="mock_aioresponse")
def mock_aioresponse_fixture():
""" "Fixture mock session"""
Expand Down Expand Up @@ -262,6 +282,32 @@ async def test_get_current_status_disconnected(
assert status.device.core.pin_status is None


@pytest.mark.asyncio
async def test_get_current_status_no_meter(
client_iometer, mock_aioresponse, status_no_meter_json
):
"""Test getting device status."""

mock_endpoint = f"http://{HOST}/v1/status"
mock_aioresponse.get(mock_endpoint, status=200, payload=status_no_meter_json)

status = await client_iometer.get_current_status()

assert isinstance(status, Status)
assert isinstance(status.meter, NullMeter)
assert status.meter.number is None
assert status.device.bridge.rssi == -30
assert status.device.bridge.version == "build-65"
assert status.device.id == "658c2b34-2017-45f2-a12b-731235f8bb97"
assert status.device.core.connection_status == "connected"
assert status.device.core.rssi == -30
assert status.device.core.version == "build-58"
assert status.device.core.power_status == "battery"
assert status.device.core.battery_level == 100
assert status.device.core.attachment_status == "attached"
assert status.device.core.pin_status is None


@pytest.mark.asyncio
async def test_timeout_error(client_iometer, mock_aioresponse):
"""Test handling of timeout errors."""
Expand Down