diff --git a/CI/tests_v2/scenarios/pod_error_scenarios/test_pod_error_scenarios.py b/CI/tests_v2/scenarios/pod_error_scenarios/test_pod_error_scenarios.py index c6afcb13..e7005cc1 100644 --- a/CI/tests_v2/scenarios/pod_error_scenarios/test_pod_error_scenarios.py +++ b/CI/tests_v2/scenarios/pod_error_scenarios/test_pod_error_scenarios.py @@ -87,25 +87,51 @@ class TestPodErrorScenarios(BaseScenarioTest): result, ns, expected_reasons=["not enough pods match", "expected 100", "found only 2 pods"] ) - def test_recovery_timeout_fails(self): + def test_recovery_timeout_fails(self, wait_for_pods_running): + """Verify Krkn reports failure when pods cannot recover within the timeout. + + Instead of relying on a very short timeout (which is racy on fast + clusters), we add a readiness probe with a long initialDelaySeconds + so replacement pods *structurally* cannot become Ready within the + recovery window. + """ ns = self.ns - before = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) - before_names = [p.metadata.name for p in before.items] + READINESS_DELAY = 30 + RECOVERY_TIMEOUT = 10 + + # Patch deployment: use Recreate strategy so both pods roll out in + # parallel, and add a readiness probe that won't pass for 30s. + # READINESS_DELAY only needs to be greater than RECOVERY_TIMEOUT. + patch_body = { + "spec": { + "strategy": {"type": "Recreate", "rollingUpdate": None}, + "template": { + "spec": { + "containers": [{ + "name": "app", + "readinessProbe": { + "httpGet": {"path": "/", "port": 80}, + "initialDelaySeconds": READINESS_DELAY, + "periodSeconds": 5, + }, + }] + } + } + } + } + self.k8s_apps.patch_namespaced_deployment( + name="krkn-pod-error-target", namespace=ns, body=patch_body + ) + wait_for_pods_running(ns, self.LABEL_SELECTOR, timeout=READINESS_DELAY + 30) result = self.run_scenario( - self.tmp_path, ns, overrides={"krkn_pod_recovery_time": 1} + self.tmp_path, ns, overrides={"krkn_pod_recovery_time": RECOVERY_TIMEOUT} ) assert_kraken_failure(result, context=f"namespace={ns}", tmp_path=self.tmp_path) - - # Verify no hang and logs contain namespace and expected timeout/recovery errors + self.assert_failure_logs_contain( - result, ns, expected_reasons=["timeout", "recover"] + result, ns, expected_reasons=["unrecovered"] ) - - # Verify at least one target pod name is in the logs - combined_lower = ((result.stdout or "") + "\n" + (result.stderr or "")).lower() - found_pod = any(name.lower() in combined_lower for name in before_names) - assert found_pod, f"None of the target pods {before_names} were found in failure logs" @pytest.mark.no_workload def test_invalid_namespace_pattern_fails(self):