outposts: fix compare_ports to support both service and container ports

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>
This commit is contained in:
Jens Langhammer 2022-02-08 17:40:49 +01:00
parent f44cf06d22
commit c63e1c9b87

View file

@ -1,6 +1,7 @@
"""k8s utils"""
from pathlib import Path
from kubernetes.client.models.v1_container_port import V1ContainerPort
from kubernetes.client.models.v1_service_port import V1ServicePort
from kubernetes.config.incluster_config import SERVICE_TOKEN_FILENAME
@ -16,19 +17,28 @@ def get_namespace() -> str:
return "default"
def compare_port(current: V1ServicePort, reference: V1ServicePort) -> bool:
def compare_port(
current: V1ServicePort | V1ContainerPort, reference: V1ServicePort | V1ContainerPort
) -> bool:
"""Compare a single port"""
if current.name != reference.name:
return False
# We only care about the target port
if current.target_port != reference.target_port:
return False
if current.protocol != reference.protocol:
return False
if isinstance(current, V1ServicePort) and isinstance(reference, V1ServicePort):
# We only care about the target port
if current.target_port != reference.target_port:
return False
if isinstance(current, V1ContainerPort) and isinstance(reference, V1ContainerPort):
# We only care about the target port
if current.container_port != reference.container_port:
return False
return True
def compare_ports(current: list[V1ServicePort], reference: list[V1ServicePort]):
def compare_ports(
current: list[V1ServicePort | V1ContainerPort], reference: list[V1ServicePort | V1ContainerPort]
):
"""Compare ports of a list"""
if len(current) != len(reference):
raise NeedsRecreate()