feat: adding deployment commands

Author: diegorubin

lifeguard_k8s/infrastructure/__init__.py

@@ -0,0 +1,21 @@
+from kubernetes import client, config
+from lifeguard_k8s.settings import (
+    LIFEGUARD_KUBERNETES_CONFIG,
+)
+
+
+def _load_config():
+    if LIFEGUARD_KUBERNETES_CONFIG:
+        config.load_kube_config(LIFEGUARD_KUBERNETES_CONFIG)
+    else:
+        config.load_incluster_config()
+
+
+def get_client():
+    _load_config()
+    return client.CoreV1Api()
+
+
+def get_apps_client():
+    _load_config()
+    return client.AppsV1Api()

lifeguard_k8s/infrastructure/deployments.py

@@ -0,0 +1,8 @@
+from lifeguard_k8s.infrastructure import get_apps_client
+
+
+def scale_a_deployment(namespace, deployment_name, replicas):
+    apps_v1 = get_apps_client()
+    apps_v1.patch_namespaced_deployment_scale(
+        deployment_name, namespace, {"spec": {"replicas": replicas}}
+    )

lifeguard_k8s/infrastructure/namespaces.py

@@ -0,0 +1,41 @@
+from lifeguard_k8s.infrastructure import get_client, get_apps_client
+
+
+def get_namespace_infos(namespace):
+    """
+    Return current main infos of a namespace
+    """
+    infos = {"pods": [], "deployments": []}
+
+    v1 = get_client()
+    apps_v1 = get_apps_client()
+
+    pods = v1.list_namespaced_pod(namespace)
+    deployments = apps_v1.list_namespaced_deployment(namespace)
+
+    for pod in pods.items:
+        infos["pods"].append(
+            {
+                "name": pod.metadata.name,
+                "status": pod.status.phase,
+                "containers": [
+                    {
+                        "name": container.name,
+                        "ready": container.ready,
+                        "restart_count": container.restart_count,
+                    }
+                    for container in pod.status.container_statuses
+                ],
+            }
+        )
+
+    for deployment in deployments.items:
+        infos["deployments"].append(
+            {
+                "name": deployment.metadata.name,
+                "replicas": deployment.status.replicas,
+                "ready_replicas": deployment.status.ready_replicas,
+                "unavailable_replicas": deployment.status.unavailable_replicas,
+            }
+        )
+    return infos

lifeguard_k8s/infrastructure/pods.py

@@ -1,6 +1,5 @@
-from kubernetes import client, config
+from lifeguard_k8s.infrastructure import get_client
 from lifeguard_k8s.settings import (
-    LIFEGUARD_KUBERNETES_CONFIG,
     LIFEGUARD_KUBERNETES_READ_LOG_MAX_SIZE,
 )
 
@@ -24,46 +23,10 @@ def _exists_success_pod_after_job(job_pod, pods):
     return False
 
 
-def _get_clients():
-    if LIFEGUARD_KUBERNETES_CONFIG:
-        config.load_kube_config(LIFEGUARD_KUBERNETES_CONFIG)
-    else:
-        config.load_incluster_config()
-
-    return client.CoreV1Api()
-
-
-def get_namespace_infos(namespace):
-    """
-    Return current main infos of a namespace
-    """
-    infos = {"pods": []}
-
-    v1 = _get_clients()
-    pods = v1.list_namespaced_pod(namespace)
-
-    for pod in pods.items:
-        infos["pods"].append(
-            {
-                "name": pod.metadata.name,
-                "status": pod.status.phase,
-                "containers": [
-                    {
-                        "name": container.name,
-                        "ready": container.ready,
-                        "restart_count": container.restart_count,
-                    }
-                    for container in pod.status.container_statuses
-                ],
-            }
-        )
-    return infos
-
-
 def get_not_running_pods(namespace):
     not_running_pods = []
 
-    v1 = _get_clients()
+    v1 = get_client()
     pods = v1.list_namespaced_pod(namespace)
 
     for pod in pods.items:
@@ -80,12 +43,12 @@ def get_not_running_pods(namespace):
 
 
 def delete_a_pod(namespace, pod_name):
-    v1 = _get_clients()
+    v1 = get_client()
     v1.delete_namespaced_pod(pod_name, namespace)
 
 
 def get_events_from_pod(namespace, pod_name):
-    v1 = _get_clients()
+    v1 = get_client()
     events = v1.list_namespaced_event(
         namespace, field_selector=f"involvedObject.name={pod_name}"
     )
@@ -109,6 +72,6 @@ def get_last_error_event_from_pod(namespace, pod_name):
 
 
 def get_logs_from_pod(namespace, pod_name):
-    v1 = _get_clients()
+    v1 = get_client()
     log = v1.read_namespaced_pod_log(pod_name, namespace)
     return log[-LIFEGUARD_KUBERNETES_READ_LOG_MAX_SIZE:]

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name="lifeguard-k8s",
-    version="1.2.1",
+    version="1.3.0",
     url="https://github.com/LifeguardSystem/lifeguard-k8s",
     author="Diego Rubin",
     author_email="[email protected]",

tests/infrastructure/__init__.py

@@ -0,0 +1,30 @@
+from unittest.mock import MagicMock
+
+
+def build_pod(status, container_status, pod_name, kind="ReplicaSet"):
+    container_statuses = MagicMock(name="container_statuses")
+    container_statuses.ready = container_status
+    container_statuses.restart_count = 0
+    container_statuses.name = "container_name"
+
+    owner_reference = MagicMock(name="owner_reference")
+    owner_reference.kind = kind
+    owner_reference.name = pod_name
+
+    pod = MagicMock(name="pod")
+    pod.status = MagicMock(name="status")
+    pod.status.phase = status
+    pod.status.container_statuses = [container_statuses]
+    pod.metadata.name = pod_name
+    pod.metadata.owner_references = [owner_reference]
+
+    return pod
+
+
+def build_deployment(name, replicas, ready_replicas):
+    deployment = MagicMock(name="deployment")
+    deployment.metadata.name = name
+    deployment.status.replicas = replicas
+    deployment.status.ready_replicas = ready_replicas
+    deployment.status.unavailable_replicas = 0
+    return deployment

tests/infrastructure/test_deployments.py

@@ -0,0 +1,20 @@
+from unittest import TestCase
+from unittest.mock import patch
+
+from lifeguard_k8s.infrastructure.deployments import scale_a_deployment
+
+
+class InfrastructureDeploymentsTests(TestCase):
+    @patch("lifeguard_k8s.infrastructure.config")
+    @patch("lifeguard_k8s.infrastructure.client")
+    def test_scale_a_deployment(self, mock_client, _mock_config):
+        patch_namespaced_deployment_scale = (
+            mock_client.AppsV1Api.return_value.patch_namespaced_deployment_scale
+        )
+        patch_namespaced_deployment_scale.return_value = ""
+
+        scale_a_deployment("namespace", "deployment_name", 1)
+
+        patch_namespaced_deployment_scale.assert_called_with(
+            "deployment_name", "namespace", {"spec": {"replicas": 1}}
+        )

tests/infrastructure/test_namespaces.py

@@ -0,0 +1,49 @@
+from unittest import TestCase
+from unittest.mock import patch
+
+from tests.infrastructure import build_pod, build_deployment
+
+from lifeguard_k8s.infrastructure.namespaces import get_namespace_infos
+
+
+class InfrastructureNamespacesTests(TestCase):
+    @patch("lifeguard_k8s.infrastructure.config")
+    @patch("lifeguard_k8s.infrastructure.client")
+    def test_get_namespace_infos(self, mock_client, _mock_config):
+        pod = build_pod("Running", True, "pod_name")
+        deployment = build_deployment("deployment_name", 1, 1)
+
+        mock_client.CoreV1Api.return_value.list_namespaced_pod.return_value.items = [
+            pod
+        ]
+
+        mock_client.AppsV1Api.return_value.list_namespaced_deployment.return_value.items = [
+            deployment
+        ]
+
+        self.assertEqual(
+            get_namespace_infos("namespace"),
+            {
+                "deployments": [
+                    {
+                        "name": "deployment_name",
+                        "ready_replicas": 1,
+                        "replicas": 1,
+                        "unavailable_replicas": 0,
+                    }
+                ],
+                "pods": [
+                    {
+                        "containers": [
+                            {
+                                "name": "container_name",
+                                "ready": True,
+                                "restart_count": 0,
+                            }
+                        ],
+                        "name": "pod_name",
+                        "status": "Running",
+                    }
+                ],
+            },
+        )