mirror of
https://github.com/Cian-H/I-Form_Server_Node_Deployer.git
synced 2025-12-23 06:32:08 +00:00
Created wrapper class for ip addresses
This commit is contained in:
@@ -1,5 +1,4 @@
|
|||||||
from fnmatch import fnmatch
|
from fnmatch import fnmatch
|
||||||
import ipaddress
|
|
||||||
from typing import Annotated, Optional
|
from typing import Annotated, Optional
|
||||||
|
|
||||||
from docker.types import Mount
|
from docker.types import Mount
|
||||||
@@ -9,14 +8,10 @@ from .cli import cli_spinner
|
|||||||
from .config import config
|
from .config import config
|
||||||
from .create_img import create_img
|
from .create_img import create_img
|
||||||
from .debug import debug_guard
|
from .debug import debug_guard
|
||||||
|
from .ip_interface import IPAddress
|
||||||
from .utils import ensure_build_dir
|
from .utils import ensure_build_dir
|
||||||
|
|
||||||
|
|
||||||
# When PEP695 is supported this line should be:
|
|
||||||
# type IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
|
|
||||||
IPAddress = ipaddress._IPAddressBase
|
|
||||||
|
|
||||||
|
|
||||||
def filter_validation_response(response: str) -> str:
|
def filter_validation_response(response: str) -> str:
|
||||||
"""Filters out erroneous warnings from the validation response
|
"""Filters out erroneous warnings from the validation response
|
||||||
|
|
||||||
@@ -136,7 +131,7 @@ def create_ignition_disk(
|
|||||||
"-ip",
|
"-ip",
|
||||||
help="IP address of the switch to connect to",
|
help="IP address of the switch to connect to",
|
||||||
prompt=True,
|
prompt=True,
|
||||||
parser=ipaddress.ip_address,
|
parser=IPAddress,
|
||||||
),
|
),
|
||||||
] = None,
|
] = None,
|
||||||
switch_port: Annotated[
|
switch_port: Annotated[
|
||||||
|
|||||||
@@ -1,4 +1,3 @@
|
|||||||
import ipaddress
|
|
||||||
import json
|
import json
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Annotated, Optional
|
from typing import Annotated, Optional
|
||||||
@@ -9,12 +8,9 @@ from .autoignition import json_to_img
|
|||||||
from .cli import cli_spinner
|
from .cli import cli_spinner
|
||||||
from .config import config
|
from .config import config
|
||||||
from .debug import debug_guard
|
from .debug import debug_guard
|
||||||
|
from .ip_interface import IPAddress
|
||||||
from .utils import ensure_build_dir
|
from .utils import ensure_build_dir
|
||||||
|
|
||||||
# When PEP695 is supported this line should be:
|
|
||||||
# type IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
|
|
||||||
IPAddress = ipaddress._IPAddressBase
|
|
||||||
|
|
||||||
|
|
||||||
def load_template() -> dict:
|
def load_template() -> dict:
|
||||||
"""Loads the default template for the ignition configuration
|
"""Loads the default template for the ignition configuration
|
||||||
@@ -121,7 +117,7 @@ def create_img(
|
|||||||
"-ip",
|
"-ip",
|
||||||
help="IP address of the switch to connect to",
|
help="IP address of the switch to connect to",
|
||||||
prompt=True,
|
prompt=True,
|
||||||
parser=ipaddress.ip_address,
|
parser=IPAddress,
|
||||||
),
|
),
|
||||||
] = None,
|
] = None,
|
||||||
switch_port: Annotated[
|
switch_port: Annotated[
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ def debug_guard(f: Callable) -> Callable:
|
|||||||
debug_f = get_debug_f(f)
|
debug_f = get_debug_f(f)
|
||||||
if kwargs.get("debug", False):
|
if kwargs.get("debug", False):
|
||||||
# Snoop depth is set to compensate for wrapper stack frames
|
# Snoop depth is set to compensate for wrapper stack frames
|
||||||
return debug_f(*args, **kwargs) # noqa: F821 #* ss is installed in debug_mode
|
return debug_f(*args, **kwargs)
|
||||||
else:
|
else:
|
||||||
return f(*args, **kwargs)
|
return f(*args, **kwargs)
|
||||||
|
|
||||||
|
|||||||
46
src/node_deployer/ip_interface.py
Normal file
46
src/node_deployer/ip_interface.py
Normal file
@@ -0,0 +1,46 @@
|
|||||||
|
from ipaddress import IPv4Address, IPv6Address, ip_address
|
||||||
|
|
||||||
|
|
||||||
|
class IPAddress:
|
||||||
|
def __init__(self, *args, **kwargs) -> None:
|
||||||
|
self.obj: IPv4Address | IPv6Address = ip_address(*args, **kwargs)
|
||||||
|
to_passthrough = (
|
||||||
|
"compressed",
|
||||||
|
"exploded",
|
||||||
|
"is_global",
|
||||||
|
"is_link_local",
|
||||||
|
"is_loopback",
|
||||||
|
"is_multicast",
|
||||||
|
"is_private",
|
||||||
|
"is_reserved",
|
||||||
|
"is_unspecified",
|
||||||
|
"max_prefixlen",
|
||||||
|
"packed",
|
||||||
|
"reverse_pointer",
|
||||||
|
"version",
|
||||||
|
)
|
||||||
|
for attrname in to_passthrough:
|
||||||
|
self._passthrough(attrname)
|
||||||
|
|
||||||
|
def _passthrough(self, attrname: str) -> None:
|
||||||
|
"""Passes through an attribute from the underlying IPv4Address or IPv6Address object
|
||||||
|
|
||||||
|
Args:
|
||||||
|
attrname (str): The name of the attribute to pass through
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
AttributeError: If the attribute is a method
|
||||||
|
"""
|
||||||
|
attr = getattr(self.obj, attrname)
|
||||||
|
if callable(attr):
|
||||||
|
raise AttributeError(f"Passthrough is unavailable for methods ({attrname})")
|
||||||
|
setattr(self, attrname, attr)
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
return str(self.obj)
|
||||||
|
|
||||||
|
def __repr__(self) -> str:
|
||||||
|
return repr(self.obj)
|
||||||
|
|
||||||
|
def __bool__(self) -> bool:
|
||||||
|
return not self.obj.is_unspecified
|
||||||
Reference in New Issue
Block a user