diff --git a/tests/components/squeezebox/conftest.py b/tests/components/squeezebox/conftest.py index e82d370049e..2b72ea46f1a 100644 --- a/tests/components/squeezebox/conftest.py +++ b/tests/components/squeezebox/conftest.py @@ -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) diff --git a/tests/components/squeezebox/snapshots/test_init.ambr b/tests/components/squeezebox/snapshots/test_init.ambr index afd90d026de..03678ef4ff8 100644 --- a/tests/components/squeezebox/snapshots/test_init.ambr +++ b/tests/components/squeezebox/snapshots/test_init.ambr @@ -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': , 'serial_number': None, diff --git a/tests/components/squeezebox/test_config_flow.py b/tests/components/squeezebox/test_config_flow.py index 1b0aa98996a..7ce86d7d24d 100644 --- a/tests/components/squeezebox/test_config_flow.py +++ b/tests/components/squeezebox/test_config_flow.py @@ -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] diff --git a/tests/components/squeezebox/test_media_player.py b/tests/components/squeezebox/test_media_player.py index 804478dce10..b0d8499312f 100644 --- a/tests/components/squeezebox/test_media_player.py +++ b/tests/components/squeezebox/test_media_player.py @@ -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) )