Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-14 23:28:42 +00:00
Code Activity

Replace repeated patches in config_flow_test with fixtures for Squeezebox (#153032)

Browse Source
This commit is contained in:
peteS-UK
2026-02-13 11:00:24 +00:00
committed by GitHub
parent 54b0393ebe
commit 44d5ecc926
4 changed files with 405 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
tests/components/squeezebox/conftest.py
View File

@@ -14,6 +14,9 @@ from homeassistant.components.squeezebox.browse_media import (
SQUEEZEBOX_ID_BY_TYPE,
)
from homeassistant.components.squeezebox.const import (
CONF_HTTPS,
CONF_VOLUME_STEP,
DOMAIN,
STATUS_QUERY_LIBRARYNAME,
STATUS_QUERY_MAC,
STATUS_QUERY_UUID,
@@ -32,15 +35,17 @@ from homeassistant.components.squeezebox.const import (
)
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from tests.common import MockConfigEntry
CONF_VOLUME_STEP = "volume_step"
TEST_VOLUME_STEP = 10
VOLUME_STEP = 10
BROWSE_LIMIT = 10
TEST_HOST = "1.2.3.4"
TEST_PORT = "9000"
TEST_USE_HTTPS = False
HOST = "1.1.1.1"
PORT = 9000
SERVER_UUIDS = [
"12345678-1234-1234-1234-123456789012",
"87654321-4321-4321-4321-210987654321",
@@ -61,7 +66,7 @@ FAKE_VALID_ITEM_ID = "1234"
FAKE_INVALID_ITEM_ID = "4321"
FAKE_IP = "42.42.42.42"
FAKE_UUID = "deadbeefdeadbeefbeefdeafbddeef42"
FAKE_PORT = 9000
FAKE_VERSION = "42.0"
@@ -90,7 +95,7 @@ FAKE_QUERY_RESPONSE = {
"modelname": "SqueezeLite-HA-Addon",
"playerindex": "status",
"model": "squeezelite",
"uuid": FAKE_UUID,
"uuid": SERVER_UUIDS[0],
"canpoweroff": 1,
"ip": "192.168.78.86:57700",
"displaytype": "none",
@@ -114,6 +119,76 @@ def mock_setup_entry() -> Generator[AsyncMock]:
yield mock_setup_entry
@pytest.fixture
def mock_config_entry():
"""Fixture that returns a mock config entry with UUID."""
return MockConfigEntry(
domain=DOMAIN,
unique_id=SERVER_UUIDS[0],
data={CONF_HOST: HOST, CONF_PORT: PORT, CONF_HTTPS: False},
)
@pytest.fixture
def mock_server():
"""Fixture to mock pysqueezebox.Server per test run."""
# Patch without autospec, so we can add arbitrary attributes
with patch("homeassistant.components.squeezebox.config_flow.Server") as server_cls:
server_mock = server_cls.return_value
# async methods
server_mock.async_query = AsyncMock()
yield server_mock
@pytest.fixture
def mock_discover_success():
"""Fixture to simulate successful async_discover."""
async def _mock_discover(callback):
class DummyServer:
host = "1.1.1.1"
port = 9000
uuid = SERVER_UUIDS[0] # Ensure UUID is defined or imported
callback(DummyServer())
return [DummyServer()]
return _mock_discover
@pytest.fixture
def mock_discover_failure():
"""Simulate failed discovery without raising unhandled exceptions."""
async def _failed_discover(callback):
# Simulate no servers found, no callback triggered
return []
return _failed_discover
@pytest.fixture
def patch_discover():
"""Patch the async_discover function to prevent actual network calls."""
async def _mock_discover(callback):
return []
with patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
side_effect=_mock_discover,
):
yield
@pytest.fixture
def dhcp_info():
"""Fixture for DHCP discovery data."""
return DhcpServiceInfo(ip=HOST, macaddress="aabbccddeeff", hostname="any")
@pytest.fixture
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Add the squeezebox mock config entry to hass."""
@@ -121,12 +196,12 @@ def config_entry(hass: HomeAssistant) -> MockConfigEntry:
domain=const.DOMAIN,
unique_id=SERVER_UUIDS[0],
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
CONF_HOST: HOST,
CONF_PORT: PORT,
const.CONF_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_VOLUME_STEP: TEST_VOLUME_STEP,
CONF_VOLUME_STEP: VOLUME_STEP,
},
)
config_entry.add_to_hass(hass)

2
tests/components/squeezebox/snapshots/test_init.ambr
View File

@@ -65,7 +65,7 @@
'manufacturer': 'https://lyrion.org/ / Ralph Irving & Adrian Smith',
'model': 'Lyrion Music Server/SqueezeLite',
'model_id': 'LMS',
'name': '1.2.3.4',
'name': '1.1.1.1',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,

880
tests/components/squeezebox/test_config_flow.py
View File

@@ -3,7 +3,6 @@
from http import HTTPStatus
from unittest.mock import patch
from pysqueezebox import Server
import pytest
from homeassistant import config_entries
@@ -20,81 +19,21 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .conftest import BROWSE_LIMIT, HOST, PORT, SERVER_UUIDS, VOLUME_STEP
from tests.common import MockConfigEntry
HOST = "1.1.1.1"
HOST2 = "2.2.2.2"
PORT = 9000
UUID = "test-uuid"
UNKNOWN_ERROR = "1234"
BROWSE_LIMIT = 10
VOLUME_STEP = 1
USER_INPUT = {
CONF_HOST: HOST,
}
async def mock_discover(_discovery_callback):
"""Mock discovering a Logitech Media Server."""
_discovery_callback(Server(None, HOST, PORT, uuid=UUID))
async def mock_failed_discover(_discovery_callback):
"""Mock unsuccessful discovery by doing nothing."""
async def patch_async_query_unauthorized(self, *args):
"""Mock an unauthorized query."""
self.http_status = HTTPStatus.UNAUTHORIZED
return False
async def test_user_form(hass: HomeAssistant) -> None:
"""Test user-initiated flow, including discovery and the edit step."""
with (
patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
),
patch(
"homeassistant.components.squeezebox.async_setup_entry",
return_value=True,
) as mock_setup_entry,
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover,
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
assert CONF_HOST in result["data_schema"].schema
for key in result["data_schema"].schema:
if key == CONF_HOST:
assert key.description == {"suggested_value": HOST}
# test the edit step
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
EDIT_INPUT = {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
@pytest.mark.usefixtures("mock_setup_entry")
@@ -108,7 +47,7 @@ async def test_options_form(hass: HomeAssistant) -> None:
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
unique_id=UUID,
unique_id=SERVER_UUIDS[0],
domain=DOMAIN,
options={CONF_BROWSE_LIMIT: 1000, CONF_VOLUME_STEP: 5},
)
@@ -137,252 +76,198 @@ async def test_options_form(hass: HomeAssistant) -> None:
}
async def test_user_form_timeout(hass: HomeAssistant) -> None:
"""Test we handle server search timeout and allow manual entry."""
# First flow: simulate timeout
async def test_user_flow_duplicate_entry(
hass: HomeAssistant,
mock_server,
mock_setup_entry,
mock_config_entry,
mock_discover_success,
) -> None:
"""Test user-initiated flow with existing entry (duplicate)."""
entry = mock_config_entry
entry.add_to_hass(hass)
mock_server.async_query.side_effect = query_success
mock_server.http_status = HTTPStatus.OK
with (
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_failed_discover,
mock_discover_success,
),
patch("homeassistant.components.squeezebox.config_flow.TIMEOUT", 0.1),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_server_found"}
# Second flow: simulate successful discovery
with (
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover,
),
patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
),
patch(
"homeassistant.components.squeezebox.async_setup_entry",
return_value=True,
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "no_server_found"}
async def test_user_form_duplicate(hass: HomeAssistant) -> None:
"""Test duplicate discovered servers are skipped."""
with (
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover,
),
patch("homeassistant.components.squeezebox.config_flow.TIMEOUT", 0.1),
patch(
"homeassistant.components.squeezebox.async_setup_entry",
return_value=True,
),
):
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=UUID,
data={CONF_HOST: HOST, CONF_PORT: PORT, CONF_HTTPS: False},
)
entry.add_to_hass(hass)
@pytest.mark.parametrize(
("discover_fixture", "expect_error", "expect_entry"),
[
("mock_discover_success", None, True),
("mock_discover_failure", "no_server_found", False),
],
)
async def test_user_flow_discovery_variants(
hass: HomeAssistant,
mock_server,
mock_setup_entry,
mock_discover_success,
mock_discover_failure,
discover_fixture,
expect_error,
expect_entry,
) -> None:
"""Test user-initiated flow variants: normal discovery and timeout."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_server_found"}
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
async def patch_async_query(self, *args):
self.http_status = HTTPStatus.UNAUTHORIZED
return False
with (
patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
),
patch(
"homeassistant.components.squeezebox.async_setup_entry",
return_value=True,
),
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover,
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
with patch(
"homeassistant.components.squeezebox.config_flow.Server.async_query",
new=patch_async_query,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
async def test_form_validate_exception(hass: HomeAssistant) -> None:
"""Test we handle exception."""
with (
patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
),
patch(
"homeassistant.components.squeezebox.async_setup_entry",
return_value=True,
),
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover,
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
with patch(
"homeassistant.components.squeezebox.config_flow.Server.async_query",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error, then succeed after retry."""
# Start the flow
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "edit"}
discover_func = (
mock_discover_success
if discover_fixture == "mock_discover_success"
else mock_discover_failure
)
assert result["type"] is FlowResultType.FORM
# First attempt: simulate cannot connect
with patch(
"pysqueezebox.Server.async_query",
return_value=False,
mock_server.async_query.side_effect = lambda *args, **kwargs: {
"uuid": SERVER_UUIDS[0]
}
mock_server.http_status = HTTPStatus.OK
with (
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
discover_func,
),
patch("homeassistant.components.squeezebox.config_flow.TIMEOUT", 0.1),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
},
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
# We should still be in a form, with an error
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
if expect_error:
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": expect_error}
return
# Second attempt: simulate a successful connection
with patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "edit"
assert CONF_HOST in result["data_schema"].schema
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], EDIT_INPUT
)
if expect_entry:
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == HOST
assert result2["data"] == EDIT_INPUT
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def query_success(*args, **kwargs):
"""Simulate successful query returning UUID."""
return {"uuid": SERVER_UUIDS[0]}
async def query_cannot_connect(*args, **kwargs):
"""Simulate connection failure."""
return False # Simulate failure; set status separately
async def query_unauthorized(*args, **kwargs):
"""Simulate unauthorized access."""
return False # Simulate failure; set status separately
class SqueezeError(Exception):
"""Custom exception to simulate unexpected query failure."""
async def query_exception(*args, **kwargs):
"""Simulate unexpected exception."""
raise SqueezeError("Unexpected error")
@pytest.mark.parametrize(
("discovery_data", "query_behavior", "http_status", "expect_error"),
[
(
{CONF_HOST: HOST, CONF_PORT: PORT, "uuid": SERVER_UUIDS[0]},
query_success,
HTTPStatus.OK,
None,
), # UUID present, success
(
{CONF_HOST: HOST, CONF_PORT: PORT, CONF_HTTPS: False},
query_unauthorized,
HTTPStatus.UNAUTHORIZED,
"invalid_auth",
), # No UUID, unauthorized
(
{CONF_HOST: HOST, CONF_PORT: PORT, CONF_HTTPS: False},
query_cannot_connect,
HTTPStatus.BAD_GATEWAY,
"cannot_connect",
), # No UUID, connection failure
(
{CONF_HOST: HOST, CONF_PORT: PORT, CONF_HTTPS: False},
query_exception,
None,
"unknown",
), # No UUID, unexpected exception
],
)
async def test_discovery_flow_variants(
hass: HomeAssistant,
mock_server,
mock_setup_entry,
discovery_data,
query_behavior,
http_status,
expect_error,
) -> None:
"""Test integration discovery flow with and without UUID."""
# Inject behavior into mock_server
mock_server.async_query.side_effect = query_behavior
mock_server.http_status = http_status
# Start flow
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data=discovery_data,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "edit"
# First configure attempt
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
if expect_error:
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": expect_error}
# Recovery attempt
mock_server.async_query.side_effect = query_success
mock_server.http_status = HTTPStatus.OK
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
@@ -391,17 +276,136 @@ async def test_form_cannot_connect(hass: HomeAssistant) -> None:
CONF_HTTPS: False,
},
)
result = result3
else:
result = result2
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST # the flow uses host as title
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["context"]["unique_id"] == UUID
# Final assertions
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["context"]["unique_id"] == SERVER_UUIDS[0]
async def test_dhcp_discovery_flow_success(
hass: HomeAssistant,
mock_server,
mock_setup_entry,
mock_discover_success,
dhcp_info,
) -> None:
"""Test DHCP discovery flow with successful discovery and query."""
# Inject successful query behavior
mock_server.async_query.side_effect = query_success
mock_server.http_status = HTTPStatus.OK
with (
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover_success,
),
patch("homeassistant.components.squeezebox.config_flow.TIMEOUT", 0.1),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp_info,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "edit"
# Final configure step
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], EDIT_INPUT
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "1.1.1.1"
assert result2["data"] == EDIT_INPUT
async def test_dhcp_discovery_existing_player(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
dhcp_info: DhcpServiceInfo,
) -> None:
"""Test that we properly ignore known players during DHCP discovery."""
# Register a squeezebox media_player entity with the same MAC unique_id
entity_registry.async_get_or_create(
domain="media_player",
platform=DOMAIN,
unique_id=format_mac("aabbccddeeff"),
)
# Fire DHCP discovery for the same MAC
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp_info,
)
# Flow should abort because the player is already known
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("query_behavior", "http_status", "expected_error"),
[
(query_unauthorized, HTTPStatus.UNAUTHORIZED, "invalid_auth"),
(query_cannot_connect, HTTPStatus.BAD_GATEWAY, "cannot_connect"),
(query_exception, None, "unknown"),
],
)
async def test_flow_errors_and_recovery(
hass: HomeAssistant,
mock_setup_entry,
mock_server,
query_behavior,
http_status,
expected_error,
patch_discover,
) -> None:
"""Test config flow error handling and recovery."""
# Start flow
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
assert result["step_id"] == "edit"
# Inject error
mock_server.async_query.side_effect = query_behavior
mock_server.http_status = http_status
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], EDIT_INPUT
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": expected_error}
# Recover
mock_server.async_query.side_effect = None
mock_server.async_query.return_value = {"uuid": SERVER_UUIDS[0]}
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], EDIT_INPUT
)
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == HOST
assert result3["data"] == EDIT_INPUT
async def test_form_missing_uuid(hass: HomeAssistant) -> None:
@@ -435,7 +439,7 @@ async def test_form_missing_uuid(hass: HomeAssistant) -> None:
# Second attempt: simulate a successful connection
with patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
return_value={"uuid": SERVER_UUIDS[0]},
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -457,254 +461,4 @@ async def test_form_missing_uuid(hass: HomeAssistant) -> None:
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["context"]["unique_id"] == UUID
async def test_discovery(hass: HomeAssistant) -> None:
"""Test handling of discovered server, then completing the flow."""
# Initial discovery: server responds with a uuid
with patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={CONF_HOST: HOST, CONF_PORT: PORT, "uuid": UUID},
)
# Discovery puts us into the edit step
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
# Complete the edit step with user input
with patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
# Flow should now complete with a config entry
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["context"]["unique_id"] == UUID
async def test_discovery_no_uuid(hass: HomeAssistant) -> None:
"""Test discovery without uuid first fails, then succeeds when uuid is available."""
# Initial discovery: no uuid returned
with patch(
"pysqueezebox.Server.async_query",
new=patch_async_query_unauthorized,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={CONF_HOST: HOST, CONF_PORT: PORT, CONF_HTTPS: False},
)
# Flow shows the edit form
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
# First attempt to complete: still no uuid → error on the form
with patch(
"pysqueezebox.Server.async_query",
new=patch_async_query_unauthorized,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
# Second attempt: now the server responds with a uuid
with patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
# Flow should now complete successfully
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["context"]["unique_id"] == UUID
async def test_dhcp_discovery(hass: HomeAssistant) -> None:
"""Test we can process discovery from dhcp and complete the flow."""
with (
patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
),
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_discover,
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DhcpServiceInfo(
ip=HOST,
macaddress="aabbccddeeff",
hostname="any",
),
)
# DHCP discovery puts us into the edit step
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
# Complete the edit step with user input
with patch(
"pysqueezebox.Server.async_query",
return_value={"uuid": UUID},
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
# Flow should now complete with a config entry
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
}
assert result["context"]["unique_id"] == UUID
async def test_dhcp_discovery_no_server_found(hass: HomeAssistant) -> None:
"""Test we can handle dhcp discovery when no server is found."""
with (
patch(
"homeassistant.components.squeezebox.config_flow.async_discover",
mock_failed_discover,
),
patch("homeassistant.components.squeezebox.config_flow.TIMEOUT", 0.1),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DhcpServiceInfo(
ip=HOST,
macaddress="aabbccddeeff",
hostname="any",
),
)
# First step: user form with only host
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
# Provide just the host to move into edit step
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: HOST},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
# Now try to complete the edit step with full schema
with patch(
"pysqueezebox.Server.async_query",
return_value=None,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: HOST,
CONF_PORT: PORT,
CONF_USERNAME: "",
CONF_PASSWORD: "",
CONF_HTTPS: False,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "edit"
assert result["errors"] == {"base": "cannot_connect"}
async def test_dhcp_discovery_existing_player(
hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None:
"""Test that we properly ignore known players during dhcp discover."""
# Register a squeezebox media_player entity with the same MAC unique_id
entity_registry.async_get_or_create(
domain="media_player",
platform=DOMAIN,
unique_id=format_mac("aabbccddeeff"),
)
# Now fire a DHCP discovery for the same MAC
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DhcpServiceInfo(
ip="1.1.1.1",
macaddress="aabbccddeeff",
hostname="any",
),
)
# Because the player is already known, the flow should abort
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert result["context"]["unique_id"] == SERVER_UUIDS[0]

6
tests/components/squeezebox/test_media_player.py
View File

@@ -72,7 +72,7 @@ from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity_registry import EntityRegistry
from homeassistant.util.dt import utcnow
from .conftest import FAKE_VALID_ITEM_ID, TEST_MAC, TEST_VOLUME_STEP
from .conftest import FAKE_VALID_ITEM_ID, TEST_MAC, VOLUME_STEP
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
@@ -218,7 +218,7 @@ async def test_squeezebox_volume_up(
blocking=True,
)
configured_player.async_set_volume.assert_called_once_with(
str(configured_player.volume + TEST_VOLUME_STEP)
str(configured_player.volume + VOLUME_STEP)
)
@@ -234,7 +234,7 @@ async def test_squeezebox_volume_down(
blocking=True,
)
configured_player.async_set_volume.assert_called_once_with(
str(configured_player.volume - TEST_VOLUME_STEP)
str(configured_player.volume - VOLUME_STEP)
)