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

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

@@ -37,6 +37,7 @@
 from lib.pif import PIF
 from lib.sr import SR
 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 +809,19 @@ 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)

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

@@ -2,7 +2,7 @@
 
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set, strtobool
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
     from lib.host import Host
@@ -51,6 +51,13 @@ def network_uuid(self) -> str:
         assert uuid is not None, "unexpected PIF without network-uuid"
         return uuid
 
+    def vlan(self) -> Optional[int]:
+        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/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

@@ -22,6 +22,27 @@ def empty_network(host: Host) -> Generator[Network, None, None]:
     finally:
         net.destroy()
 
+
+@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):
+    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()
+
 @pytest.fixture(params=[])
 def bond_devices(request: pytest.FixtureRequest) -> list[str]:
     return request.param