From e4590539e1775e4ce9e988df5059373c567349a2 Mon Sep 17 00:00:00 2001 From: Martin Hjelmare Date: Mon, 27 Apr 2020 11:36:24 +0200 Subject: [PATCH] Make ps4 config flow tests robust (#34749) --- tests/components/ps4/test_config_flow.py | 351 +++++++++++++++-------- 1 file changed, 232 insertions(+), 119 deletions(-) diff --git a/tests/components/ps4/test_config_flow.py b/tests/components/ps4/test_config_flow.py index 06f10da4f2b..2bad2d1e281 100644 --- a/tests/components/ps4/test_config_flow.py +++ b/tests/components/ps4/test_config_flow.py @@ -1,6 +1,7 @@ """Define tests for the PlayStation 4 config flow.""" from asynctest import patch from pyps4_2ndscreen.errors import CredentialTimeout +import pytest from homeassistant import data_entry_flow from homeassistant.components import ps4 @@ -73,22 +74,40 @@ MOCK_LOCATION = location.LocationInfo( ) +@pytest.fixture(name="location_info", autouse=True) +def location_info_fixture(): + """Mock location info.""" + with patch( + "homeassistant.components.ps4.config_flow.location.async_detect_location_info", + return_value=MOCK_LOCATION, + ): + yield + + +@pytest.fixture(name="ps4_setup", autouse=True) +def ps4_setup_fixture(): + """Patch ps4 setup entry.""" + with patch( + "homeassistant.components.ps4.async_setup_entry", return_value=True, + ): + yield + + async def test_full_flow_implementation(hass): """Test registering an implementation and flow works.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.location = MOCK_LOCATION - manager = hass.config_entries - # User Step Started, results in Step Creds with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): - result = await flow.async_step_user() + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "creds" # Step Creds results with form in Step Mode. with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): - result = await flow.async_step_creds({}) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "mode" @@ -96,7 +115,9 @@ async def test_full_flow_implementation(hass): with patch( "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}] ): - result = await flow.async_step_mode(MOCK_AUTO) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "link" @@ -104,44 +125,30 @@ async def test_full_flow_implementation(hass): with patch("pyps4_2ndscreen.Helper.link", return_value=(True, True)), patch( "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}] ): - result = await flow.async_step_link(MOCK_CONFIG) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_CONFIG + ) assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["data"][CONF_TOKEN] == MOCK_CREDS assert result["data"]["devices"] == [MOCK_DEVICE] assert result["title"] == MOCK_TITLE - await hass.async_block_till_done() - - # Add entry using result data. - mock_data = { - CONF_TOKEN: result["data"][CONF_TOKEN], - "devices": result["data"]["devices"], - } - entry = MockConfigEntry(domain=ps4.DOMAIN, data=mock_data) - entry.add_to_manager(manager) - - # Check if entry exists. - assert len(manager.async_entries()) == 1 - # Check if there is a device config in entry. - assert len(entry.data["devices"]) == 1 - async def test_multiple_flow_implementation(hass): """Test multiple device flows.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.location = MOCK_LOCATION - manager = hass.config_entries - # User Step Started, results in Step Creds with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): - result = await flow.async_step_user() + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "creds" # Step Creds results with form in Step Mode. with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): - result = await flow.async_step_creds({}) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "mode" @@ -150,7 +157,9 @@ async def test_multiple_flow_implementation(hass): "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}, {"host-ip": MOCK_HOST_ADDITIONAL}], ): - result = await flow.async_step_mode(MOCK_AUTO) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "link" @@ -159,26 +168,20 @@ async def test_multiple_flow_implementation(hass): "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}, {"host-ip": MOCK_HOST_ADDITIONAL}], ): - result = await flow.async_step_link(MOCK_CONFIG) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_CONFIG + ) assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["data"][CONF_TOKEN] == MOCK_CREDS assert result["data"]["devices"] == [MOCK_DEVICE] assert result["title"] == MOCK_TITLE - await hass.async_block_till_done() - - # Add entry using result data. - mock_data = { - CONF_TOKEN: result["data"][CONF_TOKEN], - "devices": result["data"]["devices"], - } - entry = MockConfigEntry(domain=ps4.DOMAIN, data=mock_data) - entry.add_to_manager(manager) - # Check if entry exists. - assert len(manager.async_entries()) == 1 + entries = hass.config_entries.async_entries() + assert len(entries) == 1 # Check if there is a device config in entry. - assert len(entry.data["devices"]) == 1 + entry_1 = entries[0] + assert len(entry_1.data["devices"]) == 1 # Test additional flow. @@ -187,13 +190,17 @@ async def test_multiple_flow_implementation(hass): "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}, {"host-ip": MOCK_HOST_ADDITIONAL}], ): - result = await flow.async_step_user() + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "creds" # Step Creds results with form in Step Mode. with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): - result = await flow.async_step_creds({}) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "mode" @@ -202,7 +209,9 @@ async def test_multiple_flow_implementation(hass): "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}, {"host-ip": MOCK_HOST_ADDITIONAL}], ): - result = await flow.async_step_mode(MOCK_AUTO) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "link" @@ -211,44 +220,40 @@ async def test_multiple_flow_implementation(hass): "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}, {"host-ip": MOCK_HOST_ADDITIONAL}], ), patch("pyps4_2ndscreen.Helper.link", return_value=(True, True)): - result = await flow.async_step_link(MOCK_CONFIG_ADDITIONAL) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_CONFIG_ADDITIONAL + ) assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["data"][CONF_TOKEN] == MOCK_CREDS assert len(result["data"]["devices"]) == 1 assert result["title"] == MOCK_TITLE - await hass.async_block_till_done() - - mock_data = { - CONF_TOKEN: result["data"][CONF_TOKEN], - "devices": result["data"]["devices"], - } - - # Update config entries with result data - entry = MockConfigEntry(domain=ps4.DOMAIN, data=mock_data) - entry.add_to_manager(manager) - manager.async_update_entry(entry) - # Check if there are 2 entries. - assert len(manager.async_entries()) == 2 - # Check if there is device config in entry. - assert len(entry.data["devices"]) == 1 + entries = hass.config_entries.async_entries() + assert len(entries) == 2 + # Check if there is device config in the last entry. + entry_2 = entries[-1] + assert len(entry_2.data["devices"]) == 1 + + # Check that entry 1 is different from entry 2. + assert entry_1 is not entry_2 async def test_port_bind_abort(hass): """Test that flow aborted when cannot bind to ports 987, 997.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - with patch("pyps4_2ndscreen.Helper.port_bind", return_value=MOCK_UDP_PORT): reason = "port_987_bind_error" - result = await flow.async_step_user(user_input=None) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["reason"] == reason with patch("pyps4_2ndscreen.Helper.port_bind", return_value=MOCK_TCP_PORT): reason = "port_997_bind_error" - result = await flow.async_step_user(user_input=None) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["reason"] == reason @@ -256,48 +261,69 @@ async def test_port_bind_abort(hass): async def test_duplicate_abort(hass): """Test that Flow aborts when found devices already configured.""" MockConfigEntry(domain=ps4.DOMAIN, data=MOCK_DATA).add_to_hass(hass) - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.creds = MOCK_CREDS + + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" + + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" with patch( "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}] ): - result = await flow.async_step_link(user_input=None) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["reason"] == "devices_configured" async def test_additional_device(hass): """Test that Flow can configure another device.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.creds = MOCK_CREDS - manager = hass.config_entries - # Mock existing entry. entry = MockConfigEntry(domain=ps4.DOMAIN, data=MOCK_DATA) - entry.add_to_manager(manager) - # Check that only 1 entry exists - assert len(manager.async_entries()) == 1 + entry.add_to_hass(hass) + + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" + + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" with patch( "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}, {"host-ip": MOCK_HOST_ADDITIONAL}], - ), patch("pyps4_2ndscreen.Helper.link", return_value=(True, True)): - result = await flow.async_step_link(MOCK_CONFIG_ADDITIONAL) + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) + + with patch("pyps4_2ndscreen.Helper.link", return_value=(True, True)): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_CONFIG_ADDITIONAL + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY assert result["data"][CONF_TOKEN] == MOCK_CREDS assert len(result["data"]["devices"]) == 1 assert result["title"] == MOCK_TITLE - # Add New Entry - entry = MockConfigEntry(domain=ps4.DOMAIN, data=MOCK_DATA) - entry.add_to_manager(manager) - - # Check that there are 2 entries - assert len(manager.async_entries()) == 2 - async def test_0_pin(hass): """Test Pin with leading '0' is passed correctly.""" @@ -337,63 +363,121 @@ async def test_0_pin(hass): async def test_no_devices_found_abort(hass): """Test that failure to find devices aborts flow.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" + + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" with patch("pyps4_2ndscreen.Helper.has_devices", return_value=[]): - result = await flow.async_step_link() + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["reason"] == "no_devices_found" async def test_manual_mode(hass): """Test host specified in manual mode is passed to Step Link.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.location = MOCK_LOCATION + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" + + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" # Step Mode with User Input: manual, results in Step Link. with patch( - "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": flow.m_device}] + "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}] ): - result = await flow.async_step_mode(MOCK_MANUAL) - assert flow.m_device == MOCK_HOST + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_MANUAL + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "link" async def test_credential_abort(hass): """Test that failure to get credentials aborts flow.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" with patch("pyps4_2ndscreen.Helper.get_creds", return_value=None): - result = await flow.async_step_creds({}) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["reason"] == "credential_error" async def test_credential_timeout(hass): """Test that Credential Timeout shows error.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" with patch("pyps4_2ndscreen.Helper.get_creds", side_effect=CredentialTimeout): - result = await flow.async_step_creds({}) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" assert result["errors"] == {"base": "credential_timeout"} async def test_wrong_pin_error(hass): """Test that incorrect pin throws an error.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.location = MOCK_LOCATION + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" - with patch("pyps4_2ndscreen.Helper.link", return_value=(True, False)), patch( + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" + + with patch( "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}] ): - result = await flow.async_step_link(MOCK_CONFIG) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) + + with patch("pyps4_2ndscreen.Helper.link", return_value=(True, False)): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_CONFIG + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "link" assert result["errors"] == {"base": "login_failed"} @@ -401,14 +485,31 @@ async def test_wrong_pin_error(hass): async def test_device_connection_error(hass): """Test that device not connected or on throws an error.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass - flow.location = MOCK_LOCATION + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" - with patch("pyps4_2ndscreen.Helper.link", return_value=(False, True)), patch( + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" + + with patch( "pyps4_2ndscreen.Helper.has_devices", return_value=[{"host-ip": MOCK_HOST}] ): - result = await flow.async_step_link(MOCK_CONFIG) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_AUTO + ) + + with patch("pyps4_2ndscreen.Helper.link", return_value=(False, True)): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_CONFIG + ) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "link" assert result["errors"] == {"base": "not_ready"} @@ -416,12 +517,24 @@ async def test_device_connection_error(hass): async def test_manual_mode_no_ip_error(hass): """Test no IP specified in manual mode throws an error.""" - flow = ps4.PlayStation4FlowHandler() - flow.hass = hass + with patch("pyps4_2ndscreen.Helper.port_bind", return_value=None): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "creds" - mock_input = {"Config Mode": "Manual Entry"} + with patch("pyps4_2ndscreen.Helper.get_creds", return_value=MOCK_CREDS): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["step_id"] == "mode" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={"Config Mode": "Manual Entry"} + ) - result = await flow.async_step_mode(mock_input) assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["step_id"] == "mode" assert result["errors"] == {CONF_IP_ADDRESS: "no_ipaddress"}