Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 58 additions & 0 deletions ocp_resources/multi_network_policy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
from ocp_resources.resource import NamespacedResource


class MultiNetworkPolicy(NamespacedResource):
"""
MultiNetworkPolicy object.

API reference:
https://docs.openshift.com/container-platform/4.14/networking/multiple_networks/configuring-multi-network-policy.html
"""

api_group = NamespacedResource.ApiGroup.K8S_CNI_CNCF_IO

def __init__(
self,
network_name=None,
policy_types=None,
ingress=None,
egress=None,
pod_selector=None,
**kwargs,
):
"""
Create and manage MultiNetworkPolicy

Args:
network_name (str): The name of the NetworkAttachmentDefinition that the policy will impact.
policy_types (list, optional): One or more of the valid ip policies.
ingress (list, optional): list containing a dictionary specifying the allowed "from" parameters.
egress (list, optional): list containing a dictionary specifying the allowed "to" parameters.
pod_selector (dict): Map a label to match.
"""
super().__init__(**kwargs)
self.network_name = network_name
self.policy_types = policy_types
self.pod_selector = pod_selector
self.ingress = ingress
self.egress = egress

def to_dict(self):
super().to_dict()
if not self.yaml_file:
if not self.network_name:
raise ValueError("Passing yaml_file or parameter 'network_name' is required")
if self.pod_selector is None:
raise ValueError("Passing yaml_file or parameter 'pod_selector' is required")

self.res["metadata"]["annotations"] = {
f"{NamespacedResource.ApiGroup.K8S_V1_CNI_CNCF_IO}/policy-for": f"{self.namespace}/{self.network_name}"
}
self.res["spec"] = {}
self.res["spec"]["podSelector"] = self.pod_selector
if self.policy_types:
self.res["spec"]["policyTypes"] = self.policy_types
if self.ingress is not None:
self.res["spec"]["ingress"] = self.ingress
if self.egress is not None:
self.res["spec"]["egress"] = self.egress