Revize cdc18553

re #9421 Restructured the test folder

client/src/usb_detector/api_client.py

@@ -49 +49 @@
 
 
 def _resend_cached_payloads():
-    retries = min(_config.max_retries, len(_cache))
+    retries = min(_config.cache_max_retries, len(_cache))
     logging.info(f"emptying the cache ({retries} records)")
     for _ in range(0, retries):
         payload = _cache.pop()

client/tests/api_client/test_api_client_set_config.py

@@ -1 @@
-from client.src.usb_detector import api_client
-
-from unittest import mock
-
-
[e-mail]('client.src.usb_detector.api_client._init_cache')
-def test_api_client_set_config_1(_init_cache_mock):
-    class Config:
-        def __init__(self):
-            self.server_url = "127.0.0.1"
-            self.server_port = "54444"
-            self.server_endpoint = "/api/v1/usb-logs"
-
-    config = Config()
-    api_client.api_client_set_config(config)
-
-    assert api_client._config is config
-    assert api_client._uri == "127.0.0.1:54444/api/v1/usb-logs"

client/tests/api_client/test_cache_failed_payload.py

@@ -1 @@
-from client.src.usb_detector import api_client
-
-
-class CacheMock:
-
-    def __init__(self):
-        self._data = []
-
-    def pop(self):
-        return self._data.pop()
-
-    def append(self, payload):
-        self._data.append(payload)
-
-    def __len__(self):
-        return len(self._data)
-
-
-class ConfigMock:
-
-    def __init__(self):
-        self.cache_max_entries = 5
-
-
-def test_cache_failed_payload_1():
-    payload_mock = {
-        "vendor_id": 1,
-        "product_id": 2
-    }
-    cache = CacheMock()
-    config = ConfigMock()
-
-    api_client._cache = cache
-    api_client._config = config
-
-    for _ in range(0, config.cache_max_entries + 1):
-        api_client._cache_failed_payload(payload_mock)
-
-    assert len(api_client._cache) == config.cache_max_entries
-
-
-def test_cache_failed_payload_2():
-    payload_mock = {
-        "vendor_id": 1,
-        "product_id": 2
-    }
-    cache = CacheMock()
-    config = ConfigMock()
-
-    api_client._cache = cache
-    api_client._config = config
-    count = int(config.cache_max_entries / 2)
-
-    for _ in range(0, count):
-        api_client._cache_failed_payload(payload_mock)
-
-    assert len(api_client._cache) == count
-

client/tests/api_client/test_resend_cached_payloads.py

@@ -1 @@
-from client.src.usb_detector import api_client
-
-from unittest import mock
-
-
-class CacheMock:
-
-    def __init__(self):
-        self._data = []
-
-    def pop(self):
-        return self._data.pop()
-
-    def append(self, payload):
-        self._data.append(payload)
-
-    def __len__(self):
-        return len(self._data)
-
-
-class ConfigMock:
-
-    def __init__(self):
-        self.cache_max_entries = 5
-        self.max_retries = 3
-
-
-payload_mock = {
-    "device": {
-        "vendor_id": 1,
-        "product_id": 2
-    }
-}
-
-
[e-mail]('client.src.usb_detector.api_client.send_data')
-def test_resend_cached_payloads_1(send_data_mock):
-    cache = CacheMock()
-    config = ConfigMock()
-
-    api_client._cache = cache
-    api_client._config = config
-
-    for _ in range(0, config.cache_max_entries + 1):
-        api_client._cache_failed_payload(payload_mock)
-
-    api_client._resend_cached_payloads()
-    send_data_mock.assert_called()
-    assert len(api_client._cache) == 2
-
-    api_client._resend_cached_payloads()
-    assert len(api_client._cache) == 0
-
-    api_client._resend_cached_payloads()
-    assert len(api_client._cache) == 0
-
-
[e-mail]('client.src.usb_detector.api_client.send_data')
-def test_resend_cached_payloads_2(send_data_mock):
-    cache = CacheMock()
-    config = ConfigMock()
-
-    api_client._cache = cache
-    api_client._config = config
-
-    api_client._resend_cached_payloads()
-    send_data_mock.assert_not_called()
-    assert len(api_client._cache) == 0
-
-
[e-mail]('client.src.usb_detector.api_client.send_data')
-def test_resend_cached_payloads_3(send_data_mock):
-    cache = CacheMock()
-    config = ConfigMock()
-
-    api_client._cache = cache
-    api_client._config = config
-
-    for _ in range(0, 2):
-        api_client._cache_failed_payload(payload_mock)
-
-    assert len(api_client._cache) == 2
-    api_client._resend_cached_payloads()
-    send_data_mock.assert_called()
-    assert len(api_client._cache) == 0
-

client/tests/api_client/test_send_data.py

@@ -1 @@
-from client.src.usb_detector import api_client
-
-import requests
-from unittest import mock
-
-
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
-def test_send_data_1(_cache_failed_payload_mock):
-    payload_mock = {
-        "device": {
-            "vendor_id": 1,
-            "product_id": 2
-        }
-    }
-
-    api_client._uri = None
-    api_client.send_data(payload_mock)
-
-    args = _cache_failed_payload_mock.call_args.args
-    _cache_failed_payload_mock.assert_called()
-    assert args[0] == payload_mock
-
-
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
-def test_send_data_2(_cache_failed_payload_mock):
-    payload_mock = {
-        "device": {
-            "vendor_id": 1,
-            "product_id": 2
-        }
-    }
-
-    api_client._uri = "127.0.0.1:54444/api/v1/usb-logs"
-    api_client.send_data(payload_mock)
-
-    args = _cache_failed_payload_mock.call_args.args
-    _cache_failed_payload_mock.assert_called()
-    assert args[0] == payload_mock
-
-
[e-mail]('client.src.usb_detector.api_client.requests.post')
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
-def test_send_data_3(_cache_failed_payload_mock, post_mock):
-    payload_mock = {
-        "device": {
-            "vendor_id": 1,
-            "product_id": 2
-        }
-    }
-
-    api_client._uri = "127.0.0.1:54444/api/v1/usb-logs"
-    api_client.send_data(payload_mock)
-
-    _cache_failed_payload_mock.assert_not_called()
-
-
[e-mail]('client.src.usb_detector.api_client.requests.post')
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
-def test_send_data_4(_cache_failed_payload_mock, post_mock):
-    payload_mock = {
-        "device": {
-            "vendor_id": 1,
-            "product_id": 2
-        }
-    }
-    post_mock.side_effect = requests.exceptions.HTTPError()
-
-    api_client._uri = "127.0.0.1:54444/api/v1/usb-logs"
-    api_client.send_data(payload_mock)
-
-    args = _cache_failed_payload_mock.call_args.args
-    _cache_failed_payload_mock.assert_called()
-    assert args[0] == payload_mock

client/tests/detector/test_get_connected_devices.py

@@ -1 @@
-from client.src.usb_detector import detector
-
-
-def test_get_connected_devices_1():
-    assert detector._get_connected_devices([], []) == []
-
-
-def test_get_connected_devices_2():
-    assert detector._get_connected_devices([], None) == []
-
-
-def test_get_connected_devices_3():
-    assert detector._get_connected_devices(None, []) == []
-
-
-def test_get_connected_devices_4():
-    assert detector._get_connected_devices(None, None) == []
-
-
-def test_get_connected_devices_5():
-    assert detector._get_connected_devices([1, 2, 3], None) == [1, 2, 3]
-
-
-def test_get_connected_devices_6():
-    assert detector._get_connected_devices(None, [4, 5, 6]) == []
-
-
-def test_get_connected_devices_7():
-    assert detector._get_connected_devices([], [4, 5, 6]) == []
-
-
-def test_get_connected_devices_8():
-    assert detector._get_connected_devices([1, 2, 3], []) == [1, 2, 3]
-
-
-def test_get_connected_devices_9():
-    assert detector._get_connected_devices([1, 2, 3], [2, 3]) == [1]
-
-
-def test_get_connected_devices_10():
-    detected_devices = [
-        {
-            "vendor_id": 1,
-            "product_id": 2,
-            "serial_number": 3
-        },
-        {
-            "vendor_id": 4,
-            "product_id": 5,
-            "serial_number": 5
-        }
-    ]
-    last_connected_devices = [
-        {
-            "vendor_id": 1,
-            "product_id": 2,
-            "serial_number": 3
-        }
-    ]
-    assert detector._get_connected_devices(detected_devices, last_connected_devices) == [detected_devices[1]]
-    assert detector._get_connected_devices(detected_devices, None) == detected_devices
-    assert detector._get_connected_devices(detected_devices, []) == detected_devices

client/tests/detector/test_get_disconnected_devices.py

@@ -1 @@
-from client.src.usb_detector import detector
-
-
-def test_get_disconnected_devices_1():
-    assert detector._get_disconnected_devices([], []) == []
-
-
-def test_get_disconnected_devices_2():
-    assert detector._get_disconnected_devices([], None) == []
-
-
-def test_get_disconnected_devices_3():
-    assert detector._get_disconnected_devices(None, []) == []
-
-
-def test_get_disconnected_devices_4():
-    assert detector._get_disconnected_devices(None, None) == []
-
-
-def test_get_disconnected_devices_5():
-    assert detector._get_disconnected_devices([1, 2, 3, 4, 5], None) == []
-
-
-def test_get_disconnected_devices_6():
-    assert detector._get_disconnected_devices([1, 2, 3, 4, 5], []) == []
-
-
-def test_get_disconnected_devices_7():
-    assert detector._get_disconnected_devices([1, 3, 5], [1, 2, 3, 4, 5]) == [2, 4]
-
-
-def test_get_disconnected_devices_8():
-    assert detector._get_disconnected_devices([], [1, 2, 3, 4, 5]) == [1, 2, 3, 4, 5]
-
-
-def test_get_disconnected_devices_9():
-    assert detector._get_disconnected_devices([], [1, 2, 3, 4, 5]) == [1, 2, 3, 4, 5]
-
-
-def test_get_disconnected_devices_10():
-    last_connected_devices = [
-        {
-            "vendor_id": 1,
-            "product_id": 2,
-            "serial_number": 3
-        },
-        {
-            "vendor_id": 4,
-            "product_id": 5,
-            "serial_number": 5
-        }
-    ]
-    detected_devices = [
-        {
-            "vendor_id": 1,
-            "product_id": 2,
-            "serial_number": 3
-        }
-    ]
-    assert detector._get_disconnected_devices(detected_devices, last_connected_devices) == [last_connected_devices[1]]
-    assert detector._get_disconnected_devices(detected_devices, None) == []
-    assert detector._get_disconnected_devices(detected_devices, []) == []
-    assert detector._get_disconnected_devices([], last_connected_devices) == last_connected_devices
-    assert detector._get_disconnected_devices(None, last_connected_devices) == last_connected_devices
-
-

client/tests/detector/test_notify_listeners.py

@@ -1 @@
-from unittest.mock import Mock
-
-from client.src.usb_detector import detector
-
-
-def test_notify_listeners_1():
-    mocks = [Mock() for _ in range(10)]
-    detector._notify_listeners([mock.listener for mock in mocks], [1, 2, 3])
-    for mock in mocks:
-        mock.listener.assert_called()
-
-
-def test_notify_listeners_2():
-    mocks = [Mock() for _ in range(10)]
-    detector._notify_listeners([mock.listener for mock in mocks], [])
-    for mock in mocks:
-        mock.listener.assert_not_called()
-
-
-def test_notify_listeners_3():
-    mocks = [Mock() for _ in range(10)]
-    detector._notify_listeners([mock.listener for mock in mocks], None)
-    for mock in mocks:
-        mock.listener.assert_not_called()

client/tests/detector/test_register_listener.py

@@ -1 @@
-from client.src.usb_detector import detector
-
-
-def test_register_listener_1():
-    def connected_handler():
-        pass
-
-    detector.register_listener(connected_handler)
-
-    assert connected_handler in detector._listeners_connected
-    assert connected_handler not in detector._listeners_disconnected
-
-
-def test_register_listener_2():
-    def connected_handler():
-        pass
-
-    detector.register_listener(connected_handler, connected=True)
-
-    assert connected_handler in detector._listeners_connected
-    assert connected_handler not in detector._listeners_disconnected
-
-
-def test_register_listener_3():
-    def connected_handler():
-        pass
-
-    detector.register_listener(connected_handler, connected=False)
-
-    assert connected_handler in detector._listeners_disconnected
-    assert connected_handler not in detector._listeners_connected

client/tests/detector/test_update.py

@@ -1 @@
-from unittest import mock
-from unittest.mock import Mock
-
-from client.src.usb_detector import detector
-
-
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[1, 2, 3])
-def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
-    connected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
-
-    disconnected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
-
-    detector._update()
-    assert detector._last_connected_devices == [1, 2, 3]
-    for listener_mock in connected_mocks:
-        listener_mock.listener.assert_called()
-    for listener_mock in disconnected_mocks:
-        listener_mock.listener.assert_not_called()
-
-
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[1, 2, 3])
-def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
-    detector._last_connected_devices = [1, 2, 3]
-    connected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
-
-    disconnected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
-
-    detector._update()
-    assert detector._last_connected_devices == [1, 2, 3]
-    for listener_mock in connected_mocks:
-        listener_mock.listener.assert_not_called()
-    for listener_mock in disconnected_mocks:
-        listener_mock.listener.assert_not_called()
-
-
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[])
-def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
-    detector._last_connected_devices = [1, 2, 3]
-    connected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
-
-    disconnected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
-
-    detector._update()
-    assert detector._last_connected_devices == []
-    for listener_mock in connected_mocks:
-        listener_mock.listener.assert_not_called()
-    for listener_mock in disconnected_mocks:
-        listener_mock.listener.assert_called()
-
-
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[2, 3, 4])
-def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
-    detector._last_connected_devices = [1, 2, 3, 5]
-    connected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
-
-    disconnected_mocks = [Mock() for _ in range(10)]
-    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
-
-    detector._update()
-    assert detector._last_connected_devices == [2, 3, 4]
-    for listener_mock in connected_mocks:
-        listener_mock.listener.assert_called()
-    for listener_mock in disconnected_mocks:
-        listener_mock.listener.assert_called()

client/tests/detector/test_usb_detector_set_config.py

@@ -1 @@
-from client.src.usb_detector import detector
-
-
-def test_usb_detector_set_config_1():
-    assert None is detector._config
-
-
-def test_usb_detector_set_config_2():
-    class Config:
-        pass
-
-    config = Config()
-    detector.usb_detector_set_config(config)
-    assert config is detector._config

client/tests/event_listener/test_get_metadata.py

@@ -1 @@
-import getpass
-import platform
-from datetime import datetime
-
-from client.src.usb_detector import event_listener
-
-
-def test_get_metadata_1():
-    expected_metadata = {
-        "username": getpass.getuser(),
-        "hostname": platform.uname().node,
-        "timestamp": str(datetime.now()).split('.')[0]
-    }
-
-    actual_metadata = event_listener._get_metadata()
-
-    assert actual_metadata["username"] == expected_metadata["username"]
-    assert actual_metadata["hostname"] == expected_metadata["hostname"]
-    assert actual_metadata["timestamp"] == expected_metadata["timestamp"]

client/tests/event_listener/test_send_payload_to_server.py

@@ -1 @@
-import platform
-import getpass
-from unittest import mock
-from datetime import datetime
-
-import client.src.usb_detector.event_listener
-
-
[e-mail]('client.src.usb_detector.event_listener.send_data')
-def test_send_payload_to_server_1(send_data_mock):
-    device_mock = {
-        "vendor_id": 1,
-        "product_id": 2
-    }
-    metadata_mock = {
-        "username": getpass.getuser(),
-        "hostname": platform.uname().node,
-        "timestamp": str(datetime.now()).split('.')[0]
-    }
-    status_mock = "connected"
-
-    metadata_mock["device"] = device_mock
-    metadata_mock["status"] = status_mock
-
-    client.src.usb_detector.event_listener._send_payload_to_server(device_mock, status_mock)
-
-    args = send_data_mock.call_args.args
-    send_data_mock.assert_called()
-    assert args[0] == metadata_mock

client/tests/test_usb_detector/test_api_client_set_config.py

@@ +1 @@
+from client.src.usb_detector import api_client
+
+from unittest import mock
+
+
[e-mail]('client.src.usb_detector.api_client._init_cache')
+def test_api_client_set_config_1(_init_cache_mock):
+    class Config:
+        def __init__(self):
+            self.server_url = "127.0.0.1"
+            self.server_port = "54444"
+            self.server_endpoint = "/api/v1/usb-logs"
+
+    config = Config()
+    api_client.api_client_set_config(config)
+
+    assert api_client._config is config
+    assert api_client._uri == "127.0.0.1:54444/api/v1/usb-logs"

client/tests/test_usb_detector/test_cache_failed_payload.py

@@ +1 @@
+from client.src.usb_detector import api_client
+
+
+class CacheMock:
+
+    def __init__(self):
+        self._data = []
+
+    def pop(self):
+        return self._data.pop()
+
+    def append(self, payload):
+        self._data.append(payload)
+
+    def __len__(self):
+        return len(self._data)
+
+
+class ConfigMock:
+
+    def __init__(self):
+        self.cache_max_entries = 5
+
+
+def test_cache_failed_payload_1():
+    payload_mock = {
+        "vendor_id": 1,
+        "product_id": 2
+    }
+    cache = CacheMock()
+    config = ConfigMock()
+
+    api_client._cache = cache
+    api_client._config = config
+
+    for _ in range(0, config.cache_max_entries + 1):
+        api_client._cache_failed_payload(payload_mock)
+
+    assert len(api_client._cache) == config.cache_max_entries
+
+
+def test_cache_failed_payload_2():
+    payload_mock = {
+        "vendor_id": 1,
+        "product_id": 2
+    }
+    cache = CacheMock()
+    config = ConfigMock()
+
+    api_client._cache = cache
+    api_client._config = config
+    count = int(config.cache_max_entries / 2)
+
+    for _ in range(0, count):
+        api_client._cache_failed_payload(payload_mock)
+
+    assert len(api_client._cache) == count
+

client/tests/test_usb_detector/test_get_connected_devices.py

@@ +1 @@
+from client.src.usb_detector import detector
+
+
+def test_get_connected_devices_1():
+    assert detector._get_connected_devices([], []) == []
+
+
+def test_get_connected_devices_2():
+    assert detector._get_connected_devices([], None) == []
+
+
+def test_get_connected_devices_3():
+    assert detector._get_connected_devices(None, []) == []
+
+
+def test_get_connected_devices_4():
+    assert detector._get_connected_devices(None, None) == []
+
+
+def test_get_connected_devices_5():
+    assert detector._get_connected_devices([1, 2, 3], None) == [1, 2, 3]
+
+
+def test_get_connected_devices_6():
+    assert detector._get_connected_devices(None, [4, 5, 6]) == []
+
+
+def test_get_connected_devices_7():
+    assert detector._get_connected_devices([], [4, 5, 6]) == []
+
+
+def test_get_connected_devices_8():
+    assert detector._get_connected_devices([1, 2, 3], []) == [1, 2, 3]
+
+
+def test_get_connected_devices_9():
+    assert detector._get_connected_devices([1, 2, 3], [2, 3]) == [1]
+
+
+def test_get_connected_devices_10():
+    detected_devices = [
+        {
+            "vendor_id": 1,
+            "product_id": 2,
+            "serial_number": 3
+        },
+        {
+            "vendor_id": 4,
+            "product_id": 5,
+            "serial_number": 5
+        }
+    ]
+    last_connected_devices = [
+        {
+            "vendor_id": 1,
+            "product_id": 2,
+            "serial_number": 3
+        }
+    ]
+    assert detector._get_connected_devices(detected_devices, last_connected_devices) == [detected_devices[1]]
+    assert detector._get_connected_devices(detected_devices, None) == detected_devices
+    assert detector._get_connected_devices(detected_devices, []) == detected_devices

client/tests/test_usb_detector/test_get_disconnected_devices.py

@@ +1 @@
+from client.src.usb_detector import detector
+
+
+def test_get_disconnected_devices_1():
+    assert detector._get_disconnected_devices([], []) == []
+
+
+def test_get_disconnected_devices_2():
+    assert detector._get_disconnected_devices([], None) == []
+
+
+def test_get_disconnected_devices_3():
+    assert detector._get_disconnected_devices(None, []) == []
+
+
+def test_get_disconnected_devices_4():
+    assert detector._get_disconnected_devices(None, None) == []
+
+
+def test_get_disconnected_devices_5():
+    assert detector._get_disconnected_devices([1, 2, 3, 4, 5], None) == []
+
+
+def test_get_disconnected_devices_6():
+    assert detector._get_disconnected_devices([1, 2, 3, 4, 5], []) == []
+
+
+def test_get_disconnected_devices_7():
+    assert detector._get_disconnected_devices([1, 3, 5], [1, 2, 3, 4, 5]) == [2, 4]
+
+
+def test_get_disconnected_devices_8():
+    assert detector._get_disconnected_devices([], [1, 2, 3, 4, 5]) == [1, 2, 3, 4, 5]
+
+
+def test_get_disconnected_devices_9():
+    assert detector._get_disconnected_devices([], [1, 2, 3, 4, 5]) == [1, 2, 3, 4, 5]
+
+
+def test_get_disconnected_devices_10():
+    last_connected_devices = [
+        {
+            "vendor_id": 1,
+            "product_id": 2,
+            "serial_number": 3
+        },
+        {
+            "vendor_id": 4,
+            "product_id": 5,
+            "serial_number": 5
+        }
+    ]
+    detected_devices = [
+        {
+            "vendor_id": 1,
+            "product_id": 2,
+            "serial_number": 3
+        }
+    ]
+    assert detector._get_disconnected_devices(detected_devices, last_connected_devices) == [last_connected_devices[1]]
+    assert detector._get_disconnected_devices(detected_devices, None) == []
+    assert detector._get_disconnected_devices(detected_devices, []) == []
+    assert detector._get_disconnected_devices([], last_connected_devices) == last_connected_devices
+    assert detector._get_disconnected_devices(None, last_connected_devices) == last_connected_devices
+
+

client/tests/test_usb_detector/test_get_metadata.py

@@ +1 @@
+import getpass
+import platform
+from datetime import datetime
+
+from client.src.usb_detector import event_listener
+
+
+def test_get_metadata_1():
+    expected_metadata = {
+        "username": getpass.getuser(),
+        "hostname": platform.uname().node,
+        "timestamp": str(datetime.now()).split('.')[0]
+    }
+
+    actual_metadata = event_listener._get_metadata()
+
+    assert actual_metadata["username"] == expected_metadata["username"]
+    assert actual_metadata["hostname"] == expected_metadata["hostname"]
+    assert actual_metadata["timestamp"] == expected_metadata["timestamp"]

client/tests/test_usb_detector/test_notify_listeners.py

@@ +1 @@
+from unittest.mock import Mock
+
+from client.src.usb_detector import detector
+
+
+def test_notify_listeners_1():
+    mocks = [Mock() for _ in range(10)]
+    detector._notify_listeners([mock.listener for mock in mocks], [1, 2, 3])
+    for mock in mocks:
+        mock.listener.assert_called()
+
+
+def test_notify_listeners_2():
+    mocks = [Mock() for _ in range(10)]
+    detector._notify_listeners([mock.listener for mock in mocks], [])
+    for mock in mocks:
+        mock.listener.assert_not_called()
+
+
+def test_notify_listeners_3():
+    mocks = [Mock() for _ in range(10)]
+    detector._notify_listeners([mock.listener for mock in mocks], None)
+    for mock in mocks:
+        mock.listener.assert_not_called()

client/tests/test_usb_detector/test_read_connected_devices.py

@@ +1 @@
+from client.src.usb_detector.usb_reader import read_connected_devices
+
+
+def test_read_connected_devices_1():
+    # This test assumes that there is no USB plugged in.
+    assert read_connected_devices() == []

client/tests/test_usb_detector/test_register_listener.py

@@ +1 @@
+from client.src.usb_detector import detector
+
+
+def test_register_listener_1():
+    def connected_handler():
+        pass
+
+    detector.register_listener(connected_handler)
+
+    assert connected_handler in detector._listeners_connected
+    assert connected_handler not in detector._listeners_disconnected
+
+
+def test_register_listener_2():
+    def connected_handler():
+        pass
+
+    detector.register_listener(connected_handler, connected=True)
+
+    assert connected_handler in detector._listeners_connected
+    assert connected_handler not in detector._listeners_disconnected
+
+
+def test_register_listener_3():
+    def connected_handler():
+        pass
+
+    detector.register_listener(connected_handler, connected=False)
+
+    assert connected_handler in detector._listeners_disconnected
+    assert connected_handler not in detector._listeners_connected

client/tests/test_usb_detector/test_resend_cached_payloads.py

@@ +1 @@
+from client.src.usb_detector import api_client
+
+from unittest import mock
+
+
+class CacheMock:
+
+    def __init__(self):
+        self._data = []
+
+    def pop(self):
+        return self._data.pop()
+
+    def append(self, payload):
+        self._data.append(payload)
+
+    def __len__(self):
+        return len(self._data)
+
+
+class ConfigMock:
+
+    def __init__(self):
+        self.cache_max_entries = 5
+        self.cache_max_retries = 3
+
+
+payload_mock = {
+    "device": {
+        "vendor_id": 1,
+        "product_id": 2
+    }
+}
+
+
[e-mail]('client.src.usb_detector.api_client.send_data')
+def test_resend_cached_payloads_1(send_data_mock):
+    cache = CacheMock()
+    config = ConfigMock()
+
+    api_client._cache = cache
+    api_client._config = config
+
+    for _ in range(0, config.cache_max_entries + 1):
+        api_client._cache_failed_payload(payload_mock)
+
+    api_client._resend_cached_payloads()
+    send_data_mock.assert_called()
+    assert len(api_client._cache) == 2
+
+    api_client._resend_cached_payloads()
+    assert len(api_client._cache) == 0
+
+    api_client._resend_cached_payloads()
+    assert len(api_client._cache) == 0
+
+
[e-mail]('client.src.usb_detector.api_client.send_data')
+def test_resend_cached_payloads_2(send_data_mock):
+    cache = CacheMock()
+    config = ConfigMock()
+
+    api_client._cache = cache
+    api_client._config = config
+
+    api_client._resend_cached_payloads()
+    send_data_mock.assert_not_called()
+    assert len(api_client._cache) == 0
+
+
[e-mail]('client.src.usb_detector.api_client.send_data')
+def test_resend_cached_payloads_3(send_data_mock):
+    cache = CacheMock()
+    config = ConfigMock()
+
+    api_client._cache = cache
+    api_client._config = config
+
+    for _ in range(0, 2):
+        api_client._cache_failed_payload(payload_mock)
+
+    assert len(api_client._cache) == 2
+    api_client._resend_cached_payloads()
+    send_data_mock.assert_called()
+    assert len(api_client._cache) == 0
+

client/tests/test_usb_detector/test_send_data.py

@@ +1 @@
+from client.src.usb_detector import api_client
+
+import requests
+from unittest import mock
+
+
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
+def test_send_data_1(_cache_failed_payload_mock):
+    payload_mock = {
+        "device": {
+            "vendor_id": 1,
+            "product_id": 2
+        }
+    }
+
+    api_client._uri = None
+    api_client.send_data(payload_mock)
+
+    args = _cache_failed_payload_mock.call_args.args
+    _cache_failed_payload_mock.assert_called()
+    assert args[0] == payload_mock
+
+
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
+def test_send_data_2(_cache_failed_payload_mock):
+    payload_mock = {
+        "device": {
+            "vendor_id": 1,
+            "product_id": 2
+        }
+    }
+
+    api_client._uri = "127.0.0.1:54444/api/v1/usb-logs"
+    api_client.send_data(payload_mock)
+
+    args = _cache_failed_payload_mock.call_args.args
+    _cache_failed_payload_mock.assert_called()
+    assert args[0] == payload_mock
+
+
[e-mail]('client.src.usb_detector.api_client.requests.post')
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
+def test_send_data_3(_cache_failed_payload_mock, post_mock):
+    payload_mock = {
+        "device": {
+            "vendor_id": 1,
+            "product_id": 2
+        }
+    }
+
+    api_client._uri = "127.0.0.1:54444/api/v1/usb-logs"
+    api_client.send_data(payload_mock)
+
+    _cache_failed_payload_mock.assert_not_called()
+
+
[e-mail]('client.src.usb_detector.api_client.requests.post')
[e-mail]('client.src.usb_detector.api_client._cache_failed_payload')
+def test_send_data_4(_cache_failed_payload_mock, post_mock):
+    payload_mock = {
+        "device": {
+            "vendor_id": 1,
+            "product_id": 2
+        }
+    }
+    post_mock.side_effect = requests.exceptions.HTTPError()
+
+    api_client._uri = "127.0.0.1:54444/api/v1/usb-logs"
+    api_client.send_data(payload_mock)
+
+    args = _cache_failed_payload_mock.call_args.args
+    _cache_failed_payload_mock.assert_called()
+    assert args[0] == payload_mock

client/tests/test_usb_detector/test_send_payload_to_server.py

@@ +1 @@
+import platform
+import getpass
+from unittest import mock
+from datetime import datetime
+
+import client.src.usb_detector.event_listener
+
+
[e-mail]('client.src.usb_detector.event_listener.send_data')
+def test_send_payload_to_server_1(send_data_mock):
+    device_mock = {
+        "vendor_id": 1,
+        "product_id": 2
+    }
+    metadata_mock = {
+        "username": getpass.getuser(),
+        "hostname": platform.uname().node,
+        "timestamp": str(datetime.now()).split('.')[0]
+    }
+    status_mock = "connected"
+
+    metadata_mock["device"] = device_mock
+    metadata_mock["status"] = status_mock
+
+    client.src.usb_detector.event_listener._send_payload_to_server(device_mock, status_mock)
+
+    args = send_data_mock.call_args.args
+    send_data_mock.assert_called()
+    assert args[0] == metadata_mock

client/tests/test_usb_detector/test_update.py

@@ +1 @@
+from unittest import mock
+from unittest.mock import Mock
+
+from client.src.usb_detector import detector
+
+
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[1, 2, 3])
+def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
+    connected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
+
+    disconnected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
+
+    detector._update()
+    assert detector._last_connected_devices == [1, 2, 3]
+    for listener_mock in connected_mocks:
+        listener_mock.listener.assert_called()
+    for listener_mock in disconnected_mocks:
+        listener_mock.listener.assert_not_called()
+
+
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[1, 2, 3])
+def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
+    detector._last_connected_devices = [1, 2, 3]
+    connected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
+
+    disconnected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
+
+    detector._update()
+    assert detector._last_connected_devices == [1, 2, 3]
+    for listener_mock in connected_mocks:
+        listener_mock.listener.assert_not_called()
+    for listener_mock in disconnected_mocks:
+        listener_mock.listener.assert_not_called()
+
+
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[])
+def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
+    detector._last_connected_devices = [1, 2, 3]
+    connected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
+
+    disconnected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
+
+    detector._update()
+    assert detector._last_connected_devices == []
+    for listener_mock in connected_mocks:
+        listener_mock.listener.assert_not_called()
+    for listener_mock in disconnected_mocks:
+        listener_mock.listener.assert_called()
+
+
[e-mail]('client.src.usb_detector.detector._store_connected_devices')
[e-mail]('client.src.usb_detector.detector.read_connected_devices', return_value=[2, 3, 4])
+def test_update_1(read_connected_devices_mock, _store_connected_devices_mock):
+    detector._last_connected_devices = [1, 2, 3, 5]
+    connected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_connected = [listener_mock.listener for listener_mock in connected_mocks]
+
+    disconnected_mocks = [Mock() for _ in range(10)]
+    detector._listeners_disconnected = [listener_mock.listener for listener_mock in disconnected_mocks]
+
+    detector._update()
+    assert detector._last_connected_devices == [2, 3, 4]
+    for listener_mock in connected_mocks:
+        listener_mock.listener.assert_called()
+    for listener_mock in disconnected_mocks:
+        listener_mock.listener.assert_called()

client/tests/test_usb_detector/test_usb_detector_set_config.py

@@ +1 @@
+from client.src.usb_detector import detector
+
+
+def test_usb_detector_set_config_1():
+    assert None is detector._config
+
+
+def test_usb_detector_set_config_2():
+    class Config:
+        pass
+
+    config = Config()
+    detector.usb_detector_set_config(config)
+    assert config is detector._config

client/tests/usb_reader/test_read_connected_devices.py

@@ -1 @@
-from client.src.usb_detector.usb_reader import read_connected_devices
-
-
-def test_read_connected_devices_1():
-    # This test assumes that there is no USB plugged in.
-    assert read_connected_devices() == []