daemon: updated core.nodes to avoid using deprecated type hinting

daemon/core/nodes/base.py

@@ -9,7 +9,7 @@ import threading
 from dataclasses import dataclass, field
 from pathlib import Path
 from threading import RLock
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type, Union
+from typing import TYPE_CHECKING, Optional, Union
 
 import netaddr
 
@@ -29,10 +29,10 @@ if TYPE_CHECKING:
     from core.configservice.base import ConfigService
     from core.services.coreservices import CoreService
 
-    CoreServices = List[Union[CoreService, Type[CoreService]]]
+    CoreServices = list[Union[CoreService, type[CoreService]]]
-    ConfigServiceType = Type[ConfigService]
+    ConfigServiceType = type[ConfigService]
 
-PRIVATE_DIRS: List[Path] = [Path("/var/run"), Path("/var/log")]
+PRIVATE_DIRS: list[Path] = [Path("/var/run"), Path("/var/log")]
 
 
 @dataclass
@@ -64,7 +64,7 @@ class Position:
         self.z = z
         return True
 
-    def get(self) -> Tuple[float, float, float]:
+    def get(self) -> tuple[float, float, float]:
         """
         Retrieve x,y,z position.
 
@@ -88,7 +88,7 @@ class Position:
         self.lat = lat
         self.alt = alt
 
-    def get_geo(self) -> Tuple[float, float, float]:
+    def get_geo(self) -> tuple[float, float, float]:
         """
         Retrieve current geo position lon, lat, alt.
 
@@ -113,9 +113,9 @@ class NodeOptions:
 class CoreNodeOptions(NodeOptions):
     model: str = "PC"
     """model is used for providing a default set of services"""
-    services: List[str] = field(default_factory=list)
+    services: list[str] = field(default_factory=list)
     """services to start within node"""
-    config_services: List[str] = field(default_factory=list)
+    config_services: list[str] = field(default_factory=list)
     """config services to start within node"""
     directory: Path = None
     """directory to define node, defaults to path under the session directory"""
@@ -152,7 +152,7 @@ class NodeBase(abc.ABC):
         self.server: "DistributedServer" = server
         self.model: Optional[str] = None
         self.services: CoreServices = []
-        self.ifaces: Dict[int, CoreInterface] = {}
+        self.ifaces: dict[int, CoreInterface] = {}
         self.iface_id: int = 0
         self.position: Position = Position()
         self.up: bool = False
@@ -201,7 +201,7 @@ class NodeBase(abc.ABC):
     def host_cmd(
         self,
         args: str,
-        env: Dict[str, str] = None,
+        env: dict[str, str] = None,
         cwd: Path = None,
         wait: bool = True,
         shell: bool = False,
@@ -246,7 +246,7 @@ class NodeBase(abc.ABC):
         """
         return self.position.set(x=x, y=y, z=z)
 
-    def getposition(self) -> Tuple[float, float, float]:
+    def getposition(self) -> tuple[float, float, float]:
         """
         Return an (x,y,z) tuple representing this object's position.
 
@@ -331,7 +331,7 @@ class NodeBase(abc.ABC):
             raise CoreError(f"node({self.name}) does not have interface({iface_id})")
         return self.ifaces[iface_id]
 
-    def get_ifaces(self, control: bool = True) -> List[CoreInterface]:
+    def get_ifaces(self, control: bool = True) -> list[CoreInterface]:
         """
         Retrieve sorted list of interfaces, optionally do not include control
         interfaces.
@@ -395,7 +395,7 @@ class CoreNodeBase(NodeBase):
             will run on, default is None for localhost
         """
         super().__init__(session, _id, name, server, options)
-        self.config_services: Dict[str, "ConfigService"] = {}
+        self.config_services: dict[str, "ConfigService"] = {}
         self.directory: Optional[Path] = None
         self.tmpnodedir: bool = False
 
@@ -481,7 +481,7 @@ class CoreNodeBase(NodeBase):
             raise CoreError(f"node({self.name}) already has service({name})")
         self.config_services[name] = service_class(self)
 
-    def set_service_config(self, name: str, data: Dict[str, str]) -> None:
+    def set_service_config(self, name: str, data: dict[str, str]) -> None:
         """
         Sets configuration service custom config data.
 
@@ -574,7 +574,7 @@ class CoreNode(CoreNodeBase):
         self.directory: Optional[Path] = options.directory
         self.ctrlchnlname: Path = self.session.directory / self.name
         self.pid: Optional[int] = None
-        self._mounts: List[Tuple[Path, Path]] = []
+        self._mounts: list[tuple[Path, Path]] = []
         self.node_net_client: LinuxNetClient = self.create_node_net_client(
             self.session.use_ovs()
         )
@@ -928,7 +928,7 @@ class CoreNetworkBase(NodeBase):
         mtu = self.session.options.get_int("mtu")
         self.mtu: int = mtu if mtu > 0 else DEFAULT_MTU
         self.brname: Optional[str] = None
-        self.linked: Dict[CoreInterface, Dict[CoreInterface, bool]] = {}
+        self.linked: dict[CoreInterface, dict[CoreInterface, bool]] = {}
         self.linked_lock: threading.Lock = threading.Lock()
 
     def attach(self, iface: CoreInterface) -> None:

daemon/core/nodes/docker.py

@@ -4,7 +4,7 @@ import shlex
 from dataclasses import dataclass, field
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import TYPE_CHECKING, Dict, List, Tuple
+from typing import TYPE_CHECKING
 
 from core.emulator.distributed import DistributedServer
 from core.errors import CoreCommandError, CoreError
@@ -23,9 +23,9 @@ DOCKER: str = "docker"
 class DockerOptions(CoreNodeOptions):
     image: str = "ubuntu"
     """image used when creating container"""
-    binds: List[Tuple[str, str]] = field(default_factory=list)
+    binds: list[tuple[str, str]] = field(default_factory=list)
     """bind mount source and destinations to setup within container"""
-    volumes: List[Tuple[str, str, bool, bool]] = field(default_factory=list)
+    volumes: list[tuple[str, str, bool, bool]] = field(default_factory=list)
     """
     volume mount source, destination, unique, delete to setup within container
 
@@ -74,8 +74,8 @@ class DockerNode(CoreNode):
         options = options or DockerOptions()
         super().__init__(session, _id, name, server, options)
         self.image: str = options.image
-        self.binds: List[Tuple[str, str]] = options.binds
+        self.binds: list[tuple[str, str]] = options.binds
-        self.volumes: Dict[str, DockerVolume] = {}
+        self.volumes: dict[str, DockerVolume] = {}
         for src, dst, unique, delete in options.volumes:
             src_name = self._unique_name(src) if unique else src
             self.volumes[src] = DockerVolume(src_name, dst, unique, delete)

daemon/core/nodes/interface.py

@@ -5,7 +5,7 @@ virtual ethernet classes that implement the interfaces available under Linux.
 import logging
 import math
 from pathlib import Path
-from typing import TYPE_CHECKING, Callable, Dict, List, Optional
+from typing import TYPE_CHECKING, Callable, Optional
 
 import netaddr
 
@@ -114,8 +114,8 @@ class CoreInterface:
         self.up: bool = False
         self.mtu: int = mtu
         self.net: Optional[CoreNetworkBase] = None
-        self.ip4s: List[netaddr.IPNetwork] = []
+        self.ip4s: list[netaddr.IPNetwork] = []
-        self.ip6s: List[netaddr.IPNetwork] = []
+        self.ip6s: list[netaddr.IPNetwork] = []
         self.mac: Optional[netaddr.EUI] = None
         # placeholder position hook
         self.poshook: Callable[[CoreInterface], None] = lambda x: None
@@ -133,7 +133,7 @@ class CoreInterface:
     def host_cmd(
         self,
         args: str,
-        env: Dict[str, str] = None,
+        env: dict[str, str] = None,
         cwd: Path = None,
         wait: bool = True,
         shell: bool = False,
@@ -235,7 +235,7 @@ class CoreInterface:
         """
         return next(iter(self.ip6s), None)
 
-    def ips(self) -> List[netaddr.IPNetwork]:
+    def ips(self) -> list[netaddr.IPNetwork]:
         """
         Retrieve a list of all ip4 and ip6 addresses combined.
 

daemon/core/nodes/lxd.py

@@ -5,7 +5,7 @@ import time
 from dataclasses import dataclass, field
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import TYPE_CHECKING, Dict, List, Tuple
+from typing import TYPE_CHECKING
 
 from core.emulator.data import InterfaceData, LinkOptions
 from core.emulator.distributed import DistributedServer
@@ -24,9 +24,9 @@ if TYPE_CHECKING:
 class LxcOptions(CoreNodeOptions):
     image: str = "ubuntu"
     """image used when creating container"""
-    binds: List[Tuple[str, str]] = field(default_factory=list)
+    binds: list[tuple[str, str]] = field(default_factory=list)
     """bind mount source and destinations to setup within container"""
-    volumes: List[Tuple[str, str, bool, bool]] = field(default_factory=list)
+    volumes: list[tuple[str, str, bool, bool]] = field(default_factory=list)
     """
     volume mount source, destination, unique, delete to setup within container
 
@@ -74,7 +74,7 @@ class LxcNode(CoreNode):
             args = f"{BASH} -c {shlex.quote(args)}"
         return f"nsenter -t {self.pid} -m -u -i -p -n {args}"
 
-    def _get_info(self) -> Dict:
+    def _get_info(self) -> dict:
         args = f"lxc list {self.name} --format json"
         output = self.host_cmd(args)
         data = json.loads(output)

daemon/core/nodes/network.py

@@ -6,7 +6,7 @@ import logging
 import threading
 from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Dict, List, Optional, Type
+from typing import TYPE_CHECKING, Optional
 
 import netaddr
 
@@ -51,7 +51,7 @@ class NftablesQueue:
         # this lock protects cmds and updates lists
         self.lock: threading.Lock = threading.Lock()
         # list of pending nftables commands
-        self.cmds: List[str] = []
+        self.cmds: list[str] = []
         # list of WLANs requiring update
         self.updates: utils.SetQueue = utils.SetQueue()
 
@@ -226,7 +226,7 @@ class CoreNetwork(CoreNetworkBase):
     def host_cmd(
         self,
         args: str,
-        env: Dict[str, str] = None,
+        env: dict[str, str] = None,
         cwd: Path = None,
         wait: bool = True,
         shell: bool = False,
@@ -448,7 +448,7 @@ class GreTapBridge(CoreNetwork):
             self.gretap = None
         super().shutdown()
 
-    def add_ips(self, ips: List[str]) -> None:
+    def add_ips(self, ips: list[str]) -> None:
         """
         Set the remote tunnel endpoint. This is a one-time method for
         creating the GreTap device, which requires the remoteip at startup.
@@ -512,7 +512,7 @@ class CtrlNet(CoreNetwork):
     policy: NetworkPolicy = NetworkPolicy.ACCEPT
     # base control interface index
     CTRLIF_IDX_BASE: int = 99
-    DEFAULT_PREFIX_LIST: List[str] = [
+    DEFAULT_PREFIX_LIST: list[str] = [
         "172.16.0.0/24 172.16.1.0/24 172.16.2.0/24 172.16.3.0/24 172.16.4.0/24",
         "172.17.0.0/24 172.17.1.0/24 172.17.2.0/24 172.17.3.0/24 172.17.4.0/24",
         "172.18.0.0/24 172.18.1.0/24 172.18.2.0/24 172.18.3.0/24 172.18.4.0/24",
@@ -734,7 +734,7 @@ class WlanNode(CoreNetwork):
             iface.poshook = self.wireless_model.position_callback
             iface.setposition()
 
-    def setmodel(self, wireless_model: Type["WirelessModel"], config: Dict[str, str]):
+    def setmodel(self, wireless_model: type["WirelessModel"], config: dict[str, str]):
         """
         Sets the mobility and wireless model.
 
@@ -753,12 +753,12 @@ class WlanNode(CoreNetwork):
             self.mobility = wireless_model(session=self.session, _id=self.id)
             self.mobility.update_config(config)
 
-    def update_mobility(self, config: Dict[str, str]) -> None:
+    def update_mobility(self, config: dict[str, str]) -> None:
         if not self.mobility:
             raise CoreError(f"no mobility set to update for node({self.name})")
         self.mobility.update_config(config)
 
-    def updatemodel(self, config: Dict[str, str]) -> None:
+    def updatemodel(self, config: dict[str, str]) -> None:
         if not self.wireless_model:
             raise CoreError(f"no model set to update for node({self.name})")
         logger.debug(
@@ -768,7 +768,7 @@ class WlanNode(CoreNetwork):
         for iface in self.get_ifaces():
             iface.setposition()
 
-    def links(self, flags: MessageFlags = MessageFlags.NONE) -> List[LinkData]:
+    def links(self, flags: MessageFlags = MessageFlags.NONE) -> list[LinkData]:
         """
         Retrieve all link data.
 

daemon/core/nodes/physical.py

@@ -4,7 +4,7 @@ PhysicalNode class for including real systems in the emulated network.
 
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING, List, Optional, Tuple
+from typing import TYPE_CHECKING, Optional
 
 import netaddr
 
@@ -52,7 +52,7 @@ class Rj45Node(CoreNodeBase):
         )
         self.iface.transport_type = TransportType.RAW
         self.old_up: bool = False
-        self.old_addrs: List[Tuple[str, Optional[str]]] = []
+        self.old_addrs: list[tuple[str, Optional[str]]] = []
 
     def startup(self) -> None:
         """
@@ -159,7 +159,7 @@ class Rj45Node(CoreNodeBase):
         """
         # TODO: save/restore the PROMISC flag
         self.old_up = False
-        self.old_addrs: List[Tuple[str, Optional[str]]] = []
+        self.old_addrs: list[tuple[str, Optional[str]]] = []
         localname = self.iface.localname
         output = self.net_client.address_show(localname)
         for line in output.split("\n"):

daemon/core/nodes/wireless.py

@@ -7,7 +7,7 @@ import logging
 import math
 import secrets
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Dict, List, Set, Tuple
+from typing import TYPE_CHECKING
 
 from core.config import ConfigBool, ConfigFloat, ConfigInt, Configuration
 from core.emulator.data import LinkData, LinkOptions
@@ -41,7 +41,7 @@ KEY_LOSS: str = "loss"
 
 
 def calc_distance(
-    point1: Tuple[float, float, float], point2: Tuple[float, float, float]
+    point1: tuple[float, float, float], point2: tuple[float, float, float]
 ) -> float:
     a = point1[0] - point2[0]
     b = point1[1] - point2[1]
@@ -51,7 +51,7 @@ def calc_distance(
     return math.hypot(math.hypot(a, b), c)
 
 
-def get_key(node1_id: int, node2_id: int) -> Tuple[int, int]:
+def get_key(node1_id: int, node2_id: int) -> tuple[int, int]:
     return (node1_id, node2_id) if node1_id < node2_id else (node2_id, node1_id)
 
 
@@ -65,7 +65,7 @@ class WirelessLink:
 
 
 class WirelessNode(CoreNetworkBase):
-    options: List[Configuration] = [
+    options: list[Configuration] = [
         ConfigBool(
             id=KEY_ENABLED, default="1" if CONFIG_ENABLED else "0", label="Enabled?"
         ),
@@ -87,7 +87,7 @@ class WirelessNode(CoreNetworkBase):
         ),
         ConfigFloat(id=KEY_LOSS, default=str(CONFIG_LOSS), label="Loss Initial"),
     ]
-    devices: Set[str] = set()
+    devices: set[str] = set()
 
     @classmethod
     def add_device(cls) -> str:
@@ -111,8 +111,8 @@ class WirelessNode(CoreNetworkBase):
         options: NodeOptions = None,
     ):
         super().__init__(session, _id, name, server, options)
-        self.bridges: Dict[int, Tuple[CoreInterface, str]] = {}
+        self.bridges: dict[int, tuple[CoreInterface, str]] = {}
-        self.links: Dict[Tuple[int, int], WirelessLink] = {}
+        self.links: dict[tuple[int, int], WirelessLink] = {}
         self.position_enabled: bool = CONFIG_ENABLED
         self.bandwidth: int = CONFIG_BANDWIDTH
         self.delay: int = CONFIG_DELAY
@@ -321,7 +321,7 @@ class WirelessNode(CoreNetworkBase):
     def adopt_iface(self, iface: CoreInterface, name: str) -> None:
         raise CoreError(f"{type(self)} does not support adopt interface")
 
-    def get_config(self) -> Dict[str, Configuration]:
+    def get_config(self) -> dict[str, Configuration]:
         config = {x.id: x for x in copy.copy(self.options)}
         config[KEY_ENABLED].default = "1" if self.position_enabled else "0"
         config[KEY_RANGE].default = str(self.max_range)
@@ -333,7 +333,7 @@ class WirelessNode(CoreNetworkBase):
         config[KEY_JITTER].default = str(self.jitter)
         return config
 
-    def set_config(self, config: Dict[str, str]) -> None:
+    def set_config(self, config: dict[str, str]) -> None:
         logger.info("wireless config: %s", config)
         self.position_enabled = config[KEY_ENABLED] == "1"
         self.max_range = float(config[KEY_RANGE])