diff --git a/TODO b/TODO index e6ca00c..377b77e 100644 --- a/TODO +++ b/TODO @@ -1,3 +1,9 @@ -- в данный момент маска никуда не прходит -- обработка виртуальных интерфейсов в yaml -- добавление bridge и vlan сущностей \ No newline at end of file +- CI для тестов +- CI для автоматической генерации topology +- CI для автоматической генерации документации (если хватит сил) +- Флаг для генерации "чистых" yaml файлов +- Добавление к examples yaml файлов (?) + - тогда надо и диаграммы, а тогда и сами могут прогнать, лишние файлы +- Актуализировать flake +- проверить реальность yaml +- валидация сейчас раскинута повсюду \ No newline at end of file diff --git a/examples/2hostsswitch/table.csv b/examples/2hostsswitch/table.csv index 82122b3..366d076 100644 --- a/examples/2hostsswitch/table.csv +++ b/examples/2hostsswitch/table.csv @@ -1,12 +1,10 @@ -Name,Role,Adapter,Interface,Master Interface,Network,VLAN,Network IP,Mask,Device IP,Default Gateway -PC1,Host,Adapter1,eth1,vlan7,A,,,,, -PC1,Host,,vlan7,,,7,10.10.10.0,/24,10.10.10.7,0.0.0.0 -PC2,Host,Adapter1,eth1,vlan9,B,,,,, -PC2,Host,,vlan9,,,9,10.10.10.0,/24,10.10.10.9,0.0.0.0 -com,Switch,Adapter1,eth1,vlan7,A,7,,,, -com,Switch,Adapter2,eth2,vlan9,B,9,,,, -com,Switch,,vlan7,bridge,,,,,, -com,Switch,,vlan9,bridge,,,,,, -com,Switch,,bridge,,,,,,, - -## REMAKE \ No newline at end of file +Name,Role,Adapter,Interface,Parent Interface,Slave Interfaces,Network,VLAN,Network IP,Mask,Device IP,Default Gateway +PC1,Host,Adapter1,eth1,,,A,,,,, +PC1,Host,,vlan7,eth1,,,7,10.10.10.0,/24,10.10.10.7,0.0.0.0 +PC2,Host,Adapter1,eth1,,,B,,,,, +PC2,Host,,vlan9,eth1,,,9,10.10.10.0,/24,10.10.10.9,0.0.0.0 +com,Switch,Adapter1,eth1,,,A,7,,,, +com,Switch,Adapter2,eth2,,,B,9,,,, +com,Switch,,vlan7,eth1,,,,,,, +com,Switch,,vlan9,eth2,,,,,,, +com,Switch,,bridge,,"vlan7,vlan9",,,,,,wa \ No newline at end of file diff --git a/src/netdiag/domain/models.py b/src/netdiag/domain/models.py index 609ed64..7c9b372 100644 --- a/src/netdiag/domain/models.py +++ b/src/netdiag/domain/models.py @@ -1,6 +1,7 @@ -from typing import Any, Dict, List, Optional +from typing import Dict, List, Optional +import logging -# =========================== +# TODO rewrite to dataclasses with validation class Interface: @@ -13,13 +14,13 @@ class Interface: network: Optional[str] subnet_mask: Optional[str] default_gateway: Optional[str] + vlan: Optional[str] # --- device: "Device" # set by Device.add_interface() when the interface is added to a device def __init__( self, name: str, - itype: Optional[str] = None, adapter: Optional[str] = None, slave_interfaces: Optional[List[str]] = None, parent_interface: Optional[str] = None, @@ -27,11 +28,10 @@ class Interface: network: Optional[str] = None, subnet_mask: Optional[str] = None, default_gateway: Optional[str] = None, + vlan: Optional[str] = None, ): if not isinstance(name, str): raise ValueError("Interface 'name' must be a string") - if itype is not None and not isinstance(itype, str): - raise ValueError("Interface 'itype' must be a string or None") if ip_address is not None and not isinstance(ip_address, str): raise ValueError("Interface 'ip_address' must be a string or None") if network is not None and not isinstance(network, str): @@ -48,33 +48,43 @@ class Interface: ) if parent_interface is not None and not isinstance(parent_interface, str): raise ValueError("Interface 'parent_interface' must be a string or None") + if vlan is not None and not isinstance(vlan, str): + raise ValueError("Interface 'vlan' must be a string or None") - itype = None if itype == "" else itype + # edit this adapter = None if adapter == "" else adapter + slave_interfaces = None if slave_interfaces == [] else slave_interfaces + parent_interface = None if parent_interface == "" else parent_interface + vlan = None if vlan == "" else vlan - if itype is None and adapter is None: - itype = "virtual" - if adapter is not None: - if itype is not None and itype != "physical": - raise ValueError( - "Interface with an adapter must have 'itype' set to 'physical'" - ) - itype = "physical" + logging.debug( + f"Creating interface '{name}' with adapter='{adapter}', slave_interfaces='{slave_interfaces}', parent_interface='{parent_interface}'" + ) - if itype not in (None, "physical", "virtual", "bridge", "vlan"): + # looks like shit + if ( + (adapter is not None) + + (slave_interfaces is not None) + + (parent_interface is not None) + ) != 1: raise ValueError( - "Interface 'itype' must be one of 'physical', 'virtual', 'bridge', 'vlan', or None" + "Interface must have exactly one of 'adapter', 'slave_interfaces', or 'parent_interface' defined" ) - # print( - # f"Creating interface '{name}' with itype='{itype}', adapter='{adapter}', slave_interfaces='{slave_interfaces}', parent_interface='{parent_interface}', ip_address='{ip_address}', network='{network}', subnet_mask='{subnet_mask}', default_gateway='{default_gateway}'" - # ) - - if itype == "bridge" and not slave_interfaces: - raise ValueError("Bridge interfaces must have 'slave_interfaces' defined") - if itype == "vlan" and not parent_interface: - raise ValueError("VLAN interfaces must have 'parent_interface' defined") + if slave_interfaces is not None: + itype = "bridge" + network = None + elif parent_interface is not None: + if vlan is None: + raise ValueError( + "Interface with 'parent_interface' must have 'vlan' defined" + ) + itype = "vlan" + network = None + else: + itype = "physical" + # sort all this values self.name = name self.itype = itype self.ip_address = ip_address @@ -84,9 +94,10 @@ class Interface: self.adapter = adapter self.slave_interfaces = slave_interfaces self.parent_interface = parent_interface + self.vlan = vlan def __repr__(self) -> str: - return f"Interface(name={self.name}, itype={self.itype}, ip_address={self.ip_address}, network={self.network}, default_gateway={self.default_gateway}, subnet_mask={self.subnet_mask}, adapter={self.adapter}, slave_interfaces={self.slave_interfaces})" + return f"Interface(name={self.name}, itype={self.itype}, ip_address={self.ip_address}, network={self.network}, default_gateway={self.default_gateway}, subnet_mask={self.subnet_mask}, adapter={self.adapter}, slave_interfaces={self.slave_interfaces}, vlan={self.vlan})" # =========================== @@ -113,7 +124,6 @@ class Device: def rm_interface(self, interface: Interface): if interface.name in self.interfaces: del self.interfaces[interface.name] - # interface.device = None # clear the device attribute of the interface del interface else: raise ValueError( @@ -142,8 +152,7 @@ class Switch(Device): class Network: name: str interfaces: List[Interface] - vlan: Optional[str] # need a proper parse - network_ip: Optional[str] + network_ip: Optional[str] # а что он делает? def __init__( self, diff --git a/src/netdiag/output/d2.py b/src/netdiag/output/d2.py index 757722c..1168711 100644 --- a/src/netdiag/output/d2.py +++ b/src/netdiag/output/d2.py @@ -10,42 +10,15 @@ from py_d2.connection import Direction # https://d2lang.com/tour/themes/ THEME_NUMBER = 200 -""" -PC2 - -PC2 - -PC3 - -PC3.adapter1 : { -shape: parallelogram -} - -PC3.adapter2 : { -shape: parallelogram -} - -netw_A : { -shape: cloud -} -netw_B : { -shape: cloud -} - -PC1.adapter1 -- netw_A -PC2.adapter1 -- netw_A -PC3.adapter1 -- netw_A - -PC1.adapter2 -- netw_B -PC3.adapter2 -- netw_B - -""" - def _check_d2_installed() -> bool: return shutil.which("d2") is not None +def _check_magick_installed() -> bool: + return shutil.which("magick") is not None + + def generate_d2_diagram(topology: Topology, output_path: Path) -> None: shapes = [] connections = [] @@ -80,7 +53,33 @@ def generate_d2_diagram(topology: Topology, output_path: Path) -> None: with open(output_path, "w", encoding="utf-8") as f: f.write(str(diagram)) - _generate_picture(output_path, output_path.with_suffix(".png")) + _generate_picture( + output_path, output_path.with_suffix(".png") + ) # output_path not used + + _run_magick_command( + output_path.with_suffix(".svg"), output_path.with_suffix(".png") + ) + + +def _run_magick_command(input_path: Path, output_path: Path) -> None: + if not _check_magick_installed(): + raise EnvironmentError( + "ImageMagick is not installed or 'magick' command is not found in PATH. Please install ImageMagick to use this feature." + ) + + res = subprocess.run( + ["magick", "convert", str(input_path), str(output_path)], + check=True, + capture_output=True, + ) + + if res.returncode != 0: + raise RuntimeError( + f"Image conversion failed (code {res.returncode})\n" + f"Stdout: {res.stdout.decode()}\n" + f"Stderr: {res.stderr.decode()}" + ) def _generate_picture(diagram: Path, output_path: Path) -> None: diff --git a/src/netdiag/output/file_convert.py b/src/netdiag/output/file_convert.py index 63c4971..f6f5244 100644 --- a/src/netdiag/output/file_convert.py +++ b/src/netdiag/output/file_convert.py @@ -60,7 +60,6 @@ def make_yaml(topology: Topology, output_path: Path) -> None: "name": interface.name, "members": interface.slave_interfaces, "ip": ip, - "network": interface.network if interface.network else None, "gateway": ( interface.default_gateway if interface.default_gateway @@ -75,12 +74,12 @@ def make_yaml(topology: Topology, output_path: Path) -> None: "name": interface.name, "parent": interface.parent_interface, "ip": ip, - "network": interface.network if interface.network else None, "gateway": ( interface.default_gateway if interface.default_gateway else None ), + "id": interface.vlan, } ) continue diff --git a/src/netdiag/output/graphviz.py b/src/netdiag/output/graphviz.py index fea6f11..391f18f 100644 --- a/src/netdiag/output/graphviz.py +++ b/src/netdiag/output/graphviz.py @@ -3,6 +3,10 @@ from pathlib import Path import graphviz import shutil +# TODO: +# - find ithers implementations of graphviz +# - actualize code + def _check_graphviz_installed() -> bool: return shutil.which("dot") is not None diff --git a/src/netdiag/parse/convert_raw.py b/src/netdiag/parse/convert_raw.py index 5662fcf..eec7607 100644 --- a/src/netdiag/parse/convert_raw.py +++ b/src/netdiag/parse/convert_raw.py @@ -15,7 +15,6 @@ name_matching = { "DEVICE_TYPE": "Role", "DEVICE_NAME": "Name", "ADAPTER": "Adapter", - "INTERFACE TYPE": "Interface Type", "MASTER": "Master Interface", "SLAVES": "Slave Interfaces", "PARENT": "Parent Interface", @@ -83,13 +82,13 @@ def add_interfaces(devices: list[Device], raw_devices: list[RawDevices]) -> None interface = Interface( name=_get_from_field(raw_device.fields, "INTERFACE_NAME"), - itype=_get_from_field(raw_device.fields, "INTERFACE TYPE"), adapter=_get_from_field(raw_device.fields, "ADAPTER"), slave_interfaces=( _get_from_field(raw_device.fields, "SLAVES").split(",") if _get_from_field(raw_device.fields, "SLAVES") else None ), + vlan=_get_from_field(raw_device.fields, "VLAN"), parent_interface=_get_from_field(raw_device.fields, "PARENT"), ip_address=_get_from_field(raw_device.fields, "IP_ADDRESS"), network=_get_from_field(raw_device.fields, "NETWORK_NAME"), @@ -109,21 +108,16 @@ def parse_networks(raw_devices: list[RawDevices]) -> list[Network]: if not network_name: continue - vlan = (raw_device.fields.get(name_matching["VLAN"], "").strip(),) network_ip = (raw_device.fields.get(name_matching["NETWORK_IP"], "").strip(),) if network_name in [n.name for n in networks]: # update if already exists network = next(n for n in networks if n.name == network_name) - - if not network.vlan and vlan: - network.vlan = vlan if not network.network_ip and network_ip: network.network_ip = network_ip continue network = Network( name=network_name, - vlan=raw_device.fields.get(name_matching["VLAN"], "").strip(), network_ip=raw_device.fields.get(name_matching["NETWORK_IP"], "").strip(), ) networks.append(network)