mirror of
https://github.com/esphome/aioesphomeapi.git
synced 2024-11-21 11:55:11 +01:00
Add pytest unit testing (#64)
This commit is contained in:
Otto Winter
GitHub
parent
06a254c82e
commit
3a7a47f649
5
.github/workflows/ci.yml
vendored
5
.github/workflows/ci.yml
vendored
@ -22,6 +22,8 @@ jobs:
|
||||
name: Check import order with isort
|
||||
- id: mypy
|
||||
name: Check typing with mypy
|
||||
- id: pytest
|
||||
name: Run tests with pytest
|
||||
steps:
|
||||
- uses: actions/checkout@v2
|
||||
- name: Set up Python
|
||||
@ -52,6 +54,7 @@ jobs:
|
||||
echo "::add-matcher::.github/workflows/matchers/pylint.json"
|
||||
echo "::add-matcher::.github/workflows/matchers/isort.json"
|
||||
echo "::add-matcher::.github/workflows/matchers/mypy.json"
|
||||
echo "::add-matcher::.github/workflows/matchers/pytest.json"
|
||||
|
||||
- run: flake8 aioesphomeapi
|
||||
if: ${{ matrix.id == 'flake8' }}
|
||||
@ -63,3 +66,5 @@ jobs:
|
||||
if: ${{ matrix.id == 'isort' }}
|
||||
- run: mypy aioesphomeapi
|
||||
if: ${{ matrix.id == 'mypy' }}
|
||||
- run: pytest -vv --tb=native tests
|
||||
if: ${{ matrix.id == 'pytest' }}
|
||||
|
||||
20
.github/workflows/matchers/pytest.json
vendored
Normal file
20
.github/workflows/matchers/pytest.json
vendored
Normal file
@ -0,0 +1,20 @@
|
||||
{
|
||||
"problemMatcher": [
|
||||
{
|
||||
"owner": "pytest",
|
||||
"fileLocation": "absolute",
|
||||
"pattern": [
|
||||
{
|
||||
"regexp": "^\\s+File \"(.*)\", line (\\d+), in (.*)$",
|
||||
"file": 1,
|
||||
"line": 2
|
||||
},
|
||||
{
|
||||
"regexp": "^\\s+(.*)$",
|
||||
"message": 1
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
10
README.rst
10
README.rst
@ -104,9 +104,17 @@ For development is recommended to use a Python virtual environment (``venv``).
|
||||
|
||||
.. code:: bash
|
||||
|
||||
# Setup virtualenv (optional)
|
||||
$ python3 -m venv .
|
||||
$ source bin/activate
|
||||
$ python3 setup.py develop
|
||||
# Install aioesphomeapi and development depenencies
|
||||
$ pip3 install -e .
|
||||
$ pip3 install -r requirements_test.txt
|
||||
|
||||
# Run linters & test
|
||||
$ script/lint
|
||||
# Update protobuf _pb2.py definitions (requires a protobuf compiler installation)
|
||||
$ script/gen-protoc
|
||||
|
||||
License
|
||||
-------
|
||||
|
||||
@ -375,13 +375,15 @@ class APIClient:
|
||||
if stop:
|
||||
req.stop = stop
|
||||
else:
|
||||
req.has_legacy_command = True
|
||||
if stop:
|
||||
req.legacy_command = LegacyCoverCommand.STOP
|
||||
req.has_legacy_command = True
|
||||
elif position == 1.0:
|
||||
req.legacy_command = LegacyCoverCommand.OPEN
|
||||
else:
|
||||
req.has_legacy_command = True
|
||||
elif position == 0.0:
|
||||
req.legacy_command = LegacyCoverCommand.CLOSE
|
||||
req.has_legacy_command = True
|
||||
assert self._connection is not None
|
||||
await self._connection.send_message(req)
|
||||
|
||||
|
||||
@ -7,6 +7,8 @@ from typing import Any, Awaitable, Callable, List, Optional, cast
|
||||
|
||||
from google.protobuf import message
|
||||
|
||||
import aioesphomeapi.host_resolver as hr
|
||||
|
||||
from .api_pb2 import ( # type: ignore
|
||||
ConnectRequest,
|
||||
ConnectResponse,
|
||||
@ -20,7 +22,6 @@ from .api_pb2 import ( # type: ignore
|
||||
PingResponse,
|
||||
)
|
||||
from .core import MESSAGE_TYPE_TO_PROTO, APIConnectionError
|
||||
from .host_resolver import ZeroconfInstanceType, async_resolve_host
|
||||
from .model import APIVersion
|
||||
from .util import bytes_to_varuint, varuint_to_bytes
|
||||
|
||||
@ -35,7 +36,7 @@ class ConnectionParams:
|
||||
password: Optional[str]
|
||||
client_info: str
|
||||
keepalive: float
|
||||
zeroconf_instance: ZeroconfInstanceType
|
||||
zeroconf_instance: hr.ZeroconfInstanceType
|
||||
|
||||
|
||||
class APIConnection:
|
||||
@ -112,7 +113,7 @@ class APIConnection:
|
||||
raise APIConnectionError(f"Already connected for {self.log_name}!")
|
||||
|
||||
try:
|
||||
coro = async_resolve_host(
|
||||
coro = hr.async_resolve_host(
|
||||
self._params.eventloop,
|
||||
self._params.address,
|
||||
self._params.port,
|
||||
|
||||
@ -4,12 +4,11 @@ from dataclasses import dataclass
|
||||
from typing import List, Tuple, Union, cast
|
||||
|
||||
import zeroconf
|
||||
from zeroconf import Zeroconf
|
||||
from zeroconf.asyncio import AsyncZeroconf
|
||||
import zeroconf.asyncio
|
||||
|
||||
from .core import APIConnectionError
|
||||
|
||||
ZeroconfInstanceType = Union[Zeroconf, AsyncZeroconf, None]
|
||||
ZeroconfInstanceType = Union[zeroconf.Zeroconf, zeroconf.asyncio.AsyncZeroconf, None]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@ -49,18 +48,18 @@ async def _async_resolve_host_zeroconf( # pylint: disable=too-many-branches
|
||||
# Use or create zeroconf instance, ensure it's an AsyncZeroconf
|
||||
if zeroconf_instance is None:
|
||||
try:
|
||||
zc = AsyncZeroconf()
|
||||
zc = zeroconf.asyncio.AsyncZeroconf()
|
||||
except Exception:
|
||||
raise APIConnectionError(
|
||||
"Cannot start mDNS sockets, is this a docker container without "
|
||||
"host network mode?"
|
||||
)
|
||||
do_close = True
|
||||
elif isinstance(zeroconf_instance, AsyncZeroconf):
|
||||
elif isinstance(zeroconf_instance, zeroconf.asyncio.AsyncZeroconf):
|
||||
zc = zeroconf_instance
|
||||
do_close = False
|
||||
elif isinstance(zeroconf_instance, Zeroconf):
|
||||
zc = AsyncZeroconf(zc=zeroconf_instance)
|
||||
elif isinstance(zeroconf_instance, zeroconf.Zeroconf):
|
||||
zc = zeroconf.asyncio.AsyncZeroconf(zc=zeroconf_instance)
|
||||
do_close = False
|
||||
else:
|
||||
raise ValueError(
|
||||
|
||||
@ -168,9 +168,9 @@ class CoverState(EntityState):
|
||||
)
|
||||
|
||||
def is_closed(self, api_version: APIVersion) -> bool:
|
||||
if api_version >= APIVersion(1, 1):
|
||||
return self.position == 0.0
|
||||
if api_version < APIVersion(1, 1):
|
||||
return self.legacy_state == LegacyCoverState.CLOSED
|
||||
return self.position == 0.0
|
||||
|
||||
|
||||
# ==================== FAN ====================
|
||||
@ -398,7 +398,7 @@ class ClimateState(EntityState):
|
||||
)
|
||||
custom_fan_mode: str = ""
|
||||
preset: Optional[ClimatePreset] = converter_field(
|
||||
default=ClimatePreset.HOME, converter=ClimatePreset.convert
|
||||
default=ClimatePreset.NONE, converter=ClimatePreset.convert
|
||||
)
|
||||
custom_preset: str = ""
|
||||
|
||||
|
||||
@ -4,3 +4,6 @@ flake8==3.9.2
|
||||
isort==5.9.2
|
||||
mypy==0.910
|
||||
types-protobuf==3.17.0
|
||||
pytest>=6.2.4,<7
|
||||
pytest-asyncio>=0.15.1,<1
|
||||
mock>=4.0.3,<5
|
||||
|
||||
@ -3,8 +3,9 @@
|
||||
cd "$(dirname "$0")/.."
|
||||
set -euxo pipefail
|
||||
|
||||
black --safe aioesphomeapi
|
||||
black --safe aioesphomeapi tests
|
||||
pylint aioesphomeapi
|
||||
flake8 aioesphomeapi
|
||||
isort aioesphomeapi
|
||||
isort aioesphomeapi tests
|
||||
mypy aioesphomeapi
|
||||
pytest tests
|
||||
|
||||
3
setup.py
3
setup.py
@ -42,9 +42,10 @@ setup(
|
||||
description='Python API for interacting with ESPHome devices.',
|
||||
long_description=long_description,
|
||||
license=PROJECT_LICENSE,
|
||||
packages=find_packages(),
|
||||
packages=find_packages("aioesphomeapi"),
|
||||
include_package_data=True,
|
||||
zip_safe=False,
|
||||
install_requires=REQUIRES,
|
||||
python_requires='>=3.7',
|
||||
test_suite="tests",
|
||||
)
|
||||
|
||||
474
tests/test_client.py
Normal file
474
tests/test_client.py
Normal file
@ -0,0 +1,474 @@
|
||||
from sys import version
|
||||
from mock import MagicMock, AsyncMock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from aioesphomeapi.api_pb2 import (
|
||||
BinarySensorStateResponse,
|
||||
CameraImageRequest,
|
||||
CameraImageResponse,
|
||||
ClimateCommandRequest,
|
||||
CoverCommandRequest,
|
||||
ExecuteServiceArgument,
|
||||
ExecuteServiceRequest,
|
||||
FanCommandRequest,
|
||||
LightCommandRequest,
|
||||
ListEntitiesBinarySensorResponse,
|
||||
ListEntitiesDoneResponse,
|
||||
ListEntitiesServicesResponse,
|
||||
NumberCommandRequest,
|
||||
SwitchCommandRequest,
|
||||
)
|
||||
from aioesphomeapi.client import APIClient
|
||||
from aioesphomeapi.model import (
|
||||
APIVersion,
|
||||
BinarySensorInfo,
|
||||
BinarySensorState,
|
||||
CameraState,
|
||||
ClimateFanMode,
|
||||
ClimateMode,
|
||||
ClimatePreset,
|
||||
ClimateSwingMode,
|
||||
FanDirection,
|
||||
FanSpeed,
|
||||
LegacyCoverCommand,
|
||||
UserService,
|
||||
UserServiceArg,
|
||||
UserServiceArgType,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(event_loop):
|
||||
client = APIClient(
|
||||
eventloop=event_loop,
|
||||
address="fake.address",
|
||||
port=6052,
|
||||
password=None,
|
||||
)
|
||||
with patch.object(client, "_connection") as conn:
|
||||
conn.is_connected = True
|
||||
conn.is_authenticated = True
|
||||
yield client
|
||||
|
||||
|
||||
def patch_response_complex(client: APIClient, messages):
|
||||
async def patched(req, app, stop, timeout=5.0):
|
||||
resp = []
|
||||
for msg in messages:
|
||||
if app(msg):
|
||||
resp.append(msg)
|
||||
if stop(msg):
|
||||
break
|
||||
else:
|
||||
raise ValueError("Response never stopped")
|
||||
return resp
|
||||
|
||||
client._connection.send_message_await_response_complex = patched
|
||||
|
||||
|
||||
def patch_response_callback(client: APIClient):
|
||||
on_message = None
|
||||
|
||||
async def patched(req, callback):
|
||||
nonlocal on_message
|
||||
on_message = callback
|
||||
|
||||
client._connection.send_message_callback_response = patched
|
||||
|
||||
async def ret(send):
|
||||
on_message(send)
|
||||
|
||||
return ret
|
||||
|
||||
|
||||
def patch_send(client: APIClient):
|
||||
send = client._connection.send_message = AsyncMock()
|
||||
return send
|
||||
|
||||
|
||||
def patch_api_version(client: APIClient, version: APIVersion):
|
||||
client._connection.api_version = version
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"input, output",
|
||||
[
|
||||
(
|
||||
[ListEntitiesBinarySensorResponse(), ListEntitiesDoneResponse()],
|
||||
([BinarySensorInfo()], []),
|
||||
),
|
||||
(
|
||||
[ListEntitiesServicesResponse(), ListEntitiesDoneResponse()],
|
||||
([], [UserService()]),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_list_entities(auth_client, input, output):
|
||||
patch_response_complex(auth_client, input)
|
||||
resp = await auth_client.list_entities_services()
|
||||
assert resp == output
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subscribe_states(auth_client):
|
||||
send = patch_response_callback(auth_client)
|
||||
on_state = MagicMock()
|
||||
await auth_client.subscribe_states(on_state)
|
||||
on_state.assert_not_called()
|
||||
|
||||
await send(BinarySensorStateResponse())
|
||||
on_state.assert_called_once_with(BinarySensorState())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subscribe_states_camera(auth_client):
|
||||
send = patch_response_callback(auth_client)
|
||||
on_state = MagicMock()
|
||||
await auth_client.subscribe_states(on_state)
|
||||
await send(CameraImageResponse(key=1, data=b"asdf"))
|
||||
on_state.assert_not_called()
|
||||
|
||||
await send(CameraImageResponse(key=1, data=b"qwer", done=True))
|
||||
on_state.assert_called_once_with(CameraState(key=1, data=b"asdfqwer"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(dict(key=1), dict(key=1)),
|
||||
(
|
||||
dict(key=1, position=1.0),
|
||||
dict(
|
||||
key=1, has_legacy_command=True, legacy_command=LegacyCoverCommand.OPEN
|
||||
),
|
||||
),
|
||||
(
|
||||
dict(key=1, position=0.0),
|
||||
dict(
|
||||
key=1, has_legacy_command=True, legacy_command=LegacyCoverCommand.CLOSE
|
||||
),
|
||||
),
|
||||
(
|
||||
dict(key=1, stop=True),
|
||||
dict(
|
||||
key=1, has_legacy_command=True, legacy_command=LegacyCoverCommand.STOP
|
||||
),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_cover_command_legacy(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
patch_api_version(auth_client, APIVersion(1, 0))
|
||||
|
||||
await auth_client.cover_command(**cmd)
|
||||
send.assert_called_once_with(CoverCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(dict(key=1), dict(key=1)),
|
||||
(dict(key=1, position=0.5), dict(key=1, has_position=True, position=0.5)),
|
||||
(dict(key=1, position=0.0), dict(key=1, has_position=True, position=0.0)),
|
||||
(dict(key=1, stop=True), dict(key=1, stop=True)),
|
||||
(
|
||||
dict(key=1, position=1.0, tilt=0.8),
|
||||
dict(key=1, has_position=True, position=1.0, has_tilt=True, tilt=0.8),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_cover_command(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
patch_api_version(auth_client, APIVersion(1, 1))
|
||||
|
||||
await auth_client.cover_command(**cmd)
|
||||
send.assert_called_once_with(CoverCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(dict(key=1), dict(key=1)),
|
||||
(dict(key=1, state=True), dict(key=1, has_state=True, state=True)),
|
||||
(
|
||||
dict(key=1, speed=FanSpeed.LOW),
|
||||
dict(key=1, has_speed=True, speed=FanSpeed.LOW),
|
||||
),
|
||||
(
|
||||
dict(key=1, speed_level=10),
|
||||
dict(key=1, has_speed_level=True, speed_level=10),
|
||||
),
|
||||
(
|
||||
dict(key=1, oscillating=False),
|
||||
dict(key=1, has_oscillating=True, oscillating=False),
|
||||
),
|
||||
(
|
||||
dict(key=1, direction=FanDirection.REVERSE),
|
||||
dict(key=1, has_direction=True, direction=FanDirection.REVERSE),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_fan_command(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
|
||||
await auth_client.fan_command(**cmd)
|
||||
send.assert_called_once_with(FanCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(dict(key=1), dict(key=1)),
|
||||
(dict(key=1, state=True), dict(key=1, has_state=True, state=True)),
|
||||
(dict(key=1, brightness=0.8), dict(key=1, has_brightness=True, brightness=0.8)),
|
||||
(
|
||||
dict(key=1, rgb=(0.1, 0.5, 1.0)),
|
||||
dict(key=1, has_rgb=True, red=0.1, green=0.5, blue=1.0),
|
||||
),
|
||||
(dict(key=1, white=0.0), dict(key=1, has_white=True, white=0.0)),
|
||||
(
|
||||
dict(key=1, color_temperature=0.0),
|
||||
dict(key=1, has_color_temperature=True, color_temperature=0.0),
|
||||
),
|
||||
(
|
||||
dict(key=1, transition_length=0.1),
|
||||
dict(key=1, has_transition_length=True, transition_length=100),
|
||||
),
|
||||
(
|
||||
dict(key=1, flash_length=0.1),
|
||||
dict(key=1, has_flash_length=True, flash_length=100),
|
||||
),
|
||||
(dict(key=1, effect="special"), dict(key=1, has_effect=True, effect="special")),
|
||||
],
|
||||
)
|
||||
async def test_light_command(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
|
||||
await auth_client.light_command(**cmd)
|
||||
send.assert_called_once_with(LightCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(dict(key=1, state=False), dict(key=1, state=False)),
|
||||
(dict(key=1, state=True), dict(key=1, state=True)),
|
||||
],
|
||||
)
|
||||
async def test_switch_command(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
|
||||
await auth_client.switch_command(**cmd)
|
||||
send.assert_called_once_with(SwitchCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(
|
||||
dict(key=1, preset=ClimatePreset.HOME),
|
||||
dict(key=1, has_legacy_away=True, legacy_away=False),
|
||||
),
|
||||
(
|
||||
dict(key=1, preset=ClimatePreset.AWAY),
|
||||
dict(key=1, has_legacy_away=True, legacy_away=True),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_climate_command_legacy(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
patch_api_version(auth_client, APIVersion(1, 4))
|
||||
|
||||
await auth_client.climate_command(**cmd)
|
||||
send.assert_called_once_with(ClimateCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(
|
||||
dict(key=1, mode=ClimateMode.HEAT),
|
||||
dict(key=1, has_mode=True, mode=ClimateMode.HEAT),
|
||||
),
|
||||
(
|
||||
dict(key=1, target_temperature=21.0),
|
||||
dict(key=1, has_target_temperature=True, target_temperature=21.0),
|
||||
),
|
||||
(
|
||||
dict(key=1, target_temperature_low=21.0),
|
||||
dict(key=1, has_target_temperature_low=True, target_temperature_low=21.0),
|
||||
),
|
||||
(
|
||||
dict(key=1, target_temperature_high=21.0),
|
||||
dict(key=1, has_target_temperature_high=True, target_temperature_high=21.0),
|
||||
),
|
||||
(
|
||||
dict(key=1, fan_mode=ClimateFanMode.LOW),
|
||||
dict(key=1, has_fan_mode=True, fan_mode=ClimateFanMode.LOW),
|
||||
),
|
||||
(
|
||||
dict(key=1, swing_mode=ClimateSwingMode.OFF),
|
||||
dict(key=1, has_swing_mode=True, swing_mode=ClimateSwingMode.OFF),
|
||||
),
|
||||
(
|
||||
dict(key=1, custom_fan_mode="asdf"),
|
||||
dict(key=1, has_custom_fan_mode=True, custom_fan_mode="asdf"),
|
||||
),
|
||||
(
|
||||
dict(key=1, preset=ClimatePreset.AWAY),
|
||||
dict(key=1, has_preset=True, preset=ClimatePreset.AWAY),
|
||||
),
|
||||
(
|
||||
dict(key=1, custom_preset="asdf"),
|
||||
dict(key=1, has_custom_preset=True, custom_preset="asdf"),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_climate_command(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
patch_api_version(auth_client, APIVersion(1, 5))
|
||||
|
||||
await auth_client.climate_command(**cmd)
|
||||
send.assert_called_once_with(ClimateCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, req",
|
||||
[
|
||||
(dict(key=1, state=0.0), dict(key=1, state=0.0)),
|
||||
(dict(key=1, state=100.0), dict(key=1, state=100.0)),
|
||||
],
|
||||
)
|
||||
async def test_number_command(auth_client, cmd, req):
|
||||
send = patch_send(auth_client)
|
||||
|
||||
await auth_client.number_command(**cmd)
|
||||
send.assert_called_once_with(NumberCommandRequest(**req))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_service(auth_client):
|
||||
send = patch_send(auth_client)
|
||||
patch_api_version(auth_client, APIVersion(1, 3))
|
||||
|
||||
service = UserService(
|
||||
name="my_service",
|
||||
key=1,
|
||||
args=[
|
||||
UserServiceArg(name="arg1", type=UserServiceArgType.BOOL),
|
||||
UserServiceArg(name="arg2", type=UserServiceArgType.INT),
|
||||
UserServiceArg(name="arg3", type=UserServiceArgType.FLOAT),
|
||||
UserServiceArg(name="arg4", type=UserServiceArgType.STRING),
|
||||
UserServiceArg(name="arg5", type=UserServiceArgType.BOOL_ARRAY),
|
||||
UserServiceArg(name="arg6", type=UserServiceArgType.INT_ARRAY),
|
||||
UserServiceArg(name="arg7", type=UserServiceArgType.FLOAT_ARRAY),
|
||||
UserServiceArg(name="arg8", type=UserServiceArgType.STRING_ARRAY),
|
||||
],
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
await auth_client.execute_service(service, data={})
|
||||
|
||||
await auth_client.execute_service(
|
||||
service,
|
||||
data={
|
||||
"arg1": False,
|
||||
"arg2": 42,
|
||||
"arg3": 99.0,
|
||||
"arg4": "asdf",
|
||||
"arg5": [False, True, False],
|
||||
"arg6": [42, 10, 9],
|
||||
"arg7": [0.0, -100.0],
|
||||
"arg8": [],
|
||||
},
|
||||
)
|
||||
send.assert_called_once_with(
|
||||
ExecuteServiceRequest(
|
||||
key=1,
|
||||
args=[
|
||||
ExecuteServiceArgument(bool_=False),
|
||||
ExecuteServiceArgument(int_=42),
|
||||
ExecuteServiceArgument(float_=99.0),
|
||||
ExecuteServiceArgument(string_="asdf"),
|
||||
ExecuteServiceArgument(bool_array=[False, True, False]),
|
||||
ExecuteServiceArgument(int_array=[42, 10, 9]),
|
||||
ExecuteServiceArgument(float_array=[0.0, -100.0]),
|
||||
ExecuteServiceArgument(string_array=[]),
|
||||
],
|
||||
)
|
||||
)
|
||||
send.reset_mock()
|
||||
|
||||
patch_api_version(auth_client, APIVersion(1, 2))
|
||||
service = UserService(
|
||||
name="my_service",
|
||||
key=2,
|
||||
args=[
|
||||
UserServiceArg(name="arg1", type=UserServiceArgType.BOOL),
|
||||
UserServiceArg(name="arg2", type=UserServiceArgType.INT),
|
||||
],
|
||||
)
|
||||
|
||||
# Test legacy_int
|
||||
await auth_client.execute_service(
|
||||
service,
|
||||
data={
|
||||
"arg1": False,
|
||||
"arg2": 42,
|
||||
},
|
||||
)
|
||||
send.assert_called_once_with(
|
||||
ExecuteServiceRequest(
|
||||
key=2,
|
||||
args=[
|
||||
ExecuteServiceArgument(bool_=False),
|
||||
ExecuteServiceArgument(legacy_int=42),
|
||||
],
|
||||
)
|
||||
)
|
||||
send.reset_mock()
|
||||
|
||||
# Test arg order
|
||||
await auth_client.execute_service(
|
||||
service,
|
||||
data={
|
||||
"arg2": 42,
|
||||
"arg1": False,
|
||||
},
|
||||
)
|
||||
send.assert_called_once_with(
|
||||
ExecuteServiceRequest(
|
||||
key=2,
|
||||
args=[
|
||||
ExecuteServiceArgument(bool_=False),
|
||||
ExecuteServiceArgument(legacy_int=42),
|
||||
],
|
||||
)
|
||||
)
|
||||
send.reset_mock()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_single_image(auth_client):
|
||||
send = patch_send(auth_client)
|
||||
|
||||
await auth_client.request_single_image()
|
||||
send.assert_called_once_with(CameraImageRequest(single=True, stream=False))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_image_stream(auth_client):
|
||||
send = patch_send(auth_client)
|
||||
|
||||
await auth_client.request_image_stream()
|
||||
send.assert_called_once_with(CameraImageRequest(single=False, stream=True))
|
||||
63
tests/test_connection.py
Normal file
63
tests/test_connection.py
Normal file
@ -0,0 +1,63 @@
|
||||
import asyncio
|
||||
import socket
|
||||
from mock import AsyncMock, MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from aioesphomeapi.api_pb2 import ConnectResponse, HelloResponse
|
||||
from aioesphomeapi.connection import APIConnection, ConnectionParams
|
||||
from aioesphomeapi.core import APIConnectionError
|
||||
from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def connection_params() -> ConnectionParams:
|
||||
return ConnectionParams(
|
||||
eventloop=asyncio.get_event_loop(),
|
||||
address="fake.address",
|
||||
port=6052,
|
||||
password=None,
|
||||
client_info="Tests client",
|
||||
keepalive=15.0,
|
||||
zeroconf_instance=None,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def conn(connection_params) -> APIConnection:
|
||||
async def on_stop():
|
||||
pass
|
||||
|
||||
return APIConnection(connection_params, on_stop)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def resolve_host():
|
||||
with patch("aioesphomeapi.host_resolver.async_resolve_host") as func:
|
||||
func.return_value = AddrInfo(
|
||||
family=socket.AF_INET,
|
||||
type=socket.SOCK_STREAM,
|
||||
proto=socket.IPPROTO_TCP,
|
||||
sockaddr=IPv4Sockaddr("10.0.0.512", 6052),
|
||||
)
|
||||
yield func
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def socket_socket():
|
||||
with patch("socket.socket") as func:
|
||||
yield func
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect(conn, resolve_host, socket_socket, event_loop):
|
||||
with patch.object(event_loop, "sock_connect"), patch(
|
||||
"asyncio.open_connection", return_value=(None, None)
|
||||
), patch.object(conn, "run_forever"), patch.object(
|
||||
conn, "_start_ping"
|
||||
), patch.object(
|
||||
conn, "send_message_await_response", return_value=HelloResponse()
|
||||
):
|
||||
await conn.connect()
|
||||
|
||||
assert conn.is_connected
|
||||
147
tests/test_host_resolver.py
Normal file
147
tests/test_host_resolver.py
Normal file
@ -0,0 +1,147 @@
|
||||
import socket
|
||||
from mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
import aioesphomeapi.host_resolver as hr
|
||||
from aioesphomeapi.core import APIConnectionError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def async_zeroconf():
|
||||
with patch("zeroconf.asyncio.AsyncZeroconf") as klass:
|
||||
yield klass.return_value
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def addr_infos():
|
||||
return [
|
||||
hr.AddrInfo(
|
||||
family=socket.AF_INET,
|
||||
type=socket.SOCK_STREAM,
|
||||
proto=socket.IPPROTO_TCP,
|
||||
sockaddr=hr.IPv4Sockaddr(address="10.0.0.42", port=6052),
|
||||
),
|
||||
hr.AddrInfo(
|
||||
family=socket.AF_INET6,
|
||||
type=socket.SOCK_STREAM,
|
||||
proto=socket.IPPROTO_TCP,
|
||||
sockaddr=hr.IPv6Sockaddr(
|
||||
address="2001:db8:85a3::8a2e:370:7334",
|
||||
port=6052,
|
||||
flowinfo=0,
|
||||
scope_id=0,
|
||||
),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_host_zeroconf(async_zeroconf, addr_infos):
|
||||
info = MagicMock()
|
||||
info.addresses_by_version.return_value = [
|
||||
b"\n\x00\x00*",
|
||||
b" \x01\r\xb8\x85\xa3\x00\x00\x00\x00\x8a.\x03ps4",
|
||||
]
|
||||
async_zeroconf.async_get_service_info = AsyncMock(return_value=info)
|
||||
async_zeroconf.async_close = AsyncMock()
|
||||
|
||||
ret = await hr._async_resolve_host_zeroconf("asdf", 6052)
|
||||
|
||||
async_zeroconf.async_get_service_info.assert_called_once_with(
|
||||
"_esphomelib._tcp.local.", "asdf._esphomelib._tcp.local.", 3000
|
||||
)
|
||||
async_zeroconf.async_close.assert_called_once_with()
|
||||
|
||||
assert ret == addr_infos
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_host_zeroconf_empty(async_zeroconf):
|
||||
async_zeroconf.async_get_service_info = AsyncMock(return_value=None)
|
||||
async_zeroconf.async_close = AsyncMock()
|
||||
|
||||
ret = await hr._async_resolve_host_zeroconf("asdf.local", 6052)
|
||||
assert ret == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_host_getaddrinfo(addr_infos):
|
||||
eventloop = AsyncMock()
|
||||
eventloop.getaddrinfo.return_value = [
|
||||
(
|
||||
socket.AF_INET,
|
||||
socket.SOCK_STREAM,
|
||||
socket.IPPROTO_TCP,
|
||||
"canon1",
|
||||
("10.0.0.42", 6052),
|
||||
),
|
||||
(
|
||||
socket.AF_INET6,
|
||||
socket.SOCK_STREAM,
|
||||
socket.IPPROTO_TCP,
|
||||
"canon2",
|
||||
("2001:db8:85a3::8a2e:370:7334", 6052, 0, 0),
|
||||
),
|
||||
(-1, socket.SOCK_STREAM, socket.IPPROTO_TCP, "canon3", ("10.0.0.42", 6052)),
|
||||
]
|
||||
ret = await hr._async_resolve_host_getaddrinfo(eventloop, "example.com", 6052)
|
||||
|
||||
assert ret == addr_infos
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_host_getaddrinfo_oserror():
|
||||
eventloop = AsyncMock()
|
||||
eventloop.getaddrinfo.side_effect = OSError()
|
||||
with pytest.raises(APIConnectionError):
|
||||
await hr._async_resolve_host_getaddrinfo(eventloop, "example.com", 6052)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
|
||||
async def test_resolve_host_mdns(resolve_addr, resolve_zc, addr_infos):
|
||||
resolve_zc.return_value = addr_infos
|
||||
ret = await hr.async_resolve_host(None, "example.local", 6052)
|
||||
|
||||
resolve_zc.assert_called_once_with("example", 6052, zeroconf_instance=None)
|
||||
resolve_addr.assert_not_called()
|
||||
assert ret == addr_infos[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
|
||||
async def test_resolve_host_mdns_empty(resolve_addr, resolve_zc, addr_infos):
|
||||
resolve_zc.return_value = []
|
||||
resolve_addr.return_value = addr_infos
|
||||
ret = await hr.async_resolve_host(None, "example.local", 6052)
|
||||
|
||||
resolve_zc.assert_called_once_with("example", 6052, zeroconf_instance=None)
|
||||
resolve_addr.assert_called_once_with(None, "example.local", 6052)
|
||||
assert ret == addr_infos[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
|
||||
async def test_resolve_host_addrinfo(resolve_addr, resolve_zc, addr_infos):
|
||||
resolve_addr.return_value = addr_infos
|
||||
ret = await hr.async_resolve_host(None, "example.com", 6052)
|
||||
|
||||
resolve_zc.assert_not_called()
|
||||
resolve_addr.assert_called_once_with(None, "example.com", 6052)
|
||||
assert ret == addr_infos[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
|
||||
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
|
||||
async def test_resolve_host_addrinfo_empty(resolve_addr, resolve_zc, addr_infos):
|
||||
resolve_addr.return_value = []
|
||||
with pytest.raises(APIConnectionError):
|
||||
await hr.async_resolve_host(None, "example.com", 6052)
|
||||
|
||||
resolve_zc.assert_not_called()
|
||||
resolve_addr.assert_called_once_with(None, "example.com", 6052)
|
||||
302
tests/test_model.py
Normal file
302
tests/test_model.py
Normal file
@ -0,0 +1,302 @@
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional
|
||||
|
||||
import pytest
|
||||
|
||||
from aioesphomeapi.api_pb2 import (
|
||||
BinarySensorStateResponse,
|
||||
ClimateStateResponse,
|
||||
CoverStateResponse,
|
||||
DeviceInfoResponse,
|
||||
FanStateResponse,
|
||||
HomeassistantServiceMap,
|
||||
HomeassistantServiceResponse,
|
||||
LightStateResponse,
|
||||
ListEntitiesBinarySensorResponse,
|
||||
ListEntitiesClimateResponse,
|
||||
ListEntitiesCoverResponse,
|
||||
ListEntitiesFanResponse,
|
||||
ListEntitiesLightResponse,
|
||||
ListEntitiesNumberResponse,
|
||||
ListEntitiesSensorResponse,
|
||||
ListEntitiesServicesArgument,
|
||||
ListEntitiesServicesResponse,
|
||||
ListEntitiesSwitchResponse,
|
||||
ListEntitiesTextSensorResponse,
|
||||
NumberStateResponse,
|
||||
SensorStateResponse,
|
||||
ServiceArgType,
|
||||
SwitchStateResponse,
|
||||
TextSensorStateResponse,
|
||||
)
|
||||
from aioesphomeapi.model import (
|
||||
APIIntEnum,
|
||||
APIModelBase,
|
||||
APIVersion,
|
||||
BinarySensorInfo,
|
||||
BinarySensorState,
|
||||
ClimateInfo,
|
||||
ClimatePreset,
|
||||
ClimateState,
|
||||
CoverInfo,
|
||||
CoverState,
|
||||
DeviceInfo,
|
||||
FanInfo,
|
||||
FanState,
|
||||
HomeassistantServiceCall,
|
||||
LegacyCoverState,
|
||||
LightInfo,
|
||||
LightState,
|
||||
NumberInfo,
|
||||
NumberState,
|
||||
SensorInfo,
|
||||
SensorState,
|
||||
SwitchInfo,
|
||||
SwitchState,
|
||||
TextSensorInfo,
|
||||
TextSensorState,
|
||||
UserService,
|
||||
UserServiceArg,
|
||||
UserServiceArgType,
|
||||
converter_field,
|
||||
)
|
||||
|
||||
|
||||
class DummyIntEnum(APIIntEnum):
|
||||
DEFAULT = 0
|
||||
MY_VAL = 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input, output",
|
||||
[
|
||||
(0, DummyIntEnum.DEFAULT),
|
||||
(1, DummyIntEnum.MY_VAL),
|
||||
(2, None),
|
||||
(-1, None),
|
||||
(DummyIntEnum.DEFAULT, DummyIntEnum.DEFAULT),
|
||||
(DummyIntEnum.MY_VAL, DummyIntEnum.MY_VAL),
|
||||
],
|
||||
)
|
||||
def test_api_int_enum_convert(input, output):
|
||||
v = DummyIntEnum.convert(input)
|
||||
assert v == output
|
||||
assert v is None or isinstance(v, DummyIntEnum)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input, output",
|
||||
[
|
||||
([], []),
|
||||
([1], [DummyIntEnum.MY_VAL]),
|
||||
([0, 1], [DummyIntEnum.DEFAULT, DummyIntEnum.MY_VAL]),
|
||||
([-1], []),
|
||||
([0, -1], [DummyIntEnum.DEFAULT]),
|
||||
([DummyIntEnum.DEFAULT], [DummyIntEnum.DEFAULT]),
|
||||
],
|
||||
)
|
||||
def test_api_int_enum_convert_list(input, output):
|
||||
v = DummyIntEnum.convert_list(input)
|
||||
assert v == output
|
||||
assert all(isinstance(x, DummyIntEnum) for x in v)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DummyAPIModel(APIModelBase):
|
||||
val1: int = 0
|
||||
val2: Optional[DummyIntEnum] = converter_field(
|
||||
default=DummyIntEnum.DEFAULT, converter=DummyIntEnum.convert
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ListAPIModel(APIModelBase):
|
||||
val: List[DummyAPIModel] = field(default_factory=list)
|
||||
|
||||
|
||||
def test_api_model_base_converter():
|
||||
assert DummyAPIModel().val2 == DummyIntEnum.DEFAULT
|
||||
assert isinstance(DummyAPIModel().val2, DummyIntEnum)
|
||||
assert DummyAPIModel(val2=0).val2 == DummyIntEnum.DEFAULT
|
||||
assert isinstance(DummyAPIModel().val2, DummyIntEnum)
|
||||
assert DummyAPIModel(val2=-1).val2 is None
|
||||
|
||||
|
||||
def test_api_model_base_to_dict():
|
||||
assert DummyAPIModel().to_dict() == {
|
||||
"val1": 0,
|
||||
"val2": 0,
|
||||
}
|
||||
assert DummyAPIModel(val1=-1, val2=1).to_dict() == {
|
||||
"val1": -1,
|
||||
"val2": 1,
|
||||
}
|
||||
assert ListAPIModel(val=[DummyAPIModel()]).to_dict() == {
|
||||
"val": [
|
||||
{
|
||||
"val1": 0,
|
||||
"val2": 0,
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def test_api_model_base_from_dict():
|
||||
assert DummyAPIModel.from_dict({}) == DummyAPIModel()
|
||||
assert (
|
||||
DummyAPIModel.from_dict(
|
||||
{
|
||||
"val1": -1,
|
||||
"val2": -1,
|
||||
}
|
||||
)
|
||||
== DummyAPIModel(val1=-1, val2=None)
|
||||
)
|
||||
assert (
|
||||
DummyAPIModel.from_dict(
|
||||
{
|
||||
"val1": -1,
|
||||
"unknown": 100,
|
||||
}
|
||||
)
|
||||
== DummyAPIModel(val1=-1)
|
||||
)
|
||||
assert ListAPIModel.from_dict({}) == ListAPIModel()
|
||||
assert ListAPIModel.from_dict({"val": []}) == ListAPIModel()
|
||||
|
||||
|
||||
def test_api_model_base_from_pb():
|
||||
class DummyPB:
|
||||
def __init__(self, val1=0, val2=0):
|
||||
self.val1 = val1
|
||||
self.val2 = val2
|
||||
|
||||
assert DummyAPIModel.from_pb(DummyPB()) == DummyAPIModel()
|
||||
assert DummyAPIModel.from_pb(DummyPB(val1=-1, val2=-1)) == DummyAPIModel(
|
||||
val1=-1, val2=None
|
||||
)
|
||||
|
||||
|
||||
def test_api_version_ord():
|
||||
assert APIVersion(1, 0) == APIVersion(1, 0)
|
||||
assert APIVersion(1, 0) < APIVersion(1, 1)
|
||||
assert APIVersion(1, 1) <= APIVersion(1, 1)
|
||||
assert APIVersion(1, 0) < APIVersion(2, 0)
|
||||
assert not (APIVersion(2, 1) <= APIVersion(2, 0))
|
||||
assert APIVersion(2, 1) > APIVersion(2, 0)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"model, pb",
|
||||
[
|
||||
(DeviceInfo, DeviceInfoResponse),
|
||||
(BinarySensorInfo, ListEntitiesBinarySensorResponse),
|
||||
(BinarySensorState, BinarySensorStateResponse),
|
||||
(CoverInfo, ListEntitiesCoverResponse),
|
||||
(CoverState, CoverStateResponse),
|
||||
(FanInfo, ListEntitiesFanResponse),
|
||||
(FanState, FanStateResponse),
|
||||
(LightInfo, ListEntitiesLightResponse),
|
||||
(LightState, LightStateResponse),
|
||||
(SensorInfo, ListEntitiesSensorResponse),
|
||||
(SensorState, SensorStateResponse),
|
||||
(SwitchInfo, ListEntitiesSwitchResponse),
|
||||
(SwitchState, SwitchStateResponse),
|
||||
(TextSensorInfo, ListEntitiesTextSensorResponse),
|
||||
(TextSensorState, TextSensorStateResponse),
|
||||
(ClimateInfo, ListEntitiesClimateResponse),
|
||||
(ClimateState, ClimateStateResponse),
|
||||
(NumberInfo, ListEntitiesNumberResponse),
|
||||
(NumberState, NumberStateResponse),
|
||||
(HomeassistantServiceCall, HomeassistantServiceResponse),
|
||||
(UserServiceArg, ListEntitiesServicesArgument),
|
||||
(UserService, ListEntitiesServicesResponse),
|
||||
],
|
||||
)
|
||||
def test_basic_pb_conversions(model, pb):
|
||||
assert model.from_pb(pb()) == model()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"state, version, out",
|
||||
[
|
||||
(CoverState(legacy_state=LegacyCoverState.OPEN), (1, 0), False),
|
||||
(CoverState(legacy_state=LegacyCoverState.CLOSED), (1, 0), True),
|
||||
(CoverState(position=1.0), (1, 1), False),
|
||||
(CoverState(position=0.5), (1, 1), False),
|
||||
(CoverState(position=0.0), (1, 1), True),
|
||||
],
|
||||
)
|
||||
def test_cover_state_legacy_state(state, version, out):
|
||||
assert state.is_closed(APIVersion(*version)) is out
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"state, version, out",
|
||||
[
|
||||
(ClimateInfo(legacy_supports_away=False), (1, 4), []),
|
||||
(
|
||||
ClimateInfo(legacy_supports_away=True),
|
||||
(1, 4),
|
||||
[ClimatePreset.HOME, ClimatePreset.AWAY],
|
||||
),
|
||||
(ClimateInfo(supported_presets=[ClimatePreset.HOME]), (1, 4), []),
|
||||
(ClimateInfo(supported_presets=[], legacy_supports_away=True), (1, 5), []),
|
||||
(
|
||||
ClimateInfo(supported_presets=[ClimatePreset.HOME]),
|
||||
(1, 5),
|
||||
[ClimatePreset.HOME],
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_climate_info_supported_presets_compat(state, version, out):
|
||||
assert state.supported_presets_compat(APIVersion(*version)) == out
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"state, version, out",
|
||||
[
|
||||
(ClimateState(legacy_away=False), (1, 4), ClimatePreset.HOME),
|
||||
(ClimateState(legacy_away=True), (1, 4), ClimatePreset.AWAY),
|
||||
(
|
||||
ClimateState(legacy_away=True, preset=ClimatePreset.HOME),
|
||||
(1, 4),
|
||||
ClimatePreset.AWAY,
|
||||
),
|
||||
(ClimateState(preset=ClimatePreset.HOME), (1, 5), ClimatePreset.HOME),
|
||||
(ClimateState(preset=ClimatePreset.BOOST), (1, 5), ClimatePreset.BOOST),
|
||||
(
|
||||
ClimateState(legacy_away=True, preset=ClimatePreset.BOOST),
|
||||
(1, 5),
|
||||
ClimatePreset.BOOST,
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_climate_state_preset_compat(state, version, out):
|
||||
assert state.preset_compat(APIVersion(*version)) == out
|
||||
|
||||
|
||||
def test_homeassistant_service_map_conversion():
|
||||
assert HomeassistantServiceCall.from_pb(
|
||||
HomeassistantServiceResponse(
|
||||
data=[HomeassistantServiceMap(key="key", value="value")]
|
||||
)
|
||||
) == HomeassistantServiceCall(data={"key": "value"})
|
||||
assert HomeassistantServiceCall.from_dict(
|
||||
{"data": {"key": "value"}}
|
||||
) == HomeassistantServiceCall(data={"key": "value"})
|
||||
|
||||
|
||||
def test_user_service_conversion():
|
||||
assert UserService.from_pb(
|
||||
ListEntitiesServicesResponse(
|
||||
args=[
|
||||
ListEntitiesServicesArgument(
|
||||
name="arg", type=ServiceArgType.SERVICE_ARG_TYPE_INT
|
||||
)
|
||||
]
|
||||
)
|
||||
) == UserService(args=[UserServiceArg(name="arg", type=UserServiceArgType.INT)])
|
||||
assert UserService.from_dict({"args": [{"name": "arg", "type": 1}]}) == UserService(
|
||||
args=[UserServiceArg(name="arg", type=UserServiceArgType.INT)]
|
||||
)
|
||||
22
tests/test_util.py
Normal file
22
tests/test_util.py
Normal file
@ -0,0 +1,22 @@
|
||||
import pytest
|
||||
|
||||
from aioesphomeapi import util
|
||||
|
||||
VARUINT_TESTCASES = [
|
||||
(0, b"\x00"),
|
||||
(42, b"\x2a"),
|
||||
(127, b"\x7f"),
|
||||
(128, b"\x80\x01"),
|
||||
(300, b"\xac\x02"),
|
||||
(65536, b"\x80\x80\x04"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("val, encoded", VARUINT_TESTCASES)
|
||||
def test_varuint_to_bytes(val, encoded):
|
||||
assert util.varuint_to_bytes(val) == encoded
|
||||
|
||||
|
||||
@pytest.mark.parametrize("val, encoded", VARUINT_TESTCASES)
|
||||
def test_bytes_to_varuint(val, encoded):
|
||||
assert util.bytes_to_varuint(encoded) == val
|
||||
Loading…
Reference in New Issue
Block a user