Move warning classes from protocols to device module

aioraven/device.py

@@ -38,6 +38,28 @@
 from aioraven.data import TimeCluster
 
 
+class RAVEnWarning(Warning):
+    """The RAVEn device has generated a warning message."""
+
+    def __init__(self, message: str) -> None:
+        """
+        Construct a RAVEnWarning.
+
+        :param message: The reason for which the device is warning us.
+        """
+        super().__init__('Warning from RAVEn device: ' + message)
+
+
+class UnknownRAVEnCommandWarning(RAVEnWarning):
+    """A recently executed command is not supported by the RAVEn device."""
+
+    MESSAGE = 'Unknown command'
+
+    def __init__(self) -> None:
+        """Construct a UnknownRAVEnCommandWarning."""
+        super().__init__(self.MESSAGE)
+
+
 class RAVEnBaseDevice:
     """RAVEn device command implementation."""
 

aioraven/protocols.py

@@ -11,31 +11,11 @@
 import warnings
 import xml.etree.ElementTree as Et
 
+from aioraven.device import RAVEnWarning
+from aioraven.device import UnknownRAVEnCommandWarning
 from aioraven.reader import RAVEnReader
 
 
-class DeviceWarning(Warning):
-    """The device has generated a warning message."""
-
-    def __init__(self, message: str) -> None:
-        """
-        Construct a DeviceWarning.
-
-        :param message: The reason for which the device is warning us.
-        """
-        super().__init__('Warning from RAVEn device: ' + message)
-
-
-class UnknownCommandWarning(DeviceWarning):
-    """A recently executed command is not supported by the device."""
-
-    MESSAGE = 'Unknown command'
-
-    def __init__(self) -> None:
-        """Construct a UnknownCommandWarning."""
-        super().__init__(self.MESSAGE)
-
-
 class RAVEnReaderProtocol(Protocol):
     """Deserialize data fragments from a RAVEn device."""
 
@@ -104,12 +84,12 @@ def data_received(self, data: bytes) -> None:
                     try:
                         e = next(iter(element), None)
                         if e is not None and e.text is not None:
-                            if e.text == UnknownCommandWarning.MESSAGE:
-                                warnings.warn(UnknownCommandWarning())
+                            if e.text == UnknownRAVEnCommandWarning.MESSAGE:
+                                warnings.warn(UnknownRAVEnCommandWarning())
                             else:
-                                warnings.warn(DeviceWarning(e.text))
+                                warnings.warn(RAVEnWarning(e.text))
                         else:
-                            warnings.warn(DeviceWarning('Unknown warning'))
+                            warnings.warn(RAVEnWarning('Unknown warning'))
                     except Warning as err:
                         self._reader.set_exception(err)
                 else:

test/test_data.py

@@ -29,8 +29,8 @@
 from aioraven.data import ScheduledEvent
 from aioraven.data import ScheduleInfo
 from aioraven.data import TimeCluster
-from aioraven.protocols import DeviceWarning
-from aioraven.protocols import UnknownCommandWarning
+from aioraven.device import RAVEnWarning
+from aioraven.device import UnknownRAVEnCommandWarning
 from aioraven.streams import RAVEnNetworkDevice
 from iso4217 import Currency
 import pytest
@@ -809,7 +809,7 @@ async def test_device_warning_generic():
                     await asyncio.wait_for(dut.get_meter_list(), timeout=0.05)
 
         assert len(w) == 1
-        assert issubclass(w[-1].category, DeviceWarning)
+        assert issubclass(w[-1].category, RAVEnWarning)
         assert 'Something unexpected happened' in str(w[-1].message)
 
 
@@ -824,10 +824,10 @@ async def test_device_warning_generic_error():
     }
 
     with warnings.catch_warnings():
-        warnings.simplefilter('error', DeviceWarning)
+        warnings.simplefilter('error', RAVEnWarning)
         async with mock_device(responses) as (host, port):
             async with RAVEnNetworkDevice(host, port) as dut:
-                with pytest.raises(DeviceWarning):
+                with pytest.raises(RAVEnWarning):
                     await dut.get_meter_list()
 
 
@@ -842,8 +842,8 @@ async def test_device_warning_unknown_command():
     }
 
     with warnings.catch_warnings():
-        warnings.simplefilter('error', DeviceWarning)
+        warnings.simplefilter('error', RAVEnWarning)
         async with mock_device(responses) as (host, port):
             async with RAVEnNetworkDevice(host, port) as dut:
-                with pytest.raises(UnknownCommandWarning):
+                with pytest.raises(UnknownRAVEnCommandWarning):
                     await dut.get_meter_list()