refactored logics and added unit test for --api-server-timeout argument

kube_downscaler/helper.py

@@ -65,9 +65,9 @@ def _matches_absolute_time_spec(time: datetime.datetime, match: Match):
     return time_from <= time <= time_to
 
 
-def get_kube_api():
+def get_kube_api(timeout: int):
     config = pykube.KubeConfig.from_env()
-    api = pykube.HTTPClient(config)
+    api = pykube.HTTPClient(config, timeout=timeout)
     return api
 
 

kube_downscaler/scaler.py

@@ -1357,11 +1357,10 @@ def scale(
     enable_events: bool = False,
     matching_labels: FrozenSet[Pattern] = frozenset(),
 ):
-    api = helper.get_kube_api()
+    api = helper.get_kube_api(api_server_timeout)
 
     now = datetime.datetime.now(datetime.timezone.utc)
     forced_uptime = pods_force_uptime(api, namespaces)
-    pykube.http.DEFAULT_HTTP_TIMEOUT=api_server_timeout
 
     for clazz in RESOURCE_CLASSES:
         plural = clazz.endpoint

tests/test_scaler.py

@@ -12,6 +12,41 @@
 from kube_downscaler.scaler import scale_down_jobs
 from kube_downscaler.scaler import scale_up_jobs
 
+def test_scale_custom_timeout(monkeypatch):
+    api_server_timeout = 15  # Defined by the user
+    api = MagicMock()
+    api.timeout = 15  # Expected timeout
+
+    mock_get_kube_api = MagicMock(return_value=api)
+    monkeypatch.setattr(
+        "kube_downscaler.scaler.helper.get_kube_api", mock_get_kube_api
+    )
+
+    scale(
+        namespaces=frozenset({"default"}),
+        upscale_period="never",
+        downscale_period="never",
+        default_uptime="never",
+        default_downtime="always",
+        upscale_target_only=False,
+        include_resources=frozenset(["pods"]),
+        exclude_namespaces=frozenset(),
+        exclude_deployments=frozenset(),
+        dry_run=False,
+        grace_period=300,
+        admission_controller="",
+        constrained_downscaler=False,
+        api_server_timeout=api_server_timeout,
+        downtime_replicas=0,
+        deployment_time_annotation=None,
+        enable_events=False,
+        matching_labels=frozenset(),
+    )
+
+    # ensure get_kube_api is called with the correct timeout value
+    mock_get_kube_api.assert_called_once_with(api_server_timeout)
+    # ensure timeout value is correctly set on the returned object
+    assert api.timeout == api_server_timeout
 
 def test_scaler_always_up(monkeypatch):
     api = MagicMock()