lib: add more abstractions for Host, Network, PIF, Tunnel, VIF, VLAN

For several tests related to network (mostly traffic rules for now), we need 
more classes and functions to manipulate xapi objects.

- create and manipulate Tunnel
- create and manipulate VLAN
- additional properties on Network, PIF, VIF
- additional function on Host to get service status

Signed-off-by: Sebastien Rodot <sebastien.rodot@vates.tech>

Author: semarie

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)

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

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,

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)

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)

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)