tests

validity/data_backends.py

@@ -56,7 +56,7 @@ def start_polling(self, devices) -> tuple[list[Generator], set[DescriptiveError]
         for poller, device_group in groupby(devices, key=lambda device: device.poller):
             if poller is None:
                 no_poller_errors.update(
-                    DescriptiveError(device=device, error="No poller bound") for device in device_group
+                    DescriptiveError(device=str(device), error="No poller bound") for device in device_group
                 )
             else:
                 result_generators.append(poller.get_backend().poll(device_group))
@@ -71,9 +71,7 @@ def fetch(self, device_filter: Q | None = None):
                 if cmd_result.errored:
                     errors.add(cmd_result.descriptive_error)
                 cmd_result.write_on_disk(dir_name)
-            polling_info = PollingInfo(
-                devices_polled=devices.count(), errors=errors, partial_sync=device_filter is not None
-            )
+            polling_info = PollingInfo(devices_polled=devices.count(), errors=errors, partial_sync=not device_filter)
             self.write_metainfo(dir_name, polling_info)
             yield dir_name
 

validity/tests/test_data_backends.py

@@ -0,0 +1,45 @@
+from functools import partial
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import Mock
+
+import yaml
+
+from validity.data_backends import PollingBackend
+from validity.pollers.result import PollingInfo
+
+
+def test_write_metainfo():
+    with TemporaryDirectory() as dir_name:
+        backend = PollingBackend("/")
+        polling_info = PollingInfo(devices_polled=5, errors=[])
+        backend.write_metainfo(dir_name, polling_info)
+        file = Path(dir_name) / "polling_info.yaml"
+        assert yaml.safe_load(file.read_text()) == polling_info.model_dump(exclude_defaults=True)
+
+
+def test_start_polling():
+    def poll(devices, poller):
+        poller.devices = list(devices)
+
+    backend = PollingBackend("/")
+    poller1 = Mock(name="poller1")
+    poller1.get_backend.return_value.poll = partial(poll, poller=poller1)
+    poller2 = Mock(name="poller2")
+    poller2.get_backend.return_value.poll = partial(poll, poller=poller2)
+    devices = [
+        Mock(poller=poller1),
+        Mock(poller=poller1),
+        Mock(poller=poller2),
+        Mock(poller=None),
+        Mock(poller=None),
+        Mock(poller=None),
+    ]
+    result_generators, errors = backend.start_polling(devices)
+    assert len(result_generators) == 2
+    assert poller1.devices == devices[:2]
+    assert poller2.devices == devices[2:3]
+    assert len(errors) == 3
+    device_strings = {str(device): device for device in devices}
+    for error in errors:
+        assert device_strings[error.device].poller is None