some update with new structure
This commit is contained in:
@@ -0,0 +1,182 @@
|
|||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import List, Optional, Dict
|
||||||
|
|
||||||
|
|
||||||
|
# ===========================
|
||||||
|
# Interface Classes
|
||||||
|
# ===========================
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Interface:
|
||||||
|
name: str
|
||||||
|
ip_address: Optional[str] = None
|
||||||
|
mac_address: Optional[str] = None
|
||||||
|
speed: Optional[str] = None
|
||||||
|
duplex: Optional[str] = None
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
self.name = fields.get("name")
|
||||||
|
self.ip_address = fields.get("ip")
|
||||||
|
self.mac_address = fields.get("mac")
|
||||||
|
self.speed = fields.get("speed")
|
||||||
|
self.duplex = fields.get("duplex")
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if not isinstance(self.name, str) or not self.name:
|
||||||
|
raise ValueError("Interface 'name' must be a non-empty string")
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class VirtualInterface(Interface):
|
||||||
|
vlan_id: int = 0
|
||||||
|
parent_physical: Optional[Interface] = None
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
super().__init__(fields)
|
||||||
|
self.vlan_id = fields.get("vlanID", fields.get("vlan_id", 0))
|
||||||
|
self.parent_physical = fields.get(
|
||||||
|
"parent_physical", fields.get("parentPhysical", fields.get("parent"))
|
||||||
|
)
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
super().__post_init__()
|
||||||
|
if not isinstance(self.vlan_id, int) or self.vlan_id < 0:
|
||||||
|
raise ValueError(
|
||||||
|
"VirtualInterface 'vlan_id' must be a non-negative integer"
|
||||||
|
)
|
||||||
|
if not isinstance(self.parent_physical, Interface):
|
||||||
|
raise TypeError(
|
||||||
|
"VirtualInterface 'parent_physical' must reference an Interface"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ===========================
|
||||||
|
# Network / Subnet
|
||||||
|
# ===========================
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Network:
|
||||||
|
cidr: str
|
||||||
|
gateway: Optional[str] = None
|
||||||
|
interfaces: List[Interface] = field(default_factory=list)
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
self.cidr = fields.get("cidr")
|
||||||
|
self.gateway = fields.get("gateway")
|
||||||
|
interfaces = fields.get("interfaces", [])
|
||||||
|
if isinstance(interfaces, str):
|
||||||
|
interfaces = [iface for iface in interfaces.split(",") if iface]
|
||||||
|
self.interfaces = interfaces
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if not isinstance(self.cidr, str) or not self.cidr:
|
||||||
|
raise ValueError("Network 'cidr' must be a non-empty string")
|
||||||
|
if self.gateway is not None and not isinstance(self.gateway, str):
|
||||||
|
raise ValueError("Network 'gateway' must be a string")
|
||||||
|
|
||||||
|
|
||||||
|
# ===========================
|
||||||
|
# Device Base Class (Abstract)
|
||||||
|
# ===========================
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Device:
|
||||||
|
name: str
|
||||||
|
mgmt_ip: Optional[str] = None
|
||||||
|
interfaces: List[Interface] = field(default_factory=list)
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
self.name = fields.get("name")
|
||||||
|
self.mgmt_ip = fields.get("mgmtIP")
|
||||||
|
self.interfaces = [f for f in fields.get("interfaces").split(",") if f]
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if not isinstance(self.name, str) or not self.name:
|
||||||
|
raise ValueError("Device 'name' must be a non-empty string")
|
||||||
|
if self.mgmt_ip is not None and not isinstance(self.mgmt_ip, str):
|
||||||
|
raise ValueError("Device 'mgmtIP' must be a string")
|
||||||
|
|
||||||
|
|
||||||
|
# ===========================
|
||||||
|
# Host / Router / Switch
|
||||||
|
# ===========================
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Host(Device):
|
||||||
|
operating_system: Optional[str] = None
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
super().__init__(fields)
|
||||||
|
self.operating_system = fields.get("operatingSystem")
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
super().__post_init__()
|
||||||
|
if not isinstance(self.operating_system, str):
|
||||||
|
raise ValueError("Host 'operatingSystem' must be a string")
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Router(Device):
|
||||||
|
routing_table: Dict[str, str] = field(default_factory=dict)
|
||||||
|
routing_protocols: List[str] = field(default_factory=list)
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
super().__init__(fields)
|
||||||
|
self.routing_table = fields.get("routingTable", {})
|
||||||
|
self.routing_protocols = fields.get("routingProtocols", [])
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
super().__post_init__()
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Switch(Device):
|
||||||
|
mac_table: Dict[str, str] = field(default_factory=dict)
|
||||||
|
vlan_database: Dict[int, str] = field(default_factory=dict)
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
super().__init__(fields)
|
||||||
|
self.mac_table = fields.get("macTable", {})
|
||||||
|
self.vlan_database = fields.get("vlanDatabase", {})
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
super().__post_init__()
|
||||||
|
|
||||||
|
|
||||||
|
# ===========================
|
||||||
|
# Physical Link (Wire)
|
||||||
|
# ===========================
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Wire:
|
||||||
|
id: str
|
||||||
|
endpoints: List[Interface] # Must contain exactly 2 interfaces
|
||||||
|
bandwidth: Optional[str] = None
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
self.id = fields.get("name")
|
||||||
|
self.endpoints = [ep for ep in fields.get("endpoints").split(",") if ep]
|
||||||
|
self.bandwidth = fields.get("bandwidth")
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if not isinstance(self.id, str) or not self.id:
|
||||||
|
raise ValueError("Wire 'id' must be a non-empty string")
|
||||||
|
if len(self.endpoints) != 2:
|
||||||
|
raise ValueError(
|
||||||
|
"Wire 'endpoints' must contain exactly 2 Interface objects"
|
||||||
|
)
|
||||||
|
if not all(isinstance(i, Interface) for i in self.endpoints):
|
||||||
|
raise TypeError("Wire 'endpoints' must be Interface objects")
|
||||||
+52
-100
@@ -1,6 +1,27 @@
|
|||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from typing import List, Optional, Dict
|
from typing import List, Optional, Dict
|
||||||
|
|
||||||
|
# ===========================
|
||||||
|
# Network / Subnet
|
||||||
|
# ===========================
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Network:
|
||||||
|
name: str
|
||||||
|
vlan_id : Optional[int] = None
|
||||||
|
ip: Optional[str] = None
|
||||||
|
subnet_mask: Optional[str] = None
|
||||||
|
|
||||||
|
def __init__(self, fields):
|
||||||
|
self.name = fields.get("name")
|
||||||
|
self.vlan_id = fields.get("vlan", fields.get("vlan_id"))
|
||||||
|
self.ip = fields.get("ip")
|
||||||
|
self.subnet_mask = fields.get("mask", fields.get("subnet_mask"))
|
||||||
|
self.__post_init__()
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if not isinstance(self.name, str) or not self.name:
|
||||||
|
raise ValueError("Network 'name' must be a non-empty string")
|
||||||
|
|
||||||
# ===========================
|
# ===========================
|
||||||
# Interface Classes
|
# Interface Classes
|
||||||
@@ -9,18 +30,33 @@ from typing import List, Optional, Dict
|
|||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Interface:
|
class Interface:
|
||||||
|
class IP:
|
||||||
|
def __init__(self, ip_str: str, default_gateway: Optional[str] = None):
|
||||||
|
self.ip_str = ip_str
|
||||||
|
self.default_gateway = default_gateway
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.ip_str
|
||||||
|
|
||||||
name: str
|
name: str
|
||||||
ip_address: Optional[str] = None
|
mode: Optional[str] = None
|
||||||
|
network: Optional[Network] = None
|
||||||
|
ips: Optional[List[IP]] = None
|
||||||
mac_address: Optional[str] = None
|
mac_address: Optional[str] = None
|
||||||
speed: Optional[str] = None
|
speed: Optional[str] = None
|
||||||
duplex: Optional[str] = None
|
|
||||||
|
|
||||||
def __init__(self, fields):
|
def __init__(self, fields):
|
||||||
self.name = fields.get("name")
|
self.name = fields.get("name")
|
||||||
self.ip_address = fields.get("ip")
|
ips = fields.get("ips", [])
|
||||||
|
if isinstance(ips, str):
|
||||||
|
ips = [self.IP(ip_str=ip.strip()) for ip in ips.split(",") if ip.strip()]
|
||||||
|
self.ips = ips
|
||||||
|
|
||||||
|
# возможно передаём просто field
|
||||||
|
self.network = Network(fields.get("network", {})) if fields.get("network") else None
|
||||||
|
|
||||||
self.mac_address = fields.get("mac")
|
self.mac_address = fields.get("mac")
|
||||||
self.speed = fields.get("speed")
|
self.speed = fields.get("speed")
|
||||||
self.duplex = fields.get("duplex")
|
|
||||||
self.__post_init__()
|
self.__post_init__()
|
||||||
|
|
||||||
def __post_init__(self):
|
def __post_init__(self):
|
||||||
@@ -28,58 +64,6 @@ class Interface:
|
|||||||
raise ValueError("Interface 'name' must be a non-empty string")
|
raise ValueError("Interface 'name' must be a non-empty string")
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class VirtualInterface(Interface):
|
|
||||||
vlan_id: int = 0
|
|
||||||
parent_physical: Optional[Interface] = None
|
|
||||||
|
|
||||||
def __init__(self, fields):
|
|
||||||
super().__init__(fields)
|
|
||||||
self.vlan_id = fields.get("vlanID", fields.get("vlan_id", 0))
|
|
||||||
self.parent_physical = fields.get(
|
|
||||||
"parent_physical", fields.get("parentPhysical", fields.get("parent"))
|
|
||||||
)
|
|
||||||
self.__post_init__()
|
|
||||||
|
|
||||||
def __post_init__(self):
|
|
||||||
super().__post_init__()
|
|
||||||
if not isinstance(self.vlan_id, int) or self.vlan_id < 0:
|
|
||||||
raise ValueError(
|
|
||||||
"VirtualInterface 'vlan_id' must be a non-negative integer"
|
|
||||||
)
|
|
||||||
if not isinstance(self.parent_physical, Interface):
|
|
||||||
raise TypeError(
|
|
||||||
"VirtualInterface 'parent_physical' must reference an Interface"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# ===========================
|
|
||||||
# Network / Subnet
|
|
||||||
# ===========================
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Network:
|
|
||||||
cidr: str
|
|
||||||
gateway: Optional[str] = None
|
|
||||||
interfaces: List[Interface] = field(default_factory=list)
|
|
||||||
|
|
||||||
def __init__(self, fields):
|
|
||||||
self.cidr = fields.get("cidr")
|
|
||||||
self.gateway = fields.get("gateway")
|
|
||||||
interfaces = fields.get("interfaces", [])
|
|
||||||
if isinstance(interfaces, str):
|
|
||||||
interfaces = [iface for iface in interfaces.split(",") if iface]
|
|
||||||
self.interfaces = interfaces
|
|
||||||
self.__post_init__()
|
|
||||||
|
|
||||||
def __post_init__(self):
|
|
||||||
if not isinstance(self.cidr, str) or not self.cidr:
|
|
||||||
raise ValueError("Network 'cidr' must be a non-empty string")
|
|
||||||
if self.gateway is not None and not isinstance(self.gateway, str):
|
|
||||||
raise ValueError("Network 'gateway' must be a string")
|
|
||||||
|
|
||||||
|
|
||||||
# ===========================
|
# ===========================
|
||||||
# Device Base Class (Abstract)
|
# Device Base Class (Abstract)
|
||||||
# ===========================
|
# ===========================
|
||||||
@@ -88,20 +72,17 @@ class Network:
|
|||||||
@dataclass
|
@dataclass
|
||||||
class Device:
|
class Device:
|
||||||
name: str
|
name: str
|
||||||
mgmt_ip: Optional[str] = None
|
|
||||||
interfaces: List[Interface] = field(default_factory=list)
|
interfaces: List[Interface] = field(default_factory=list)
|
||||||
|
|
||||||
def __init__(self, fields):
|
def __init__(self, fields):
|
||||||
self.name = fields.get("name")
|
self.name = fields.get("name")
|
||||||
self.mgmt_ip = fields.get("mgmtIP")
|
# тут точно понять, какие поля передаются
|
||||||
self.interfaces = [f for f in fields.get("interfaces").split(",") if f]
|
self.interfaces = [Interface(iface) for iface in fields.get("interfaces", [])]
|
||||||
self.__post_init__()
|
self.__post_init__()
|
||||||
|
|
||||||
def __post_init__(self):
|
def __post_init__(self):
|
||||||
if not isinstance(self.name, str) or not self.name:
|
if not isinstance(self.name, str) or not self.name:
|
||||||
raise ValueError("Device 'name' must be a non-empty string")
|
raise ValueError("Device 'name' must be a non-empty string")
|
||||||
if self.mgmt_ip is not None and not isinstance(self.mgmt_ip, str):
|
|
||||||
raise ValueError("Device 'mgmtIP' must be a string")
|
|
||||||
|
|
||||||
|
|
||||||
# ===========================
|
# ===========================
|
||||||
@@ -111,28 +92,26 @@ class Device:
|
|||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Host(Device):
|
class Host(Device):
|
||||||
operating_system: Optional[str] = None
|
# operating_system: Optional[str] = None
|
||||||
|
|
||||||
def __init__(self, fields):
|
def __init__(self, fields):
|
||||||
super().__init__(fields)
|
super().__init__(fields)
|
||||||
self.operating_system = fields.get("operatingSystem")
|
# self.operating_system = fields.get("operatingSystem")
|
||||||
self.__post_init__()
|
self.__post_init__()
|
||||||
|
|
||||||
def __post_init__(self):
|
def __post_init__(self):
|
||||||
super().__post_init__()
|
super().__post_init__()
|
||||||
if not isinstance(self.operating_system, str):
|
|
||||||
raise ValueError("Host 'operatingSystem' must be a string")
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Router(Device):
|
class Router(Device):
|
||||||
routing_table: Dict[str, str] = field(default_factory=dict)
|
# routing_table: Dict[str, str] = field(default_factory=dict)
|
||||||
routing_protocols: List[str] = field(default_factory=list)
|
# routing_protocols: List[str] = field(default_factory=list)
|
||||||
|
|
||||||
def __init__(self, fields):
|
def __init__(self, fields):
|
||||||
super().__init__(fields)
|
super().__init__(fields)
|
||||||
self.routing_table = fields.get("routingTable", {})
|
# self.routing_table = fields.get("routingTable", {})
|
||||||
self.routing_protocols = fields.get("routingProtocols", [])
|
# self.routing_protocols = fields.get("routingProtocols", [])
|
||||||
self.__post_init__()
|
self.__post_init__()
|
||||||
|
|
||||||
def __post_init__(self):
|
def __post_init__(self):
|
||||||
@@ -141,42 +120,15 @@ class Router(Device):
|
|||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Switch(Device):
|
class Switch(Device):
|
||||||
mac_table: Dict[str, str] = field(default_factory=dict)
|
# mac_table: Dict[str, str] = field(default_factory=dict)
|
||||||
vlan_database: Dict[int, str] = field(default_factory=dict)
|
# vlan_database: Dict[int, str] = field(default_factory=dict)
|
||||||
|
|
||||||
def __init__(self, fields):
|
def __init__(self, fields):
|
||||||
super().__init__(fields)
|
super().__init__(fields)
|
||||||
self.mac_table = fields.get("macTable", {})
|
# self.mac_table = fields.get("macTable", {})
|
||||||
self.vlan_database = fields.get("vlanDatabase", {})
|
# self.vlan_database = fields.get("vlanDatabase", {})
|
||||||
self.__post_init__()
|
self.__post_init__()
|
||||||
|
|
||||||
def __post_init__(self):
|
def __post_init__(self):
|
||||||
super().__post_init__()
|
super().__post_init__()
|
||||||
|
|
||||||
|
|
||||||
# ===========================
|
|
||||||
# Physical Link (Wire)
|
|
||||||
# ===========================
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Wire:
|
|
||||||
id: str
|
|
||||||
endpoints: List[Interface] # Must contain exactly 2 interfaces
|
|
||||||
bandwidth: Optional[str] = None
|
|
||||||
|
|
||||||
def __init__(self, fields):
|
|
||||||
self.id = fields.get("name")
|
|
||||||
self.endpoints = [ep for ep in fields.get("endpoints").split(",") if ep]
|
|
||||||
self.bandwidth = fields.get("bandwidth")
|
|
||||||
self.__post_init__()
|
|
||||||
|
|
||||||
def __post_init__(self):
|
|
||||||
if not isinstance(self.id, str) or not self.id:
|
|
||||||
raise ValueError("Wire 'id' must be a non-empty string")
|
|
||||||
if len(self.endpoints) != 2:
|
|
||||||
raise ValueError(
|
|
||||||
"Wire 'endpoints' must contain exactly 2 Interface objects"
|
|
||||||
)
|
|
||||||
if not all(isinstance(i, Interface) for i in self.endpoints):
|
|
||||||
raise TypeError("Wire 'endpoints' must be Interface objects")
|
|
||||||
|
|||||||
Reference in New Issue
Block a user