fix: harden test suite for real K8s — poll_until, retry loops, SDK methods

Author: amanjain

docs/NORTHSTAR.txt

@@ -6688,6 +6688,53 @@ TEST ISOLATION + CLEANUP
     isolation as unit tests.
 
 
+REAL K8S TESTING RULES — HARD-WON, NON-NEGOTIABLE
+──────────────────────────────────────────────────
+These rules prevent flaky CI failures on real K8s (--backend real).
+Every new test touching real K8s MUST follow these.
+
+    1. NEVER use raw k8s.update_deployment/service/statefulset() in tests
+       that run on real K8s. Deployments have controllers that race to
+       mutate resourceVersion → 409 Conflict. Use SDK methods (ps.scale,
+       deploy.deploy) which have built-in retry logic.
+
+    2. NEVER assert immediately after delete. K8s deletion is async.
+       Use poll_until(fn=_get_or_none, predicate=lambda r: r is None).
+
+    3. NEVER use time.sleep() for stabilization. Use poll_until() or
+       wait_for_port_open(). Sleeps waste CI time and still flake.
+
+    4. NEVER create Deployments/Services with raw k8s.create_*() in tests
+       that run on real K8s. Use factory.deploy.deploy() which sets
+       correct labels for cleanup.
+
+    5. Contract test updates MUST use read-modify-write + retry loop:
+           for _attempt in range(5):
+               current = k8s.get_deployment(ns, name)
+               current["spec"]["replicas"] = 3
+               try:
+                   k8s.update_deployment(ns, name, current)
+                   break
+               except FakeConflictError:
+                   continue
+
+    6. Namespace tests MUST clean up — use request.addfinalizer() and
+       k8ut- prefix so session cleanup catches strays.
+
+    7. _wipe_namespace_resources() MUST wait for ALL resource types
+       (Deployment, StatefulSet, Service, Ingress, NetworkPolicy, PVC,
+       ConfigMap, Secret) — not just a subset.
+
+    8. Know which fixtures route to real K8s:
+       - factory, k8s_client → real with --backend real
+       - app_factory, deployed_factory, fake_k8s → always FakeK8sClient
+
+    Shared E2E helpers (tests/e2e/helpers.py):
+    poll_until, wait_for_pods_ready, wait_for_port_open,
+    wait_for_sts_ready, wait_for_service_deleted,
+    wait_for_lb_address, wait_for_no_pods, find_running_pod
+
+
 UNIT TEST EXAMPLES
 ──────────────────
     def test_apps_create(fake_client):

tests/apps/test_destroy.py

@@ -90,42 +90,10 @@ def test_destroy_cleans_up_deployments_and_services(
     ) -> None:
         """Destroy also deletes deployments and services with matching labels."""
         apps_service.create("myapi")
+        # Use SDK deploy (creates Deployment + Service with correct labels)
+        factory.deploy.deploy("myapi", image="nginx:1.27", wait=False)
         k8s = factory.k8s
         ns = factory.namespace
-        # Simulate a deployment and service existing for this app
-        app_labels = {
-            f"{factory.domain}/managed-by": factory.prefix,
-            f"{factory.domain}/app": "myapi",
-        }
-        k8s.create_deployment(
-            ns,
-            {
-                "apiVersion": "apps/v1",
-                "kind": "Deployment",
-                "metadata": {"name": f"{factory.prefix}-myapi-web", "labels": dict(app_labels)},
-                "spec": {
-                    "replicas": 1,
-                    "selector": {"matchLabels": {"app": "myapi-web"}},
-                    "template": {
-                        "metadata": {"labels": {"app": "myapi-web"}},
-                        "spec": {"containers": [{"name": "web", "image": "nginx"}]},
-                    },
-                },
-            },
-        )
-        k8s.create_service(
-            ns,
-            {
-                "apiVersion": "v1",
-                "kind": "Service",
-                "metadata": {"name": f"{factory.prefix}-myapi-web", "labels": dict(app_labels)},
-                "spec": {
-                    "type": "ClusterIP",
-                    "selector": {"app": "myapi-web"},
-                    "ports": [{"port": 80, "targetPort": 8080, "protocol": "TCP"}],
-                },
-            },
-        )
         apps_service.destroy("myapi")
         # Verify everything is gone
         assert k8s.list_deployments(ns, labels={f"{factory.domain}/app": "myapi"}) == []

tests/apps/test_info.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import json
 from unittest.mock import patch
 
 import pytest
@@ -38,21 +37,10 @@ def test_info_with_formation_data(
     ) -> None:
         """info() parses formation JSON data into ProcessFormation."""
         apps_service.create("myapi")
-        # Write formation data directly
-        k8s = factory.k8s
-        ns = factory.namespace
-        p = factory.prefix
-        cm = k8s.get_configmap(ns, f"{p}-formation-myapi")
-        cm["data"] = {
-            "web": json.dumps(
-                {
-                    "replicas": 3,
-                    "command": "gunicorn app:app",
-                    "command_source": "manual",
-                }
-            ),
-        }
-        k8s.update_configmap(ns, f"{p}-formation-myapi", cm)
+        factory.deploy.deploy("myapi", image="nginx:1.27", wait=False)
+        # Use SDK methods (handles 409 retries on real K8s)
+        factory.ps.scale("myapi", {"web": 3})
+        factory.ps.set_commands("myapi", {"web": "gunicorn app:app"}, no_restart=True)
         app = apps_service.info("myapi")
         assert "web" in app.process_types
         assert app.process_types["web"].replicas == 3

tests/apps/test_rename.py

@@ -77,42 +77,10 @@ def test_rename_updates_deployments_and_services(
     ) -> None:
         """Rename updates labels on deployments and services too."""
         apps_service.create("oldapp")
+        # Use SDK deploy (creates Deployment + Service with correct labels)
+        factory.deploy.deploy("oldapp", image="nginx:1.27", wait=False)
         k8s = factory.k8s
         ns = factory.namespace
-        # Simulate deployment and service
-        app_labels = {
-            f"{factory.domain}/managed-by": factory.prefix,
-            f"{factory.domain}/app": "oldapp",
-        }
-        k8s.create_deployment(
-            ns,
-            {
-                "apiVersion": "apps/v1",
-                "kind": "Deployment",
-                "metadata": {"name": f"{factory.prefix}-oldapp-web", "labels": dict(app_labels)},
-                "spec": {
-                    "replicas": 1,
-                    "selector": {"matchLabels": {"app": "oldapp-web"}},
-                    "template": {
-                        "metadata": {"labels": {"app": "oldapp-web"}},
-                        "spec": {"containers": [{"name": "web", "image": "nginx"}]},
-                    },
-                },
-            },
-        )
-        k8s.create_service(
-            ns,
-            {
-                "apiVersion": "v1",
-                "kind": "Service",
-                "metadata": {"name": f"{factory.prefix}-oldapp-web", "labels": dict(app_labels)},
-                "spec": {
-                    "type": "ClusterIP",
-                    "selector": {"app": "oldapp-web"},
-                    "ports": [{"port": 80, "targetPort": 8080, "protocol": "TCP"}],
-                },
-            },
-        )
         apps_service.rename("oldapp", "newapp")
         # Verify labels were updated
         deps = k8s.list_deployments(ns, labels={f"{factory.domain}/app": "newapp"})

tests/apps/test_status.py

@@ -2,14 +2,13 @@
 
 from __future__ import annotations
 
-import time
-
 import pytest
 from click.testing import CliRunner
 
 from kuberoku.cli.main import cli
 from kuberoku.exceptions import AppNotFoundError
 from kuberoku.factory import KuberokuFactory
+from tests.e2e.helpers import poll_until
 from tests.services.conftest import _create_fake_pod
 
 
@@ -36,12 +35,14 @@ def test_status_with_deploy(
         if request.config.getoption("--backend") == "fake":
             _create_fake_pod(factory, "myapi", "web", "pod-a", image="nginx:1.27")
         else:
-            # Real K8s: deployment creates its own pod — wait for it
-            for _ in range(30):
-                st = factory.apps.status("myapi")
-                if st.dynos:
-                    break
-                time.sleep(1)
+            # Real K8s: deployment creates its own pod — poll until visible
+            poll_until(
+                fn=lambda: factory.apps.status("myapi"),
+                predicate=lambda st: len(st.dynos) >= 1,
+                timeout=60,
+                interval=1,
+                message="Pod never appeared in status after deploy",
+            )
         st = factory.apps.status("myapi")
         assert len(st.dynos) >= 1
         assert st.dynos[0].process_type == "web"
@@ -105,6 +106,15 @@ def test_status_cli_with_release_and_processes(
         factory.deploy.deploy("myapi", image="nginx:1.27", wait=False)
         if request.config.getoption("--backend") == "fake":
             _create_fake_pod(factory, "myapi", "web", "pod-a", image="nginx:1.27")
+        else:
+            # Real K8s: wait for pod before checking CLI output
+            poll_until(
+                fn=lambda: factory.apps.status("myapi"),
+                predicate=lambda st: len(st.dynos) >= 1,
+                timeout=60,
+                interval=1,
+                message="Pod never appeared in status after deploy",
+            )
         factory.config.set("myapi", {"DB": "pg://x"})
         factory.config.set("myapi", {"KEY": "secret"}, secret=True)
         factory.addons.create("myapi", "postgres")

tests/conftest.py

@@ -226,23 +226,37 @@ def _wipe_namespace_resources(ns_name: str) -> None:
             remaining = 0
             with contextlib.suppress(Exception):
                 remaining += len(
-                    net_v1.list_namespaced_network_policy(ns_name, label_selector=label_sel).items
+                    apps_v1.list_namespaced_deployment(ns_name, label_selector=label_sel).items
                 )
             with contextlib.suppress(Exception):
                 remaining += len(
-                    core.list_namespaced_config_map(ns_name, label_selector=label_sel).items
+                    apps_v1.list_namespaced_stateful_set(ns_name, label_selector=label_sel).items
                 )
             with contextlib.suppress(Exception):
                 remaining += len(
-                    core.list_namespaced_secret(ns_name, label_selector=label_sel).items
+                    core.list_namespaced_service(ns_name, label_selector=label_sel).items
                 )
             with contextlib.suppress(Exception):
                 remaining += len(
-                    core.list_namespaced_service(ns_name, label_selector=label_sel).items
+                    net_v1.list_namespaced_ingress(ns_name, label_selector=label_sel).items
                 )
             with contextlib.suppress(Exception):
                 remaining += len(
-                    apps_v1.list_namespaced_deployment(ns_name, label_selector=label_sel).items
+                    net_v1.list_namespaced_network_policy(ns_name, label_selector=label_sel).items
+                )
+            with contextlib.suppress(Exception):
+                remaining += len(
+                    core.list_namespaced_persistent_volume_claim(
+                        ns_name, label_selector=label_sel
+                    ).items
+                )
+            with contextlib.suppress(Exception):
+                remaining += len(
+                    core.list_namespaced_config_map(ns_name, label_selector=label_sel).items
+                )
+            with contextlib.suppress(Exception):
+                remaining += len(
+                    core.list_namespaced_secret(ns_name, label_selector=label_sel).items
                 )
             if remaining == 0:
                 return

tests/contract/test_deployment_crud.py

@@ -11,6 +11,7 @@
 import pytest
 
 from tests.backends.fake import FakeConflictError, FakeNotFoundError
+from tests.e2e.helpers import poll_until
 
 
 def _deployment_body(
@@ -86,8 +87,20 @@ def test_delete_then_get_raises(self, k8s_client, tmp_namespace):
         body = _deployment_body("dep-del")
         k8s_client.create_deployment(tmp_namespace, body)
         k8s_client.delete_deployment(tmp_namespace, "dep-del")
-        with pytest.raises(FakeNotFoundError):
-            k8s_client.get_deployment(tmp_namespace, "dep-del")
+
+        def _get_or_none() -> dict[str, Any] | None:
+            try:
+                return k8s_client.get_deployment(tmp_namespace, "dep-del")
+            except FakeNotFoundError:
+                return None
+
+        poll_until(
+            fn=_get_or_none,
+            predicate=lambda result: result is None,
+            timeout=30,
+            interval=0.5,
+            message="Deployment dep-del still exists after deletion",
+        )
 
     def test_update_changes_labels(self, k8s_client, tmp_namespace):
         body = _deployment_body("dep-upd", extra_meta_labels={"version": "v1"})

tests/contract/test_ingress_crud.py

@@ -8,7 +8,6 @@
 from __future__ import annotations
 
 import contextlib
-import time
 from typing import Any
 
 import pytest
@@ -74,13 +73,19 @@ def test_get_not_found_raises(self, k8s_client, tmp_namespace) -> None:
 
 
 class TestIngressUpdate:
-    def test_update_changes_rules(self, k8s_client, tmp_namespace, request) -> None:
+    def test_update_changes_rules(self, k8s_client, tmp_namespace) -> None:
         k8s_client.create_ingress(tmp_namespace, _ingress_body("my-ingress", "old.com"))
-        if request.config.getoption("--backend") == "real":
-            time.sleep(1)  # K8s controller mutates ingress after create
-        ingress = k8s_client.get_ingress(tmp_namespace, "my-ingress")
-        ingress["spec"]["rules"][0]["host"] = "new.com"
-        k8s_client.update_ingress(tmp_namespace, "my-ingress", ingress)
+        # Read-modify-write with retry: real K8s ingress controller may mutate
+        for _attempt in range(5):
+            ingress = k8s_client.get_ingress(tmp_namespace, "my-ingress")
+            ingress["spec"]["rules"][0]["host"] = "new.com"
+            try:
+                k8s_client.update_ingress(tmp_namespace, "my-ingress", ingress)
+                break
+            except FakeConflictError:
+                continue
+        else:
+            pytest.fail("Ingress update failed after 5 retries due to conflict")
         updated = k8s_client.get_ingress(tmp_namespace, "my-ingress")
         assert updated["spec"]["rules"][0]["host"] == "new.com"
 
@@ -102,8 +107,20 @@ class TestIngressDelete:
     def test_delete_removes(self, k8s_client, tmp_namespace) -> None:
         k8s_client.create_ingress(tmp_namespace, _ingress_body("my-ingress"))
         k8s_client.delete_ingress(tmp_namespace, "my-ingress")
-        with pytest.raises(FakeNotFoundError):
-            k8s_client.get_ingress(tmp_namespace, "my-ingress")
+
+        def _get_or_none() -> dict[str, Any] | None:
+            try:
+                return k8s_client.get_ingress(tmp_namespace, "my-ingress")
+            except FakeNotFoundError:
+                return None
+
+        poll_until(
+            fn=_get_or_none,
+            predicate=lambda result: result is None,
+            timeout=30,
+            interval=0.5,
+            message="Ingress my-ingress still exists after deletion",
+        )
 
     def test_delete_not_found_raises(self, k8s_client, tmp_namespace) -> None:
         with pytest.raises(FakeNotFoundError):