Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Fix converter. change method to class method. add logging.
Browse files Browse the repository at this point in the history
  • Loading branch information
jpkrajewski committed Mar 13, 2024
1 parent 101c19e commit 6bb54e4
Show file tree
Hide file tree
Showing 2 changed files with 72 additions and 36 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ class LanVpnDhcpServerParcel(_ParcelBase):
validation_alias=AliasPath("data", "addressPool"),
description="Configure IPv4 prefix range of the DHCP address pool",
)
exclude: Union[Global[List[IPv4Address]], Variable, Default[None]] = Field(
exclude: Union[Global[List[IPv4Address]], Global[List[str]], Variable, Default[None]] = Field(
default=Default[None](value=None),
validation_alias=AliasPath("data", "exclude"),
description="Configure IPv4 address to exclude from DHCP address pool",
Expand Down
106 changes: 71 additions & 35 deletions catalystwan/utils/config_migration/converters/feature_template/dhcp.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,27 @@
import logging
from copy import deepcopy
from ipaddress import IPv4Address, IPv4Network
from typing import List

from catalystwan.api.configuration_groups.parcel import Global, as_global, as_variable
from catalystwan.api.configuration_groups.parcel import Global, Variable, as_global, as_variable
from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import (
LanVpnDhcpServerParcel,
SubnetMask,
)

logger = logging.getLogger(__name__)


class DhcpTemplateConverter:
supported_template_types = ("dhcp", "cisco_dhcp_server")
supported_template_types = ("dhcp", "cisco_dhcp_server", "dhcp-server")

variable_address_pool = "{{dhcp_1_addressPool_networkAddress}}"
variable_subnet_mask = "{{dhcp_1_addressPool_subnetMask}}"
variable_mac_address = "{{{{dhcp_1_staticLease_{}_macAddress}}}}"
variable_ip = "{{{{dhcp_1_staticLease_{}_ip}}}}"

@staticmethod
def create_parcel(name: str, description: str, template_values: dict) -> LanVpnDhcpServerParcel:
@classmethod
def create_parcel(cls, name: str, description: str, template_values: dict) -> LanVpnDhcpServerParcel:
"""
Create a LanVpnDhcpServerParcel object based on the provided parameters.
Expand All @@ -27,21 +35,6 @@ def create_parcel(name: str, description: str, template_values: dict) -> LanVpnD
"""

def convert_str_list_to_ipv4_list(d: dict, key: str) -> None:
"""
Convert a list of strings representing IPv4 addresses to a list of IPv4Address objects.
Args:
d (dict): The dictionary containing the key-value pair to be converted.
key (str): The key in the dictionary representing the list of strings.
Returns:
None. The function modifies the dictionary in-place by
replacing the list of strings with a list of IPv4Address objects.
"""
if str_list := d.get(key, as_global([])).value:
d[key] = Global[List[IPv4Address]](value=[IPv4Address(ip) for ip in str_list])

values = deepcopy(template_values)
values.update(values.pop("options", {}))

Expand All @@ -51,22 +44,32 @@ def convert_str_list_to_ipv4_list(d: dict, key: str) -> None:
address = as_global(network.network_address)
mask = as_global(str(network.netmask), SubnetMask)
values["address_pool"] = {"network_address": address, "subnet_mask": mask}

for key in ("exclude", "dns_servers", "tftp_servers"):
convert_str_list_to_ipv4_list(values, key)

if static_leases := values.get("static_lease", []):
mac_address_variable = "{{{{dhcp_1_staticLease_{}_macAddress}}}}"
ip_variable = "{{{{dhcp_1_staticLease_{}_ip}}}}"
static_lease = []
for i, entry in enumerate(static_leases):
static_lease.append(
{
"mac_address": entry.get("mac_address", as_variable(mac_address_variable.format(i + 1))),
"ip": entry.get("ip", as_variable(ip_variable.format(i + 1))),
}
)
values["static_lease"] = static_lease
else:
logger.warning(
"No address pool specified for DHCP server parcel."
"Assiging variable: dhcp_1_addressPool_networkAddress and dhcp_1_addressPool_subnetMask."
)
values["address_pool"] = {
"network_address": as_variable(cls.variable_address_pool),
"subnet_mask": as_variable(cls.variable_subnet_mask),
}

for entry in values.get("option_code", []):
cls._convert_str_list_to_ipv4_list(entry, "ip")

for key in ("dns_servers", "tftp_servers"):
cls._convert_str_list_to_ipv4_list(values, key)

static_lease = []
for i, entry in enumerate(values.get("static_lease", [])):
mac_address, ip = cls._get_mac_address_and_ip(entry, i)
static_lease.append(
{
"mac_address": mac_address,
"ip": ip,
}
)
values["static_lease"] = static_lease

parcel_values = {
"parcel_name": name,
Expand All @@ -75,3 +78,36 @@ def convert_str_list_to_ipv4_list(d: dict, key: str) -> None:
}

return LanVpnDhcpServerParcel(**parcel_values) # type: ignore

@classmethod
def _convert_str_list_to_ipv4_list(cls, d: dict, key: str) -> None:
"""
Convert a list of strings representing IPv4 addresses to a list of IPv4Address objects.
Args:
d (dict): The dictionary containing the key-value pair to be converted.
key (str): The key in the dictionary representing the list of strings.
Returns:
None. The function modifies the dictionary in-place by
replacing the list of strings with a list of IPv4Address objects.
"""
if str_list := d.get(key, as_global([])).value:
d[key] = Global[List[IPv4Address]](value=[IPv4Address(ip) for ip in str_list])

@classmethod
def _get_mac_address_and_ip(cls, entry: dict, i: int) -> tuple:
mac_address = entry.get("mac_address", as_variable(cls.variable_mac_address.format(i + 1)))
ip = entry.get("ip", as_variable(cls.variable_ip.format(i + 1)))
if isinstance(mac_address, Variable):
logger.warning(
f"No MAC address specified for static lease {i + 1}."
f"Assigning variable: {cls.variable_mac_address.format(i + 1)}"
)
if isinstance(ip, Variable):
logger.warning(
f"No IP address specified for static lease {i + 1}."
f"Assigning variable: {cls.variable_ip.format(i + 1)}"
)

return mac_address, ip

0 comments on commit 6bb54e4

Please sign in to comment.