fix: add linting conditions

nervous_sensors/connection_manager.py

@@ -92,8 +92,11 @@ async def manage_connection(self, sensor):
 
     async def manage_battery_level(self):
         while not self._stop_event.is_set():
-            await asyncio.sleep(120)
-            self.print_battery_level()
+            try:
+                await asyncio.wait_for(self._stop_event.wait(), timeout=120)
+                self.print_battery_level()
+            except TimeoutError:
+                self.print_battery_level()
 
     # Sensors parallel actions
 

nervous_sensors/nervous_sensor.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 
 from bleak import BleakClient, BleakScanner
-from bleak.exc import BleakDeviceNotFoundError, BleakError
+from bleak.exc import BleakError
 
 from .cli_utils import RESET, get_color
 from .codec import time_encode
@@ -100,7 +100,7 @@ async def stop_notifications(self) -> bool:
             await self._client.stop_notify("00002a19-0000-1000-8000-00805f9b34fb")
             await self._client.stop_notify("6e400003-b5a3-f393-e0a9-e50e24dcca9e")
             return True
-        except (BleakError, KeyError, AttributeError, ValueError):
+        except:
             return False
 
     async def connect(self):
@@ -132,7 +132,7 @@ async def connect(self):
 
             self._client = None
 
-        except BleakDeviceNotFoundError:
+        except:
             if connection_was_established:
                 self._connection_manager.on_sensor_disconnect(self)
             else:

pyproject.toml

@@ -29,6 +29,7 @@ optional = true
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.2.2"
+pytest-asyncio = "^0.23.8"
 pre-commit = "^3.7.1"
 
 [tool.setuptools.packages.find]
@@ -44,6 +45,7 @@ lint.select = ["I", "E", "F", "C"]
 
 [tool.ruff.lint.per-file-ignores]
 "nervous_sensors/pb2.py" = ["F821", "E501"]
+"nervous_sensors/nervous_sensor.py" = ["E722"]
 
 [build-system]
 requires = ["poetry-core"]

tests/test_connection_manager.py

@@ -0,0 +1,48 @@
+import asyncio
+from unittest.mock import Mock
+
+import pytest
+
+from nervous_sensors.connection_manager import ConnectionManager
+from nervous_sensors.nervous_sensor import NervousSensor
+
+
+@pytest.fixture
+def mock_nervous_sensor1():
+    mock_sensor = Mock(spec=NervousSensor)
+    mock_sensor.is_connected.return_value = False
+    return mock_sensor
+
+
+@pytest.fixture
+def mock_nervous_sensor2():
+    mock_sensor = Mock(spec=NervousSensor)
+    mock_sensor.is_connected.return_value = False
+    return mock_sensor
+
+
+@pytest.fixture
+def mock_nervous_sensor3():
+    mock_sensor = Mock(spec=NervousSensor)
+    mock_sensor.is_connected.return_value = False
+    return mock_sensor
+
+
+@pytest.mark.asyncio
+async def test_start_connection(mock_nervous_sensor1, mock_nervous_sensor2, mock_nervous_sensor3):
+    manager = ConnectionManager(sensor_names=[], gui=False, folder=False, lsl=False, parallel_connection_authorized=3)
+    sensors = [mock_nervous_sensor1, mock_nervous_sensor2, mock_nervous_sensor3]
+    manager._sensors = sensors
+
+    try:
+        await asyncio.wait_for(manager.start(), timeout=1)
+    except asyncio.TimeoutError:
+        pass
+
+    for mock_sensor in sensors:
+        assert mock_sensor.connect.call_count >= 1
+
+
+@pytest.mark.asyncio
+async def test_stop_all_notifications():
+    assert True

tests/test_nervous_sensor.py

@@ -0,0 +1,66 @@
+import asyncio
+from asyncio import TaskGroup
+from unittest.mock import Mock
+
+import pytest
+from bleak import BleakClient, BleakScanner, BLEDevice
+
+from nervous_sensors.connection_manager import ConnectionManager
+from nervous_sensors.nervous_sensor import NervousSensor
+
+
+@pytest.fixture
+def mock_bleak_client():
+    return Mock(spec=BleakClient)
+
+
+@pytest.fixture
+def mock_bleak_scanner():
+    return Mock(spec=BleakScanner)
+
+
+@pytest.fixture
+def mock_bleak_device():
+    return Mock(spec=BLEDevice)
+
+
+@pytest.fixture
+def mock_connection_manager():
+    return Mock(spec=ConnectionManager)
+
+
+@pytest.mark.asyncio
+async def tes_connect(mock_connection_manager, mock_bleak_client, mock_bleak_scanner, mock_bleak_device):
+    sensor = NervousSensor("ECG73BA", "2023-03-01 12:00:00", 10, mock_connection_manager)
+    sensor._client = mock_bleak_client
+    sensor._scanner = mock_bleak_scanner
+
+    mock_bleak_scanner.find_device_by_name.return_value = mock_bleak_device
+    mock_bleak_client.connect.return_value = True
+
+    async def disconnection_event():
+        await asyncio.sleep(1)
+        await sensor.disconnect()
+
+    async with TaskGroup() as tg:
+        tg.create_task(sensor.connect())
+        tg.create_task(disconnection_event())
+
+    mock_connection_manager.on_sensor_connect.assert_called_once_with(sensor)
+    mock_connection_manager.on_sensor_fail_to_connect.assert_not_called()
+    mock_connection_manager.on_sensor_disconnect.assert_called_once_with(sensor)
+
+
+@pytest.mark.asyncio
+async def tes_connect_fail(mock_connection_manager, mock_bleak_client, mock_bleak_scanner):
+    sensor = NervousSensor("ECG73BA", "2023-03-01 12:00:00", 10, mock_connection_manager)
+    sensor._client = mock_bleak_client
+    sensor._scanner = mock_bleak_scanner
+
+    mock_bleak_scanner.find_device_by_name.return_value = None
+    mock_bleak_client.connect.return_value = False
+    await sensor.connect()
+
+    mock_connection_manager.on_sensor_connect.assert_not_called()
+    mock_connection_manager.on_sensor_fail_to_connect.assert_called_once_with(sensor)
+    mock_connection_manager.on_sensor_disconnect.assert_not_called()