afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

added type hints to class variables in nodes/physical.py

Browse source
This commit is contained in:
Blake Harnden 2020-05-25 11:33:59 -07:00
parent f95a8113c9
commit 8fed201fd8
1 changed files with 14 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
daemon/core/nodes/physical.py
View file

@ -5,7 +5,7 @@ PhysicalNode class for including real systems in the emulated network.
import logging import logging
import os import os
import threading import threading
from typing import IO, TYPE_CHECKING, List, Optional from typing import IO, TYPE_CHECKING, List, Optional, Tuple
from core import utils from core import utils
from core.constants import MOUNT_BIN, UMOUNT_BIN from core.constants import MOUNT_BIN, UMOUNT_BIN
@ -23,7 +23,7 @@ if TYPE_CHECKING:
class PhysicalNode(CoreNodeBase): class PhysicalNode(CoreNodeBase):
def __init__( def __init__(
self, self,
session, session: "Session",
_id: int = None, _id: int = None,
name: str = None, name: str = None,
nodedir: str = None, nodedir: str = None,
@ -33,10 +33,10 @@ class PhysicalNode(CoreNodeBase):
super().__init__(session, _id, name, start, server) super().__init__(session, _id, name, start, server)
if not self.server: if not self.server:
raise CoreError("physical nodes must be assigned to a remote server") raise CoreError("physical nodes must be assigned to a remote server")
self.nodedir = nodedir self.nodedir: Optional[str] = nodedir
self.up = start self.up: bool = start
self.lock = threading.RLock() self.lock: threading.RLock = threading.RLock()
self._mounts = [] self._mounts: List[Tuple[str, str]] = []
if start: if start:
self.startup() self.startup()
@ -112,7 +112,7 @@ class PhysicalNode(CoreNodeBase):
logging.exception("trying to delete unknown address: %s", addr) logging.exception("trying to delete unknown address: %s", addr)
if self.up: if self.up:
self.net_client.delete_address(interface.name, str(addr)) self.net_client.delete_address(interface.name, addr)
def adoptnetif( def adoptnetif(
self, netif: CoreInterface, ifindex: int, hwaddr: str, addrlist: List[str] self, netif: CoreInterface, ifindex: int, hwaddr: str, addrlist: List[str]
@ -256,8 +256,8 @@ class Rj45Node(CoreNodeBase, CoreInterface):
network. network.
""" """
apitype = NodeTypes.RJ45 apitype: NodeTypes = NodeTypes.RJ45
type = "rj45" type: str = "rj45"
def __init__( def __init__(
self, self,
@ -281,11 +281,11 @@ class Rj45Node(CoreNodeBase, CoreInterface):
""" """
CoreNodeBase.__init__(self, session, _id, name, start, server) CoreNodeBase.__init__(self, session, _id, name, start, server)
CoreInterface.__init__(self, session, self, name, name, mtu, server) CoreInterface.__init__(self, session, self, name, name, mtu, server)
self.lock = threading.RLock() self.lock: threading.RLock = threading.RLock()
self.ifindex = None self.ifindex: Optional[int] = None
self.transport_type = "raw" self.transport_type: str = "raw"
self.old_up = False self.old_up: bool = False
self.old_addrs = [] self.old_addrs: List[str] = []
if start: if start:
self.startup() self.startup()