|
1 | 1 | """Test network manager.""" |
2 | 2 |
|
3 | 3 | import asyncio |
| 4 | +from collections.abc import AsyncGenerator |
4 | 5 | from ipaddress import IPv4Address, IPv6Address |
| 6 | +from typing import Any |
5 | 7 | from unittest.mock import AsyncMock, patch |
6 | 8 |
|
7 | 9 | from dbus_fast import DBusError, Variant |
|
10 | 12 | from supervisor.const import CoreState |
11 | 13 | from supervisor.coresys import CoreSys |
12 | 14 | from supervisor.dbus.const import InterfaceMethod |
| 15 | +from supervisor.dbus.network.connection import NetworkConnection |
13 | 16 | from supervisor.dbus.network.setting import ( |
14 | 17 | CONF_ATTR_802_WIRELESS_SECURITY, |
15 | 18 | CONF_ATTR_802_WIRELESS_SECURITY_PSK, |
16 | 19 | ) |
17 | 20 | from supervisor.dbus.network.setting.generate import get_connection_from_interface |
18 | | -from supervisor.exceptions import HostNetworkError, HostNotSupportedError |
| 21 | +from supervisor.exceptions import ( |
| 22 | + HostNetworkActivationFailedError, |
| 23 | + HostNetworkError, |
| 24 | + HostNotSupportedError, |
| 25 | +) |
19 | 26 | from supervisor.homeassistant.const import WSEvent, WSType |
20 | | -from supervisor.host.const import MulticastDnsMode, WifiMode |
| 27 | +from supervisor.host.configuration import WifiConfig |
| 28 | +from supervisor.host.const import AuthMethod, MulticastDnsMode, WifiMode |
21 | 29 |
|
22 | 30 | from tests.dbus_service_mocks.base import DBusServiceMock |
23 | 31 | from tests.dbus_service_mocks.network_active_connection import ( |
@@ -64,6 +72,44 @@ async def fixture_device_wlan0_service( |
64 | 72 | ] |
65 | 73 |
|
66 | 74 |
|
| 75 | +@pytest.fixture(name="activated_connections") |
| 76 | +async def fixture_activated_connections( |
| 77 | + coresys: CoreSys, |
| 78 | +) -> AsyncGenerator[list[NetworkConnection]]: |
| 79 | + """Record active connections returned by NetworkManager activation calls.""" |
| 80 | + activated: list[NetworkConnection] = [] |
| 81 | + network = coresys.dbus.network |
| 82 | + activate_connection = network.activate_connection |
| 83 | + add_and_activate_connection = network.add_and_activate_connection |
| 84 | + |
| 85 | + async def _activate_connection(*args: Any) -> NetworkConnection: |
| 86 | + con = await activate_connection(*args) |
| 87 | + activated.append(con) |
| 88 | + return con |
| 89 | + |
| 90 | + async def _add_and_activate_connection(*args: Any) -> tuple[Any, NetworkConnection]: |
| 91 | + settings, con = await add_and_activate_connection(*args) |
| 92 | + activated.append(con) |
| 93 | + return settings, con |
| 94 | + |
| 95 | + with ( |
| 96 | + patch.object(network, "activate_connection", new=_activate_connection), |
| 97 | + patch.object( |
| 98 | + network, "add_and_activate_connection", new=_add_and_activate_connection |
| 99 | + ), |
| 100 | + ): |
| 101 | + yield activated |
| 102 | + |
| 103 | + |
| 104 | +def _assert_connection_shut_down(con: NetworkConnection) -> None: |
| 105 | + """Assert an activated connection and its child objects are shut down.""" |
| 106 | + assert con.is_shutdown |
| 107 | + assert not con.is_connected |
| 108 | + assert con.settings.is_shutdown |
| 109 | + assert con.ipv4.is_shutdown |
| 110 | + assert con.ipv6.is_shutdown |
| 111 | + |
| 112 | + |
67 | 113 | async def test_load( |
68 | 114 | coresys: CoreSys, |
69 | 115 | network_manager_service: NetworkManagerService, |
@@ -625,3 +671,74 @@ async def test_apply_changes_v2_reenable_reuses_profile( |
625 | 671 | "/org/freedesktop/NetworkManager/Devices/1", |
626 | 672 | "/", |
627 | 673 | ) in network_manager_service.ActivateConnection.calls |
| 674 | + |
| 675 | + |
| 676 | +@pytest.mark.parametrize( |
| 677 | + ("interface_name", "wifi"), |
| 678 | + [ |
| 679 | + pytest.param("eth0", None, id="update"), |
| 680 | + pytest.param( |
| 681 | + "wlan0", |
| 682 | + WifiConfig( |
| 683 | + WifiMode.INFRASTRUCTURE, "MY_TEST", AuthMethod.WPA_PSK, "password", None |
| 684 | + ), |
| 685 | + id="create", |
| 686 | + ), |
| 687 | + ], |
| 688 | +) |
| 689 | +async def test_apply_changes_shuts_down_activated_connection( |
| 690 | + coresys: CoreSys, |
| 691 | + activated_connections: list[NetworkConnection], |
| 692 | + interface_name: str, |
| 693 | + wifi: WifiConfig | None, |
| 694 | +): |
| 695 | + """Test the connection returned by activation is shut down once activated.""" |
| 696 | + await coresys.host.network.load() |
| 697 | + |
| 698 | + interface = coresys.host.network.get(interface_name) |
| 699 | + interface.enabled = True |
| 700 | + interface.wifi = wifi |
| 701 | + await coresys.host.network.apply_changes(interface) |
| 702 | + |
| 703 | + assert len(activated_connections) == 1 |
| 704 | + _assert_connection_shut_down(activated_connections[0]) |
| 705 | + # The interface tracks the active connection with its own object |
| 706 | + assert not coresys.dbus.network.get("eth0").connection.is_shutdown |
| 707 | + |
| 708 | + |
| 709 | +async def test_apply_changes_shuts_down_failed_connection( |
| 710 | + coresys: CoreSys, |
| 711 | + active_connection_service: ActiveConnectionService, |
| 712 | + activated_connections: list[NetworkConnection], |
| 713 | +): |
| 714 | + """Test the connection returned by activation is shut down if activation fails.""" |
| 715 | + await coresys.host.network.load() |
| 716 | + |
| 717 | + original_state = active_connection_service.fixture.state |
| 718 | + active_connection_service.fixture.state = 4 # DEACTIVATED |
| 719 | + try: |
| 720 | + with pytest.raises(HostNetworkActivationFailedError): |
| 721 | + await coresys.host.network.apply_changes(coresys.host.network.get("eth0")) |
| 722 | + finally: |
| 723 | + active_connection_service.fixture.state = original_state |
| 724 | + |
| 725 | + assert len(activated_connections) == 1 |
| 726 | + _assert_connection_shut_down(activated_connections[0]) |
| 727 | + |
| 728 | + |
| 729 | +async def test_apply_changes_background_shuts_down_activated_connection( |
| 730 | + coresys: CoreSys, |
| 731 | + activated_connections: list[NetworkConnection], |
| 732 | +): |
| 733 | + """Test the connection is shut down once a background activation finishes.""" |
| 734 | + await coresys.host.network.load() |
| 735 | + |
| 736 | + job_id = await coresys.host.network.apply_changes( |
| 737 | + coresys.host.network.get("eth0"), background_activation=True |
| 738 | + ) |
| 739 | + job = coresys.jobs.get_job(job_id) |
| 740 | + while not job.done: |
| 741 | + await asyncio.sleep(0) |
| 742 | + |
| 743 | + assert len(activated_connections) == 1 |
| 744 | + _assert_connection_shut_down(activated_connections[0]) |
0 commit comments