test(k8s): register retry classifier suite-wide to kill real retry-sleeps

The k8s handlers wrap kubernetes-client calls in the resilience retry layer; the
K8sRetryClassifier (registered during provider bootstrap in production) marks
404/4xx ApiExceptions as non-retryable. Unit/mocked tests skip bootstrap, so
terminal errors were retried with real 1s time.sleep backoff — a handful of tests
dominated the suite. Add a tree-wide autouse fixture (tests/providers/k8s/
conftest.py) that registers the classifier, mirroring production. Also fix
test_job_deleted_not_found_no_warning_flood to raise a real ApiException(404)
instead of a bare Exception('Not Found') so the classifier recognises it and the
handler fails fast — no is_not_found override or retry-backoff hack needed. k8s
leg ~40s -> ~18s, 1764 passing, no behaviour change.

Author: fgogolli

tests/providers/k8s/conftest.py

@@ -0,0 +1,39 @@
+"""Shared fixtures for the k8s provider test tree.
+
+The k8s handlers wrap kubernetes-client calls in the resilience retry layer.
+Whether an error is retried is decided by the *retry classifier registry*: the
+``K8sRetryClassifier`` marks 404/400/401/403 (and similar terminal API errors)
+as non-retryable so the handler fails fast instead of retrying.
+
+In production that classifier is registered during provider bootstrap
+(``K8sProviderStrategy.initialize`` → ``registration.py``).  Unit and mocked
+tests construct handlers directly and never run that bootstrap, so without this
+fixture the registry is empty: every exception — including the 404s that
+mocked tests deliberately provoke — is treated as retryable and retried with
+real ``time.sleep`` backoff (``base_delay`` defaults to 1s).  A handful of
+tests then spend seconds sleeping through retries they should never attempt,
+dominating the k8s suite's wall-clock.
+
+Registering the classifier for the whole k8s tree makes those terminal errors
+fail fast (matching production behaviour) and removes the spurious retry
+sleeps.  It is autouse + function-scoped with a clean teardown so no test
+leaks classifier state into another.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _register_k8s_retry_classifier():
+    """Register K8sRetryClassifier for each k8s test (mirrors provider bootstrap)."""
+    from orb.infrastructure.resilience.retry_classifier_registry import (
+        clear_classifiers,
+        register_retry_classifier,
+    )
+    from orb.providers.k8s.resilience.retry_classifier import K8sRetryClassifier
+
+    register_retry_classifier(K8sRetryClassifier())
+    yield
+    clear_classifiers()

tests/providers/k8s/unit/handlers/test_status_resolver_perf.py

@@ -714,9 +714,14 @@ def test_job_deleted_not_found_no_warning_flood(caplog: Any) -> None:
     """A not-found Job on read_job_status must log at DEBUG, not WARNING."""
     import logging
 
+    # Raise a real ApiException(404): the K8sRetryClassifier recognises it as
+    # non-retryable so the handler fails fast — no retry backoff, no sleep.
+    from kubernetes.client.exceptions import ApiException
+
     from orb.providers.k8s.infrastructure.handlers.job_handler import K8sJobHandler
 
-    not_found_exc = Exception("Not Found")
+    not_found_exc = ApiException(status=404)
+    not_found_exc.status = 404
 
     core_v1 = MagicMock()
     core_v1.list_namespaced_pod.return_value = SimpleNamespace(items=[])
@@ -731,9 +736,9 @@ def test_job_deleted_not_found_no_warning_flood(caplog: Any) -> None:
     logger_mock = MagicMock()
     handler = K8sJobHandler(kubernetes_client=client, config=config, logger=logger_mock)
 
-    # Make is_not_found return True for our exception.
-    handler.is_not_found = lambda exc: True  # type: ignore[method-assign]
-
+    # No is_not_found override or retry-backoff hack needed: the real
+    # ApiException(404) is recognised as not-found by the handler and as
+    # non-retryable by K8sRetryClassifier, so it fails fast on the first attempt.
     request = _req(provider_api="Job", deployment_name="job-1")
 
     with caplog.at_level(logging.WARNING):

tests/providers/k8s/unit/services/test_metrics_otel_integration.py

@@ -241,8 +241,17 @@ def test_set_active_pods_absolute(self) -> None:
             metrics.set_active_pods(namespace="ns", count=10)
             metrics.set_active_pods(namespace="ns", count=4)  # should be 4, not 14
             text = _scrape(reg)
-            assert "4" in text or "4.0" in text
-            assert "14" not in text
+            # Assert on the actual gauge line, not a naked substring: the full
+            # scrape (HELP/target_info/other series) can incidentally contain
+            # "14", which made the old substring check flaky under xdist.
+            ns_lines = [
+                line
+                for line in text.split("\n")
+                if "orb_k8s_active_pods" in line and 'namespace="ns"' in line
+            ]
+            assert ns_lines, f"No orb_k8s_active_pods{{namespace=ns}} line in scrape:\n{text}"
+            value = float(ns_lines[0].rsplit(" ", 1)[1])
+            assert value == 4.0, f"Expected absolute value 4.0 (not cumulative 14), got {value}"
         finally:
             provider.shutdown()