add traffic rules tests:

- simple VIF rule (add/delete with simple vm.start/destroy cycle)
- simple Network rule (add/delete with simple vm.start/destroy cycle)
- migrate with simple VIF rule
- migrate with Network rule rule
- VLAN with simple VIF rule
- VLAN with simple Network rule
- Tunnel with simple VIF rule
- Tunnel with simple Network rule

fixtures:
- add connected_hosts_with_xo: returns the list of hosts which are already connected to xo-cli
- add VLAN fixture: returns a configured VLAN network

lib
- add VLAN abstraction

Signed-off-by: Sebastien Marie <semarie@kapouay.eu.org>

Author: semarie

conftest.py

@@ -267,6 +267,11 @@ def hosts_with_xo(hosts, registered_xo_cli):
             logging.info("<<< Disconnect host %s" % h)
             h.xo_server_remove()
 
+@pytest.fixture(scope='session')
+def connected_hosts_with_xo(hosts: list[Host], registered_xo_cli):
+    connected = [h for h in hosts if not h.skip_xo_config and h.xo_server_connected()]
+    yield connected
+
 @pytest.fixture(scope='session')
 def hostA1(hosts):
     """ Master of first pool (pool A). """

jobs.py

@@ -470,6 +470,8 @@
     "tests/migration/test_host_evacuate.py::TestHostEvacuateWithNetwork",
     # not really broken but has complex prerequisites (3 NICs)
     "tests/network/test_bond.py::test_bond",
+    # not really broken but has complex prerequisites (xo-cli + 2 hosts)
+    "tests/network/test_traffic_rules.py",
     # running quicktest on zfsvol generates dangling TAP devices that are hard to
     # cleanup. Bug needs to be fixed before enabling quicktest on zfsvol.
     "tests/storage/zfsvol/test_zfsvol_sr.py::TestZfsvolVm::test_quicktest",

lib/host.py

@@ -36,7 +36,9 @@
 from lib.network import Network
 from lib.pif import PIF
 from lib.sr import SR
+from lib.tunnel import Tunnel
 from lib.vdi import VDI
+from lib.vlan import VLAN
 from lib.vm import VM
 from lib.xo import xo_cli, xo_object_exists
 
@@ -808,3 +810,35 @@ def create_network(self, label: str, description: Optional[str] = None) -> Netwo
         logging.info(f"New Network: {uuid}")
 
         return Network(self, uuid)
+
+    def create_vlan(self, network: Network, pif: PIF, vlan: int) -> VLAN:
+        args: dict[str, str | bool] = {
+            '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] = {
+            '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)

lib/network.py

@@ -49,3 +49,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

lib/pif.py

@@ -51,6 +51,18 @@ 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 = self.param_get('VLAN')
+        if vlan is None:
+            return None
+        else:
+            return int(vlan)
+
     def reconfigure_ip(self, mode: str) -> None:
         self.host.xe("pif-reconfigure-ip", {
             "uuid": self.uuid,

lib/tunnel.py

@@ -0,0 +1,47 @@
+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
+
+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
+
+    def param_get(self, param_name, key=None, accept_unknown_key=False):
+        return _param_get(self.host, Tunnel.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    def param_set(self, param_name, value, key=None):
+        _param_set(self.host, Tunnel.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_add(self, param_name, value, key=None):
+        _param_add(self.host, Tunnel.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_clear(self, param_name):
+        _param_clear(self.host, Tunnel.xe_prefix, self.uuid, param_name)
+
+    def param_remove(self, param_name, key, accept_unknown_key=False):
+        _param_remove(self.host, Tunnel.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    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)

lib/vif.py

@@ -1,6 +1,7 @@
 import logging
 
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
+from lib.network import Network
 
 class VIF:
     xe_prefix = "vif"
@@ -42,6 +43,11 @@ def mac_address(self) -> str:
         assert mac_address is not None, "VIF must have a MAC address"
         return mac_address
 
+    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)
+
     def plug(self):
         logging.info("Plugging VIF %s on VM %s", self.param_get('device'), self.vm.uuid)
         self.vm.host.xe('vif-plug', {'uuid': self.uuid})

lib/vlan.py

@@ -0,0 +1,52 @@
+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
+
+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
+
+    def param_get(self, param_name, key=None, accept_unknown_key=False):
+        return _param_get(self.host, VLAN.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    def param_set(self, param_name, value, key=None):
+        _param_set(self.host, VLAN.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_add(self, param_name, value, key=None):
+        _param_add(self.host, VLAN.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_clear(self, param_name):
+        _param_clear(self.host, VLAN.xe_prefix, self.uuid, param_name)
+
+    def param_remove(self, param_name, key, accept_unknown_key=False):
+        _param_remove(self.host, VLAN.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    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)

tests/network/conftest.py

@@ -1,9 +1,15 @@
+from __future__ import annotations
+
 import pytest
 
 import logging
 
+from lib.bond import Bond
 from lib.host import Host
 from lib.network import Network
+from lib.tunnel import Tunnel
+from lib.vlan import VLAN
+from lib.xo import xo_cli
 
 from typing import Generator
 
@@ -14,14 +20,7 @@ def host_no_sdn_controller(host: Host):
         pytest.skip("This test requires an XCP-ng with no SDN controller")
 
 
-@pytest.fixture(scope='module')
-def empty_network(host: Host) -> Generator[Network, None, None]:
-    try:
-        net = host.create_network(label="empty_network for tests")
-        yield net
-    finally:
-        net.destroy()
-
+# ---- Bond ----
 @pytest.fixture(params=[])
 def bond_devices(request: pytest.FixtureRequest) -> list[str]:
     return request.param
@@ -31,7 +30,7 @@ def bond_mode(request: pytest.FixtureRequest) -> str:
     return request.param
 
 @pytest.fixture
-def bond(host: Host, empty_network: Network, bond_devices: list[str], bond_mode: str):
+def bond(host: Host, empty_network: Network, bond_devices: list[str], bond_mode: str) -> Generator[Bond, None, None]:
     pifs = []
     logging.info(f"bond: resolve PIFs on {host.hostname_or_ip} using \
         {[(pif.network_uuid(), pif.param_get('device')) for pif in host.pifs()]}")
@@ -44,3 +43,106 @@ def bond(host: Host, empty_network: Network, bond_devices: list[str], bond_mode:
         yield bond
     finally:
         bond.destroy()
+
+
+# ---- Network ----
+@pytest.fixture(scope='module')
+def empty_network(host: Host) -> Generator[Network, None, None]:
+    try:
+        net = host.create_network(label="empty_network for tests")
+        yield net
+    finally:
+        net.destroy()
+
+
+# ---- Tunnel ----
+@pytest.fixture(params=["eth0"])
+def tunnel_device(request: pytest.FixtureRequest) -> str:
+    return request.param
+
+@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(
+    connected_hosts_with_xo: list[Host],
+    tunnel_device: str, tunnel_protocol: str, tunnel_encryption: bool,
+) -> Generator[Tunnel, None, None]:
+    host = connected_hosts_with_xo[0]
+
+    # check system requirements
+    if host.ssh_with_result("rpm -q openvswitch-ipsec").returncode != 0:
+        pytest.skip("'tunnel' fixture requires configuration, see https://docs.xen-orchestra.com/sdn_controller")
+
+    logging.info(f"tunnel: resolve PIF on {host.hostname_or_ip} using \
+        {[(pif.network_uuid(), pif.param_get('device')) for pif in host.pifs()]}")
+
+    [pif] = host.pifs(device=tunnel_device)
+    if pif.ip_configuration_mode() == "None":
+        pytest.skip(f"'tunnel' fixture requires tunnel_device={tunnel_device} to have configured IP")
+
+    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',
+    })
+    tunnel = None
+    network = None
+    try:
+        # get Tunnel from PIF
+        tunnel_uuid = host.xe('tunnel-list', {
+            'transport-PIF': pif.uuid,
+        }, minimal=True)
+        tunnel = Tunnel(host, tunnel_uuid)
+
+        # get Network from Tunnel
+        network_uuid = tunnel.access_PIF().network_uuid()
+        network = Network(host, network_uuid)
+
+        yield tunnel
+    finally:
+        if tunnel is not None:
+            tunnel.destroy()
+
+        if network is not None:
+            # sdnController.createPrivateNetwork might have create several Tunnel (one per host)
+            # so get all Tunnel attached to Network and destroy them
+            for pif_uuid in network.pif_uuids():
+                tunnel_uuid = host.xe('tunnel-list', {
+                    'access-PIF': pif_uuid,
+                }, minimal=True)
+                tunnel = Tunnel(host, tunnel_uuid)
+                tunnel.destroy()
+
+            # finally destroy the Network
+            network.destroy()
+
+
+# ---- VLAN ----
+@pytest.fixture(params=["eth0"])
+def vlan_device(request: pytest.FixtureRequest) -> str:
+    return request.param
+
+@pytest.fixture(params=[0])
+def vlan_tag(request: pytest.FixtureRequest) -> int:
+    return request.param
+
+@pytest.fixture
+def vlan(host: Host, empty_network: Network, vlan_tag: int, vlan_device: str) -> 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()]}")
+
+    [pif] = host.pifs(device=vlan_device)
+    vlan = host.create_vlan(empty_network, pif, vlan_tag)
+    try:
+        yield vlan
+    finally:
+        vlan.destroy()