jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-18 10:47:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use fixtures in UniFi switch tests (#118831)

Browse Source
This commit is contained in:
Robert Svensson 2024-06-04 21:01:03 +02:00 committed by GitHub
parent c2e245f9d4
commit b4f6325278
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 211 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

413
tests/components/unifi/test_switch.py
View File

@ -35,9 +35,9 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_registry import RegistryEntryDisabler from homeassistant.helpers.entity_registry import RegistryEntryDisabler
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
from .test_hub import CONTROLLER_HOST, ENTRY_CONFIG, SITE, setup_unifi_integration from .test_hub import CONTROLLER_HOST
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import async_fire_time_changed
from tests.test_util.aiohttp import AiohttpClientMocker from tests.test_util.aiohttp import AiohttpClientMocker
CLIENT_1 = { CLIENT_1 = {
@ -760,77 +760,50 @@ WLAN = {
} }
async def test_no_clients( @pytest.mark.parametrize("client_payload", [[CONTROLLER_HOST]])
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker @pytest.mark.parametrize("device_payload", [[DEVICE_1]])
) -> None: @pytest.mark.usefixtures("config_entry_setup")
"""Test the update_clients function when no clients are found.""" async def test_hub_not_client(hass: HomeAssistant) -> None:
await setup_unifi_integration( """Test that the cloud key doesn't become a switch."""
hass, assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
aioclient_mock, assert hass.states.get("switch.cloud_key") is None
options={
@pytest.mark.parametrize("client_payload", [[CLIENT_1]])
@pytest.mark.parametrize("device_payload", [[DEVICE_1]])
@pytest.mark.parametrize(
"site_payload",
[[{"desc": "Site name", "name": "site_id", "role": "not admin", "_id": "1"}]],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_not_admin(hass: HomeAssistant) -> None:
"""Test that switch platform only work on an admin account."""
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
@pytest.mark.parametrize(
"config_entry_options",
[
{
CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]],
CONF_TRACK_CLIENTS: False, CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False, CONF_TRACK_DEVICES: False,
CONF_DPI_RESTRICTIONS: False, }
}, ],
) )
@pytest.mark.parametrize("client_payload", [[CLIENT_4]])
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 @pytest.mark.parametrize("clients_all_payload", [[BLOCKED, UNBLOCKED, CLIENT_1]])
@pytest.mark.parametrize("dpi_app_payload", [DPI_APPS])
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
async def test_hub_not_client( @pytest.mark.usefixtures("config_entry_setup")
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test that the cloud key doesn't become a switch."""
await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_CLIENTS: False, CONF_TRACK_DEVICES: False},
clients_response=[CONTROLLER_HOST],
devices_response=[DEVICE_1],
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
cloudkey = hass.states.get("switch.cloud_key")
assert cloudkey is None
async def test_not_admin(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test that switch platform only work on an admin account."""
site = deepcopy(SITE)
site[0]["role"] = "not admin"
await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_CLIENTS: False, CONF_TRACK_DEVICES: False},
sites=site,
clients_response=[CLIENT_1],
devices_response=[DEVICE_1],
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_switches( async def test_switches(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker, aioclient_mock: AiohttpClientMocker,
config_entry_setup,
) -> None: ) -> None:
"""Test the update_items function with some clients.""" """Test the update_items function with some clients."""
config_entry = await setup_unifi_integration( config_entry = config_entry_setup
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]],
CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False,
},
clients_response=[CLIENT_4],
clients_all_response=[BLOCKED, UNBLOCKED, CLIENT_1],
dpigroup_response=DPI_GROUPS,
dpiapp_response=DPI_APPS,
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 3 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 3
@ -906,19 +879,15 @@ async def test_switches(
assert aioclient_mock.mock_calls[1][2] == {"enabled": True} assert aioclient_mock.mock_calls[1][2] == {"enabled": True}
async def test_remove_switches( @pytest.mark.parametrize(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_unifi_websocket "config_entry_options", [{CONF_BLOCK_CLIENT: [UNBLOCKED["mac"]]}]
) -> None: )
@pytest.mark.parametrize("client_payload", [[UNBLOCKED]])
@pytest.mark.parametrize("dpi_app_payload", [DPI_APPS])
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
@pytest.mark.usefixtures("config_entry_setup")
async def test_remove_switches(hass: HomeAssistant, mock_unifi_websocket) -> None:
"""Test the update_items function with some clients.""" """Test the update_items function with some clients."""
await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_BLOCK_CLIENT: [UNBLOCKED["mac"]]},
clients_response=[UNBLOCKED],
dpigroup_response=DPI_GROUPS,
dpiapp_response=DPI_APPS,
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
assert hass.states.get("switch.block_client_2") is not None assert hass.states.get("switch.block_client_2") is not None
@ -939,21 +908,26 @@ async def test_remove_switches(
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_block_switches( @pytest.mark.parametrize(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_unifi_websocket "config_entry_options",
) -> None: [
"""Test the update_items function with some clients.""" {
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]], CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]],
CONF_TRACK_CLIENTS: False, CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False, CONF_TRACK_DEVICES: False,
}, }
clients_response=[UNBLOCKED], ],
clients_all_response=[BLOCKED], )
) @pytest.mark.parametrize("client_payload", [[UNBLOCKED]])
@pytest.mark.parametrize("clients_all_payload", [[BLOCKED]])
async def test_block_switches(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
config_entry_setup,
) -> None:
"""Test the update_items function with some clients."""
config_entry = config_entry_setup
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
@ -1006,20 +980,13 @@ async def test_block_switches(
} }
@pytest.mark.parametrize("dpi_app_payload", [DPI_APPS])
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
@pytest.mark.usefixtures("config_entry_setup")
async def test_dpi_switches( async def test_dpi_switches(
hass: HomeAssistant, hass: HomeAssistant, mock_unifi_websocket, websocket_mock
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
websocket_mock,
) -> None: ) -> None:
"""Test the update_items function with some clients.""" """Test the update_items function with some clients."""
await setup_unifi_integration(
hass,
aioclient_mock,
dpigroup_response=DPI_GROUPS,
dpiapp_response=DPI_APPS,
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
dpi_switch = hass.states.get("switch.block_media_streaming") dpi_switch = hass.states.get("switch.block_media_streaming")
@ -1050,17 +1017,13 @@ async def test_dpi_switches(
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
@pytest.mark.parametrize("dpi_app_payload", [DPI_APPS])
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
@pytest.mark.usefixtures("config_entry_setup")
async def test_dpi_switches_add_second_app( async def test_dpi_switches_add_second_app(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_unifi_websocket hass: HomeAssistant, mock_unifi_websocket
) -> None: ) -> None:
"""Test the update_items function with some clients.""" """Test the update_items function with some clients."""
await setup_unifi_integration(
hass,
aioclient_mock,
dpigroup_response=DPI_GROUPS,
dpiapp_response=DPI_APPS,
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
assert hass.states.get("switch.block_media_streaming").state == STATE_ON assert hass.states.get("switch.block_media_streaming").state == STATE_ON
@ -1109,43 +1072,29 @@ async def test_dpi_switches_add_second_app(
@pytest.mark.parametrize( @pytest.mark.parametrize(
("entity_id", "test_data", "outlet_index", "expected_switches"), ("device_payload", "entity_id", "outlet_index", "expected_switches"),
[ [
( ([OUTLET_UP1], "plug_outlet_1", 1, 1),
"plug_outlet_1", ([PDU_DEVICE_1], "dummy_usp_pdu_pro_usb_outlet_1", 1, 2),
OUTLET_UP1, ([PDU_DEVICE_1], "dummy_usp_pdu_pro_outlet_2", 2, 2),
1,
1,
),
(
"dummy_usp_pdu_pro_usb_outlet_1",
PDU_DEVICE_1,
1,
2,
),
(
"dummy_usp_pdu_pro_outlet_2",
PDU_DEVICE_1,
2,
2,
),
], ],
) )
async def test_outlet_switches( async def test_outlet_switches(
hass: HomeAssistant, hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker, aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket, mock_unifi_websocket,
config_entry_setup,
device_payload,
websocket_mock, websocket_mock,
entity_id: str, entity_id: str,
test_data: any,
outlet_index: int, outlet_index: int,
expected_switches: int, expected_switches: int,
) -> None: ) -> None:
"""Test the outlet entities.""" """Test the outlet entities."""
config_entry = await setup_unifi_integration( config_entry = config_entry_setup
hass, aioclient_mock, devices_response=[test_data]
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == expected_switches assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == expected_switches
# Validate state object # Validate state object
switch_1 = hass.states.get(f"switch.{entity_id}") switch_1 = hass.states.get(f"switch.{entity_id}")
assert switch_1 is not None assert switch_1 is not None
@ -1153,14 +1102,14 @@ async def test_outlet_switches(
assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.OUTLET assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.OUTLET
# Update state object # Update state object
device_1 = deepcopy(test_data) device_1 = deepcopy(device_payload[0])
device_1["outlet_table"][outlet_index - 1]["relay_state"] = False device_1["outlet_table"][outlet_index - 1]["relay_state"] = False
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1) mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(f"switch.{entity_id}").state == STATE_OFF assert hass.states.get(f"switch.{entity_id}").state == STATE_OFF
# Turn off outlet # Turn off outlet
device_id = test_data["device_id"] device_id = device_payload[0]["device_id"]
aioclient_mock.clear_requests() aioclient_mock.clear_requests()
aioclient_mock.put( aioclient_mock.put(
f"https://{config_entry.data[CONF_HOST]}:1234" f"https://{config_entry.data[CONF_HOST]}:1234"
@ -1229,21 +1178,22 @@ async def test_outlet_switches(
assert hass.states.get(f"switch.{entity_id}") is None assert hass.states.get(f"switch.{entity_id}") is None
async def test_new_client_discovered_on_block_control( @pytest.mark.parametrize(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, mock_unifi_websocket "config_entry_options",
) -> None: [
"""Test if 2nd update has a new client.""" {
await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: [BLOCKED["mac"]], CONF_BLOCK_CLIENT: [BLOCKED["mac"]],
CONF_TRACK_CLIENTS: False, CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False, CONF_TRACK_DEVICES: False,
CONF_DPI_RESTRICTIONS: False, CONF_DPI_RESTRICTIONS: False,
}, }
) ],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_new_client_discovered_on_block_control(
hass: HomeAssistant, mock_unifi_websocket
) -> None:
"""Test if 2nd update has a new client."""
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
assert hass.states.get("switch.block_client_1") is None assert hass.states.get("switch.block_client_1") is None
@ -1254,22 +1204,27 @@ async def test_new_client_discovered_on_block_control(
assert hass.states.get("switch.block_client_1") is not None assert hass.states.get("switch.block_client_1") is not None
@pytest.mark.parametrize(
"config_entry_options", [{CONF_BLOCK_CLIENT: [BLOCKED["mac"]]}]
)
@pytest.mark.parametrize("clients_all_payload", [[BLOCKED, UNBLOCKED]])
async def test_option_block_clients( async def test_option_block_clients(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker hass: HomeAssistant, config_entry_setup, clients_all_payload
) -> None: ) -> None:
"""Test the changes to option reflects accordingly.""" """Test the changes to option reflects accordingly."""
config_entry = await setup_unifi_integration( config_entry = config_entry_setup
hass,
aioclient_mock,
options={CONF_BLOCK_CLIENT: [BLOCKED["mac"]]},
clients_all_response=[BLOCKED, UNBLOCKED],
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
# Add a second switch # Add a second switch
hass.config_entries.async_update_entry( hass.config_entries.async_update_entry(
config_entry, config_entry,
options={CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]]}, options={
CONF_BLOCK_CLIENT: [
clients_all_payload[0]["mac"],
clients_all_payload[1]["mac"],
]
},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
@ -1277,15 +1232,15 @@ async def test_option_block_clients(
# Remove the second switch again # Remove the second switch again
hass.config_entries.async_update_entry( hass.config_entries.async_update_entry(
config_entry, config_entry,
options={CONF_BLOCK_CLIENT: [BLOCKED["mac"]]}, options={CONF_BLOCK_CLIENT: [clients_all_payload[0]["mac"]]},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
# Enable one and remove another one # Enable one and remove the other one
hass.config_entries.async_update_entry( hass.config_entries.async_update_entry(
config_entry, config_entry,
options={CONF_BLOCK_CLIENT: [UNBLOCKED["mac"]]}, options={CONF_BLOCK_CLIENT: [clients_all_payload[1]["mac"]]},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
@ -1299,21 +1254,17 @@ async def test_option_block_clients(
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_option_remove_switches( @pytest.mark.parametrize(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker "config_entry_options",
) -> None: [{CONF_TRACK_CLIENTS: False, CONF_TRACK_DEVICES: False}],
)
@pytest.mark.parametrize("client_payload", [[CLIENT_1]])
@pytest.mark.parametrize("dpi_app_payload", [DPI_APPS])
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
async def test_option_remove_switches(hass: HomeAssistant, config_entry_setup) -> None:
"""Test removal of DPI switch when options updated.""" """Test removal of DPI switch when options updated."""
config_entry = await setup_unifi_integration( config_entry = config_entry_setup
hass,
aioclient_mock,
options={
CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False,
},
clients_response=[CLIENT_1],
dpigroup_response=DPI_GROUPS,
dpiapp_response=DPI_APPS,
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
# Disable DPI Switches # Disable DPI Switches
@ -1325,17 +1276,18 @@ async def test_option_remove_switches(
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
@pytest.mark.parametrize("device_payload", [[DEVICE_1]])
async def test_poe_port_switches( async def test_poe_port_switches(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker, aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket, mock_unifi_websocket,
websocket_mock, websocket_mock,
config_entry_setup,
device_payload,
) -> None: ) -> None:
"""Test the update_items function with some clients.""" """Test PoE port entities work."""
config_entry = await setup_unifi_integration( config_entry = config_entry_setup
hass, aioclient_mock, devices_response=[DEVICE_1]
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
@ -1350,7 +1302,7 @@ async def test_poe_port_switches(
entity_registry.async_update_entity( entity_registry.async_update_entity(
entity_id="switch.mock_name_port_2_poe", disabled_by=None entity_id="switch.mock_name_port_2_poe", disabled_by=None
) )
await hass.async_block_till_done() # await hass.async_block_till_done()
async_fire_time_changed( async_fire_time_changed(
hass, hass,
@ -1365,7 +1317,7 @@ async def test_poe_port_switches(
assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.OUTLET assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.OUTLET
# Update state object # Update state object
device_1 = deepcopy(DEVICE_1) device_1 = deepcopy(device_payload[0])
device_1["port_table"][0]["poe_mode"] = "off" device_1["port_table"][0]["poe_mode"] = "off"
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1) mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -1437,17 +1389,18 @@ async def test_poe_port_switches(
assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_OFF assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_OFF
@pytest.mark.parametrize("wlan_payload", [[WLAN]])
async def test_wlan_switches( async def test_wlan_switches(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker, aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket, mock_unifi_websocket,
websocket_mock, websocket_mock,
config_entry_setup,
wlan_payload,
) -> None: ) -> None:
"""Test control of UniFi WLAN availability.""" """Test control of UniFi WLAN availability."""
config_entry = await setup_unifi_integration( config_entry = config_entry_setup
hass, aioclient_mock, wlans_response=[WLAN]
)
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
@ -1462,7 +1415,7 @@ async def test_wlan_switches(
assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.SWITCH assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.SWITCH
# Update state object # Update state object
wlan = deepcopy(WLAN) wlan = deepcopy(wlan_payload[0])
wlan["enabled"] = False wlan["enabled"] = False
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan) mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -1472,7 +1425,7 @@ async def test_wlan_switches(
aioclient_mock.clear_requests() aioclient_mock.clear_requests()
aioclient_mock.put( aioclient_mock.put(
f"https://{config_entry.data[CONF_HOST]}:1234" f"https://{config_entry.data[CONF_HOST]}:1234"
f"/api/s/{config_entry.data[CONF_SITE_ID]}/rest/wlanconf/{WLAN['_id']}", f"/api/s/{config_entry.data[CONF_SITE_ID]}/rest/wlanconf/{wlan['_id']}",
) )
await hass.services.async_call( await hass.services.async_call(
@ -1505,15 +1458,11 @@ async def test_wlan_switches(
assert hass.states.get("switch.ssid_1").state == STATE_OFF assert hass.states.get("switch.ssid_1").state == STATE_OFF
async def test_port_forwarding_switches( @pytest.mark.parametrize(
hass: HomeAssistant, "port_forward_payload",
entity_registry: er.EntityRegistry, [
aioclient_mock: AiohttpClientMocker, [
mock_unifi_websocket, {
websocket_mock,
) -> None:
"""Test control of UniFi port forwarding."""
_data = {
"_id": "5a32aa4ee4b0412345678911", "_id": "5a32aa4ee4b0412345678911",
"dst_port": "12345", "dst_port": "12345",
"enabled": True, "enabled": True,
@ -1525,10 +1474,20 @@ async def test_port_forwarding_switches(
"site_id": "5a32aa4ee4b0412345678910", "site_id": "5a32aa4ee4b0412345678910",
"src": "any", "src": "any",
} }
config_entry = await setup_unifi_integration( ]
hass, aioclient_mock, port_forward_response=[_data.copy()] ],
) )
async def test_port_forwarding_switches(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
websocket_mock,
config_entry_setup,
port_forward_payload,
) -> None:
"""Test control of UniFi port forwarding."""
config_entry = config_entry_setup
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
ent_reg_entry = entity_registry.async_get("switch.unifi_network_plex") ent_reg_entry = entity_registry.async_get("switch.unifi_network_plex")
@ -1542,7 +1501,7 @@ async def test_port_forwarding_switches(
assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.SWITCH assert switch_1.attributes.get(ATTR_DEVICE_CLASS) == SwitchDeviceClass.SWITCH
# Update state object # Update state object
data = _data.copy() data = port_forward_payload[0].copy()
data["enabled"] = False data["enabled"] = False
mock_unifi_websocket(message=MessageKey.PORT_FORWARD_UPDATED, data=data) mock_unifi_websocket(message=MessageKey.PORT_FORWARD_UPDATED, data=data)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -1562,7 +1521,7 @@ async def test_port_forwarding_switches(
blocking=True, blocking=True,
) )
assert aioclient_mock.call_count == 1 assert aioclient_mock.call_count == 1
data = _data.copy() data = port_forward_payload[0].copy()
data["enabled"] = False data["enabled"] = False
assert aioclient_mock.mock_calls[0][2] == data assert aioclient_mock.mock_calls[0][2] == data
@ -1574,7 +1533,7 @@ async def test_port_forwarding_switches(
blocking=True, blocking=True,
) )
assert aioclient_mock.call_count == 2 assert aioclient_mock.call_count == 2
assert aioclient_mock.mock_calls[1][2] == _data assert aioclient_mock.mock_calls[1][2] == port_forward_payload[0]
# Availability signalling # Availability signalling
@ -1587,18 +1546,19 @@ async def test_port_forwarding_switches(
assert hass.states.get("switch.unifi_network_plex").state == STATE_OFF assert hass.states.get("switch.unifi_network_plex").state == STATE_OFF
# Remove entity on deleted message # Remove entity on deleted message
mock_unifi_websocket(message=MessageKey.PORT_FORWARD_DELETED, data=_data) mock_unifi_websocket(
message=MessageKey.PORT_FORWARD_DELETED, data=port_forward_payload[0]
)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_updating_unique_id( @pytest.mark.parametrize(
hass: HomeAssistant, "device_payload",
entity_registry: er.EntityRegistry, [
aioclient_mock: AiohttpClientMocker, [
) -> None: OUTLET_UP1,
"""Verify outlet control and poe control unique ID update works.""" {
poe_device = {
"board_rev": 3, "board_rev": 3,
"device_id": "mock-id", "device_id": "mock-id",
"ip": "10.0.0.1", "ip": "10.0.0.1",
@ -1625,34 +1585,35 @@ async def test_updating_unique_id(
"up": True, "up": True,
}, },
], ],
} },
]
config_entry = MockConfigEntry( ],
domain=UNIFI_DOMAIN, )
data=ENTRY_CONFIG, async def test_updating_unique_id(
source="test", hass: HomeAssistant,
options={}, entity_registry: er.EntityRegistry,
entry_id="1", config_entry_factory,
) config_entry,
device_payload,
) -> None:
"""Verify outlet control and poe control unique ID update works."""
entity_registry.async_get_or_create( entity_registry.async_get_or_create(
SWITCH_DOMAIN, SWITCH_DOMAIN,
UNIFI_DOMAIN, UNIFI_DOMAIN,
f'{poe_device["mac"]}-poe-1', f'{device_payload[0]["mac"]}-outlet-1',
suggested_object_id="switch_port_1_poe",
config_entry=config_entry,
)
entity_registry.async_get_or_create(
SWITCH_DOMAIN,
UNIFI_DOMAIN,
f'{OUTLET_UP1["mac"]}-outlet-1',
suggested_object_id="plug_outlet_1", suggested_object_id="plug_outlet_1",
config_entry=config_entry, config_entry=config_entry,
) )
entity_registry.async_get_or_create(
await setup_unifi_integration( SWITCH_DOMAIN,
hass, aioclient_mock, devices_response=[poe_device, OUTLET_UP1] UNIFI_DOMAIN,
f'{device_payload[1]["mac"]}-poe-1',
suggested_object_id="switch_port_1_poe",
config_entry=config_entry,
) )
await config_entry_factory()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2 assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
assert hass.states.get("switch.switch_port_1_poe")
assert hass.states.get("switch.plug_outlet_1") assert hass.states.get("switch.plug_outlet_1")
assert hass.states.get("switch.switch_port_1_poe")