Merge remote-tracking branch 'origin/main' into gui_development

This commit is contained in:
Cian Hughes
2023-11-08 12:55:41 +00:00
8 changed files with 134 additions and 16 deletions

3
docs/src/ip_interface.md Normal file
View File

@@ -0,0 +1,3 @@
# node_deployer.ip_interface
::: node_deployer.ip_interface

View File

@@ -41,6 +41,7 @@ nav:
- create_img: src/create_img.md
- create_disk: src/create_disk.md
- debug: src/debug.md
- ip_interface: src/ip_interface.md
- utils: src/utils.md
- Reference:
# - FAQ: faq.md

View File

@@ -27,9 +27,10 @@ mkdocs-material = {extras = ["all"], version = "^9.4.8"}
ruff = "^0.1.1"
black = "^23.10.1"
snoop = "^0.4.3"
pytest = "^7.4.3"
mypy = "^1.6.1"
docker-stubs = {git = "https://github.com/rdozier-work/docker-stubs"}
pytest = "^7.4.3"
hypothesis = "^6.88.3"
[tool.poetry.group.docs.dependencies]
mkdocs = "^1.5.3"

View File

@@ -1,5 +1,4 @@
from fnmatch import fnmatch
import ipaddress
from typing import Annotated, Optional
from docker.types import Mount
@@ -9,14 +8,10 @@ from .cli import cli_spinner
from .config import config
from .create_img import create_img
from .debug import debug_guard
from .ip_interface import IPAddress
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:
"""Filters out erroneous warnings from the validation response
@@ -136,7 +131,7 @@ def create_ignition_disk(
"-ip",
help="IP address of the switch to connect to",
prompt=True,
parser=ipaddress.ip_address,
parser=IPAddress,
),
] = None,
switch_port: Annotated[

View File

@@ -1,4 +1,3 @@
import ipaddress
import json
from pathlib import Path
from typing import Annotated, Optional
@@ -9,12 +8,9 @@ from .autoignition import json_to_img
from .cli import cli_spinner
from .config import config
from .debug import debug_guard
from .ip_interface import IPAddress
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:
"""Loads the default template for the ignition configuration
@@ -121,7 +117,7 @@ def create_img(
"-ip",
help="IP address of the switch to connect to",
prompt=True,
parser=ipaddress.ip_address,
parser=IPAddress,
),
] = None,
switch_port: Annotated[

View File

@@ -44,7 +44,7 @@ def debug_guard(f: Callable) -> Callable:
debug_f = get_debug_f(f)
if kwargs.get("debug", False):
# 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:
return f(*args, **kwargs)

View File

@@ -0,0 +1,67 @@
from ipaddress import IPv4Address, IPv6Address, ip_address
class IPAddress:
def __init__(self, *args, **kwargs) -> None:
self.obj: IPv4Address | IPv6Address = ip_address(*args, **kwargs)
@property
def compressed(self) -> str:
return self.obj.compressed
@property
def exploded(self) -> str:
return self.obj.exploded
@property
def is_global(self) -> bool:
return self.obj.is_global
@property
def is_link_local(self) -> bool:
return self.obj.is_link_local
@property
def is_loopback(self) -> bool:
return self.obj.is_loopback
@property
def is_multicast(self) -> bool:
return self.obj.is_multicast
@property
def is_private(self) -> bool:
return self.obj.is_private
@property
def is_reserved(self) -> bool:
return self.obj.is_reserved
@property
def is_unspecified(self) -> bool:
return self.obj.is_unspecified
@property
def max_prefixlen(self) -> int:
return self.obj.max_prefixlen
@property
def packed(self) -> bytes:
return self.obj.packed
@property
def reverse_pointer(self) -> str:
return self.obj.reverse_pointer
@property
def version(self) -> int:
return self.obj.version
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

View File

@@ -1,10 +1,13 @@
import atexit
import filecmp
from ipaddress import IPv4Address, IPv6Address
import os
from pathlib import Path
import pickle
import shutil
from hypothesis import given
from hypothesis import strategies as st
from node_deployer.config import config
import tomllib
@@ -33,7 +36,7 @@ def cleanup():
atexit.register(cleanup)
from node_deployer import autoignition, create_disk, create_img # noqa: E402
from node_deployer import autoignition, create_disk, create_img, ip_interface # noqa: E402
with open(config.PROJECT_ROOT / "tests/data/node_deployer/test_args.toml", "rb") as f:
@@ -42,6 +45,58 @@ with open(config.PROJECT_ROOT / "tests/data/node_deployer/test_args.toml", "rb")
TEST_DATA_DIR = config.PROJECT_ROOT / "tests/data/node_deployer"
class TestIPInterface:
TEST_ATTRS = (
"compressed",
"exploded",
"is_global",
"is_link_local",
"is_loopback",
"is_multicast",
"is_private",
"is_reserved",
"is_unspecified",
"max_prefixlen",
"packed",
"reverse_pointer",
"version",
)
@given(st.ip_addresses(v=4))
def test_ipv4_parsing(self, ip: IPv4Address):
ip_str = str(ip)
test_result = ip_interface.IPAddress(ip_str)
assert test_result.obj == ip
@given(st.ip_addresses(v=4))
def test_ipv4_attr_passthrough(self, ip: IPv4Address):
ip_addr = ip_interface.IPAddress(str(ip))
for attr in self.TEST_ATTRS:
assert getattr(ip_addr, attr) == getattr(ip_addr.obj, attr)
@given(st.ip_addresses(v=4))
def test_ipv4_bool(self, ip: IPv4Address):
ip_addr = ip_interface.IPAddress(str(ip))
assert bool(ip_addr) != ip.is_unspecified
@given(st.ip_addresses(v=6))
def test_ipv6_parsing(self, ip: IPv6Address):
ip_str = str(ip)
test_result = ip_interface.IPAddress(ip_str)
assert test_result.obj == ip
@given(st.ip_addresses(v=6))
def test_ipv6_attr_passthrough(self, ip: IPv6Address):
ip_addr = ip_interface.IPAddress(str(ip))
for attr in self.TEST_ATTRS:
assert getattr(ip_addr, attr) == getattr(ip_addr.obj, attr)
@given(st.ip_addresses(v=6))
def test_ipv6_bool(self, ip: IPv6Address):
ip_addr = ip_interface.IPAddress(str(ip))
assert bool(ip_addr) != ip.is_unspecified
class TestAutoignition:
def test_json_to_img(self, tmp_path: Path):
tmp_path.mkdir(parents=True, exist_ok=True)