Make ps4 config flow tests robust (#34749)

This commit is contained in:
Martin Hjelmare
2020-04-27 11:36:24 +02:00
committed by GitHub
parent bec8781558
commit e4590539e1

View File

@ -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"}