From 42350f2916aff708ef6c06f720afad418ebe29b3 Mon Sep 17 00:00:00 2001 From: Shahaf Bahar <110476642+sbahar619@users.noreply.github.com> Date: Thu, 31 Oct 2024 08:50:54 +0200 Subject: [PATCH] Add UDN layer3 class (#2184) * Add UDN layer3 class * Move constants from global scope to class scope * Remove redundant function call * Fix key error * Add missing value argument validation * Use dictionary access instead of class attributes * Make parameter optional in a method * Correct the docstring for the role parameter * Improve error message for missing required arguments * Refactor to follow conventions * Revert "Refactor to follow conventions" This reverts commit 5d4b9427f51a93b9360e4fa559d7355fd6bf74bf. * Refactor type annotation * Refactor condition statement * Refactor docstring for layer3 subnets --- ocp_resources/user_defined_network.py | 61 +++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/ocp_resources/user_defined_network.py b/ocp_resources/user_defined_network.py index d7fd69ca7f..7d7f87e934 100644 --- a/ocp_resources/user_defined_network.py +++ b/ocp_resources/user_defined_network.py @@ -128,3 +128,64 @@ def to_dict(self) -> None: if self.ipam_lifecycle: _layer2["ipamLifecycle"] = self.ipam_lifecycle + + +class Layer3UserDefinedNetwork(UserDefinedNetwork): + """ + UserDefinedNetwork layer3 object. + + API reference: + https://ovn-kubernetes.io/api-reference/userdefinednetwork-api-spec/#layer3config + """ + + LAYER3: str = "Layer3" + + def __init__( + self, + role: Optional[str] = None, + mtu: Optional[int] = None, + subnets: Optional[List[Dict[str, Any]]] = None, + join_subnets: Optional[List[str]] = None, + **kwargs, + ): + """ + Create and manage UserDefinedNetwork with layer3 configuration + + Args: + role (Optional[str]): role describes the network role in the pod. + mtu (Optional[int]): mtu is the maximum transmission unit for a network. + subnets (Optional[List[Dict]]): subnets are used for the pod network across the cluster, each expecting: + - `cidr` (str): IP range in CIDR notation. + - `hostSubnet` (Optional[int]): Host-specific subnet. + API reference: + https://ovn-kubernetes.io/api-reference/userdefinednetwork-api-spec/#layer3subnet + join_subnets (Optional[List[str]]): join_subnets are used inside the OVN network topology. + """ + super().__init__( + topology=self.LAYER3, + **kwargs, + ) + self.role = role + self.mtu = mtu + self.subnets = subnets + self.join_subnets = join_subnets + + def to_dict(self) -> None: + super().to_dict() + if not self.kind_dict and not self.yaml_file: + if not self.role: + raise MissingRequiredArgumentError(argument="role") + if not self.subnets: + raise MissingRequiredArgumentError(argument="subnets") + + self.res["spec"][self.LAYER3.lower()] = {"role": self.role} + _layer3 = self.res["spec"][self.LAYER3.lower()] + + if self.mtu: + _layer3["mtu"] = self.mtu + + if self.join_subnets: + _layer3["joinSubnets"] = self.join_subnets + + if self.subnets: + _layer3["subnets"] = self.subnets