tests from #33

tests/core/test_reader.py

@@ -9,15 +9,18 @@ class MockReader(reader.Reader):
     def __init__(self, raise_on_enter=False):
         self.raise_on_enter = raise_on_enter
 
+    def read_one(self):
+        raise NotImplemented
+
     def __call__(self):
         raise NotImplemented
 
-    def __enter__(self):
+    def open(self):
         if self.raise_on_enter:
             raise reader.UnableToRead()
         self.entered = True
 
-    def __exit__(self, *_args):
+    def close(self):
         self.exited = True
 
 
@@ -107,6 +110,31 @@ def passive_read(n):
     )
 
 
+@pytest.fixture
+def mock_sensor_temp_failure(mock_serial):
+    def passive_read(n):
+        if n == 1:
+            # first return garbage data (bad checksum)
+            return (
+                b"BM\x00\x1c"  # expected header
+                + b"\0" * 26  # payload (to total 32 bytes)
+                + b"\x00\xFF"  # checksum
+            )
+        else:
+            # then behave like the original stub again
+            return (
+                b"BM\x00\x1c"  # expected header
+                + b".........................."  # payload
+                + b"\x05W"  # checksum
+            )
+
+    mock_serial.stub(
+        name="passive_read",
+        receive_bytes=b"BM\xe2\x00\x00\x01q",
+        send_fn=passive_read,
+    )
+
+
 @pytest.fixture
 def sensor_reader_factory(monkeypatch, mock_sensor):
     def factory(
@@ -153,6 +181,26 @@ def test_sensor_reader(mock_sensor, sensor_reader_factory):
     assert mock_sensor.stubs["sleep"].called
 
 
+def test_sensor_reader_read_one(mock_sensor, sensor_reader_factory):
+    sensor_reader = sensor_reader_factory()
+
+    with sensor_reader as r:
+        obs = sensor_reader.read_one()
+
+    # check data was read
+    assert obs.pm10 == 11822
+
+
+def test_sensor_reader_read_one_raw(mock_sensor, sensor_reader_factory):
+    sensor_reader = sensor_reader_factory()
+
+    with sensor_reader as r:
+        raw_data = sensor_reader.read_one(raw=True)
+
+    # check data was read
+    assert raw_data.data
+
+
 def test_sensor_reader_sleep(sensor_reader_factory, mock_sleep):
     sensor_reader = sensor_reader_factory(
         samples=2,  # try to read twice
@@ -204,6 +252,23 @@ def test_sensor_reader_warm_up(
     assert len(obs) == 1
 
 
+def test_sensor_reader_temp_failure(
+    mock_sensor,
+    sensor_reader_factory,
+    mock_sensor_temp_failure,
+):
+    sensor_reader = sensor_reader_factory()
+
+    with sensor_reader as r:
+        obs = list(r())
+
+    # check one sample still acquired
+    assert len(obs) == 1
+
+    # check two samples were attempted
+    assert mock_sensor.stubs["passive_read"].calls == 2
+
+
 def test_sensor_reader_sensor_mismatch(mock_sensor, sensor_reader_factory):
     sensor_reader = sensor_reader_factory()
 
@@ -271,6 +336,18 @@ def test_message_reader():
     assert len(values) == 10
 
 
+def test_message_reader_read_one():
+    message_reader = reader.MessageReader(
+        path=captured_data,
+        sensor=Sensor["PMS3003"],
+    )
+
+    with message_reader:
+        obs = message_reader.read_one()
+
+    assert obs.pm10 == 0
+
+
 def test_message_reader_closed():
     message_reader = reader.MessageReader(
         path=captured_data,