diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index 8bb9e1b..f3733f0 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -11,5 +11,4 @@ Put an 'x' in the boxes that apply. --> -- [ ] Version number in \_\_init\_\_.py was properly set. - [ ] Changelog is updated. diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml index 6bbf559..d64e73d 100644 --- a/.github/workflows/pythonpackage.yml +++ b/.github/workflows/pythonpackage.yml @@ -30,18 +30,18 @@ jobs: - name: Lint with flake8 run: | python -m pip install --upgrade pip - pip install flake8 + python -m pip install flake8 flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics flake8 . --count --exit-zero --statistics - name: Lint with pylint run: | - pip install pylint - pip install -e . + python -m pip install pylint + python -m pip install -e . pylint --errors-only --score=n devolo_plc_api pylint --exit-zero --score=n --disable=C,E,R --enable=useless-suppression --ignore-patterns=.*_pb2.py devolo_plc_api devolo_plc_api - name: Lint with mypy run: | - pip install mypy + python -m pip install mypy types-protobuf mypy devolo_plc_api test: @@ -60,7 +60,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -e .[test] + python -m pip install -e .[test] - name: Test with pytest run: | pytest --cov=devolo_plc_api @@ -88,7 +88,7 @@ jobs: - name: Coveralls run: | python -m pip install --upgrade pip - pip install wheel coveralls + python -m pip install wheel coveralls export COVERALLS_REPO_TOKEN=${{ secrets.COVERALLS_TOKEN }} coveralls - name: Clean up coverage diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 220d8ac..59d2bb0 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,14 +1,14 @@ repos: - repo: https://github.com/pre-commit/mirrors-yapf - rev: '' + rev: 'v0.31.0' hooks: - id: yapf - repo: https://github.com/pycqa/isort - rev: '' + rev: '5.9.3' hooks: - id: isort - repo: https://github.com/pre-commit/pre-commit-hooks - rev: '' + rev: 'v4.0.1' hooks: - id: end-of-file-fixer - id: trailing-whitespace diff --git a/README.md b/README.md index bcb238c..9932355 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ Other versions and even other operating systems might work. Feel free to tell us * pytest 5.4.3 * pytest-asyncio 0.14.0 * pytest-mock 3.2.0 +* pytest-httpx 0.12 * asynctest 0.13.0 ## Versioning diff --git a/devolo_plc_api/__init__.py b/devolo_plc_api/__init__.py index dd9b22c..eddea92 100644 --- a/devolo_plc_api/__init__.py +++ b/devolo_plc_api/__init__.py @@ -1 +1,10 @@ -__version__ = "0.5.1" +try: + from importlib.metadata import PackageNotFoundError, version +except ImportError: + from importlib_metadata import PackageNotFoundError, version # type: ignore[no-redef] + +try: + __version__ = version("devolo_plc_api") +except PackageNotFoundError: + # package is not installed - e.g. pulled and run locally + __version__ = "0.0.0" diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0e182db..d6a4865 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,11 +4,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [v0.5.1] - 2021/12/19 +## [v0.5.2] - 2021/09/01 + +### Changed + +- Use newer dependency versions + +## [v0.5.1] - 2021/01/19 ### Fixed - React correctly on different connection errors + ## [v0.5.0] - 2020/12/21 ### Changed diff --git a/setup.py b/setup.py index 2ae084a..11c76d3 100644 --- a/setup.py +++ b/setup.py @@ -4,8 +4,6 @@ from setuptools import find_packages, setup from setuptools.command.develop import develop -from devolo_plc_api import __version__ - with open("README.md", "r") as fh: long_description = fh.read() @@ -24,7 +22,7 @@ def run(self): setup( name="devolo_plc_api", - version=__version__, + use_scm_version=True, author="Markus Bong, Guido Schmitz", author_email="m.bong@famabo.de, guido.schmitz@fedaix.de", description="devolo PLC devices in Python", @@ -39,7 +37,8 @@ def run(self): "Operating System :: OS Independent", ], install_requires=[ - "httpx>=0.14,<0.17", + "httpx>=0.14,<0.20", + "importlib-metadata;python_version<'3.8'", "protobuf", "zeroconf>=0.27.0", ], @@ -52,8 +51,10 @@ def run(self): "pytest", "pytest-asyncio", "pytest-cov", + "pytest-httpx>=0.12", "pytest-mock", ], }, + setup_requires=["setuptools_scm"], python_requires='>=3.7', ) diff --git a/tests/fixtures/device.py b/tests/fixtures/device.py index 954474e..af8a848 100644 --- a/tests/fixtures/device.py +++ b/tests/fixtures/device.py @@ -15,7 +15,7 @@ @pytest.fixture() -def mock_device(mocker, request): +def mock_device(request): device = Device(ip=request.cls.ip) device._info = deepcopy(request.cls.device_info) device._loop = Mock() @@ -24,13 +24,12 @@ def mock_device(mocker, request): device._zeroconf = Mock() device._zeroconf.close = lambda: None yield device - device.disconnect() @pytest.fixture() def mock_service_browser(mocker): mocker.patch("zeroconf.ServiceBrowser.__init__", MockServiceBrowser.__init__) - mocker.patch("zeroconf.ServiceBrowser.cancel", return_value=None) + mocker.patch("zeroconf.ServiceBrowser.cancel") @pytest.fixture() diff --git a/tests/fixtures/device_api.py b/tests/fixtures/device_api.py index 69cb0f0..051a83f 100644 --- a/tests/fixtures/device_api.py +++ b/tests/fixtures/device_api.py @@ -2,26 +2,20 @@ from unittest.mock import patch import pytest -from httpx import AsyncClient, Response +from httpx import AsyncClient from devolo_plc_api.device_api.deviceapi import DeviceApi -try: - from unittest.mock import AsyncMock -except ImportError: - from asynctest import CoroutineMock as AsyncMock - @pytest.fixture() -def device_api(request, feature): - with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=AsyncMock(return_value=Response)), \ - patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=AsyncMock(return_value=Response)), \ - patch("asyncio.get_running_loop", asyncio.new_event_loop): - asyncio.new_event_loop() +async def device_api(request, feature): + with patch("asyncio.get_running_loop", asyncio.new_event_loop): request.cls.device_info["_dvl-deviceapi._tcp.local."]["properties"]["Features"] = feature - yield DeviceApi(request.cls.ip, AsyncClient(), request.cls.device_info["_dvl-deviceapi._tcp.local."]) + async with AsyncClient() as client: + device_api = DeviceApi(request.cls.ip, client, request.cls.device_info["_dvl-deviceapi._tcp.local."]) + yield device_api @pytest.fixture() def mock_device_api(mocker): - mocker.patch("devolo_plc_api.device_api.deviceapi.DeviceApi.__init__", return_value=None) + mocker.patch("devolo_plc_api.device_api.deviceapi.DeviceApi") diff --git a/tests/fixtures/plcnet_api.py b/tests/fixtures/plcnet_api.py index 030a739..d5a4641 100644 --- a/tests/fixtures/plcnet_api.py +++ b/tests/fixtures/plcnet_api.py @@ -2,25 +2,20 @@ from unittest.mock import patch import pytest -from httpx import AsyncClient, Response +from httpx import AsyncClient from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi -try: - from unittest.mock import AsyncMock -except ImportError: - from asynctest import CoroutineMock as AsyncMock - @pytest.fixture() -def plcnet_api(request): - with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=AsyncMock(return_value=Response)), \ - patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=AsyncMock(return_value=Response)), \ - patch("asyncio.get_running_loop", asyncio.new_event_loop): +async def plcnet_api(request): + with patch("asyncio.get_running_loop", asyncio.new_event_loop): asyncio.new_event_loop() - yield PlcNetApi(request.cls.ip, AsyncClient(), request.cls.device_info["_dvl-plcnetapi._tcp.local."]) + async with AsyncClient() as client: + plcnet_api = PlcNetApi(request.cls.ip, client, request.cls.device_info["_dvl-plcnetapi._tcp.local."]) + yield plcnet_api @pytest.fixture() def mock_plcnet_api(mocker): - mocker.patch("devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.__init__", return_value=None) + mocker.patch("devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi") diff --git a/tests/fixtures/protobuf.py b/tests/fixtures/protobuf.py index 7690eaf..3a2462e 100644 --- a/tests/fixtures/protobuf.py +++ b/tests/fixtures/protobuf.py @@ -1,36 +1,29 @@ import pytest from httpx import ConnectTimeout -try: - from unittest.mock import AsyncMock -except ImportError: - from asynctest import CoroutineMock as AsyncMock - from ..stubs.protobuf import StubProtobuf @pytest.fixture() -def mock_protobuf(): - return StubProtobuf() +async def mock_protobuf(): + protobuf = StubProtobuf() + yield protobuf + await protobuf._session.aclose() @pytest.fixture() -def mock_get(mocker): - mocker.patch("httpx.AsyncClient.get", new=AsyncMock()) +def mock_device_unavailable(httpx_mock): + def raise_type_error(request, extensions): + raise ConnectTimeout(request=request, message=extensions) -@pytest.fixture() -def mock_post(mocker): - mocker.patch("httpx.AsyncClient.post", new=AsyncMock()) + httpx_mock.add_callback(raise_type_error) @pytest.fixture() -def mock_device_unavailable(mocker): - mocker.patch("httpx.AsyncClient.get", side_effect=ConnectTimeout(message="", request="")) - mocker.patch("httpx.AsyncClient.post", side_effect=ConnectTimeout(message="", request="")) +def mock_wrong_password(httpx_mock): + def raise_type_error(request, extensions): + raise TypeError -@pytest.fixture() -def mock_wrong_password(mocker): - mocker.patch("httpx.AsyncClient.get", side_effect=TypeError()) - mocker.patch("httpx.AsyncClient.post", side_effect=TypeError()) + httpx_mock.add_callback(raise_type_error) diff --git a/tests/stubs/protobuf.py b/tests/stubs/protobuf.py index 8e203ba..98ac9a0 100644 --- a/tests/stubs/protobuf.py +++ b/tests/stubs/protobuf.py @@ -3,6 +3,8 @@ import logging import pathlib +from httpx import AsyncClient + from devolo_plc_api.clients.protobuf import Protobuf file = pathlib.Path(__file__).parent / "../test_data.json" @@ -17,6 +19,7 @@ def __init__(self): self._loop = asyncio.new_event_loop() self._ip = test_data["ip"] self._port = 14791 + self._session = AsyncClient() self._path = test_data["device_info"]["_dvl-plcnetapi._tcp.local."]["properties"]["Path"] self._version = test_data["device_info"]["_dvl-plcnetapi._tcp.local."]["properties"]["Version"] self._user = "user" diff --git a/tests/test_device.py b/tests/test_device.py index 21694f9..1afa1f9 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -84,7 +84,7 @@ async def test__get_device_info(self, mock_device): @pytest.mark.asyncio @pytest.mark.usefixtures("mock_device_api") async def test__get_device_info_timeout(self, mock_device): - with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=AsyncMock()), \ + with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=Mock()), \ patch("asyncio.wait_for", new=AsyncMock(side_effect=asyncio.TimeoutError())): await mock_device._get_device_info() assert mock_device.device is None @@ -102,7 +102,7 @@ async def test__get_plcnet_info(self, mock_device): @pytest.mark.asyncio @pytest.mark.usefixtures("mock_plcnet_api") async def test__get_plcnet_info_timeout(self, mock_device): - with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=AsyncMock()), \ + with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=Mock()), \ patch("asyncio.wait_for", new=AsyncMock(side_effect=asyncio.TimeoutError())): await mock_device._get_plcnet_info() assert mock_device.plcnet is None diff --git a/tests/test_deviceapi.py b/tests/test_deviceapi.py index 20f2bb3..7a7bf9d 100644 --- a/tests/test_deviceapi.py +++ b/tests/test_deviceapi.py @@ -1,13 +1,6 @@ -from unittest.mock import patch - import pytest from google.protobuf.json_format import MessageToDict -try: - from unittest.mock import AsyncMock -except ImportError: - from asynctest import CoroutineMock as AsyncMock - from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_ledsettings_pb2 import LedSettingsGet, LedSettingsSetResponse from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_updatefirmware_pb2 import UpdateFirmwareCheck, UpdateFirmwareStart from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_wifinetwork_pb2 import (WifiConnectedStationsGet, @@ -32,166 +25,166 @@ def test_feature(self, device_api): @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["led"]) - async def test_async_get_led_setting(self, device_api): + async def test_async_get_led_setting(self, device_api, httpx_mock): led_setting_get = LedSettingsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=led_setting_get.SerializeToString())): - led_setting = await device_api.async_get_led_setting() - assert led_setting == MessageToDict(led_setting_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=led_setting_get.SerializeToString()) + led_setting = await device_api.async_get_led_setting() + assert led_setting == MessageToDict(led_setting_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.parametrize("feature", ["led"]) - def test_get_led_setting(self, device_api): + def test_get_led_setting(self, device_api, httpx_mock): led_setting_get = LedSettingsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=led_setting_get.SerializeToString())): - led_setting = device_api.get_led_setting() - assert led_setting == MessageToDict(led_setting_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=led_setting_get.SerializeToString()) + led_setting = device_api.get_led_setting() + assert led_setting == MessageToDict(led_setting_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["led"]) - async def test_async_set_led_setting(self, device_api): + async def test_async_set_led_setting(self, device_api, httpx_mock): led_setting_set = LedSettingsSetResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=led_setting_set.SerializeToString())): - assert await device_api.async_set_led_setting(True) + httpx_mock.add_response(data=led_setting_set.SerializeToString()) + assert await device_api.async_set_led_setting(True) @pytest.mark.parametrize("feature", ["led"]) - def test_set_led_setting(self, device_api): + def test_set_led_setting(self, device_api, httpx_mock): led_setting_set = LedSettingsSetResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=led_setting_set.SerializeToString())): - assert device_api.set_led_setting(True) + httpx_mock.add_response(data=led_setting_set.SerializeToString()) + assert device_api.set_led_setting(True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["update"]) - async def test_async_check_firmware_available(self, device_api): + async def test_async_check_firmware_available(self, device_api, httpx_mock): firmware_available = UpdateFirmwareCheck() - with patch("httpx.Response.aread", new=AsyncMock(return_value=firmware_available.SerializeToString())): - firmware = await device_api.async_check_firmware_available() - assert firmware == MessageToDict(firmware_available, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=firmware_available.SerializeToString()) + firmware = await device_api.async_check_firmware_available() + assert firmware == MessageToDict(firmware_available, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.parametrize("feature", ["update"]) - def test_check_firmware_available(self, device_api): + def test_check_firmware_available(self, device_api, httpx_mock): firmware_available = UpdateFirmwareCheck() - with patch("httpx.Response.aread", new=AsyncMock(return_value=firmware_available.SerializeToString())): - firmware = device_api.check_firmware_available() - assert firmware == MessageToDict(firmware_available, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=firmware_available.SerializeToString()) + firmware = device_api.check_firmware_available() + assert firmware == MessageToDict(firmware_available, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["update"]) - async def test_async_start_firmware_update(self, device_api): + async def test_async_start_firmware_update(self, device_api, httpx_mock): firmware_update = UpdateFirmwareStart() - with patch("httpx.Response.aread", new=AsyncMock(return_value=firmware_update.SerializeToString())): - assert await device_api.async_start_firmware_update() + httpx_mock.add_response(data=firmware_update.SerializeToString()) + assert await device_api.async_start_firmware_update() @pytest.mark.parametrize("feature", ["update"]) - def test_start_firmware_update(self, device_api): + def test_start_firmware_update(self, device_api, httpx_mock): firmware_update = UpdateFirmwareStart() - with patch("httpx.Response.aread", new=AsyncMock(return_value=firmware_update.SerializeToString())): - assert device_api.start_firmware_update() + httpx_mock.add_response(data=firmware_update.SerializeToString()) + assert device_api.start_firmware_update() @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["wifi1"]) - async def test_async_get_wifi_connected_station(self, device_api): + async def test_async_get_wifi_connected_station(self, device_api, httpx_mock): wifi_connected_stations_get = WifiConnectedStationsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_connected_stations_get.SerializeToString())): - connected_stations = await device_api.async_get_wifi_connected_station() - assert connected_stations == MessageToDict(wifi_connected_stations_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_connected_stations_get.SerializeToString()) + connected_stations = await device_api.async_get_wifi_connected_station() + assert connected_stations == MessageToDict(wifi_connected_stations_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.parametrize("feature", ["wifi1"]) - def test_get_wifi_connected_station(self, device_api): + def test_get_wifi_connected_station(self, device_api, httpx_mock): wifi_connected_stations_get = WifiConnectedStationsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_connected_stations_get.SerializeToString())): - connected_stations = device_api.get_wifi_connected_station() - assert connected_stations == MessageToDict(wifi_connected_stations_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_connected_stations_get.SerializeToString()) + connected_stations = device_api.get_wifi_connected_station() + assert connected_stations == MessageToDict(wifi_connected_stations_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["wifi1"]) - async def test_async_get_wifi_guest_access(self, device_api): + async def test_async_get_wifi_guest_access(self, device_api, httpx_mock): wifi_guest_access_get = WifiGuestAccessGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_guest_access_get.SerializeToString())): - wifi_guest_access = await device_api.async_get_wifi_guest_access() - assert wifi_guest_access == MessageToDict(wifi_guest_access_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_guest_access_get.SerializeToString()) + wifi_guest_access = await device_api.async_get_wifi_guest_access() + assert wifi_guest_access == MessageToDict(wifi_guest_access_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.parametrize("feature", ["wifi1"]) - def test_get_wifi_guest_access(self, device_api): + def test_get_wifi_guest_access(self, device_api, httpx_mock): wifi_guest_access_get = WifiGuestAccessGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_guest_access_get.SerializeToString())): - wifi_guest_access = device_api.get_wifi_guest_access() - assert wifi_guest_access == MessageToDict(wifi_guest_access_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_guest_access_get.SerializeToString()) + wifi_guest_access = device_api.get_wifi_guest_access() + assert wifi_guest_access == MessageToDict(wifi_guest_access_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["wifi1"]) - async def test_async_set_wifi_guest_access(self, device_api): + async def test_async_set_wifi_guest_access(self, device_api, httpx_mock): wifi_guest_access_set = WifiGuestAccessSetResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_guest_access_set.SerializeToString())): - assert await device_api.async_set_wifi_guest_access(True) + httpx_mock.add_response(data=wifi_guest_access_set.SerializeToString()) + assert await device_api.async_set_wifi_guest_access(True) @pytest.mark.parametrize("feature", ["wifi1"]) - def test_set_wifi_guest_access(self, device_api): + def test_set_wifi_guest_access(self, device_api, httpx_mock): wifi_guest_access_set = WifiGuestAccessSetResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_guest_access_set.SerializeToString())): - assert device_api.set_wifi_guest_access(True) + httpx_mock.add_response(data=wifi_guest_access_set.SerializeToString()) + assert device_api.set_wifi_guest_access(True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["wifi1"]) - async def test_async_get_wifi_neighbor_access_points(self, device_api): + async def test_async_get_wifi_neighbor_access_points(self, device_api, httpx_mock): wifi_neighbor_accesspoints_get = WifiNeighborAPsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_neighbor_accesspoints_get.SerializeToString())): - wifi_neighbor_access_points = await device_api.async_get_wifi_neighbor_access_points() - assert wifi_neighbor_access_points == MessageToDict(wifi_neighbor_accesspoints_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_neighbor_accesspoints_get.SerializeToString()) + wifi_neighbor_access_points = await device_api.async_get_wifi_neighbor_access_points() + assert wifi_neighbor_access_points == MessageToDict(wifi_neighbor_accesspoints_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.parametrize("feature", ["wifi1"]) - def test_get_wifi_neighbor_access_points(self, device_api): + def test_get_wifi_neighbor_access_points(self, device_api, httpx_mock): wifi_neighbor_accesspoints_get = WifiNeighborAPsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_neighbor_accesspoints_get.SerializeToString())): - wifi_neighbor_access_points = device_api.get_wifi_neighbor_access_points() - assert wifi_neighbor_access_points == MessageToDict(wifi_neighbor_accesspoints_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_neighbor_accesspoints_get.SerializeToString()) + wifi_neighbor_access_points = device_api.get_wifi_neighbor_access_points() + assert wifi_neighbor_access_points == MessageToDict(wifi_neighbor_accesspoints_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["wifi1"]) - async def test_async_get_wifi_repeated_access_points(self, device_api): + async def test_async_get_wifi_repeated_access_points(self, device_api, httpx_mock): wifi_repeated_accesspoints_get = WifiRepeatedAPsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_repeated_accesspoints_get.SerializeToString())): - wifi_repeated_access_points = await device_api.async_get_wifi_repeated_access_points() - assert wifi_repeated_access_points == MessageToDict(wifi_repeated_accesspoints_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_repeated_accesspoints_get.SerializeToString()) + wifi_repeated_access_points = await device_api.async_get_wifi_repeated_access_points() + assert wifi_repeated_access_points == MessageToDict(wifi_repeated_accesspoints_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.parametrize("feature", ["wifi1"]) - def test_get_wifi_repeated_access_points(self, device_api): + def test_get_wifi_repeated_access_points(self, device_api, httpx_mock): wifi_repeated_accesspoints_get = WifiRepeatedAPsGet() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wifi_repeated_accesspoints_get.SerializeToString())): - wifi_repeated_access_points = device_api.get_wifi_repeated_access_points() - assert wifi_repeated_access_points == MessageToDict(wifi_repeated_accesspoints_get, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=wifi_repeated_accesspoints_get.SerializeToString()) + wifi_repeated_access_points = device_api.get_wifi_repeated_access_points() + assert wifi_repeated_access_points == MessageToDict(wifi_repeated_accesspoints_get, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio @pytest.mark.parametrize("feature", ["wifi1"]) - async def test_async_start_wps(self, device_api): + async def test_async_start_wps(self, device_api, httpx_mock): wps = WifiWpsPbcStart() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wps.SerializeToString())): - assert await device_api.async_start_wps() + httpx_mock.add_response(data=wps.SerializeToString()) + assert await device_api.async_start_wps() @pytest.mark.parametrize("feature", ["wifi1"]) - def test_start_wps(self, device_api): + def test_start_wps(self, device_api, httpx_mock): wps = WifiWpsPbcStart() - with patch("httpx.Response.aread", new=AsyncMock(return_value=wps.SerializeToString())): - assert device_api.start_wps() + httpx_mock.add_response(data=wps.SerializeToString()) + assert device_api.start_wps() diff --git a/tests/test_network.py b/tests/test_network.py index 9c6fdbd..c7fafa9 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -1,4 +1,4 @@ -from unittest.mock import Mock, patch +from unittest.mock import patch import pytest from zeroconf import ServiceStateChange, Zeroconf @@ -19,9 +19,10 @@ async def test_async_discover_network(self, mocker): device = { "1234567890123456": Device(ip="123.123.123.123") } - with patch("zeroconf.Zeroconf", new=Mock()), \ + with patch("zeroconf.Zeroconf"), \ patch("asyncio.sleep", new=AsyncMock()), \ - patch("devolo_plc_api.device.Device.async_connect", new=AsyncMock()): + patch("devolo_plc_api.device.Device.async_connect"), \ + patch("zeroconf.ServiceBrowser.cancel"): network._devices = device discovered = await network.async_discover_network() assert discovered == device @@ -30,9 +31,10 @@ def test_discover_network(self, mocker): device = { "1234567890123456": Device(ip="123.123.123.123") } - with patch("zeroconf.ServiceBrowser", new=Mock()), \ - patch("time.sleep", new=Mock()), \ - patch("devolo_plc_api.device.Device.connect", new=Mock()): + with patch("zeroconf.Zeroconf"), \ + patch("time.sleep"), \ + patch("devolo_plc_api.device.Device.connect"), \ + patch("zeroconf.ServiceBrowser.cancel"): network._devices = device discovered = network.discover_network() assert discovered == device diff --git a/tests/test_plcnetapi.py b/tests/test_plcnetapi.py index f1b759b..f9943fb 100644 --- a/tests/test_plcnetapi.py +++ b/tests/test_plcnetapi.py @@ -1,13 +1,6 @@ -from unittest.mock import patch - import pytest from google.protobuf.json_format import MessageToDict -try: - from unittest.mock import AsyncMock -except ImportError: - from asynctest import CoroutineMock as AsyncMock - from devolo_plc_api.plcnet_api.devolo_idl_proto_plcnetapi_getnetworkoverview_pb2 import GetNetworkOverview from devolo_plc_api.plcnet_api.devolo_idl_proto_plcnetapi_identifydevice_pb2 import IdentifyDeviceResponse from devolo_plc_api.plcnet_api.devolo_idl_proto_plcnetapi_setuserdevicename_pb2 import SetUserDeviceNameResponse @@ -16,52 +9,51 @@ class TestDeviceApi: @pytest.mark.asyncio - async def test_async_get_network_overview(self, plcnet_api): + async def test_async_get_network_overview(self, plcnet_api, httpx_mock): network_overview = GetNetworkOverview() - with patch("httpx.Response.aread", new=AsyncMock(return_value=network_overview.SerializeToString())): - overview = await plcnet_api.async_get_network_overview() - assert overview == MessageToDict(network_overview, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=network_overview.SerializeToString()) + overview = await plcnet_api.async_get_network_overview() + assert overview == MessageToDict(network_overview, + including_default_value_fields=True, + preserving_proto_field_name=True) - def test_get_network_overview(self, plcnet_api): + def test_get_network_overview(self, plcnet_api, httpx_mock): network_overview = GetNetworkOverview() - with patch("httpx.Response.aread", new=AsyncMock(return_value=network_overview.SerializeToString())): - overview = plcnet_api.get_network_overview() - assert overview == MessageToDict(network_overview, - including_default_value_fields=True, - preserving_proto_field_name=True) + httpx_mock.add_response(data=network_overview.SerializeToString()) + overview = plcnet_api.get_network_overview() + assert overview == MessageToDict(network_overview, + including_default_value_fields=True, + preserving_proto_field_name=True) @pytest.mark.asyncio - async def test_async_identify_device_start(self, plcnet_api): + async def test_async_identify_device_start(self, plcnet_api, httpx_mock): identify_device = IdentifyDeviceResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=identify_device.SerializeToString())): - assert await plcnet_api.async_identify_device_start() + httpx_mock.add_response(data=identify_device.SerializeToString()) + assert await plcnet_api.async_identify_device_start() - def test_identify_device_start(self, plcnet_api): + def test_identify_device_start(self, plcnet_api, httpx_mock): identify_device = IdentifyDeviceResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=identify_device.SerializeToString())): - assert plcnet_api.identify_device_start() + httpx_mock.add_response(data=identify_device.SerializeToString()) + assert plcnet_api.identify_device_start() @pytest.mark.asyncio - async def test_async_identify_device_stop(self, plcnet_api): + async def test_async_identify_device_stop(self, plcnet_api, httpx_mock): identify_device = IdentifyDeviceResponse() + httpx_mock.add_response(data=identify_device.SerializeToString()) + assert await plcnet_api.async_identify_device_stop() - with patch("httpx.Response.aread", new=AsyncMock(return_value=identify_device.SerializeToString())): - assert await plcnet_api.async_identify_device_stop() - - def test_identify_device_stop(self, plcnet_api): + def test_identify_device_stop(self, plcnet_api, httpx_mock): identify_device = IdentifyDeviceResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=identify_device.SerializeToString())): - assert plcnet_api.identify_device_stop() + httpx_mock.add_response(data=identify_device.SerializeToString()) + assert plcnet_api.identify_device_stop() @pytest.mark.asyncio - async def test_async_set_user_device_name(self, plcnet_api): + async def test_async_set_user_device_name(self, plcnet_api, httpx_mock): user_device_name_set = SetUserDeviceNameResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=user_device_name_set.SerializeToString())): - assert await plcnet_api.async_set_user_device_name("Test") + httpx_mock.add_response(data=user_device_name_set.SerializeToString()) + assert await plcnet_api.async_set_user_device_name("Test") - def test_set_user_device_name(self, plcnet_api): + def test_set_user_device_name(self, plcnet_api, httpx_mock): user_device_name_set = SetUserDeviceNameResponse() - with patch("httpx.Response.aread", new=AsyncMock(return_value=user_device_name_set.SerializeToString())): - assert plcnet_api.set_user_device_name("Test") + httpx_mock.add_response(data=user_device_name_set.SerializeToString()) + assert plcnet_api.set_user_device_name("Test") diff --git a/tests/test_profobuf.py b/tests/test_profobuf.py index edb0ee1..3668613 100644 --- a/tests/test_profobuf.py +++ b/tests/test_profobuf.py @@ -19,9 +19,8 @@ def test_url(self, request, mock_protobuf): assert mock_protobuf.url == f"http://{ip}:14791/{path}/{version}/" @pytest.mark.asyncio - @pytest.mark.usefixtures("mock_get") - async def test__async_get(self, mocker, mock_protobuf): - mock_protobuf._session = httpx.AsyncClient() + async def test__async_get(self, httpx_mock, mocker, mock_protobuf): + httpx_mock.add_response() spy = mocker.spy(httpx.AsyncClient, "get") await mock_protobuf._async_get("LedSettingsGet") spy.assert_called_once() @@ -29,14 +28,12 @@ async def test__async_get(self, mocker, mock_protobuf): @pytest.mark.asyncio @pytest.mark.usefixtures("mock_wrong_password") async def test__async_get_wrong_password(self, mock_protobuf): - mock_protobuf._session = httpx.AsyncClient() with pytest.raises(DevicePasswordProtected): await mock_protobuf._async_get("LedSettingsGet") @pytest.mark.asyncio @pytest.mark.usefixtures("mock_device_unavailable") async def test__async_get_device_unavailable(self, mock_protobuf): - mock_protobuf._session = httpx.AsyncClient() with pytest.raises(DeviceUnavailable): await mock_protobuf._async_get("LedSettingsGet") @@ -46,9 +43,8 @@ def test__message_to_dict(self, mocker, mock_protobuf): spy.assert_called_once() @pytest.mark.asyncio - @pytest.mark.usefixtures("mock_post") - async def test__async_post(self, mocker, mock_protobuf): - mock_protobuf._session = httpx.AsyncClient() + async def test__async_post(self, httpx_mock, mocker, mock_protobuf): + httpx_mock.add_response() spy = mocker.spy(httpx.AsyncClient, "post") await mock_protobuf._async_post("LedSettingsGet", "") spy.assert_called_once() @@ -56,13 +52,11 @@ async def test__async_post(self, mocker, mock_protobuf): @pytest.mark.asyncio @pytest.mark.usefixtures("mock_wrong_password") async def test__async_post_wrong_password(self, mock_protobuf): - mock_protobuf._session = httpx.AsyncClient() with pytest.raises(DevicePasswordProtected): await mock_protobuf._async_post("LedSettingsGet", "") @pytest.mark.asyncio @pytest.mark.usefixtures("mock_device_unavailable") async def test__async_post_device_unavailable(self, mock_protobuf): - mock_protobuf._session = httpx.AsyncClient() with pytest.raises(DeviceUnavailable): await mock_protobuf._async_post("LedSettingsGet", "")