diff --git a/conftest.py b/conftest.py index 3b9ed38ab..dd58ee6fe 100644 --- a/conftest.py +++ b/conftest.py @@ -37,7 +37,7 @@ from lib.vbd import VBD from lib.vdi import VDI from lib.vm import VM, vm_cache_key_from_def -from lib.xo import xo_cli +from lib.xo import _allow_xo_cli, xo_cli # Import package-scoped fixtures. Although we need to define them in a separate file so that we can # then import them in individual packages to fix the buggy package scope handling by pytest, we also @@ -202,6 +202,7 @@ def pytest_collection_modifyitems(items: list[pytest.Item], config: pytest.Confi 'hostB1', 'unused_512B_disks', 'unused_4k_disks', + 'hosts_with_xo', ] # ------------- @@ -367,12 +368,15 @@ def registered_xo_cli() -> None: # The fixture is not responsible for establishing the connection. # We just check that xo-cli is currently registered try: + old_allow_xo_cli = _allow_xo_cli(True) xo_cli('server.getAll') + _allow_xo_cli(old_allow_xo_cli) except Exception as e: - raise Exception(f"Check for registered xo_cli failed: {e}") + pytest.fail(f"Check for registered xo_cli failed: {e}") @pytest.fixture(scope='session') def hosts_with_xo(hosts: list[Host], registered_xo_cli: None) -> Generator[list[Host], None, None]: + old_allow_xo_cli = _allow_xo_cli(True) for h in hosts: logging.info(">>> Connect host %s" % h) if not h.skip_xo_config: @@ -386,6 +390,7 @@ def hosts_with_xo(hosts: list[Host], registered_xo_cli: None) -> Generator[list[ if not h.skip_xo_config: logging.info("<<< Disconnect host %s" % h) h.xo_server_remove() + _allow_xo_cli(old_allow_xo_cli) @pytest.fixture(scope='session') def hostA1(hosts: list[Host]) -> Generator[Host, None, None]: diff --git a/jobs.py b/jobs.py index 36d3f4067..ac520a983 100755 --- a/jobs.py +++ b/jobs.py @@ -44,7 +44,8 @@ class JobData(TypedDict): "tests/xapi_plugins", "tests/install/test_fixtures.py", ], - "markers": "(small_vm or no_vm) and not flaky and not reboot and not complex_prerequisites", + "markers": "(small_vm or no_vm) and not flaky and not reboot " + "and not hosts_with_xo and not complex_prerequisites", }, "main-multi-unix": { "description": "a group of tests that need to run on the largest variety of VMs - unix split", @@ -79,16 +80,17 @@ class JobData(TypedDict): "network-advanced": { "description": "a group of network tests with complex prerequisites", "requirements": [ - "A pool with at least 1 host.", + "A pool with at least 1 host (if more, with same network configuration).", "At least 2 free NICs on every host.", "A small VM that can be imported on the SRs.", + "xo-cli locally installed, in $PATH, and registered to an XO instance.", ], "nb_pools": 1, "params": { "--vm": "single/small_vm", }, "paths": ["tests/network"], - "markers": "complex_prerequisites", + "markers": "complex_prerequisites or hosts_with_xo", }, "packages": { "description": "tests that packages can be installed correctly", diff --git a/lib/host.py b/lib/host.py index 05609f3a3..9b4028617 100644 --- a/lib/host.py +++ b/lib/host.py @@ -31,6 +31,8 @@ from lib.network import Network from lib.pif import PIF from lib.sr import SR +from lib.tunnel import Tunnel +from lib.vlan import VLAN from lib.vm import VM from lib.xo import xo_cli, xo_object_exists @@ -716,6 +718,9 @@ def yum_restore_saved_state(self) -> None: self.saved_packages_list = None self.saved_rollback_id = None + def service_started(self, name: str) -> bool: + return self.ssh(f'systemctl is-active {name}', check=False) == 'active' + def reboot(self, verify: bool = False) -> None: logging.info(f"[{self}] Reboot host") # Running `reboot` directly immediately disconnects the ssh session and makes the ssh client return with an @@ -1051,6 +1056,9 @@ def pifs(self, device: str | None = None) -> list[PIF]: return [PIF(uuid, self) for uuid in safe_split(self.xe("pif-list", args, minimal=True))] + def tunnels(self) -> list[Tunnel]: + return [Tunnel(self, uuid) for uuid in safe_split(self.xe("tunnel-list", {}, minimal=True))] + def create_bond(self, network: Network, pifs: list[PIF], mode: str | None = None) -> Bond: args: dict[str, str | bool | dict[str, str]] = { 'network-uuid': network.uuid, @@ -1078,3 +1086,35 @@ def create_network(self, label: str, description: str | None = None) -> Network: logging.info(f"[{self}] New Network: {uuid}") return Network(self, uuid) + + def create_vlan(self, network: Network, pif: PIF, vlan: int) -> VLAN: + args: dict[str, str | bool | dict[str, str]] = { + 'network-uuid': network.uuid, + 'pif-uuid': pif.uuid, + 'vlan': str(vlan), + } + + untagged_pif_uuid = self.xe("vlan-create", args, minimal=True) + uuid = self.xe("pif-param-get", { + "uuid": untagged_pif_uuid, + "param-name": "vlan-master-of", + }) + logging.info(f"New VLAN: {uuid} (untagged-pif: {untagged_pif_uuid})") + + return VLAN(self, uuid) + + def create_tunnel(self, network: Network, pif: PIF, protocol: str) -> Tunnel: + args: dict[str, str | bool | dict[str, str]] = { + 'network-uuid': network.uuid, + 'pif-uuid': pif.uuid, + 'protocol': protocol, + } + + access_pif_uuid = self.xe("tunnel-create", args, minimal=True) + uuid = self.xe("pif-param-get", { + "uuid": access_pif_uuid, + "param-name": "tunnel-access-PIF-of", + }) + logging.info(f"New Tunnel: {uuid} (access-pif: {access_pif_uuid})") + + return Tunnel(self, uuid) diff --git a/lib/network.py b/lib/network.py index 648965c4e..6edc6b4f3 100644 --- a/lib/network.py +++ b/lib/network.py @@ -64,3 +64,8 @@ def managed(self) -> bool: def MTU(self) -> int: return int(self.param_get('MTU') or '0') + + def bridge(self) -> str: + bridge = self.param_get('bridge') + assert bridge is not None, "network must have a bridge" + return bridge diff --git a/lib/pif.py b/lib/pif.py index edc738903..59fd9525a 100644 --- a/lib/pif.py +++ b/lib/pif.py @@ -66,6 +66,20 @@ def network_uuid(self) -> str: assert uuid is not None, "unexpected PIF without network-uuid" return uuid + def ip_configuration_mode(self) -> str: + mode = self.param_get("IP-configuration-mode") + assert mode + return mode + + def vlan(self) -> int | None: + vlan_str = self.param_get('VLAN') + assert vlan_str + vlan = int(vlan_str) + if vlan == -1: + return None + else: + return vlan + def reconfigure_ip(self, mode: str) -> None: self.host.xe("pif-reconfigure-ip", { "uuid": self.uuid, diff --git a/lib/tunnel.py b/lib/tunnel.py new file mode 100644 index 000000000..122294706 --- /dev/null +++ b/lib/tunnel.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import logging + +from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set +from lib.pif import PIF + +from typing import TYPE_CHECKING, Literal, overload + +if TYPE_CHECKING: + from lib.host import Host + +class Tunnel: + xe_prefix = "tunnel" + + def __init__(self, host: Host, uuid: str): + self.host = host + self.uuid = uuid + + @overload + def param_get(self, param_name: str, key: str | None = ..., accept_unknown_key: Literal[False] = ...) -> str: + ... + + @overload + def param_get( + self, param_name: str, key: str | None = ..., accept_unknown_key: Literal[True] = ... + ) -> str | None: + ... + + def param_get(self, param_name: str, key: str | None = None, accept_unknown_key: bool = False) -> str | None: + return _param_get(self.host, self.xe_prefix, self.uuid, param_name, key, accept_unknown_key) + + def param_set(self, param_name: str, value: str | bool | dict[str, str], key: str | None = None) -> None: + _param_set(self.host, self.xe_prefix, self.uuid, param_name, value, key) + + def param_remove(self, param_name: str, key: str, accept_unknown_key: bool = False) -> None: + _param_remove(self.host, self.xe_prefix, self.uuid, param_name, key, accept_unknown_key) + + def param_add(self, param_name: str, value: str, key: str | None = None) -> None: + _param_add(self.host, self.xe_prefix, self.uuid, param_name, value, key) + + def param_clear(self, param_name: str) -> None: + _param_clear(self.host, self.xe_prefix, self.uuid, param_name) + + def destroy(self): + logging.info(f"Destroying Tunnel: {self.uuid}") + self.host.xe('tunnel-destroy', {'uuid': self.uuid}) + + def access_pif(self) -> PIF: + uuid = self.param_get("access-PIF") + assert uuid + return PIF(uuid, self.host) + + def transport_pif(self) -> PIF: + uuid = self.param_get("transport-PIF") + assert uuid + return PIF(uuid, self.host) diff --git a/lib/vif.py b/lib/vif.py index ea26fc1fd..8fb421a02 100644 --- a/lib/vif.py +++ b/lib/vif.py @@ -4,6 +4,7 @@ import time from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set +from lib.network import Network from typing import TYPE_CHECKING, Literal, overload @@ -134,3 +135,8 @@ def configure_ipv6( gateway: str | None = None, ) -> None: self._configure("ipv6", mode, address, gateway) + + def network(self) -> Network: + network_uuid = self.param_get('network-uuid') + assert network_uuid is not None, "VIF must have a network-uuid" + return Network(self.vm.host, network_uuid) diff --git a/lib/vlan.py b/lib/vlan.py new file mode 100644 index 000000000..c61123a33 --- /dev/null +++ b/lib/vlan.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import logging + +from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set +from lib.pif import PIF + +from typing import TYPE_CHECKING, Literal, overload + +if TYPE_CHECKING: + from lib.host import Host + +class VLAN: + xe_prefix = "vlan" + + def __init__(self, host: Host, uuid: str): + self.host = host + self.uuid = uuid + + @overload + def param_get(self, param_name: str, key: str | None = ..., accept_unknown_key: Literal[False] = ...) -> str: + ... + + @overload + def param_get( + self, param_name: str, key: str | None = ..., accept_unknown_key: Literal[True] = ... + ) -> str | None: + ... + + def param_get(self, param_name: str, key: str | None = None, accept_unknown_key: bool = False) -> str | None: + return _param_get(self.host, self.xe_prefix, self.uuid, param_name, key, accept_unknown_key) + + def param_set(self, param_name: str, value: str | bool | dict[str, str], key: str | None = None) -> None: + _param_set(self.host, self.xe_prefix, self.uuid, param_name, value, key) + + def param_remove(self, param_name: str, key: str, accept_unknown_key: bool = False) -> None: + _param_remove(self.host, self.xe_prefix, self.uuid, param_name, key, accept_unknown_key) + + def param_add(self, param_name: str, value: str, key: str | None = None) -> None: + _param_add(self.host, self.xe_prefix, self.uuid, param_name, value, key) + + def param_clear(self, param_name: str) -> None: + _param_clear(self.host, self.xe_prefix, self.uuid, param_name) + + def destroy(self): + logging.info(f"Destroying VLAN: {self.uuid}") + self.host.xe('vlan-destroy', {'uuid': self.uuid}) + + def tag(self) -> int: + tag = self.param_get('tag') + assert tag + return int(tag) + + def tagged_pif(self) -> PIF: + uuid = self.param_get("tagged-PIF") + assert uuid + return PIF(uuid, self.host) + + def untagged_pif(self) -> PIF: + uuid = self.param_get("untagged-PIF") + assert uuid + return PIF(uuid, self.host) diff --git a/lib/xo.py b/lib/xo.py index 4ef8cfd59..4674426c2 100644 --- a/lib/xo.py +++ b/lib/xo.py @@ -1,3 +1,5 @@ +import pytest + import json from data import TOOLS @@ -6,6 +8,20 @@ from typing import Literal, overload +__allow_xo_cli = False +def _allow_xo_cli(value: bool) -> bool: + """ + Permit to configure the usage of xo_cli function (returns the previous value). + This function shoudln't be called directly. + If you need xo_cli(), use the hosts_with_xo fixture. + """ + global __allow_xo_cli + + old = __allow_xo_cli + __allow_xo_cli = value + + return old + @overload def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: Literal[False] = False) -> str: ... @@ -14,6 +30,9 @@ def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_js ... def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: bool = False) -> JSONType | str: + if not __allow_xo_cli: + pytest.fail("xo_cli function requires hosts_with_xo fixture usage.") + cmd = [TOOLS.get('xo-cli', 'xo-cli'), action] if action != 'list-objects' and use_json: cmd += ['--json'] diff --git a/pyproject.toml b/pyproject.toml index d6ee00e46..2594b8c25 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "pytest-dependency", "requests", "ipdb", + "rpm-version>=0.5.1", ] [dependency-groups] diff --git a/pytest.ini b/pytest.ini index 920bb67b7..14a8ac233 100644 --- a/pytest.ini +++ b/pytest.ini @@ -38,6 +38,7 @@ markers = # * Other markers reboot: tests that reboot one or more hosts. + hosts_with_xo: tests that require XO to be available via xo-cli tool. flaky: flaky tests. Usually pass, but sometimes fail unexpectedly. complex_prerequisites: tests whose prerequisites are complex and may require special attention. quicktest: runs `quicktest`. diff --git a/requirements/base.txt b/requirements/base.txt index a59856ebe..de86544b7 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -10,3 +10,4 @@ pytest>=9.1.1 pytest-dependency requests ipdb +rpm-version>=0.5.1 diff --git a/tests/network/conftest.py b/tests/network/conftest.py index 9d0e1802b..baa1e7739 100644 --- a/tests/network/conftest.py +++ b/tests/network/conftest.py @@ -2,15 +2,22 @@ import pytest +import json import logging +from rpm_version import Evr # type: ignore[import-untyped] + from data import HOST_FREE_NICS from lib.common import PackageManagerEnum from lib.host import Host from lib.network import Network +from lib.tunnel import Tunnel +from lib.typing import JSONType +from lib.vlan import VLAN from lib.vm import VM +from lib.xo import xo_cli -from typing import Generator +from typing import Generator, Literal @pytest.fixture(scope='package') def host_no_sdn_controller(host: Host) -> None: @@ -18,6 +25,73 @@ def host_no_sdn_controller(host: Host) -> None: if host.xe('sdn-controller-list', minimal=True): pytest.fail("This test requires an XCP-ng with no SDN controller") +@pytest.fixture(scope='package') +def hosts_with_traffic_rules(hosts_with_xo: list[Host]) -> Generator[list[Host], None, None]: + """A list of XCP-ng hosts with proper traffic rules configuration.""" + hosts = hosts_with_xo + + # check XO: check sdn-controller plugin (loaded + minimal version) + minimal = Evr.parse("1.3.0") + + plugin_found = False + plugins = xo_cli('plugin.get', use_json=True) + assert isinstance(plugins, list) + for plugin in plugins: + assert isinstance(plugin, dict) + if plugin.get('id') != 'sdn-controller': + continue + + plugin_found = True + loaded = plugin.get('loaded', False) + assert isinstance(loaded, bool) + if loaded: + version = plugin.get('version', '') + assert isinstance(version, str) + if minimal > Evr.parse(version): + pytest.fail(f"This test requires XO with at least sdn-controller version {minimal}") + else: + pytest.fail("This test requires XO with sdn-controller plugin loaded") + + if not plugin_found: + pytest.fail("This test requires XO with sdn-controller plugin") + + # check host: xcp-ng-xapi-plugins minimal version + minimal = Evr.parse("xcp-ng-xapi-plugins-1.17.0") + + def host_with_xcp_ng_xapi_plugins(host: Host): + # get the package version + packages = json.loads(host.xe('host-call-plugin', { + 'host-uuid': host.uuid, + 'plugin': 'updater.py', + 'fn': 'query_installed', + 'args:packages': 'xcp-ng-xapi-plugins', + }, minimal=True)) + + return minimal <= Evr.parse(packages.get('xcp-ng-xapi-plugins', '')) + + hosts = list(filter(host_with_xcp_ng_xapi_plugins, hosts)) + if len(hosts) == 0: + pytest.fail(f"This test requires hosts with at least xcp-ng-xapi-plugins version {minimal}") + + # check XO: check sdn-controller configuration: should be using xapi-plugin method for OpenFlow rules + def host_with_xapiplugin(host: Host) -> bool: + # the key 'xo:sdn-controller:of-method' is present since cycle XO 6.5c 2026-05-14 (xo-lite v0.21.0) + of_method = host.pool.param_get( + 'other-config', + key='xo:sdn-controller:of-method', + accept_unknown_key=True, + ) or 'channel' + + return of_method == 'xapi-plugin' + + hosts = list(filter(host_with_xapiplugin, hosts)) + if len(hosts) == 0: + pytest.fail("This test requires XO to use of-method=xapi-plugin " + "(see https://docs.xen-orchestra.com/xo5/configuration#sdn-controller-mode)") + + yield hosts + + # a clone of imported_vm in which we've added tcpdump # not to be used by tests directly @pytest.fixture(scope='module') @@ -50,19 +124,8 @@ def vm_with_tcpdump_scope_function(vm_with_tcpdump_scope_module: VM): yield vm vm.destroy() -@pytest.fixture(scope='function') -def empty_network(host: Host) -> Generator[Network, None, None]: - net = host.create_network(label="empty_network for tests") - - yield net - - for vif_uuid in net.vif_uuids(): - host.xe("vif-unplug", { - 'uuid': vif_uuid, - }) - - net.destroy() +# ---- Bond ---- @pytest.fixture(scope='function') def bond_lacp(host: Host, empty_network: Network): if len(HOST_FREE_NICS) < 2: @@ -110,3 +173,122 @@ def bond_balanceslb(host: Host, empty_network: Network): bond = host.create_bond(empty_network, pifs, mode="balance-slb") yield bond bond.destroy() + + +# ---- Network ---- +@pytest.fixture(scope='function') +def empty_network(host: Host) -> Generator[Network, None, None]: + net = host.create_network(label="empty_network for tests") + + yield net + + for vif_uuid in net.vif_uuids(): + host.xe("vif-unplug", { + 'uuid': vif_uuid, + }) + + net.destroy() + + +# ---- Tunnel ---- +@pytest.fixture(params=["gre", "vxlan"]) +def tunnel_protocol(request: pytest.FixtureRequest) -> str: + return request.param + +@pytest.fixture(params=[False, True]) +def tunnel_encryption(request: pytest.FixtureRequest) -> bool: + return request.param + +@pytest.fixture +def tunnel( + hosts_with_xo: list[Host], + tunnel_protocol: str, tunnel_encryption: bool, +) -> Generator[Tunnel, None, None]: + host = hosts_with_xo[0] + + # check system requirements + prepare: dict[str, Literal[True]] = {} + if not host.is_package_installed("openvswitch-ipsec"): + prepare["installed-openvswitch-ipsec"] = True + host.yum_install(["openvswitch-ipsec"]) + if not host.service_started("ipsec"): + prepare["service-ipsec"] = True + host.ssh("systemctl start ipsec") + if not host.service_started("openvswitch-ipsec"): + prepare["service-openvswitch-ipsec"] = True + host.ssh("systemctl start openvswitch-ipsec") + + # create a tunnel over the management PIF + tunnel_device = host.management_pif().device() + + logging.info(f"tunnel: resolve PIF on {host.hostname_or_ip} using \ + {[(pif.network_uuid(), pif.device()) for pif in host.pifs()]}") + + # we could have several pifs on one device (due to VLANs for example) + pifs = [pif for pif in host.pifs(device=tunnel_device) if pif.ip_configuration_mode() != "None"] + if len(pifs) == 0: + pytest.fail(f"'tunnel' fixture requires tunnel_device={tunnel_device} to have configured IP") + + # use the first usable pif + pif = pifs[0] + + existing_tunnels = [t.uuid for t in host.tunnels()] + logging.info(f"tunnel: existing tunnels: {existing_tunnels}") + + xo_cli('sdnController.createPrivateNetwork', { + 'poolIds': f"json:[\"{host.pool.uuid}\"]", + 'pifIds': f"json:[\"{pif.uuid}\"]", + 'name': 'test-tunnel', + 'description': 'tunnel for test', + 'encapsulation': tunnel_protocol, + 'encrypted': 'true' if tunnel_encryption else 'false', + }) + + # sdnController.createPrivateNetwork might have created several Tunnel (one per host) + # so get all created Tunnel + created_tunnels = list(set([t.uuid for t in host.tunnels()]) - set(existing_tunnels)) + logging.info(f"tunnel: created tunnels: {created_tunnels}") + + # yield only the first tunnel + yield Tunnel(host, created_tunnels[0]) + + # teardown created_tunnels (and associated networks) + network_uuids: set[str] = set() + + for tunnel_uuid in created_tunnels: + tunnel = Tunnel(host, tunnel_uuid) + + # get network linked to the tunnel + network_uuids.add(tunnel.access_pif().network_uuid()) + + # destroy the tunnel + tunnel.destroy() + + # destroy networks associated to destroyed tunnels + for network_uuid in network_uuids: + Network(host, network_uuid).destroy() + + # remove installed dependencies + if "service-openvswitch-ipsec" in prepare: + host.ssh("systemctl stop openvswitch-ipsec") + if "service-ipsec" in prepare: + host.ssh("systemctl stop ipsec") + if "installed-openvswitch-ipsec" in prepare: + host.yum_remove(["openvswitch-ipsec"]) + +# ---- VLAN ---- +@pytest.fixture +def vlan(host: Host, empty_network: Network) -> Generator[VLAN, None, None]: + logging.info(f"vlan: resolve PIF on {host.hostname_or_ip} using \ + {[(pif.network_uuid(), pif.param_get('device')) for pif in host.pifs()]}") + + if len(HOST_FREE_NICS) < 1: + pytest.fail("This fixture needs at least 1 free NICs") + + # randomly chosen tag + vlan_tag = 42 + + [pif] = host.pifs(device=HOST_FREE_NICS[0]) + vlan = host.create_vlan(empty_network, pif, vlan_tag) + yield vlan + vlan.destroy() diff --git a/tests/network/test_traffic_rules.py b/tests/network/test_traffic_rules.py new file mode 100644 index 000000000..dbcb74ddc --- /dev/null +++ b/tests/network/test_traffic_rules.py @@ -0,0 +1,849 @@ +from __future__ import annotations + +import pytest + +import logging +import os +from time import sleep + +from lib.common import Defer, wait_for, wait_for_not +from lib.host import Host +from lib.network import Network +from lib.sr import SR +from lib.tunnel import Tunnel +from lib.vlan import VLAN +from lib.vm import VM +from lib.xo import xo_cli + +from typing import Callable + +# Requirements: +# xo-cli (on the host running the test) is expected to be usable +# From --hosts parameter: +# - host(A1): first XCP-ng host (no traffic rules should be already present on the host) +# From --vm parameter +# - A VM to import + +# Special requirements for some tests: +# - TestVLAN needs at least 1 free NICs (see HOST_FREE_NICS in data.py) +# - TestMigrate needs second XCP-ng host in the same pool +# - TestTunnel will create encrypted tunnel (and only one could be created at a time) + +cache_ovs_vsctl_bridge_to_parent: dict[str, str] = {} + +def ovs_vsctl_bridge_to_parent(host: Host, br: str) -> str: + key = f"test({os.environ.get('PYTEST_CURRENT_TEST')})-host({host.uuid})-br({br})" + + if key not in cache_ovs_vsctl_bridge_to_parent: + cache_ovs_vsctl_bridge_to_parent[key] = host.ssh(f"ovs-vsctl br-to-parent {br}") + + return cache_ovs_vsctl_bridge_to_parent[key] + +def ofctl_dumpflows(host: Host, br: str) -> list[str]: + """ + Get the list of dump-flows installed for the bridge {br} + """ + br = ovs_vsctl_bridge_to_parent(host, br) + return host.ssh( + f"ovs-ofctl -O OpenFlow11 dump-flows '{br}' | grep -F cookie=", + ).splitlines() + +def count_of(host: Host, br: str): + """ + Return the number of OF flows in the bridge (excluding the default one) + """ + return len(ofctl_dumpflows(host, br)) - 1 + +def ofproto_trace_drop_in_port( + host: Host, br: str, flow: str, in_port: str, + vlan_tag: int | None, vlan_device: str | None, +) -> bool: + """ + Run ovs-appctl ofproto/trace program to check OpenFlow rules processing + on a specific port of a bridge. + """ + if vlan_device == in_port: + flow = f"in_port={in_port},vlan_vid={vlan_tag},{flow}" + else: + flow = f"in_port={in_port},{flow}" + + logging.debug(f"ofproto/trace port='{in_port}'") + result = host.ssh(f"ovs-appctl ofproto/trace {br} {flow}") + return result.endswith("Datapath actions: drop") + +def ofproto_trace_drop( + host: Host, br: str, flow: str, + network_br: str | None = None, + vlan_tag: int | None = None, vlan_device: str | None = None, +) -> bool: + """ + Run ovs-appctl ofproto/trace program to check OpenFlow rules processing + on all ports of a bridge. + """ + def is_not_xapi_port(portname: str) -> bool: + """ + Return False if portname has the form of "{br}_port" + it is the internal communication port for linking hosts + """ + return not portname.startswith(f"{br}_port") + + ports = host.ssh(f"ovs-vsctl list-ports {network_br or br}").splitlines() + ports = list(filter(is_not_xapi_port, ports)) + if len(ports) == 0: + # no ports on bridge, packet will pass + return False + + br = ovs_vsctl_bridge_to_parent(host, br) + logging.debug(f"ofproto/trace: dumping flows: {br}") + host.ssh(f"ovs-ofctl -O OpenFlow11 dump-flows {br}") + + return all([ + ofproto_trace_drop_in_port( + host, br, flow, port, + vlan_tag, vlan_device, + ) + for port in ports + ]) + +def sync_sdnController_action( + host: Host, + action: str, + args: dict[str, str] = {}, +) -> str: + def log_date(host: Host) -> str: + # %y : file modification date, example: "2026-08-24 15:15:58.982053917 +0200" + return host.ssh("stat -c %y /var/log/sdn-controller-plugin.log") + + hosts = {} + + # on each host, get the number of Host.call_plugin calls + for h in host.pool.hosts: + hosts[h.name()] = log_date(h) + + # run the sdnController action + logging.info(f"sdnController.{action}") + ret = xo_cli(f"sdnController.{action}", args) + + # as XO is contacting only involved hosts, so some hosts in the pool might *not* be contacted at all + # wait max 5 sec, and discard TimeoutError exception + try: + for h in host.pool.hosts: + wait_for( + lambda: log_date(h) != hosts[h.name()], + timeout_secs=5, + ) + except TimeoutError: + pass + + return ret + +def xo_vm_power_state(vm: VM, power_state: str) -> Callable[[], bool]: + """ + Return a function that return if the VM is seen by XO in the given power_state. + """ + def vm_power_state() -> bool: + lst = xo_cli('list-objects', { + 'type': 'VM', + 'uuid': vm.uuid, + }, use_json=True) + assert isinstance(lst, list) + + if len(lst) != 1: + return False + + xovm = lst[0] + assert isinstance(xovm, dict) + return (power_state == xovm.get('power_state', '?')) + + return vm_power_state + +@pytest.mark.small_vm +class TestSimple: + def test_vifRule(self, hosts_with_traffic_rules: list[Host], imported_vm: VM, defer: Defer): + host = hosts_with_traffic_rules[0] + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + + vif = vm.vifs()[0] + vifId = vif.uuid + macAddress = vif.mac_address() + hostBr = vif.network().bridge() + + assert count_of(host, hostBr) == 0, "no OF at init" + + # add OF rule (before starting VM) + sync_sdnController_action(host, 'addRule', { + 'vifId': vifId, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:80', + 'allow': 'false', + }) + + # before starting the VM, traffic pass + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + # start the VM and wait for XO to see the VM + vm.start() + vm.wait_for_os_booted() + + # right after VM booted + assert ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + # add OF rule (while running) + sync_sdnController_action(host, 'addRule', { + 'vifId': vifId, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:81', + 'allow': 'false', + }) + + # new rule added, both traffic dropped + assert ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + # delete OF rule (while running) + sync_sdnController_action(host, 'deleteRule', { + 'vifId': vifId, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:80', + }) + + # first rule deleted, traffic should pass (and 2nd rule drop traffic) + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + vm.shutdown(verify=True) + + # after shutdown, two rules removed (vif not here anymore) + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + vm.start() + vm.wait_for_os_booted() + + # after restarted, only the second rule apply + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + vm.shutdown(verify=True) + + # delete OF rule (while stopped) + sync_sdnController_action(host, 'deleteRule', { + 'vifId': vifId, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:81', + }) + + # no more traffic blocked + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=80,dl_src={macAddress}") + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + assert count_of(host, hostBr) == 0, "no OF at end" + + def test_networkRule(self, hosts_with_traffic_rules: list[Host], imported_vm: VM, defer: Defer): + host = hosts_with_traffic_rules[0] + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + networkId = host.management_network() + hostBr = Network(host, networkId).bridge() + + assert count_of(host, hostBr) == 0, "no OF at init" + + # add OF rule (before starting VM) + sync_sdnController_action(host, 'addNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + 'allow': 'false', + }) + defer( + lambda: + sync_sdnController_action(host, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + }) + ) + + # the rule is not applied as there is no interface in the network at the time + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.2") + + # start the VM + vm.start() + + # wait for XO to see the VM + vm.wait_for_os_booted() + + # the rule is applied (after booting VM) + assert ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.2") + + # add OF rule (while running) + sync_sdnController_action(host, 'addNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.2', + 'direction': 'to', + 'protocol': 'icmp', + 'allow': 'false', + }) + defer( + lambda: + sync_sdnController_action(host, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.2', + 'direction': 'to', + 'protocol': 'icmp', + }) + ) + + # both rules are applied + assert ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + assert ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.2") + + # delete OF rule (while running) + sync_sdnController_action(host, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + }) + + # second rule only is applied + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + assert ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.2") + + # restart the VM + vm.shutdown(verify=True) + vm.start() + vm.wait_for_os_booted() + + # same as previous (after VM restart) + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + assert ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.2") + + vm.shutdown(verify=True) + + # delete OF rule (while stopped) + sync_sdnController_action(host, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.2', + 'direction': 'to', + 'protocol': 'icmp', + }) + + # no more rules + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.2") + + assert count_of(host, hostBr) == 0, "no OF at end" + + +@pytest.mark.small_vm +class TestMigrate: + def test_vifRule( + self, + hosts_with_traffic_rules: list[Host], + hostA2: Host, + local_sr_on_hostA2: SR, + imported_vm: VM, + defer: Defer, + ): + hostA1 = hosts_with_traffic_rules[0] + + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + vm.start() + vm.wait_for_vm_running_and_ssh_up() + + vif = vm.vifs()[0] + vifId = vif.uuid + macAddress = vif.mac_address() + hostBr = vif.network().bridge() + + assert count_of(hostA1, hostBr) == 0, "no OF at init (on hostA1)" + assert count_of(hostA2, hostBr) == 0, "no OF at init (on hostA2)" + + # no drop before adding the rule + assert not ofproto_trace_drop(hostA1, hostBr, f"icmp,dl_src={macAddress}") + assert not ofproto_trace_drop(hostA2, hostBr, f"icmp,dl_src={macAddress}") + + # add OF rule + sync_sdnController_action(hostA1, 'addRule', { + 'vifId': vifId, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'icmp', + 'allow': 'false', + }) + try: + # drop after adding the rule + assert ofproto_trace_drop(hostA1, hostBr, f"icmp,dl_src={macAddress}") + + vm.migrate(hostA2, local_sr_on_hostA2) + + wait_for( + lambda: ofproto_trace_drop(hostA2, hostBr, f"icmp,dl_src={macAddress}"), + msg="Wait for still dropping after migrate", + timeout_secs=30, + ) + + finally: + sync_sdnController_action(hostA1, 'deleteRule', { + 'vifId': vifId, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'icmp', + }) + + # no more drop after deleting the rule + assert not ofproto_trace_drop(hostA1, hostBr, f"icmp,dl_src={macAddress}") + assert not ofproto_trace_drop(hostA2, hostBr, f"icmp,dl_src={macAddress}") + + assert count_of(hostA1, hostBr) == 0, "no OF after deleteRule (on hostA1)" + assert count_of(hostA2, hostBr) == 0, "no OF after deleteRule (on hostA2)" + + def test_networkRule( + self, + hosts_with_traffic_rules: list[Host], + hostA2: Host, + local_sr_on_hostA1: SR, + local_sr_on_hostA2: SR, + imported_vm: VM, + defer: Defer, + ): + hostA1 = hosts_with_traffic_rules[0] + + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + vm.start(on=hostA1.name()) # the test is assymetric, start with the VM on know host + vm.wait_for_vm_running_and_ssh_up() + + networkId = hostA1.management_network() + hostA1Br = Network(hostA1, networkId).bridge() + hostA2Br = Network(hostA2, networkId).bridge() + + assert count_of(hostA1, hostA1Br) == 0, "no OF at init (on hostA1)" + assert count_of(hostA2, hostA2Br) == 0, "no OF at init (on hostA2)" + + # no rule + assert not ofproto_trace_drop(hostA1, hostA1Br, "icmp,nw_dst=10.0.0.1") + assert not ofproto_trace_drop(hostA2, hostA2Br, "icmp,nw_dst=10.0.0.1") + + # add OF rule + sync_sdnController_action(hostA1, 'addNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + 'allow': 'false', + }) + try: + logging.info("check pre-migrate") + wait_for( + lambda: ofproto_trace_drop(hostA1, hostA1Br, "icmp,nw_dst=10.0.0.1"), + msg="Wait for the rule to applied on hostA1", + timeout_secs=30, + ) + + vm.migrate(hostA2, local_sr_on_hostA2) + + logging.info("check post-migrate") + wait_for( + lambda: ofproto_trace_drop(hostA2, hostA2Br, "icmp,nw_dst=10.0.0.1"), + msg="Wait for the rule to applied on hostA2", + timeout_secs=30, + ) + + finally: + sync_sdnController_action(hostA1, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + }) + + # no more rule + logging.info("check post-delete") + assert not ofproto_trace_drop(hostA1, hostA1Br, "icmp,nw_dst=10.0.0.1") + assert not ofproto_trace_drop(hostA2, hostA2Br, "icmp,nw_dst=10.0.0.1") + + assert count_of(hostA1, hostA1Br) == 0, "no OF at end (on hostA1)" + assert count_of(hostA2, hostA2Br) == 0, "no OF at end (on hostA2)" + + +@pytest.mark.complex_prerequisites +@pytest.mark.small_vm +class TestVLAN: + def test_vifRule(self, hosts_with_traffic_rules: list[Host], imported_vm: VM, empty_network: Network, + vlan: VLAN, defer: Defer): + host = hosts_with_traffic_rules[0] + network = empty_network + + vlan_tag = vlan.tag() + vlan_device = vlan.untagged_pif().device() + + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + + # get bridge of the network of the tagged PIF + hostBr = Network(host, vlan.tagged_pif().network_uuid()).bridge() + netBr = network.bridge() + logging.info(f"host bridge for vlan: {hostBr} / {netBr}") + + assert count_of(host, hostBr) == 0, "no OF at start" + + vif = vm.create_vif(1, network_uuid=network.uuid) + macAddress = vif.mac_address() + vm.start() + + # no rule applied + assert not ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=81,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + assert not ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=82,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + sync_sdnController_action(host, 'addRule', { + 'vifId': vif.uuid, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:81', + 'allow': 'false', + }) + + vm.wait_for_os_booted() + + # rule applied + assert ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=81,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + assert not ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=82,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + sync_sdnController_action(host, 'addRule', { + 'vifId': vif.uuid, + 'ipRange': '0.0.0.0/0', + 'direction': 'from', + 'protocol': 'tcp', + 'port': 'json:82', + 'allow': 'false', + }) + + assert ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=81,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + assert ofproto_trace_drop( + host, hostBr, + f"tcp,tp_src=82,dl_dst={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + sync_sdnController_action(host, 'deleteRule', { + 'vifId': vif.uuid, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:81', + }) + + assert not ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=81,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + assert ofproto_trace_drop( + host, hostBr, + f"tcp,tp_src=82,dl_dst={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + sync_sdnController_action(host, 'deleteRule', { + 'vifId': vif.uuid, + 'ipRange': '0.0.0.0/0', + 'direction': 'from', + 'protocol': 'tcp', + 'port': 'json:82', + }) + + # rule not applied + assert not ofproto_trace_drop( + host, hostBr, + f"tcp,tp_dst=81,dl_src={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + assert not ofproto_trace_drop( + host, hostBr, + f"tcp,tp_src=82,dl_dst={macAddress}", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + assert count_of(host, hostBr) == 0, "no OF at end" + + def test_networkRule(self, hosts_with_traffic_rules: list[Host], imported_vm: VM, + empty_network: Network, vlan: VLAN, defer: Defer): + host = hosts_with_traffic_rules[0] + network = empty_network + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + vlan_tag = vlan.tag() + vlan_device = vlan.untagged_pif().device() + + networkId = network.uuid + logging.info(f"networkId = {networkId}") + + try: + # get bridge of the network of the tagged PIF + hostBr = Network(host, vlan.tagged_pif().network_uuid()).bridge() + netBr = Network(host, networkId).bridge() + logging.info(f"host bridge for vlan: {hostBr} / {netBr}") + + # put one vif in the VLAN + vm.create_vif(1, network_uuid=networkId) + + assert count_of(host, hostBr) == 0, "no OF at init" + + # no rules + assert not ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + # add OF rule (before starting VM) + sync_sdnController_action(host, 'addNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + 'allow': 'false', + }) + + # XXX weird, but no OF so seems expected + logging.info("check pre-start") + wait_for_not( + lambda: ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ), + msg="Wait for the rule to not apply for now (nobody connected to the network)", + timeout_secs=30, + ) + + # start the VM + vm.start() + wait_for(xo_vm_power_state(vm, "Running"), msg="Waiting for XO to see VM is running") + + # rule is applied + logging.info("check post-start") + wait_for( + lambda: ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ), + msg="Wait for the rule to be applied", + timeout_secs=30, + ) + + vm.shutdown(verify=True, force=True) + wait_for(xo_vm_power_state(vm, "Halted"), msg="Waiting for XO to see VM is halted") + + # XXX weird, but no OF so seems expected + logging.info("check post-shutdown") + wait_for_not( + lambda: ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ), + msg="Wait for the rule to not apply for now (nobody connected to the network)", + timeout_secs=30, + ) + + finally: + # delete networkRule + sync_sdnController_action(host, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + }) + + # no rule applied + assert not ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + network_br=netBr, + vlan_tag=vlan_tag, vlan_device=vlan_device, + ) + + assert count_of(host, hostBr) == 0, "no OF at end" + + +@pytest.mark.small_vm +class TestTunnel: + def test_vifRule(self, hosts_with_traffic_rules: list[Host], imported_vm: VM, + tunnel: Tunnel, tunnel_protocol: str, defer: Defer): + host = hosts_with_traffic_rules[0] + network = Network(host, tunnel.access_pif().network_uuid()) + hostBr = network.bridge() + + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + + assert count_of(host, hostBr) == 0, "no OF at start" + + vif = vm.create_vif(1, network_uuid=network.uuid) + macAddress = vif.mac_address() + vm.start() + + # no rule applied + assert not ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}") + + sync_sdnController_action(host, 'addRule', { + 'vifId': vif.uuid, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:81', + 'allow': 'false', + }) + + # rule applied + wait_for( + lambda: ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}"), + msg="Wait for rule to apply", + timeout_secs=30, + ) + + sync_sdnController_action(host, 'deleteRule', { + 'vifId': vif.uuid, + 'ipRange': '0.0.0.0/0', + 'direction': 'to', + 'protocol': 'tcp', + 'port': 'json:81', + }) + + wait_for_not( + lambda: ofproto_trace_drop(host, hostBr, f"tcp,tp_dst=81,dl_src={macAddress}"), + msg="Wait for rule to not apply anymore", + timeout_secs=30, + ) + + assert count_of(host, hostBr) == 0, "no OF at end" + + def test_networkRule(self, hosts_with_traffic_rules: list[Host], imported_vm: VM, + tunnel: Tunnel, tunnel_protocol: str, defer: Defer): + host = hosts_with_traffic_rules[0] + network = Network(host, tunnel.access_pif().network_uuid()) + vm = imported_vm.clone() + defer(lambda: vm.destroy()) + hostBr = network.bridge() + + networkId = network.uuid + logging.info(f"networkId = {networkId}") + + try: + # put one vif in the Tunnel + vm.create_vif(1, network_uuid=networkId) + + assert count_of(host, hostBr) == 0, "no OF at init" + + # no rules + assert not ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1") + + # add OF rule (before starting VM) + sync_sdnController_action(host, 'addNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + 'allow': 'false', + }) + + # XXX weird, but no OF so seems expected + logging.info("check pre-start") + wait_for_not( + lambda: ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + ), + msg="Wait for the rule to not apply for now (nobody connected to the network)", + timeout_secs=30, + ) + + # start the VM + vm.start() + wait_for(xo_vm_power_state(vm, "Running"), msg="Waiting for XO to see VM is running") + + # rule is applied + logging.info("check post-start") + wait_for( + lambda: ofproto_trace_drop(host, hostBr, "icmp,nw_dst=10.0.0.1"), + msg="Wait for the rule to be applied", + timeout_secs=30, + ) + + vm.shutdown(verify=True, force=True) + wait_for(xo_vm_power_state(vm, "Halted"), msg="Waiting for XO to see VM is halted") + + # XXX weird, but no OF so seems expected + logging.info("check post-shutdown") + wait_for_not( + lambda: ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + ), + msg="Wait for the rule to not apply for now (nobody connected to the network)", + timeout_secs=30, + ) + + finally: + # delete networkRule + sync_sdnController_action(host, 'deleteNetworkRule', { + 'networkId': networkId, + 'ipRange': '10.0.0.1', + 'direction': 'to', + 'protocol': 'icmp', + }) + + # no rule applied + assert not ofproto_trace_drop( + host, hostBr, "icmp,nw_dst=10.0.0.1", + ) + + assert count_of(host, hostBr) == 0, "no OF at end" diff --git a/uv.lock b/uv.lock index c37d8bce6..a307dfaa5 100644 --- a/uv.lock +++ b/uv.lock @@ -1004,6 +1004,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a0/f4/c67b0b3f1b9245e8d266f0f112c500d50e5b4e83cb6f3b71b6528104182a/requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0", size = 73075, upload-time = "2026-05-14T19:25:26.443Z" }, ] +[[package]] +name = "rpm-version" +version = "0.5.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/24/f9/f4042715f9cba089926c76858c424fd9701b34f55d748a565d9d726abddb/rpm_version-0.5.1.tar.gz", hash = "sha256:8f55e499bd93d878f98d9cb5e53ccbb19794997941ebd5f3848133a9bdfadb4a", size = 29855, upload-time = "2026-07-24T21:06:12.94Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/28/5f/a90090877736926af5572f45d59155a3f4db3d444c0d6a3556c572af7734/rpm_version-0.5.1-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:fcdb88d85541a54d337daf18070e978f06df653a4ebfb08da0cb9f49cfcb5a59", size = 282363, upload-time = "2026-07-24T21:06:06.653Z" }, + { url = "https://files.pythonhosted.org/packages/26/5f/c7b6c6dd759abac18bf2634a8e9d40b95ff414fbfc91cfebeb9c3e8bc06b/rpm_version-0.5.1-cp310-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:1381a3dce1713a8320da21e193b95593989b7bf5c01559a81881e0c9f318436f", size = 312904, upload-time = "2026-07-24T21:06:08.362Z" }, + { url = "https://files.pythonhosted.org/packages/1e/56/58f114ccd1687d49bd2987bd3c911132d392a5166107e9862c92c535023e/rpm_version-0.5.1-cp310-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:54619ce73b9b8d23460a5e73fd0be1b3c2923f32252e6b42c4988b91f6ad3102", size = 315585, upload-time = "2026-07-24T21:06:09.883Z" }, + { url = "https://files.pythonhosted.org/packages/9d/45/03e4e3c0384874100ed308682a157fadc45e58b2262a54a0caf452baf295/rpm_version-0.5.1-cp310-abi3-win_amd64.whl", hash = "sha256:78fa7ed2f588a164b87c4ee2e814445aaf4932f3dc82ab82ba07e574e585f1d2", size = 167962, upload-time = "2026-07-24T21:06:11.423Z" }, +] + [[package]] name = "ruff" version = "0.16.3" @@ -1191,6 +1203,7 @@ dependencies = [ { name = "pytest" }, { name = "pytest-dependency" }, { name = "requests" }, + { name = "rpm-version" }, ] [package.dev-dependencies] @@ -1228,6 +1241,7 @@ requires-dist = [ { name = "pytest", specifier = ">=9.1.1" }, { name = "pytest-dependency" }, { name = "requests" }, + { name = "rpm-version", specifier = ">=0.5.1" }, ] [package.metadata.requires-dev]