200 lines
7.6 KiB
Python
200 lines
7.6 KiB
Python
"""Tests hors ligne du flux de configuration Arkteos."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
from homeassistant.config_entries import SOURCE_RECONFIGURE, SOURCE_USER
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.arkteos.const import CONF_HOST, CONF_PORT, DEFAULT_PORT, DOMAIN
|
|
|
|
|
|
pytestmark = pytest.mark.usefixtures("enable_custom_integrations")
|
|
|
|
|
|
class FakeClient:
|
|
"""Client factice pour valider le flux sans ouvrir de socket."""
|
|
|
|
connected_result = True
|
|
valid_frame = True
|
|
instances: list["FakeClient"] = []
|
|
|
|
def __init__(self, _host: str, _port: int) -> None:
|
|
self.connected = False
|
|
self.last_valid_frame = None
|
|
self.last_error = None
|
|
self.started = False
|
|
self.stopped = False
|
|
self.__class__.instances.append(self)
|
|
|
|
async def start(self) -> None:
|
|
self.started = True
|
|
self.connected = self.__class__.connected_result
|
|
if self.__class__.valid_frame:
|
|
self.last_valid_frame = object()
|
|
|
|
async def stop(self) -> None:
|
|
self.stopped = True
|
|
|
|
async def wait_until_connected(self, _timeout: float | None = None) -> bool:
|
|
return self.__class__.connected_result
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_client(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from custom_components.arkteos import config_flow
|
|
|
|
FakeClient.connected_result = True
|
|
FakeClient.valid_frame = True
|
|
FakeClient.instances.clear()
|
|
monkeypatch.setattr(config_flow, "ArkteosClient", FakeClient)
|
|
|
|
|
|
async def _start_user_flow(hass):
|
|
return await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
|
|
|
|
|
async def test_user_form_is_shown(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
|
|
|
|
async def test_default_port_is_9641(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
assert result["data_schema"]({CONF_HOST: "proxy.local"})[CONF_PORT] == DEFAULT_PORT
|
|
|
|
|
|
async def test_empty_host_is_refused(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "", CONF_PORT: DEFAULT_PORT}
|
|
)
|
|
assert result["errors"][CONF_HOST] == "invalid_host"
|
|
|
|
|
|
async def test_port_zero_is_refused(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: 0}
|
|
)
|
|
assert result["errors"][CONF_PORT] == "invalid_port"
|
|
|
|
|
|
async def test_port_above_range_is_refused(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: 65536}
|
|
)
|
|
assert result["errors"][CONF_PORT] == "invalid_port"
|
|
|
|
|
|
async def test_cannot_connect_is_reported(hass) -> None:
|
|
FakeClient.connected_result = False
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
|
)
|
|
assert result["errors"]["base"] == "cannot_connect"
|
|
assert FakeClient.instances[-1].stopped
|
|
|
|
|
|
async def test_validation_timeout_is_reported(hass, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from custom_components.arkteos import config_flow
|
|
|
|
FakeClient.valid_frame = False
|
|
monkeypatch.setattr(config_flow, "VALIDATION_TIMEOUT", 0.001)
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
|
)
|
|
assert result["errors"]["base"] == "timeout"
|
|
assert FakeClient.instances[-1].stopped
|
|
|
|
|
|
async def test_connection_and_valid_frame_create_entry(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Arkteos proxy.local:9641"
|
|
|
|
|
|
async def test_created_entry_contains_host_and_port(hass) -> None:
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: 9999}
|
|
)
|
|
assert result["data"] == {CONF_HOST: "proxy.local", CONF_PORT: 9999}
|
|
|
|
|
|
async def test_duplicate_host_and_port_are_refused(hass) -> None:
|
|
entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT})
|
|
entry.add_to_hass(hass)
|
|
result = await _start_user_flow(hass)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
|
|
|
|
async def test_reconfigure_succeeds(hass, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from custom_components.arkteos import config_flow
|
|
|
|
entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "old.local", CONF_PORT: DEFAULT_PORT})
|
|
entry.add_to_hass(hass)
|
|
update_entry = Mock(wraps=hass.config_entries.async_update_entry)
|
|
reload_entry = AsyncMock(return_value=True)
|
|
|
|
def create_fake_client(host: str, port: int) -> FakeClient:
|
|
client = FakeClient(host, port)
|
|
client.wait_until_connected = AsyncMock(wraps=client.wait_until_connected)
|
|
return client
|
|
|
|
client_factory = Mock(side_effect=create_fake_client)
|
|
monkeypatch.setattr(hass.config_entries, "async_update_entry", update_entry)
|
|
monkeypatch.setattr(hass.config_entries, "async_reload", reload_entry)
|
|
monkeypatch.setattr(config_flow, "ArkteosClient", client_factory)
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_RECONFIGURE, "entry_id": entry.entry_id}
|
|
)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "new.local", CONF_PORT: 9999}
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "reconfigure_successful"
|
|
assert update_entry.call_count == 1
|
|
_, update_kwargs = update_entry.call_args
|
|
assert update_kwargs["entry"] is entry
|
|
assert update_kwargs["data"] == {CONF_HOST: "new.local", CONF_PORT: 9999}
|
|
assert update_kwargs["title"] == "Arkteos new.local:9999"
|
|
reload_entry.assert_awaited_once_with(entry.entry_id)
|
|
client_factory.assert_called_once_with("new.local", 9999)
|
|
assert len(FakeClient.instances) == 1
|
|
client = FakeClient.instances[0]
|
|
assert client.started
|
|
client.wait_until_connected.assert_awaited_once()
|
|
assert client.last_valid_frame is not None
|
|
assert client.stopped
|
|
assert not any(
|
|
task.get_name() == "arkteos-client" and not task.done()
|
|
for task in __import__("asyncio").all_tasks()
|
|
)
|
|
|
|
|
|
async def test_invalid_reconfigure_is_refused(hass) -> None:
|
|
entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "old.local", CONF_PORT: DEFAULT_PORT})
|
|
entry.add_to_hass(hass)
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_RECONFIGURE, "entry_id": entry.entry_id}
|
|
)
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={CONF_HOST: "", CONF_PORT: DEFAULT_PORT}
|
|
)
|
|
assert result["errors"][CONF_HOST] == "invalid_host"
|