Upstream/aioesphomeapi
1
0
Fork 0
mirror of https://github.com/esphome/aioesphomeapi.git synced 2025-02-01 23:22:27 +01:00
Code Issues Projects Releases Wiki Activity

Add test for connecting with a password (#709)

Browse Source
This commit is contained in:
J. Nick Koston 2023-11-25 08:59:36 -06:00 committed by GitHub
parent d592f1e7ab
commit b7e319b26a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 15 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
tests/conftest.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio import asyncio
import socket import socket
from dataclasses import replace
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest import pytest
@ -64,20 +65,6 @@ def connection_params() -> ConnectionParams:
return get_mock_connection_params() return get_mock_connection_params()
@pytest.fixture
def noise_connection_params() -> ConnectionParams:
return ConnectionParams(
address="fake.address",
port=6052,
password=None,
client_info="Tests client",
keepalive=KEEP_ALIVE_INTERVAL,
zeroconf_manager=ZeroconfManager(),
noise_psk="QRTIErOb/fcE9Ukd/5qA3RGYMn0Y+p06U58SCtOXvPc=",
expected_name="test",
)
def on_stop(expected_disconnect: bool) -> None: def on_stop(expected_disconnect: bool) -> None:
pass pass
@ -88,8 +75,17 @@ def conn(connection_params: ConnectionParams) -> APIConnection:
@pytest.fixture @pytest.fixture
def noise_conn(noise_connection_params: ConnectionParams) -> APIConnection: def conn_with_password(connection_params: ConnectionParams) -> APIConnection:
return PatchableAPIConnection(noise_connection_params, on_stop, True, None) connection_params = replace(connection_params, password="password")
return PatchableAPIConnection(connection_params, on_stop, True, None)
@pytest.fixture
def noise_conn(connection_params: ConnectionParams) -> APIConnection:
connection_params = replace(
connection_params, noise_psk="QRTIErOb/fcE9Ukd/5qA3RGYMn0Y+p06U58SCtOXvPc="
)
return PatchableAPIConnection(connection_params, on_stop, True, None)
@pytest_asyncio.fixture(name="plaintext_connect_task_no_login") @pytest_asyncio.fixture(name="plaintext_connect_task_no_login")
@ -118,7 +114,7 @@ async def plaintext_connect_task_no_login(
@pytest_asyncio.fixture(name="plaintext_connect_task_with_login") @pytest_asyncio.fixture(name="plaintext_connect_task_with_login")
async def plaintext_connect_task_with_login( async def plaintext_connect_task_with_login(
conn: APIConnection, resolve_host, socket_socket, event_loop conn_with_password: APIConnection, resolve_host, socket_socket, event_loop
) -> tuple[APIConnection, asyncio.Transport, APIPlaintextFrameHelper, asyncio.Task]: ) -> tuple[APIConnection, asyncio.Transport, APIPlaintextFrameHelper, asyncio.Task]:
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
protocol: APIPlaintextFrameHelper | None = None protocol: APIPlaintextFrameHelper | None = None
@ -135,9 +131,9 @@ async def plaintext_connect_task_with_login(
with patch.object(event_loop, "sock_connect"), patch.object( with patch.object(event_loop, "sock_connect"), patch.object(
loop, "create_connection", side_effect=_create_mock_transport_protocol loop, "create_connection", side_effect=_create_mock_transport_protocol
): ):
connect_task = asyncio.create_task(connect(conn, login=True)) connect_task = asyncio.create_task(connect(conn_with_password, login=True))
await connected.wait() await connected.wait()
yield conn, transport, protocol, connect_task yield conn_with_password, transport, protocol, connect_task
@pytest_asyncio.fixture(name="api_client") @pytest_asyncio.fixture(name="api_client")