From e544fc04d57043cf40c07a25a2b9976a93fc4023 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Szczypi=C5=84ski?= Date: Sat, 8 Mar 2025 23:27:24 +0100 Subject: [PATCH 1/5] Add support for TAPO devices MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marek Szczypiński --- doc/configuration.rst | 8 ++ labgrid/driver/power/tapo.py | 101 +++++++++++++++++++++++++ pyproject.toml | 1 + tests/test_powerdriver.py | 4 + tests/test_tapo.py | 142 +++++++++++++++++++++++++++++++++++ 5 files changed, 256 insertions(+) create mode 100644 labgrid/driver/power/tapo.py create mode 100644 tests/test_tapo.py diff --git a/doc/configuration.rst b/doc/configuration.rst index a956386fd..f48ae64d8 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -243,6 +243,14 @@ Currently available are: Controls *TP-Link power strips* via `python-kasa `_. +``tapo`` + Controls *Tapo power strips and single socket devices* via `python-kasa + `_. + Requires valid TP-Link/TAPO cloud credentials to work. + See the `docstring in the module + `__ + for details. + ``tinycontrol`` Controls a tinycontrol.eu IP Power Socket via HTTP. It was tested on the *6G10A v2* model. diff --git a/labgrid/driver/power/tapo.py b/labgrid/driver/power/tapo.py new file mode 100644 index 000000000..2332f9220 --- /dev/null +++ b/labgrid/driver/power/tapo.py @@ -0,0 +1,101 @@ +"""Driver for controlling TP-Link Tapo smart plugs and power strips. + +This module provides functionality to control TP-Link Tapo smart power devices through +the kasa library. It supports both single socket devices (like P100) and multi-socket +power strips (like P300). + +Features: +- Environment-based authentication using KASA_USERNAME and KASA_PASSWORD +- Support for both single and multi-socket devices + +Requirements: +- Valid TP-Link cloud credentials (username/password) +""" + +import asyncio +import os +import sys + +from kasa import Credentials, Device, DeviceConfig, DeviceConnectionParameters, DeviceEncryptionType, DeviceFamily + + +def _get_credentials() -> Credentials: + username = os.environ.get("KASA_USERNAME") + password = os.environ.get("KASA_PASSWORD") + if username is None or password is None: + raise EnvironmentError("KASA_USERNAME or KASA_PASSWORD environment variable not set") + return Credentials(username=username, password=password) + + +def _get_connection_type() -> DeviceConnectionParameters: + # Somewhere between python-kasa 0.7.7 and 0.10.2 the API changed + # Labgrid on Python <= 3.10 uses python-kasa 0.7.7 + # Labgrid on Python >= 3.11 uses python-kasa 0.10.2 + if sys.version_info < (3, 11): + return DeviceConnectionParameters( + device_family=DeviceFamily.SmartTapoPlug, + encryption_type=DeviceEncryptionType.Klap, + https=False, + login_version=2, + ) + return DeviceConnectionParameters( + device_family=DeviceFamily.SmartTapoPlug, + encryption_type=DeviceEncryptionType.Klap, + https=False, + login_version=2, + http_port=80, + ) + + +def _get_device_config(host: str) -> DeviceConfig: + # Same as with `_get_connection_type` - python-kasa API changed + if sys.version_info < (3, 11): + return DeviceConfig( + host=host, credentials=_get_credentials(), connection_type=_get_connection_type(), uses_http=True + ) + return DeviceConfig( + host=host, credentials=_get_credentials(), connection_type=_get_connection_type() + ) + + +async def _power_set(host: str, port: str, index: str, value: bool) -> None: + """We embed the coroutines in an `async` function to minimise calls to `asyncio.run`""" + assert port is None + index = int(index) + device = await Device.connect(config=_get_device_config(host)) + await device.update() + + if device.children: + assert len(device.children) > index, "Trying to access non-existant plug socket on device" + + target = device if not device.children else device.children[index] + if value: + await target.turn_on() + else: + await target.turn_off() + await device.disconnect() + + +def power_set(host: str, port: str, index: str, value: bool) -> None: + asyncio.run(_power_set(host, port, index, value)) + + +async def _power_get(host: str, port: str, index: str) -> bool: + assert port is None + index = int(index) + device = await Device.connect(config=_get_device_config(host)) + await device.update() + + pwr_state: bool + # If the device has no children, it is a single plug socket + if not device.children: + pwr_state = device.is_on + else: + assert len(device.children) > index, "Trying to access non-existant plug socket on device" + pwr_state = device.children[index].is_on + await device.disconnect() + return pwr_state + + +def power_get(host: str, port: str, index: str) -> bool: + return asyncio.run(_power_get(host, port, index)) diff --git a/pyproject.toml b/pyproject.toml index ef6082651..181d5a281 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,6 +81,7 @@ dev = [ # additional dev dependencies "psutil>=5.8.0", + "pytest-asyncio>=0.25.3", "pytest-benchmark>=4.0.0", "pytest-cov>=3.0.0", "pytest-dependency>=0.5.1", diff --git a/tests/test_powerdriver.py b/tests/test_powerdriver.py index 0dfff1547..13ab9fdfa 100644 --- a/tests/test_powerdriver.py +++ b/tests/test_powerdriver.py @@ -299,6 +299,10 @@ def test_import_backend_tplink(self): pytest.importorskip("kasa") import labgrid.driver.power.tplink + def test_import_backend_tapo(self): + pytest.importorskip("kasa") + import labgrid.driver.power.tapo + def test_import_backend_siglent(self): pytest.importorskip("vxi11") import labgrid.driver.power.siglent diff --git a/tests/test_tapo.py b/tests/test_tapo.py new file mode 100644 index 000000000..652c1414b --- /dev/null +++ b/tests/test_tapo.py @@ -0,0 +1,142 @@ +import os +from unittest.mock import AsyncMock, patch + +import pytest + +from labgrid.driver.power.tapo import _get_credentials, _power_get, _power_set, power_get + + +@pytest.fixture +def mock_device_strip(): + device = AsyncMock() + device.children = [ + AsyncMock(is_on=True), + AsyncMock(is_on=False), + AsyncMock(is_on=True) + ] + return device + + +@pytest.fixture +def mock_device_single_plug(): + device = AsyncMock() + device.children = [] + return device + + +@pytest.fixture +def mock_env(): + os.environ['KASA_USERNAME'] = 'test_user' + os.environ['KASA_PASSWORD'] = 'test_pass' + yield + del os.environ['KASA_USERNAME'] + del os.environ['KASA_PASSWORD'] + + +class TestTapoPowerDriver: + def test_get_credentials_should_raise_value_error_when_credentials_missing(self): + # Save existing environment variables + saved_username = os.environ.pop('KASA_USERNAME', None) + saved_password = os.environ.pop('KASA_PASSWORD', None) + + try: + with pytest.raises(EnvironmentError, match="KASA_USERNAME or KASA_PASSWORD environment variable not set"): + _get_credentials() + finally: + # Restore environment variables if they existed + if saved_username is not None: + os.environ['KASA_USERNAME'] = saved_username + if saved_password is not None: + os.environ['KASA_PASSWORD'] = saved_password + + def test_credentials_valid(self, mock_env): + creds = _get_credentials() + assert creds.username == 'test_user' + assert creds.password == 'test_pass' + + @pytest.mark.asyncio + async def test_power_get_single_plug_turn_on(self, mock_device_single_plug, mock_env): + mock_device_single_plug.is_on = True + + with patch('kasa.Device.connect', return_value=mock_device_single_plug): + result = await _power_get('192.168.1.100', None, "0") + assert result is True + + @pytest.mark.asyncio + async def test_power_get_single_plug_turn_off(self, mock_device_single_plug, mock_env): + mock_device_single_plug.is_on = False + + with patch('kasa.Device.connect', return_value=mock_device_single_plug): + result = await _power_get('192.168.1.100', None, "0") + assert result is False + + @pytest.mark.asyncio + async def test_power_get_single_plug_should_not_care_for_index(self, mock_device_single_plug, mock_env): + invalid_index_ignored = "7" + mock_device_single_plug.is_on = True + + with patch('kasa.Device.connect', return_value=mock_device_single_plug): + result = await _power_get('192.168.1.100', None, invalid_index_ignored) + assert result is True + + @pytest.mark.asyncio + async def test_power_set_single_plug_turn_on(self, mock_device_single_plug, mock_env): + mock_device_single_plug.is_on = False + with patch('kasa.Device.connect', return_value=mock_device_single_plug): + await _power_set('192.168.1.100', None, "0", True) + mock_device_single_plug.turn_on.assert_called_once() + + @pytest.mark.asyncio + async def test_power_set_single_plug_turn_off(self, mock_device_single_plug, mock_env): + mock_device_single_plug.is_on = True + with patch('kasa.Device.connect', return_value=mock_device_single_plug): + await _power_set('192.168.1.100', None, "0", False) + mock_device_single_plug.turn_off.assert_called_once() + + @pytest.mark.asyncio + async def test_power_get_strip_valid_socket(self, mock_device_strip, mock_env): + with patch('kasa.Device.connect', return_value=mock_device_strip): + # Test first outlet (on) + result = await _power_get('192.168.1.100', None, "0") + assert result is True + + # Test second outlet (off) + result = await _power_get('192.168.1.100', None, "1") + assert result is False + + # Test third outlet (on) + result = await _power_get('192.168.1.100', None, "2") + assert result is True + + @pytest.mark.asyncio + async def test_power_set_strip_valid_socket(self, mock_device_strip, mock_env): + with patch('kasa.Device.connect', return_value=mock_device_strip): + await _power_set('192.168.1.100', None, "0", False) + mock_device_strip.children[0].turn_off.assert_called_once() + + await _power_set('192.168.1.100', None, "1", True) + mock_device_strip.children[1].turn_on.assert_called_once() + + def test_power_get_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): + invalid_socket = "5" + with patch('kasa.Device.connect', return_value=mock_device_strip): + with pytest.raises(AssertionError, match="Trying to access non-existant plug socket"): + power_get('192.168.1.100', None, invalid_socket) + + @pytest.mark.asyncio + async def test_power_set_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): + invalid_socket = "5" + with patch('kasa.Device.connect', return_value=mock_device_strip): + with pytest.raises(AssertionError, match="Trying to access non-existant plug socket"): + await _power_set('192.168.1.100', None, invalid_socket, True) + + def test_port_not_none_strip(self, mock_device_strip): + with patch('kasa.Device.connect', return_value=mock_device_strip): + with pytest.raises(AssertionError): + power_get('192.168.1.100', '8080', "0") + + def test_port_not_none_single_socket(self, mock_device_single_plug): + mock_device_single_plug.is_on = True + with patch('kasa.Device.connect', return_value=mock_device_single_plug): + with pytest.raises(AssertionError): + power_get('192.168.1.100', '8080', "0") From 783fa10241b75f3a1c12e24f9f5f79a30b0976cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Szczypi=C5=84ski?= Date: Mon, 7 Jul 2025 10:23:36 +0200 Subject: [PATCH 2/5] Fix regression MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marek Szczypiński --- doc/configuration.rst | 4 ---- 1 file changed, 4 deletions(-) diff --git a/doc/configuration.rst b/doc/configuration.rst index 8178fda28..43a12228e 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -249,10 +249,6 @@ Currently available are: `__ for details. -``tplink`` - Controls *TP-Link power strips* via `python-kasa - `_. - ``tapo`` Controls *Tapo power strips and single socket devices* via `python-kasa `_. From 432aefc26ede935d2d8257f8a2b82f1f39b4b5ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Szczypi=C5=84ski?= Date: Tue, 15 Jul 2025 09:41:27 +0200 Subject: [PATCH 3/5] refactor: directly test kasa version in TAPO power model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marek Szczypiński --- labgrid/driver/power/tapo.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/labgrid/driver/power/tapo.py b/labgrid/driver/power/tapo.py index 2332f9220..590b5408e 100644 --- a/labgrid/driver/power/tapo.py +++ b/labgrid/driver/power/tapo.py @@ -14,9 +14,16 @@ import asyncio import os -import sys +import kasa from kasa import Credentials, Device, DeviceConfig, DeviceConnectionParameters, DeviceEncryptionType, DeviceFamily +from packaging.version import parse + + +def _using_old_kasa_api() -> bool: + target_kasa_version = parse("0.10.0") + current_kasa_version = parse(kasa.__version__) + return current_kasa_version < target_kasa_version def _get_credentials() -> Credentials: @@ -28,10 +35,8 @@ def _get_credentials() -> Credentials: def _get_connection_type() -> DeviceConnectionParameters: - # Somewhere between python-kasa 0.7.7 and 0.10.2 the API changed - # Labgrid on Python <= 3.10 uses python-kasa 0.7.7 - # Labgrid on Python >= 3.11 uses python-kasa 0.10.2 - if sys.version_info < (3, 11): + # API changed between versions 0.9.1 and 0.10.0 of python-kasa + if _using_old_kasa_api(): return DeviceConnectionParameters( device_family=DeviceFamily.SmartTapoPlug, encryption_type=DeviceEncryptionType.Klap, @@ -49,10 +54,10 @@ def _get_connection_type() -> DeviceConnectionParameters: def _get_device_config(host: str) -> DeviceConfig: # Same as with `_get_connection_type` - python-kasa API changed - if sys.version_info < (3, 11): + if _using_old_kasa_api(): return DeviceConfig( host=host, credentials=_get_credentials(), connection_type=_get_connection_type(), uses_http=True - ) + ) # type: ignore[call-arg] return DeviceConfig( host=host, credentials=_get_credentials(), connection_type=_get_connection_type() ) From 6e70e3e3065e745aa18490f1b5d7726198c4d9ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Szczypi=C5=84ski?= Date: Tue, 11 Nov 2025 13:54:17 +0100 Subject: [PATCH 4/5] fix(tapo): add pylint disables for python-kasa version compatibility MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Suppress unexpected-keyword-arg warnings for uses_http and http_port parameters that exist in different python-kasa API versions. Additionally tapo.py and test_tapo.py are now ruff formatted. Signed-off-by: Marek Szczypiński --- labgrid/driver/power/tapo.py | 8 ++-- tests/test_tapo.py | 76 +++++++++++++++++------------------- 2 files changed, 41 insertions(+), 43 deletions(-) diff --git a/labgrid/driver/power/tapo.py b/labgrid/driver/power/tapo.py index 590b5408e..3c6601d8c 100644 --- a/labgrid/driver/power/tapo.py +++ b/labgrid/driver/power/tapo.py @@ -43,6 +43,8 @@ def _get_connection_type() -> DeviceConnectionParameters: https=False, login_version=2, ) + # http_port parameter exists in new kasa (>= 0.10.0) API but not in old versions (<= 0.9.1) + # pylint: disable-next=unexpected-keyword-arg return DeviceConnectionParameters( device_family=DeviceFamily.SmartTapoPlug, encryption_type=DeviceEncryptionType.Klap, @@ -55,12 +57,12 @@ def _get_connection_type() -> DeviceConnectionParameters: def _get_device_config(host: str) -> DeviceConfig: # Same as with `_get_connection_type` - python-kasa API changed if _using_old_kasa_api(): + # uses_http parameter exists in old kasa (<= 0.9.1) API but not in new versions (>= 0.10.0) + # pylint: disable-next=unexpected-keyword-arg return DeviceConfig( host=host, credentials=_get_credentials(), connection_type=_get_connection_type(), uses_http=True ) # type: ignore[call-arg] - return DeviceConfig( - host=host, credentials=_get_credentials(), connection_type=_get_connection_type() - ) + return DeviceConfig(host=host, credentials=_get_credentials(), connection_type=_get_connection_type()) async def _power_set(host: str, port: str, index: str, value: bool) -> None: diff --git a/tests/test_tapo.py b/tests/test_tapo.py index 652c1414b..88db095ce 100644 --- a/tests/test_tapo.py +++ b/tests/test_tapo.py @@ -9,11 +9,7 @@ @pytest.fixture def mock_device_strip(): device = AsyncMock() - device.children = [ - AsyncMock(is_on=True), - AsyncMock(is_on=False), - AsyncMock(is_on=True) - ] + device.children = [AsyncMock(is_on=True), AsyncMock(is_on=False), AsyncMock(is_on=True)] return device @@ -26,18 +22,18 @@ def mock_device_single_plug(): @pytest.fixture def mock_env(): - os.environ['KASA_USERNAME'] = 'test_user' - os.environ['KASA_PASSWORD'] = 'test_pass' + os.environ["KASA_USERNAME"] = "test_user" + os.environ["KASA_PASSWORD"] = "test_pass" yield - del os.environ['KASA_USERNAME'] - del os.environ['KASA_PASSWORD'] + del os.environ["KASA_USERNAME"] + del os.environ["KASA_PASSWORD"] class TestTapoPowerDriver: def test_get_credentials_should_raise_value_error_when_credentials_missing(self): # Save existing environment variables - saved_username = os.environ.pop('KASA_USERNAME', None) - saved_password = os.environ.pop('KASA_PASSWORD', None) + saved_username = os.environ.pop("KASA_USERNAME", None) + saved_password = os.environ.pop("KASA_PASSWORD", None) try: with pytest.raises(EnvironmentError, match="KASA_USERNAME or KASA_PASSWORD environment variable not set"): @@ -45,29 +41,29 @@ def test_get_credentials_should_raise_value_error_when_credentials_missing(self) finally: # Restore environment variables if they existed if saved_username is not None: - os.environ['KASA_USERNAME'] = saved_username + os.environ["KASA_USERNAME"] = saved_username if saved_password is not None: - os.environ['KASA_PASSWORD'] = saved_password + os.environ["KASA_PASSWORD"] = saved_password def test_credentials_valid(self, mock_env): creds = _get_credentials() - assert creds.username == 'test_user' - assert creds.password == 'test_pass' + assert creds.username == "test_user" + assert creds.password == "test_pass" @pytest.mark.asyncio async def test_power_get_single_plug_turn_on(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = True - with patch('kasa.Device.connect', return_value=mock_device_single_plug): - result = await _power_get('192.168.1.100', None, "0") + with patch("kasa.Device.connect", return_value=mock_device_single_plug): + result = await _power_get("192.168.1.100", None, "0") assert result is True @pytest.mark.asyncio async def test_power_get_single_plug_turn_off(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = False - with patch('kasa.Device.connect', return_value=mock_device_single_plug): - result = await _power_get('192.168.1.100', None, "0") + with patch("kasa.Device.connect", return_value=mock_device_single_plug): + result = await _power_get("192.168.1.100", None, "0") assert result is False @pytest.mark.asyncio @@ -75,68 +71,68 @@ async def test_power_get_single_plug_should_not_care_for_index(self, mock_device invalid_index_ignored = "7" mock_device_single_plug.is_on = True - with patch('kasa.Device.connect', return_value=mock_device_single_plug): - result = await _power_get('192.168.1.100', None, invalid_index_ignored) + with patch("kasa.Device.connect", return_value=mock_device_single_plug): + result = await _power_get("192.168.1.100", None, invalid_index_ignored) assert result is True @pytest.mark.asyncio async def test_power_set_single_plug_turn_on(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = False - with patch('kasa.Device.connect', return_value=mock_device_single_plug): - await _power_set('192.168.1.100', None, "0", True) + with patch("kasa.Device.connect", return_value=mock_device_single_plug): + await _power_set("192.168.1.100", None, "0", True) mock_device_single_plug.turn_on.assert_called_once() @pytest.mark.asyncio async def test_power_set_single_plug_turn_off(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = True - with patch('kasa.Device.connect', return_value=mock_device_single_plug): - await _power_set('192.168.1.100', None, "0", False) + with patch("kasa.Device.connect", return_value=mock_device_single_plug): + await _power_set("192.168.1.100", None, "0", False) mock_device_single_plug.turn_off.assert_called_once() @pytest.mark.asyncio async def test_power_get_strip_valid_socket(self, mock_device_strip, mock_env): - with patch('kasa.Device.connect', return_value=mock_device_strip): + with patch("kasa.Device.connect", return_value=mock_device_strip): # Test first outlet (on) - result = await _power_get('192.168.1.100', None, "0") + result = await _power_get("192.168.1.100", None, "0") assert result is True # Test second outlet (off) - result = await _power_get('192.168.1.100', None, "1") + result = await _power_get("192.168.1.100", None, "1") assert result is False # Test third outlet (on) - result = await _power_get('192.168.1.100', None, "2") + result = await _power_get("192.168.1.100", None, "2") assert result is True @pytest.mark.asyncio async def test_power_set_strip_valid_socket(self, mock_device_strip, mock_env): - with patch('kasa.Device.connect', return_value=mock_device_strip): - await _power_set('192.168.1.100', None, "0", False) + with patch("kasa.Device.connect", return_value=mock_device_strip): + await _power_set("192.168.1.100", None, "0", False) mock_device_strip.children[0].turn_off.assert_called_once() - await _power_set('192.168.1.100', None, "1", True) + await _power_set("192.168.1.100", None, "1", True) mock_device_strip.children[1].turn_on.assert_called_once() def test_power_get_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): invalid_socket = "5" - with patch('kasa.Device.connect', return_value=mock_device_strip): + with patch("kasa.Device.connect", return_value=mock_device_strip): with pytest.raises(AssertionError, match="Trying to access non-existant plug socket"): - power_get('192.168.1.100', None, invalid_socket) + power_get("192.168.1.100", None, invalid_socket) @pytest.mark.asyncio async def test_power_set_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): invalid_socket = "5" - with patch('kasa.Device.connect', return_value=mock_device_strip): + with patch("kasa.Device.connect", return_value=mock_device_strip): with pytest.raises(AssertionError, match="Trying to access non-existant plug socket"): - await _power_set('192.168.1.100', None, invalid_socket, True) + await _power_set("192.168.1.100", None, invalid_socket, True) def test_port_not_none_strip(self, mock_device_strip): - with patch('kasa.Device.connect', return_value=mock_device_strip): + with patch("kasa.Device.connect", return_value=mock_device_strip): with pytest.raises(AssertionError): - power_get('192.168.1.100', '8080', "0") + power_get("192.168.1.100", "8080", "0") def test_port_not_none_single_socket(self, mock_device_single_plug): mock_device_single_plug.is_on = True - with patch('kasa.Device.connect', return_value=mock_device_single_plug): + with patch("kasa.Device.connect", return_value=mock_device_single_plug): with pytest.raises(AssertionError): - power_get('192.168.1.100', '8080', "0") + power_get("192.168.1.100", "8080", "0") From f3c2fa75858f038264039c39b563ecac7e518bd5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Szczypi=C5=84ski?= Date: Tue, 3 Feb 2026 23:06:02 +0100 Subject: [PATCH 5/5] refactor(tapo): unit tests refactored, typo fixed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marek Szczypiński --- labgrid/driver/power/tapo.py | 4 +-- pyproject.toml | 2 +- tests/test_tapo.py | 70 ++++++++++++++++-------------------- 3 files changed, 34 insertions(+), 42 deletions(-) diff --git a/labgrid/driver/power/tapo.py b/labgrid/driver/power/tapo.py index 3c6601d8c..a92606df5 100644 --- a/labgrid/driver/power/tapo.py +++ b/labgrid/driver/power/tapo.py @@ -73,7 +73,7 @@ async def _power_set(host: str, port: str, index: str, value: bool) -> None: await device.update() if device.children: - assert len(device.children) > index, "Trying to access non-existant plug socket on device" + assert len(device.children) > index, "Trying to access non-existent plug socket on device" target = device if not device.children else device.children[index] if value: @@ -98,7 +98,7 @@ async def _power_get(host: str, port: str, index: str) -> bool: if not device.children: pwr_state = device.is_on else: - assert len(device.children) > index, "Trying to access non-existant plug socket on device" + assert len(device.children) > index, "Trying to access non-existent plug socket on device" pwr_state = device.children[index].is_on await device.disconnect() return pwr_state diff --git a/pyproject.toml b/pyproject.toml index cc33204d9..de445f780 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,7 +83,6 @@ dev = [ # additional dev dependencies "psutil>=5.8.0", - "pytest-asyncio>=0.25.3", "pytest-benchmark>=4.0.0", "pytest-cov>=3.0.0", "pytest-dependency>=0.5.1", @@ -225,6 +224,7 @@ include = [ "labgrid/driver/power/gude8031.py", "labgrid/driver/power/pe6216.py", "labgrid/driver/power/shelly_gen2.py", + "labgrid/driver/power/tapo.py", "labgrid/driver/rawnetworkinterfacedriver.py", "labgrid/protocol/**/*.py", "labgrid/remote/**/*.py", diff --git a/tests/test_tapo.py b/tests/test_tapo.py index 88db095ce..991dc4df0 100644 --- a/tests/test_tapo.py +++ b/tests/test_tapo.py @@ -3,7 +3,7 @@ import pytest -from labgrid.driver.power.tapo import _get_credentials, _power_get, _power_set, power_get +from labgrid.driver.power.tapo import _get_credentials, power_get, power_set @pytest.fixture @@ -50,89 +50,81 @@ def test_credentials_valid(self, mock_env): assert creds.username == "test_user" assert creds.password == "test_pass" - @pytest.mark.asyncio - async def test_power_get_single_plug_turn_on(self, mock_device_single_plug, mock_env): + def test_power_get_single_plug_turn_on(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = True with patch("kasa.Device.connect", return_value=mock_device_single_plug): - result = await _power_get("192.168.1.100", None, "0") + result = power_get("192.168.1.100", None, "0") assert result is True - @pytest.mark.asyncio - async def test_power_get_single_plug_turn_off(self, mock_device_single_plug, mock_env): + def test_power_get_single_plug_turn_off(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = False with patch("kasa.Device.connect", return_value=mock_device_single_plug): - result = await _power_get("192.168.1.100", None, "0") + result = power_get("192.168.1.100", None, "0") assert result is False - @pytest.mark.asyncio - async def test_power_get_single_plug_should_not_care_for_index(self, mock_device_single_plug, mock_env): + def test_power_get_single_plug_should_not_care_for_index(self, mock_device_single_plug, mock_env): invalid_index_ignored = "7" mock_device_single_plug.is_on = True with patch("kasa.Device.connect", return_value=mock_device_single_plug): - result = await _power_get("192.168.1.100", None, invalid_index_ignored) + result = power_get("192.168.1.100", None, invalid_index_ignored) assert result is True - @pytest.mark.asyncio - async def test_power_set_single_plug_turn_on(self, mock_device_single_plug, mock_env): + def test_power_set_single_plug_turn_on(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = False with patch("kasa.Device.connect", return_value=mock_device_single_plug): - await _power_set("192.168.1.100", None, "0", True) + power_set("192.168.1.100", None, "0", True) mock_device_single_plug.turn_on.assert_called_once() - @pytest.mark.asyncio - async def test_power_set_single_plug_turn_off(self, mock_device_single_plug, mock_env): + def test_power_set_single_plug_turn_off(self, mock_device_single_plug, mock_env): mock_device_single_plug.is_on = True with patch("kasa.Device.connect", return_value=mock_device_single_plug): - await _power_set("192.168.1.100", None, "0", False) + power_set("192.168.1.100", None, "0", False) mock_device_single_plug.turn_off.assert_called_once() - @pytest.mark.asyncio - async def test_power_get_strip_valid_socket(self, mock_device_strip, mock_env): + def test_power_get_strip_valid_socket(self, mock_device_strip, mock_env): with patch("kasa.Device.connect", return_value=mock_device_strip): # Test first outlet (on) - result = await _power_get("192.168.1.100", None, "0") + result = power_get("192.168.1.100", None, "0") assert result is True # Test second outlet (off) - result = await _power_get("192.168.1.100", None, "1") + result = power_get("192.168.1.100", None, "1") assert result is False # Test third outlet (on) - result = await _power_get("192.168.1.100", None, "2") + result = power_get("192.168.1.100", None, "2") assert result is True - @pytest.mark.asyncio - async def test_power_set_strip_valid_socket(self, mock_device_strip, mock_env): + def test_power_set_strip_valid_socket(self, mock_device_strip, mock_env): with patch("kasa.Device.connect", return_value=mock_device_strip): - await _power_set("192.168.1.100", None, "0", False) + power_set("192.168.1.100", None, "0", False) mock_device_strip.children[0].turn_off.assert_called_once() - await _power_set("192.168.1.100", None, "1", True) + power_set("192.168.1.100", None, "1", True) mock_device_strip.children[1].turn_on.assert_called_once() def test_power_get_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): invalid_socket = "5" - with patch("kasa.Device.connect", return_value=mock_device_strip): - with pytest.raises(AssertionError, match="Trying to access non-existant plug socket"): - power_get("192.168.1.100", None, invalid_socket) + with patch("kasa.Device.connect", return_value=mock_device_strip), \ + pytest.raises(AssertionError, match="Trying to access non-existent plug socket"): + power_get("192.168.1.100", None, invalid_socket) - @pytest.mark.asyncio - async def test_power_set_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): + def test_power_set_should_raise_assertion_error_when_invalid_index_strip(self, mock_device_strip, mock_env): invalid_socket = "5" - with patch("kasa.Device.connect", return_value=mock_device_strip): - with pytest.raises(AssertionError, match="Trying to access non-existant plug socket"): - await _power_set("192.168.1.100", None, invalid_socket, True) + with patch("kasa.Device.connect", return_value=mock_device_strip), \ + pytest.raises(AssertionError, match="Trying to access non-existent plug socket"): + power_set("192.168.1.100", None, invalid_socket, True) def test_port_not_none_strip(self, mock_device_strip): - with patch("kasa.Device.connect", return_value=mock_device_strip): - with pytest.raises(AssertionError): - power_get("192.168.1.100", "8080", "0") + with patch("kasa.Device.connect", return_value=mock_device_strip), \ + pytest.raises(AssertionError): + power_get("192.168.1.100", "8080", "0") def test_port_not_none_single_socket(self, mock_device_single_plug): mock_device_single_plug.is_on = True - with patch("kasa.Device.connect", return_value=mock_device_single_plug): - with pytest.raises(AssertionError): - power_get("192.168.1.100", "8080", "0") + with patch("kasa.Device.connect", return_value=mock_device_single_plug), \ + pytest.raises(AssertionError): + power_get("192.168.1.100", "8080", "0")