diff --git a/.mega-linter.yml b/.mega-linter.yml index 2d9a363..b3428b6 100644 --- a/.mega-linter.yml +++ b/.mega-linter.yml @@ -11,9 +11,6 @@ PYTHON_PYLINT_PRE_COMMANDS: - command: pip install pylint-pydantic venv: pylint continue_if_failed: false -# If you use ENABLE variable, all other languages/formats/tooling-formats will -# be disabled by default -# ENABLE: PYTHON_PYLINT_ARGUMENTS: # Disable import checks because it needs all the Python dependencies installed in the linter @@ -23,17 +20,16 @@ PYTHON_PYLINT_ARGUMENTS: - "--load-plugins=pylint_pydantic" PYTHON_FLAKE8_ARGUMENTS: - "--max-line-length=100" -# If you use ENABLE_LINTERS variable, all other linters will be disabled by -# default -# ENABLE_LINTERS: DISABLE_LINTERS: - SPELL_CSPELL - PYTHON_PYRIGHT # TODO this should not be disabled!! + - PYTHON_MYPY # TODO this should not be disabled!! - REPOSITORY_CHECKOV # TODO this should not be disabled!! + - COPYPASTE_JSCPD # TODO this should not be disabled!! SHOW_ELAPSED_TIME: true FILEIO_REPORTER: false -# Uncomment if you want MegaLinter to detect errors but not block CI to pass -# DISABLE_ERRORS: true + +PYTHON_BANDIT_FILTER_REGEX_EXCLUDE: test_ diff --git a/Dockerfile b/Dockerfile index c686acf..f58bae5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ FROM python:3 -RUN apt-get update && apt-get --no-install-recommends install -y bluez bluetooth sudo +RUN apt-get update && apt-get --no-install-recommends install -y bluez=5.66 bluetooth=5.50-1.2 && rm -rf /var/lib/apt/lists/* RUN adduser --disabled-password --gecos '' myuser @@ -8,7 +8,7 @@ RUN adduser --disabled-password --gecos '' myuser WORKDIR /usr/src/app COPY setup.py ./ COPY plejd ./plejd -RUN pip3 install . +RUN pip3 install --no-cache-dir . # Docker entrypoint COPY docker-entrypoint.sh . @@ -17,4 +17,9 @@ RUN chmod +x ./docker-entrypoint.sh # Switch to the new user USER myuser +# Healthcheck +HEALTHCHECK --interval=5m --timeout=3s \ + CMD curl -f http://localhost/ || exit 1 + + ENTRYPOINT ["./docker-entrypoint.sh"] \ No newline at end of file diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh old mode 100644 new mode 100755 index 732ff5e..2cda6ac --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -9,7 +9,7 @@ sudo service bluetooth start # We must wait for bluetooth service to start msg="Waiting for services to start..." time=0 -echo -n $msg +echo -n "$msg" while [[ "$(pidof start-stop-daemon)" != "" ]]; do sleep 1 time=$((time + 1)) diff --git a/ha_plejd_mqtt/bt_client.py b/ha_plejd_mqtt/bt_client.py index 0b6aa47..c24a7ed 100644 --- a/ha_plejd_mqtt/bt_client.py +++ b/ha_plejd_mqtt/bt_client.py @@ -168,7 +168,6 @@ async def send_command( response_type: Type of response """ if not self.is_connected(): - print("Not connected") error_message = "Trying to send command when not connected to Plejd mesh" logging.error(error_message) raise PlejdNotConnectedError(error_message) @@ -263,8 +262,6 @@ async def ping(self) -> bool: except (PlejdBluetoothError, PlejdNotConnectedError, PlejdTimeoutError) as err: logging.info(f"Ping operation failed: {str(err)}") return False - print("pong_data",pong_data) - print("ping_data",ping_data) if pong_data == bytearray() or not ((ping_data[0] + 1) & 0xFF) == pong_data[0]: return False @@ -369,18 +366,15 @@ async def set_plejd_time(self, plejd_device: BTDeviceInfo, time: datetime) -> bo """ timestamp = struct.pack(" bool: ) except PlejdNotConnectedError as err: logging.warning( - f"Device {self._device_info.name} is not connected, cannot turn on. Error: {str(err)}" + f"Device {self._device_info.name} is not connected, cannot turn on." + f"Error: {str(err)}" ) return False except (PlejdBluetoothError, PlejdTimeoutError) as err: logging.warning( - f"Failed to turn on device {self._device_info.name}, due to bluetooth a error. Error: {str(err)}" + f"Failed to turn on device {self._device_info.name}, due to bluetooth a error." + f"Error: {str(err)}" ) return False @@ -111,12 +118,14 @@ async def off(self) -> bool: ) except PlejdNotConnectedError as err: logging.warning( - f"Device {self._device_info.name} is not connected, cannot turn on. Error: {str(err)}" + f"Device {self._device_info.name} is not connected, cannot turn on." + f"Error: {str(err)}" ) return False except (PlejdBluetoothError, PlejdTimeoutError) as err: logging.warning( - f"Failed to turn on device {self._device_info.name}, due to bluetooth a error. Error: {str(err)}" + f"Failed to turn on device {self._device_info.name}, due to bluetooth a error." + f"Error: {str(err)}" ) return False @@ -150,12 +159,14 @@ async def brightness(self, brightness: int) -> bool: ) except PlejdNotConnectedError as err: logging.warning( - f"Device {self._device_info.name} is not connected, cannot turn on. Error: {str(err)}" + f"Device {self._device_info.name} is not connected, cannot turn on." + f"Error: {str(err)}" ) return False except (PlejdBluetoothError, PlejdTimeoutError) as err: logging.warning( - f"Failed to turn on device {self._device_info.name}, due to bluetooth a error. Error: {str(err)}" + f"Failed to turn on device {self._device_info.name}, due to bluetooth a error." + f"Error: {str(err)}" ) return False diff --git a/ha_plejd_mqtt/mdl/combined_device.py b/ha_plejd_mqtt/mdl/combined_device.py index 73ec8c4..a943180 100644 --- a/ha_plejd_mqtt/mdl/combined_device.py +++ b/ha_plejd_mqtt/mdl/combined_device.py @@ -84,9 +84,7 @@ async def start(self) -> None: raise MQTTDeviceError(error_msg) from err try: - print("Creating BT device") self._bt_device = await self._create_bt_device() - print("Created BT device") except BTDeviceError as err: error_msg = f"Failed to create BT device for {self._device_info.name}" logging.error(error_msg) diff --git a/tests/test_bt_client.py b/tests/test_bt_client.py index 2c2d398..12c9392 100644 --- a/tests/test_bt_client.py +++ b/tests/test_bt_client.py @@ -1,11 +1,18 @@ import datetime import struct + import pytest from ha_plejd_mqtt import constants -from ha_plejd_mqtt.bt_client import BTClient, PlejdBluetoothError, PlejdNotConnectedError, UnsupportedCommandError +from ha_plejd_mqtt.bt_client import ( + BTClient, + PlejdBluetoothError, + PlejdNotConnectedError, + UnsupportedCommandError, +) from ha_plejd_mqtt.mdl.bt_device_info import BTDeviceInfo from ha_plejd_mqtt.mdl.settings import API, PlejdSettings + class TestBTClient: """Test BTClient class""" @@ -25,18 +32,22 @@ def setup(self): name="device", ble_address=1, plejd_id=1, - device_type="type") + device_type="type", + ) @pytest.mark.asyncio - @pytest.mark.parametrize("is_connected,_connect,expected", [ - (True, True, True), # Already connected - (False, True, True), # Connect success - (False, False, False) # Connect failed - ]) + @pytest.mark.parametrize( + "is_connected,_connect,expected", + [ + (True, True, True), # Already connected + (False, True, True), # Connect success + (False, False, False), # Connect failed + ], + ) async def test_connect(self, mocker, is_connected, _connect, expected): """Test connect method of BTClient""" self.bt_client.is_connected = mocker.MagicMock(return_value=is_connected) - self.bt_client._connect = mocker.AsyncMock(return_value=_connect) + self.bt_client._connect = mocker.AsyncMock(return_value=_connect) self.bt_client._stay_connected_loop = mocker.MagicMock() result = await self.bt_client.connect(stay_connected=False) @@ -44,12 +55,15 @@ async def test_connect(self, mocker, is_connected, _connect, expected): assert result == expected @pytest.mark.asyncio - @pytest.mark.parametrize("is_connected,_disconnect,expected", [ - (True, True, True), # Already connected, disconnect should be successful - (False, True, True), # Not connected, disconnect should still return True - (True, False, False), # Connected, disconnect should return False - (False, False, True) # Not connected, no disconnect should be attempted - ]) + @pytest.mark.parametrize( + "is_connected,_disconnect,expected", + [ + (True, True, True), # Already connected, disconnect should be successful + (False, True, True), # Not connected, disconnect should still return True + (True, False, False), # Connected, disconnect should return False + (False, False, True), # Not connected, no disconnect should be attempted + ], + ) async def test_disconnect(self, mocker, is_connected, _disconnect, expected): """Test disconnect method of BTClient""" self.bt_client.is_connected = mocker.MagicMock(return_value=is_connected) @@ -61,28 +75,43 @@ async def test_disconnect(self, mocker, is_connected, _disconnect, expected): assert result == expected @pytest.mark.asyncio - @pytest.mark.parametrize("is_connected,command,write_request_success,expected_exception", [ - (False, 0, True, PlejdNotConnectedError), # Not connected - (True, -1, True, UnsupportedCommandError), # Invalid command - (True, constants.PlejdCommand.BLE_CMD_STATE_CHANGE, False, PlejdBluetoothError), # Write request failed - (True, constants.PlejdCommand.BLE_CMD_STATE_CHANGE, True, None), # Successful case - ]) + @pytest.mark.parametrize( + "is_connected,command,write_request_success,expected_exception", + [ + (False, 0, True, PlejdNotConnectedError), # Not connected + (True, -1, True, UnsupportedCommandError), # Invalid command + ( + True, + constants.PlejdCommand.BLE_CMD_STATE_CHANGE, + False, + PlejdBluetoothError, + ), # Write request failed + ( + True, + constants.PlejdCommand.BLE_CMD_STATE_CHANGE, + True, + None, + ), # Successful case + ], + ) async def test_send_command( - self, - mocker, - is_connected, - command, - write_request_success, - expected_exception): + self, mocker, is_connected, command, write_request_success, expected_exception + ): """Test send_command method of BTClient""" self.bt_client.is_connected = mocker.MagicMock(return_value=is_connected) self.bt_client._get_cmd_payload = mocker.MagicMock(return_value="payload") self.bt_client._client = mocker.MagicMock(address="address") - self.bt_client._encode_address = mocker.MagicMock(return_value="encoded_address") - self.bt_client._encrypt_decrypt_data = mocker.MagicMock(return_value="encrypted_data") + self.bt_client._encode_address = mocker.MagicMock( + return_value="encoded_address" + ) + self.bt_client._encrypt_decrypt_data = mocker.MagicMock( + return_value="encrypted_data" + ) self.bt_client._write_request = mocker.AsyncMock( - side_effect= - PlejdBluetoothError('Write request failed') if not write_request_success else None) + side_effect=PlejdBluetoothError("Write request failed") + if not write_request_success + else None + ) if expected_exception: with pytest.raises(expected_exception): @@ -91,40 +120,70 @@ async def test_send_command( await self.bt_client.send_command(1, command, "data", 1) @pytest.mark.asyncio - @pytest.mark.parametrize("write_request_success, read_request_success, pong_data, expected", [ - (False, True, b'\x01', False), # Write request failed - (True, False, b'\x01', False), # Read request failed - (True, True, b'', False), # Pong data is empty - (True, True, b'\x02', False), # Pong data is not the expected value - (True, True, b'\x01', True), # Successful case - ]) - async def test_ping(self, mocker, write_request_success, read_request_success, pong_data, expected): + @pytest.mark.parametrize( + "write_request_success, read_request_success, pong_data, expected", + [ + (False, True, b"\x01", False), # Write request failed + (True, False, b"\x01", False), # Read request failed + (True, True, b"", False), # Pong data is empty + (True, True, b"\x02", False), # Pong data is not the expected value + (True, True, b"\x01", True), # Successful case + ], + ) + async def test_ping( + self, mocker, write_request_success, read_request_success, pong_data, expected + ): """Test ping method of BTClient""" - ping_data = b'\x00' - mocker.patch('ha_plejd_mqtt.bt_client.randbytes', return_value=ping_data) + ping_data = b"\x00" + mocker.patch("ha_plejd_mqtt.bt_client.randbytes", return_value=ping_data) self.bt_client._write_request = mocker.AsyncMock( - side_effect=PlejdBluetoothError('Write request failed') if not write_request_success else None) + side_effect=PlejdBluetoothError("Write request failed") + if not write_request_success + else None + ) self.bt_client._read_request = mocker.AsyncMock( - side_effect=PlejdBluetoothError('Read request failed') if not read_request_success else None, return_value=pong_data) + side_effect=PlejdBluetoothError("Read request failed") + if not read_request_success + else None, + return_value=pong_data, + ) result = await self.bt_client.ping() assert result == expected @pytest.mark.asyncio - @pytest.mark.parametrize("is_connected, read_request_success, encrypted_data, expected_exception", [ - (False, True, b'\x01', PlejdNotConnectedError), # Not connected - (True, False, b'\x01', PlejdBluetoothError), # _read_request failed - (True, True, b'\x01', None), # Successful case - ]) - async def test_get_last_data(self, mocker, is_connected, read_request_success, encrypted_data, expected_exception): + @pytest.mark.parametrize( + "is_connected, read_request_success, encrypted_data, expected_exception", + [ + (False, True, b"\x01", PlejdNotConnectedError), # Not connected + (True, False, b"\x01", PlejdBluetoothError), # _read_request failed + (True, True, b"\x01", None), # Successful case + ], + ) + async def test_get_last_data( + self, + mocker, + is_connected, + read_request_success, + encrypted_data, + expected_exception, + ): """Test get_last_data method of BTClient""" self.bt_client.is_connected = mocker.MagicMock(return_value=is_connected) self.bt_client._client = mocker.MagicMock(address="address") - self.bt_client._encode_address = mocker.MagicMock(return_value="encoded_address") - self.bt_client._encrypt_decrypt_data = mocker.MagicMock(return_value="decrypted_data") + self.bt_client._encode_address = mocker.MagicMock( + return_value="encoded_address" + ) + self.bt_client._encrypt_decrypt_data = mocker.MagicMock( + return_value="decrypted_data" + ) self.bt_client._read_request = mocker.AsyncMock( - side_effect=PlejdBluetoothError('Read request failed') if not read_request_success else None, return_value=encrypted_data) + side_effect=PlejdBluetoothError("Read request failed") + if not read_request_success + else None, + return_value=encrypted_data, + ) if expected_exception: with pytest.raises(expected_exception): @@ -134,21 +193,80 @@ async def test_get_last_data(self, mocker, is_connected, read_request_success, e assert result == "decrypted_data" @pytest.mark.asyncio - @pytest.mark.parametrize("is_connected, send_command_success, get_last_data_success, last_data, expected_exception, expected_result", [ - (False, True, True, b'\x01\x02\x03\x04\x05\x06\x07\x08', PlejdNotConnectedError, None), # Not connected - (True, False, True, b'\x01\x02\x03\x04\x05\x06\x07\x08', PlejdBluetoothError, None), # send_command failed - (True, True, False, b'\x01\x02\x03\x04\x05\x06\x07\x08', PlejdBluetoothError, None), # get_last_data failed - (True, True, True, b'\x01\x02\x03\x04\x05\x06\x07\x08', UnsupportedCommandError, None), # Invalid last_data - (True, True, True, b'\x01\x02\x00\x00\x1b\x00\x00\x00\x00\x00\x00\x00', None, datetime.datetime.fromtimestamp(0)), # Successful case - ]) - async def test_get_plejd_time(self, mocker, is_connected, send_command_success, get_last_data_success, last_data, expected_exception, expected_result): + @pytest.mark.parametrize( + "is_connected," + "send_command_success," + "get_last_data_success," + "last_data," + "expected_exception," + "expected_result", + [ + ( + False, + True, + True, + b"\x01\x02\x03\x04\x05\x06\x07\x08", + PlejdNotConnectedError, + None, + ), # Not connected + ( + True, + False, + True, + b"\x01\x02\x03\x04\x05\x06\x07\x08", + PlejdBluetoothError, + None, + ), # send_command failed + ( + True, + True, + False, + b"\x01\x02\x03\x04\x05\x06\x07\x08", + PlejdBluetoothError, + None, + ), # get_last_data failed + ( + True, + True, + True, + b"\x01\x02\x03\x04\x05\x06\x07\x08", + UnsupportedCommandError, + None, + ), # Invalid last_data + ( + True, + True, + True, + b"\x01\x02\x00\x00\x1b\x00\x00\x00\x00\x00\x00\x00", + None, + datetime.datetime.fromtimestamp(0), + ), # Successful case + ], + ) + async def test_get_plejd_time( + self, + mocker, + is_connected, + send_command_success, + get_last_data_success, + last_data, + expected_exception, + expected_result, + ): """Test get_plejd_time method of BTClient""" self.bt_client.is_connected = mocker.MagicMock(return_value=is_connected) self.bt_client.send_command = mocker.AsyncMock( - side_effect=PlejdBluetoothError('send_command failed') if not send_command_success else None) + side_effect=PlejdBluetoothError("send_command failed") + if not send_command_success + else None + ) self.bt_client.get_last_data = mocker.AsyncMock( - side_effect=PlejdBluetoothError('get_last_data failed') if not get_last_data_success else None, return_value=last_data) + side_effect=PlejdBluetoothError("get_last_data failed") + if not get_last_data_success + else None, + return_value=last_data, + ) if expected_exception: with pytest.raises(expected_exception): @@ -158,18 +276,26 @@ async def test_get_plejd_time(self, mocker, is_connected, send_command_success, assert result == expected_result @pytest.mark.asyncio - @pytest.mark.parametrize("send_command_success, expected_exception, expected_result", [ - (False, PlejdNotConnectedError, False), # Not connected - (False, PlejdBluetoothError, False), # send_command failed - (True, None, True), # Successful case - ]) - async def test_set_plejd_time(self, mocker, send_command_success, expected_exception, expected_result): + @pytest.mark.parametrize( + "send_command_success, expected_exception, expected_result", + [ + (False, PlejdNotConnectedError, False), # Not connected + (False, PlejdBluetoothError, False), # send_command failed + (True, None, True), # Successful case + ], + ) + async def test_set_plejd_time( + self, mocker, send_command_success, expected_exception, expected_result + ): """Test set_plejd_time method of BTClient""" time = datetime.datetime.now() timestamp = struct.pack("= 400: mock_response.raise_for_status.side_effect = requests.HTTPError() - mocker.patch('requests.post', return_value=mock_response) + mocker.patch("requests.post", return_value=mock_response) api = PlejdAPI(self.settings) if expected == IncorrectCredentialsError: @@ -50,12 +61,15 @@ def test_get_site_no_session_token(self): with pytest.raises(PlejdNotLoggedInError): api.get_site() - @pytest.mark.parametrize("exception", [requests.RequestException, ValueError, UnknownResponseError("test")]) + @pytest.mark.parametrize( + "exception", + [requests.RequestException, ValueError, UnknownResponseError("test")], + ) def test_get_site_exceptions(self, mocker, exception): """Test get_site method of PlejdAPI when an exception is raised""" api = PlejdAPI(self.settings) api._session_token = "test_token" - mocker.patch('requests.post', side_effect=exception) + mocker.patch("requests.post", side_effect=exception) with pytest.raises(PlejdAPIError): - api.get_site() \ No newline at end of file + api.get_site()