"""Test NetwrokInterface API.""" from unittest.mock import AsyncMock, patch import pytest from supervisor.const import DOCKER_NETWORK, DOCKER_NETWORK_MASK from supervisor.coresys import CoreSys from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN @pytest.mark.asyncio async def test_api_network_info(api_client, coresys: CoreSys): """Test network manager api.""" resp = await api_client.get("/network/info") result = await resp.json() assert TEST_INTERFACE in ( inet["interface"] for inet in result["data"]["interfaces"] ) assert TEST_INTERFACE_WLAN in ( inet["interface"] for inet in result["data"]["interfaces"] ) for interface in result["data"]["interfaces"]: if interface["interface"] == TEST_INTERFACE: assert interface["primary"] assert interface["ipv4"]["gateway"] == "192.168.2.1" if interface["interface"] == TEST_INTERFACE_WLAN: assert not interface["primary"] assert interface["ipv4"] == { "address": [], "gateway": None, "method": "disabled", "nameservers": [], "ready": False, } assert interface["ipv6"] == { "address": [], "gateway": None, "method": "disabled", "nameservers": [], "ready": False, } assert result["data"]["docker"]["interface"] == DOCKER_NETWORK assert result["data"]["docker"]["address"] == str(DOCKER_NETWORK_MASK) assert result["data"]["docker"]["dns"] == str(coresys.docker.network.dns) assert result["data"]["docker"]["gateway"] == str(coresys.docker.network.gateway) @pytest.mark.asyncio async def test_api_network_interface_info(api_client): """Test network manager api.""" resp = await api_client.get(f"/network/interface/{TEST_INTERFACE}/info") result = await resp.json() assert result["data"]["ipv4"]["address"][-1] == "192.168.2.148/24" assert result["data"]["ipv4"]["gateway"] == "192.168.2.1" assert result["data"]["ipv4"]["nameservers"] == ["192.168.2.2"] assert result["data"]["ipv4"]["ready"] is True assert ( result["data"]["ipv6"]["address"][0] == "2a03:169:3df5:0:6be9:2588:b26a:a679/64" ) assert ( result["data"]["ipv6"]["address"][1] == "fd14:949b:c9cc:0:522b:8108:8ff8:cca3/64" ) assert result["data"]["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69" assert result["data"]["ipv6"]["nameservers"] == [ "2001:1620:2777:1::10", "2001:1620:2777:2::20", ] assert result["data"]["ipv6"]["ready"] is True assert result["data"]["interface"] == TEST_INTERFACE @pytest.mark.asyncio async def test_api_network_interface_info_default(api_client): """Test network manager default api.""" resp = await api_client.get("/network/interface/default/info") result = await resp.json() assert result["data"]["ipv4"]["address"][-1] == "192.168.2.148/24" assert result["data"]["ipv4"]["gateway"] == "192.168.2.1" assert result["data"]["ipv4"]["nameservers"] == ["192.168.2.2"] assert result["data"]["ipv4"]["ready"] is True assert ( result["data"]["ipv6"]["address"][0] == "2a03:169:3df5:0:6be9:2588:b26a:a679/64" ) assert ( result["data"]["ipv6"]["address"][1] == "fd14:949b:c9cc:0:522b:8108:8ff8:cca3/64" ) assert result["data"]["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69" assert result["data"]["ipv6"]["nameservers"] == [ "2001:1620:2777:1::10", "2001:1620:2777:2::20", ] assert result["data"]["ipv6"]["ready"] is True assert result["data"]["interface"] == TEST_INTERFACE @pytest.mark.asyncio async def test_api_network_interface_update( api_client, coresys: CoreSys, dbus: list[str] ): """Test network manager api.""" dbus.clear() resp = await api_client.post( f"/network/interface/{TEST_INTERFACE}/update", json={ "ipv4": { "method": "static", "nameservers": ["1.1.1.1"], "address": ["192.168.2.148/24"], "gateway": "192.168.1.1", } }, ) result = await resp.json() assert result["result"] == "ok" assert ( "/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.CheckConnectivity" in dbus ) @pytest.mark.asyncio async def test_api_network_interface_update_wifi(api_client): """Test network manager api.""" resp = await api_client.post( f"/network/interface/{TEST_INTERFACE_WLAN}/update", json={ "enabled": True, "ipv4": { "method": "static", "nameservers": ["1.1.1.1"], "address": ["192.168.2.148/24"], "gateway": "192.168.1.1", }, "wifi": {"ssid": "MY_TEST", "auth": "wpa-psk", "psk": "myWifiPassword"}, }, ) result = await resp.json() assert result["result"] == "ok" @pytest.mark.asyncio async def test_api_network_interface_update_remove(api_client): """Test network manager api.""" resp = await api_client.post( f"/network/interface/{TEST_INTERFACE}/update", json={"enabled": False}, ) result = await resp.json() assert result["result"] == "ok" @pytest.mark.asyncio async def test_api_network_interface_info_invalid(api_client): """Test network manager api.""" resp = await api_client.get("/network/interface/invalid/info") result = await resp.json() assert result["message"] assert result["result"] == "error" @pytest.mark.asyncio async def test_api_network_interface_update_invalid(api_client): """Test network manager api.""" resp = await api_client.post("/network/interface/invalid/update", json={}) result = await resp.json() assert result["message"] == "Interface invalid does not exist" resp = await api_client.post(f"/network/interface/{TEST_INTERFACE}/update", json={}) result = await resp.json() assert result["message"] == "You need to supply at least one option to update" resp = await api_client.post( f"/network/interface/{TEST_INTERFACE}/update", json={"ipv4": {"nameservers": "1.1.1.1"}}, ) result = await resp.json() assert ( result["message"] == "expected a list for dictionary value @ data['ipv4']['nameservers']. Got '1.1.1.1'" ) @pytest.mark.asyncio async def test_api_network_wireless_scan(api_client): """Test network manager api.""" with patch("asyncio.sleep", return_value=AsyncMock()): resp = await api_client.get( f"/network/interface/{TEST_INTERFACE_WLAN}/accesspoints" ) result = await resp.json() assert ["UPC4814466", "VQ@35(55720"] == [ ap["ssid"] for ap in result["data"]["accesspoints"] ] assert [47, 63] == [ap["signal"] for ap in result["data"]["accesspoints"]] @pytest.mark.asyncio async def test_api_network_reload(api_client, coresys, dbus: list[str]): """Test network manager reload api.""" dbus.clear() resp = await api_client.post("/network/reload") result = await resp.json() assert result["result"] == "ok" # Check that we forced NM to do an immediate connectivity check assert ( "/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.CheckConnectivity" in dbus ) async def test_api_network_vlan(api_client, coresys: CoreSys, dbus: list[str]): """Test creating a vlan.""" dbus.clear() resp = await api_client.post( f"/network/interface/{TEST_INTERFACE}/vlan/1", json={"ipv4": {"method": "auto"}} ) result = await resp.json() assert result["result"] == "ok" assert ( "/org/freedesktop/NetworkManager/Settings-org.freedesktop.NetworkManager.Settings.AddConnection" in dbus )