Add more abstractions for Host, Network, PIF, Tunnel, VIF, VLAN

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
 
@@ -1050,6 +1052,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,
@@ -1077,3 +1082,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

@@ -3,6 +3,7 @@
 import logging
 
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
+from lib.network import Network
 
 from typing import TYPE_CHECKING
 
@@ -64,3 +65,8 @@ def plug(self) -> None:
     def unplug(self) -> None:
         logging.info("Unplugging VIF %s on VM %s", self.param_get('device'), self.vm.uuid)
         self.vm.host.xe('vif-unplug', {'uuid': self.uuid})
+
+    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)