13577129160

Committed 27 Feb 2025 10:29PM UTC coverage: 96.397%. First build

Build # 13577129160

Build Type

Pull #534

github

Committed by

web-flow

Commit Message

Merge dcec09782 into 03be56195

Pull Request Pull Request #534: feat: Support for removing annotation from service + integration tests

Run Details

283 of 308 branches covered (91.88%)

Branch coverage included in aggregate %.

3 of 11 new or added lines in 1 file covered. (27.27%)

1028 of 1052 relevant lines covered (97.72%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.25

/app/handlers/handlers_services.py

from collections.abc import Callable

import kopf
import kubernetes

from app.handlers import success
from app.utils import to_bool


def k8s_get_twingate_resource(
    namespace: str, name: str, kapi: kubernetes.client.CustomObjectsApi | None = None
) -> dict | None:
    kapi = kapi or kubernetes.client.CustomObjectsApi()
    try:
        return kapi.get_namespaced_custom_object(
            "twingate.com", "v1beta", namespace, "twingateresources", name
        )
    except kubernetes.client.exceptions.ApiException as ex:
        if ex.status == 404:
            return None
        raise


ALLOWED_EXTRA_ANNOTATIONS: list[tuple[str, Callable]] = [
    ("name", str),
    ("alias", str),
    ("isBrowserShortcutEnabled", to_bool),
    ("securityPolicyId", str),
    ("isVisible", to_bool),
]


def service_to_twingate_resource(service_body, namespace) -> dict:
    meta = service_body.metadata
    spec = service_body.spec
    service_name = service_body.meta.name
    resource_object_name = f"{service_name}-resource"

    result: dict = {
        "apiVersion": "twingate.com/v1beta",
        "kind": "TwingateResource",
        "metadata": {
            "name": resource_object_name,
        },
        "spec": {
            "name": resource_object_name,
            "address": f"{service_name}.{namespace}.svc.cluster.local",
        },
    }

    for key, convert_f in ALLOWED_EXTRA_ANNOTATIONS:
        # TODO: Remove once we release v1.0 (see https://github.com/Twingate/kubernetes-operator/issues/530)
        if value := meta.annotations.get(f"twingate.com/resource-{key}"):
            result["spec"][key] = convert_f(value)
        if value := meta.annotations.get(f"resource.twingate.com/{key}"):
            result["spec"][key] = convert_f(value)

    protocols: dict = {
        "allowIcmp": False,
        "tcp": {"policy": "RESTRICTED", "ports": []},
        "udp": {"policy": "RESTRICTED", "ports": []},
    }
    for port_obj in spec.get("ports", []):
        port = port_obj["port"]
        if port_obj["protocol"] == "TCP":
            protocols["tcp"]["ports"].append({"start": port, "end": port})
        elif port_obj["protocol"] == "UDP":
            protocols["udp"]["ports"].append({"start": port, "end": port})

    result["spec"]["protocols"] = protocols

    return result


# TODO: Remove once we release v1.0 (see https://github.com/Twingate/kubernetes-operator/issues/530)
@kopf.on.resume("service", annotations={"twingate.com/resource": "true"})
@kopf.on.create("service", annotations={"twingate.com/resource": "true"})
@kopf.on.update("service", annotations={"twingate.com/resource": "true"})
@kopf.on.resume("service", annotations={"resource.twingate.com": "true"})
@kopf.on.create("service", annotations={"resource.twingate.com": "true"})
@kopf.on.update("service", annotations={"resource.twingate.com": "true"})
def twingate_service_create(body, spec, namespace, meta, logger, **_):
    logger.info("twingate_service_create: %s", spec)

    resource_subobject = service_to_twingate_resource(body, namespace)
    kopf.adopt(resource_subobject)

    resource_object_name = resource_subobject["metadata"]["name"]

    kapi = kubernetes.client.CustomObjectsApi()
    if existing_resource_object := k8s_get_twingate_resource(
        namespace, resource_object_name, kapi
    ):
        logger.info("TwingateResource already exists: %s", existing_resource_object)
        kapi.patch_namespaced_custom_object(
            "twingate.com",
            "v1beta",
            namespace,
            "twingateresources",
            resource_object_name,
            resource_subobject,
        )
    else:
        api_response = kapi.create_namespaced_custom_object(
            "twingate.com", "v1beta", namespace, "twingateresources", resource_subobject
        )
        logger.info("create_namespaced_custom_object response: %s", api_response)

    return success(
        message=f"Created TwingateResource {resource_subobject['spec']['name']}"
    )


@kopf.on.update("service", annotations={"resource.twingate.com": "false"})
@kopf.on.update("service", annotations={"resource.twingate.com": kopf.ABSENT})
def twingate_service_annotation_removed(body, spec, namespace, meta, logger, **_):
    logger.info("twingate_service_annotation_removed: %s", spec)

    resource_object_name = f"{body.meta.name}-resource"

    kapi = kubernetes.client.CustomObjectsApi()
    if existing_resource_object := k8s_get_twingate_resource(
        namespace, resource_object_name, kapi
    ):
        logger.info("Deleting TwingateResource: %s", existing_resource_object)
        kapi.delete_namespaced_custom_object(
            "twingate.com",
            "v1beta",
            namespace,
            "twingateresources",
            resource_object_name,
        )
        return success(message=f"Deleted TwingateResource {resource_object_name}")

    return success(message=f"TwingateResource {resource_object_name} does not exist")

1	from collections.abc import Callable	1✔
2
3	import kopf	1✔
4	import kubernetes	1✔
5
6	from app.handlers import success	1✔
7	from app.utils import to_bool	1✔
8
9
10	def k8s_get_twingate_resource(	1✔
11	namespace: str, name: str, kapi: kubernetes.client.CustomObjectsApi \| None = None
12	) -> dict \| None:
13	kapi = kapi or kubernetes.client.CustomObjectsApi()	1✔
14	try:	1✔
15	return kapi.get_namespaced_custom_object(	1✔
16	"twingate.com", "v1beta", namespace, "twingateresources", name
17	)
18	except kubernetes.client.exceptions.ApiException as ex:	1✔
19	if ex.status == 404:	1✔
20	return None	1✔
21	raise	1✔
22
23
24	ALLOWED_EXTRA_ANNOTATIONS: list[tuple[str, Callable]] = [	1✔
25	("name", str),
26	("alias", str),
27	("isBrowserShortcutEnabled", to_bool),
28	("securityPolicyId", str),
29	("isVisible", to_bool),
30	]
31
32
33	def service_to_twingate_resource(service_body, namespace) -> dict:	1✔
34	meta = service_body.metadata	1✔
35	spec = service_body.spec	1✔
36	service_name = service_body.meta.name	1✔
37	resource_object_name = f"{service_name}-resource"	1✔
38
39	result: dict = {	1✔
40	"apiVersion": "twingate.com/v1beta",
41	"kind": "TwingateResource",
42	"metadata": {
43	"name": resource_object_name,
44	},
45	"spec": {
46	"name": resource_object_name,
47	"address": f"{service_name}.{namespace}.svc.cluster.local",
48	},
49	}
50
51	for key, convert_f in ALLOWED_EXTRA_ANNOTATIONS:	1✔
52	# TODO: Remove once we release v1.0 (see https://github.com/Twingate/kubernetes-operator/issues/530)
53	if value := meta.annotations.get(f"twingate.com/resource-{key}"):	1✔
54	result["spec"][key] = convert_f(value)	1✔
55	if value := meta.annotations.get(f"resource.twingate.com/{key}"):	1!
56	result["spec"][key] = convert_f(value)	×
57
58	protocols: dict = {	1✔
59	"allowIcmp": False,
60	"tcp": {"policy": "RESTRICTED", "ports": []},
61	"udp": {"policy": "RESTRICTED", "ports": []},
62	}
63	for port_obj in spec.get("ports", []):	1✔
64	port = port_obj["port"]	1✔
65	if port_obj["protocol"] == "TCP":	1✔
66	protocols["tcp"]["ports"].append({"start": port, "end": port})	1✔
67	elif port_obj["protocol"] == "UDP":	1!
68	protocols["udp"]["ports"].append({"start": port, "end": port})	1✔
69
70	result["spec"]["protocols"] = protocols	1✔
71
72	return result	1✔
73
74
75	# TODO: Remove once we release v1.0 (see https://github.com/Twingate/kubernetes-operator/issues/530)
76	@kopf.on.resume("service", annotations={"twingate.com/resource": "true"})	1✔
77	@kopf.on.create("service", annotations={"twingate.com/resource": "true"})	1✔
78	@kopf.on.update("service", annotations={"twingate.com/resource": "true"})	1✔
79	@kopf.on.resume("service", annotations={"resource.twingate.com": "true"})	1✔
80	@kopf.on.create("service", annotations={"resource.twingate.com": "true"})	1✔
81	@kopf.on.update("service", annotations={"resource.twingate.com": "true"})	1✔
82	def twingate_service_create(body, spec, namespace, meta, logger, **_):	1✔
83	logger.info("twingate_service_create: %s", spec)	1✔
84
85	resource_subobject = service_to_twingate_resource(body, namespace)	1✔
86	kopf.adopt(resource_subobject)	1✔
87
88	resource_object_name = resource_subobject["metadata"]["name"]	1✔
89
90	kapi = kubernetes.client.CustomObjectsApi()	1✔
91	if existing_resource_object := k8s_get_twingate_resource(	1✔
92	namespace, resource_object_name, kapi
93	):
94	logger.info("TwingateResource already exists: %s", existing_resource_object)	1✔
95	kapi.patch_namespaced_custom_object(	1✔
96	"twingate.com",
97	"v1beta",
98	namespace,
99	"twingateresources",
100	resource_object_name,
101	resource_subobject,
102	)
103	else:
104	api_response = kapi.create_namespaced_custom_object(	1✔
105	"twingate.com", "v1beta", namespace, "twingateresources", resource_subobject
106	)
107	logger.info("create_namespaced_custom_object response: %s", api_response)	1✔
108
109	return success(	1✔
110	message=f"Created TwingateResource {resource_subobject['spec']['name']}"
111	)
112
113
114	@kopf.on.update("service", annotations={"resource.twingate.com": "false"})	1✔
115	@kopf.on.update("service", annotations={"resource.twingate.com": kopf.ABSENT})	1✔
116	def twingate_service_annotation_removed(body, spec, namespace, meta, logger, **_):	1✔
NEW 117	logger.info("twingate_service_annotation_removed: %s", spec)	×
118
NEW 119	resource_object_name = f"{body.meta.name}-resource"	×
120
NEW 121	kapi = kubernetes.client.CustomObjectsApi()	×
NEW 122	if existing_resource_object := k8s_get_twingate_resource(	×
123	namespace, resource_object_name, kapi
124	):
NEW 125	logger.info("Deleting TwingateResource: %s", existing_resource_object)	×
NEW 126	kapi.delete_namespaced_custom_object(	×
127	"twingate.com",
128	"v1beta",
129	namespace,
130	"twingateresources",
131	resource_object_name,
132	)
NEW 133	return success(message=f"Deleted TwingateResource {resource_object_name}")	×
134
NEW 135	return success(message=f"TwingateResource {resource_object_name} does not exist")	×

Twingate / kubernetes-operator / 13577129160

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous