Enhanced Tests For Scaled Objects Downscaling/Upscaling

chart/Chart.yaml

@@ -3,5 +3,5 @@ name: py-kube-downscaler
 description: A Helm chart for deploying py-kube-downscaler
 
 type: application
-version: 0.2.6
-appVersion: "24.8.1"
+version: 0.2.7
+appVersion: "24.8.2"

chart/templates/rbac.yaml

@@ -63,6 +63,7 @@ rules:
     - batch
   resources:
     - cronjobs
+    - jobs
   verbs:
     - get
     - watch

kube_downscaler/resources/keda.py

@@ -15,9 +15,11 @@ class ScaledObject(NamespacedAPIObject):
     @property
     def replicas(self):
         if ScaledObject.keda_pause_annotation in self.annotations:
-            if self.annotations[ScaledObject.keda_pause_annotation] == "0":
+            if self.annotations[ScaledObject.keda_pause_annotation] is None:
+                replicas = 1
+            elif self.annotations[ScaledObject.keda_pause_annotation] == "0":
                 replicas = 0
-            elif self.annotations[ScaledObject.keda_pause_annotation] != "0":
+            elif self.annotations[ScaledObject.keda_pause_annotation] != "0" and self.annotations[ScaledObject.keda_pause_annotation] is not None:
                 replicas = int(self.annotations[ScaledObject.keda_pause_annotation])
         else:
             replicas = 1

tests/test_autoscale_resource.py

@@ -935,7 +935,7 @@ def test_upscale_hpa_with_autoscaling():
     assert hpa.obj["spec"]["minReplicas"] == 4
     assert hpa.obj["metadata"]["annotations"][ORIGINAL_REPLICAS_ANNOTATION] is None
 
-    
+
 def test_downscale_pdb_minavailable_with_autoscaling():
     pdb = PodDisruptionBudget(
         None,
@@ -1002,6 +1002,7 @@ def test_upscale_pdb_minavailable_with_autoscaling():
     assert pdb.obj["spec"]["minAvailable"] == 4
     assert pdb.obj["metadata"]["annotations"][ORIGINAL_REPLICAS_ANNOTATION] is None
 
+
 def test_downscale_pdb_maxunavailable_with_autoscaling():
     pdb = PodDisruptionBudget(
         None,
@@ -1068,7 +1069,7 @@ def test_upscale_pdb_maxunavailable_with_autoscaling():
     assert pdb.obj["spec"]["maxUnavailable"] == 4
     assert pdb.obj["metadata"]["annotations"][ORIGINAL_REPLICAS_ANNOTATION] is None
 
-    
+
 def test_downscale_daemonset_with_autoscaling():
     ds = DaemonSet(
         None,
@@ -1168,7 +1169,6 @@ def test_downscale_scaledobject_with_pause_annotation_already_present():
         tzinfo=timezone.utc
     )
 
-
     autoscale_resource(
         so,
         upscale_target_only=False,
@@ -1185,7 +1185,7 @@ def test_downscale_scaledobject_with_pause_annotation_already_present():
 
     # Check if the annotations have been correctly updated
     assert so.annotations[ScaledObject.keda_pause_annotation] == "0"
-    assert so.annotations[ScaledObject.last_keda_pause_annotation_if_present] == "3"
+    assert so.replicas == 0
 
 
 def test_upscale_scaledobject_with_pause_annotation_already_present():
@@ -1206,12 +1206,10 @@ def test_upscale_scaledobject_with_pause_annotation_already_present():
         }
     )
 
-
     now = datetime.strptime("2023-08-21T10:30:00Z", "%Y-%m-%dT%H:%M:%SZ").replace(
         tzinfo=timezone.utc
     )
 
-
     autoscale_resource(
         so,
         upscale_target_only=False,
@@ -1228,18 +1226,20 @@ def test_upscale_scaledobject_with_pause_annotation_already_present():
 
     # Check if the annotations have been correctly updated for the upscale operation
     assert so.annotations[ScaledObject.keda_pause_annotation] == "3"
+    assert so.replicas == 3
     assert so.annotations.get(ScaledObject.last_keda_pause_annotation_if_present) is None
 
-def test_downscale_scaledobject_without_pause_annotation():
+
+def test_downscale_scaledobject_without_keda_pause_annotation():
     so = ScaledObject(
         None,
         {
             "metadata": {
                 "name": "scaledobject-1",
                 "namespace": "default",
                 "creationTimestamp": "2023-08-21T10:00:00Z",
+                "annotations": {}
             },
-            "spec": {}
         }
     )
 
@@ -1263,9 +1263,11 @@ def test_downscale_scaledobject_without_pause_annotation():
 
     # Check if the annotations have been correctly updated
     assert so.annotations[ScaledObject.keda_pause_annotation] == "0"
+    assert so.annotations.get(ScaledObject.last_keda_pause_annotation_if_present) is None
+    assert so.replicas == 0
 
 
-def test_upscale_scaledobject_without_pause_annotation():
+def test_upscale_scaledobject_without_keda_pause_annotation():
     so = ScaledObject(
         None,
         {
@@ -1275,20 +1277,16 @@ def test_upscale_scaledobject_without_pause_annotation():
                 "creationTimestamp": "2023-08-21T10:00:00Z",
                 "annotations": {
                     "autoscaling.keda.sh/paused-replicas": "0",
-                    "downscaler/original-pause-replicas": "3",
                     "downscaler/original-replicas": "3",
                 }
             },
-            "spec": {}
         }
     )
 
-
     now = datetime.strptime("2023-08-21T10:30:00Z", "%Y-%m-%dT%H:%M:%SZ").replace(
         tzinfo=timezone.utc
     )
 
-
     autoscale_resource(
         so,
         upscale_target_only=False,
@@ -1304,5 +1302,6 @@ def test_upscale_scaledobject_without_pause_annotation():
     )
 
     # Check if the annotations have been correctly updated for the upscale operation
-    assert so.annotations[ScaledObject.keda_pause_annotation] == "3"
-    assert so.annotations.get(ScaledObject.last_keda_pause_annotation_if_present) is None
+    assert so.annotations[ScaledObject.keda_pause_annotation] is None
+    assert so.annotations.get(ScaledObject.last_keda_pause_annotation_if_present) is None
+    assert so.replicas == 1