Add few bond tests

- add some tests for Bond
- add new abstractions (Network and Bond) with fixtures

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

Author: semarie

jobs.py

@@ -468,6 +468,8 @@
 BROKEN_TESTS = [
     # not really broken but has complex prerequisites (3 NICs on 3 different networks)
     "tests/migration/test_host_evacuate.py::TestHostEvacuateWithNetwork",
+    # not really broken but has complex prerequisites (3 NICs)
+    "tests/network/test_bond.py::test_bond",
     # 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/bond.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import logging
+
+from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set, safe_split
+from lib.pif import PIF
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from lib.host import Host
+
+class Bond:
+    xe_prefix = "bond"
+
+    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, Bond.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    def param_set(self, param_name, value, key=None):
+        _param_set(self.host, Bond.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_add(self, param_name, value, key=None):
+        _param_add(self.host, Bond.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_clear(self, param_name):
+        _param_clear(self.host, Bond.xe_prefix, self.uuid, param_name)
+
+    def param_remove(self, param_name, key, accept_unknown_key=False):
+        _param_remove(self.host, Bond.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    def destroy(self):
+        logging.info(f"Destroying bond: {self.uuid}")
+        self.host.xe('bond-destroy', {'uuid': self.uuid})
+
+    def master(self) -> PIF:
+        uuid = self.param_get('master')
+        assert uuid is not None, "no master on Bond"
+        return PIF(uuid, self.host)
+
+    def slaves(self) -> list[str]:
+        return safe_split(self.param_get('slaves'))
+
+    def mode(self) -> str | None:
+        return self.param_get("mode")

lib/common.py

@@ -51,6 +51,7 @@ class PackageManagerEnum(Enum):
     UNKNOWN = 1
     RPM = 2
     APT_GET = 3
+    APK = 4
 
 # Common VM images used in tests
 def vm_image(vm_key):

lib/host.py

@@ -16,6 +16,7 @@
 if TYPE_CHECKING:
     from lib.pool import Pool
 
+from lib.bond import Bond
 from lib.common import (
     DiskDevName,
     _param_add,
@@ -32,6 +33,7 @@
     wait_for_not,
 )
 from lib.netutil import wrap_ip
+from lib.network import Network
 from lib.pif import PIF
 from lib.sr import SR
 from lib.vdi import VDI
@@ -768,3 +770,41 @@ def lvs(self, vgName: Optional[str] = None, ignore_MGT: bool = True) -> List[str
                 continue
             ret.append(line.strip())
         return ret
+
+    def pifs(self, device: Optional[str] = None) -> list[PIF]:
+        args: Dict[str, str | bool] = {
+            "host-uuid": self.uuid,
+        }
+
+        if device is not None:
+            args["device"] = device
+
+        return [PIF(uuid, self) for uuid in safe_split(self.xe("pif-list", args, minimal=True))]
+
+    def create_bond(self, network: Network, pifs: list[PIF], mode: Optional[str] = None) -> Bond:
+        args: dict[str, str | bool] = {
+            'network-uuid': network.uuid,
+            'pif-uuids': ','.join([pif.uuid for pif in pifs]),
+        }
+
+        if mode is not None:
+            args['mode'] = mode
+
+        uuid = self.xe("bond-create", args, minimal=True)
+        logging.info(f"New Bond: {uuid}")
+
+        return Bond(self, uuid)
+
+    def create_network(self, label: str, description: Optional[str] = None) -> Network:
+        args: dict[str, str | bool] = {
+            'name-label': label,
+        }
+
+        if description is not None:
+            args['name-description'] = description
+
+        logging.info(f"Creating network '{label}'")
+        uuid = self.xe("network-create", args, minimal=True)
+        logging.info(f"New Network: {uuid}")
+
+        return Network(self, uuid)

lib/network.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import logging
+
+from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set, safe_split
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from lib.host import Host
+
+class Network:
+    xe_prefix = "network"
+
+    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, Network.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    def param_set(self, param_name, value, key=None):
+        _param_set(self.host, Network.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_add(self, param_name, value, key=None):
+        _param_add(self.host, Network.xe_prefix, self.uuid, param_name, value, key)
+
+    def param_clear(self, param_name):
+        _param_clear(self.host, Network.xe_prefix, self.uuid, param_name)
+
+    def param_remove(self, param_name, key, accept_unknown_key=False):
+        _param_remove(self.host, Network.xe_prefix, self.uuid, param_name, key, accept_unknown_key)
+
+    def destroy(self):
+        logging.info(f"Destroying network '{self.param_get('name-label')}': {self.uuid}")
+        self.host.xe('network-destroy', {'uuid': self.uuid})
+
+    def pif_uuids(self) -> list[str]:
+        return safe_split(self.param_get('PIF-uuids'), '; ')
+
+    def vif_uuids(self) -> list[str]:
+        return safe_split(self.param_get('VIF-uuids'), '; ')
+
+    def is_private(self) -> bool:
+        return len(self.pif_uuids()) == 0
+
+    def managed(self) -> bool:
+        return self.param_get('managed') == 'true'
+
+    def MTU(self) -> int:
+        return int(self.param_get('MTU') or '0')

lib/pif.py

@@ -1,9 +1,16 @@
-from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
+from __future__ import annotations
+
+from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set, strtobool
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from lib.host import Host
 
 class PIF:
     xe_prefix = "pif"
 
-    def __init__(self, uuid, host):
+    def __init__(self, uuid: str, host: Host):
         self.uuid = uuid
         self.host = host
 
@@ -26,3 +33,32 @@ def param_add(self, param_name, value, key=None):
     def param_clear(self, param_name):
         _param_clear(self.host, self.xe_prefix, self.uuid,
                      param_name)
+
+    def is_managed(self) -> bool:
+        return strtobool(self.param_get("managed"))
+
+    def is_physical(self) -> bool:
+        return strtobool(self.param_get("physical"))
+
+    def is_currently_attached(self) -> bool:
+        return strtobool(self.param_get("currently-attached"))
+
+    def is_management(self) -> bool:
+        return strtobool(self.param_get("management"))
+
+    def network_uuid(self) -> str:
+        uuid = self.param_get("network-uuid")
+        assert uuid is not None, "unexpected PIF without network-uuid"
+        return uuid
+
+    def reconfigure_ip(self, mode: str) -> None:
+        self.host.xe("pif-reconfigure-ip", {
+            "uuid": self.uuid,
+            "mode": mode,
+        })
+
+    def reconfigure_ipv6(self, mode: str) -> None:
+        self.host.xe("pif-reconfigure-ipv6", {
+            "uuid": self.uuid,
+            "mode": mode,
+        })

lib/vm.py

@@ -478,6 +478,8 @@ def detect_package_manager(self):
             return PackageManagerEnum.RPM
         elif self.file_exists('/usr/bin/apt-get'):
             return PackageManagerEnum.APT_GET
+        elif self.file_exits('/sbin/apk'):
+            return PackageManagerEnum.APK
         else:
             return PackageManagerEnum.UNKNOWN
 

tests/network/conftest.py

@@ -1,7 +1,47 @@
 import pytest
 
+import logging
+
+from lib.host import Host
+from lib.network import Network
+
+from typing import Generator
+
 @pytest.fixture(scope='package')
-def host_no_sdn_controller(host):
+def host_no_sdn_controller(host: Host):
     """ An XCP-ng with no SDN controller. """
     if host.xe('sdn-controller-list', minimal=True):
         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:
+        if net is not None:
+            net.destroy()
+
+@pytest.fixture(params=[])
+def bond_devices(request: pytest.FixtureRequest) -> list[str]:
+    return request.param
+
+@pytest.fixture(params=["lacp"])
+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):
+    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()]}")
+    for name in bond_devices:
+        [pif] = host.pifs(device=name)
+        pifs.append(pif)
+
+    bond = host.create_bond(empty_network, pifs, mode=bond_mode)
+    try:
+        yield bond
+    finally:
+        bond.destroy()