daemon: cleaned up InterfaceData class, it now leverages dataclass, removed extra bloat and no longer requires parameters as they are optional

This commit is contained in:
Blake Harnden 2020-06-09 08:48:18 -07:00
parent 6ddf1ac9a4
commit 18044f9474
5 changed files with 31 additions and 85 deletions

View file

@ -59,19 +59,17 @@ def link_interface(interface_proto: core_pb2.Interface) -> InterfaceData:
""" """
interface = None interface = None
if interface_proto: if interface_proto:
name = interface_proto.name name = interface_proto.name if interface_proto.name else None
if name == "": mac = interface_proto.mac if interface_proto.mac else None
name = None ip4 = interface_proto.ip4 if interface_proto.ip4 else None
mac = interface_proto.mac ip6 = interface_proto.ip6 if interface_proto.ip6 else None
if mac == "":
mac = None
interface = InterfaceData( interface = InterfaceData(
_id=interface_proto.id, id=interface_proto.id,
name=name, name=name,
mac=mac, mac=mac,
ip4=interface_proto.ip4, ip4=ip4,
ip4_mask=interface_proto.ip4mask, ip4_mask=interface_proto.ip4mask,
ip6=interface_proto.ip6, ip6=ip6,
ip6_mask=interface_proto.ip6mask, ip6_mask=interface_proto.ip6mask,
) )
return interface return interface

View file

@ -749,7 +749,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
node_two_id = message.get_tlv(LinkTlvs.N2_NUMBER.value) node_two_id = message.get_tlv(LinkTlvs.N2_NUMBER.value)
interface_one = InterfaceData( interface_one = InterfaceData(
_id=message.get_tlv(LinkTlvs.INTERFACE1_NUMBER.value), id=message.get_tlv(LinkTlvs.INTERFACE1_NUMBER.value),
name=message.get_tlv(LinkTlvs.INTERFACE1_NAME.value), name=message.get_tlv(LinkTlvs.INTERFACE1_NAME.value),
mac=message.get_tlv(LinkTlvs.INTERFACE1_MAC.value), mac=message.get_tlv(LinkTlvs.INTERFACE1_MAC.value),
ip4=message.get_tlv(LinkTlvs.INTERFACE1_IP4.value), ip4=message.get_tlv(LinkTlvs.INTERFACE1_IP4.value),
@ -758,7 +758,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
ip6_mask=message.get_tlv(LinkTlvs.INTERFACE1_IP6_MASK.value), ip6_mask=message.get_tlv(LinkTlvs.INTERFACE1_IP6_MASK.value),
) )
interface_two = InterfaceData( interface_two = InterfaceData(
_id=message.get_tlv(LinkTlvs.INTERFACE2_NUMBER.value), id=message.get_tlv(LinkTlvs.INTERFACE2_NUMBER.value),
name=message.get_tlv(LinkTlvs.INTERFACE2_NAME.value), name=message.get_tlv(LinkTlvs.INTERFACE2_NAME.value),
mac=message.get_tlv(LinkTlvs.INTERFACE2_MAC.value), mac=message.get_tlv(LinkTlvs.INTERFACE2_MAC.value),
ip4=message.get_tlv(LinkTlvs.INTERFACE2_IP4.value), ip4=message.get_tlv(LinkTlvs.INTERFACE2_IP4.value),

View file

@ -1,4 +1,5 @@
from typing import List, Optional, Union from dataclasses import dataclass
from typing import List, Union
import netaddr import netaddr
@ -122,87 +123,32 @@ class LinkOptions:
self.opaque = None self.opaque = None
@dataclass
class InterfaceData: class InterfaceData:
""" """
Convenience class for storing interface data. Convenience class for storing interface data.
""" """
def __init__( id: int = None
self, name: str = None
_id: int, mac: str = None
name: str, ip4: str = None
mac: str, ip4_mask: int = None
ip4: str, ip6: str = None
ip4_mask: int, ip6_mask: int = None
ip6: str,
ip6_mask: int,
) -> None:
"""
Creates an InterfaceData object.
:param _id: interface id
:param name: name for interface
:param mac: mac address
:param ip4: ipv4 address
:param ip4_mask: ipv4 bit mask
:param ip6: ipv6 address
:param ip6_mask: ipv6 bit mask
"""
self.id = _id
self.name = name
self.mac = mac
self.ip4 = ip4
self.ip4_mask = ip4_mask
self.ip6 = ip6
self.ip6_mask = ip6_mask
def has_ip4(self) -> bool:
"""
Determines if interface has an ip4 address.
:return: True if has ip4, False otherwise
"""
return all([self.ip4, self.ip4_mask])
def has_ip6(self) -> bool:
"""
Determines if interface has an ip6 address.
:return: True if has ip6, False otherwise
"""
return all([self.ip6, self.ip6_mask])
def ip4_address(self) -> Optional[str]:
"""
Retrieve a string representation of the ip4 address and netmask.
:return: ip4 string or None
"""
if self.has_ip4():
return f"{self.ip4}/{self.ip4_mask}"
else:
return None
def ip6_address(self) -> Optional[str]:
"""
Retrieve a string representation of the ip6 address and netmask.
:return: ip4 string or None
"""
if self.has_ip6():
return f"{self.ip6}/{self.ip6_mask}"
else:
return None
def get_addresses(self) -> List[str]: def get_addresses(self) -> List[str]:
""" """
Returns a list of ip4 and ip6 address when present. Returns a list of ip4 and ip6 addresses when present.
:return: list of addresses :return: list of addresses
""" """
ip4 = self.ip4_address() addresses = []
ip6 = self.ip6_address() if self.ip4 and self.ip4_mask:
return [i for i in [ip4, ip6] if i] addresses.append(f"{self.ip4}/{self.ip4_mask}")
if self.ip6 and self.ip6_mask:
addresses.append(f"{self.ip6}/{self.ip6_mask}")
return addresses
class IpPrefixes: class IpPrefixes:
@ -285,7 +231,7 @@ class IpPrefixes:
mac = utils.random_mac() mac = utils.random_mac()
return InterfaceData( return InterfaceData(
_id=inteface_id, id=inteface_id,
name=name, name=name,
ip4=ip4, ip4=ip4,
ip4_mask=ip4_mask, ip4_mask=ip4_mask,

View file

@ -29,7 +29,9 @@ _EMANE_MODELS = [
_DIR = os.path.dirname(os.path.abspath(__file__)) _DIR = os.path.dirname(os.path.abspath(__file__))
def ping(from_node, to_node, ip_prefixes, count=3): def ping(
from_node: CoreNode, to_node: CoreNode, ip_prefixes: IpPrefixes, count: int = 3
):
address = ip_prefixes.ip4_address(to_node) address = ip_prefixes.ip4_address(to_node)
try: try:
from_node.cmd(f"ping -c {count} {address}") from_node.cmd(f"ping -c {count} {address}")

View file

@ -21,7 +21,7 @@ _MOBILITY_FILE = os.path.join(_PATH, "mobility.scen")
_WIRED = [PtpNet, HubNode, SwitchNode] _WIRED = [PtpNet, HubNode, SwitchNode]
def ping(from_node, to_node, ip_prefixes): def ping(from_node: CoreNode, to_node: CoreNode, ip_prefixes: IpPrefixes):
address = ip_prefixes.ip4_address(to_node) address = ip_prefixes.ip4_address(to_node)
try: try:
from_node.cmd(f"ping -c 1 {address}") from_node.cmd(f"ping -c 1 {address}")