test(k8s): update stale StatefulSet/Job selective-release tests to assert refuse behavior

These three tests predated the deliberate refuse-subset-release change and
were never exercised until the k8s CI leg was enabled on this branch.

StatefulSet handler (tests 1 & 2): the handler now raises K8sError when
the caller-supplied victims are not the top-of-ordinal-stack set, rather
than logging a warning and proceeding with the scale patch. Silently
evicting the wrong pods would cause data loss. Updated:
- test_release_hosts_selective_non_highest_ordinal_warns_and_still_scales
  → test_release_hosts_selective_non_highest_ordinal_refused
- test_release_hosts_selective_unparseable_victim_names_warns
  → test_release_hosts_selective_unparseable_victim_names_refused
Both now assert pytest.raises(K8sError) and confirm no scale patch is issued.

Job handler (test 3): the "selective release not supported" info message the
test expected no longer exists. When parallelism is not recorded in
provider_data the handler logs "Kubernetes job release: ... (deleting whole
Job)" and deletes the Job. Updated the assertion to match the actual log
format ("job release") and clarified the docstring to describe the
no-parallelism fallback path being exercised.

No src/ changes.

Author: fgogolli

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

@@ -8,9 +8,9 @@
 * ``release_hosts`` selective path: when the victims are the top-of-stack
   ordinals, no warning is emitted and ``spec.replicas`` is patched down
   by the victim count.  Pods are never deleted directly.
-* ``release_hosts`` ordinal caveat: when victims include non-highest
-  ordinals, a WARNING is logged and the replicas patch still proceeds
-  (the controller will pick highest ordinals to evict).
+* ``release_hosts`` ordinal caveat: when victims are NOT the top-of-stack
+  ordinals, a ``K8sError`` is raised and no scale patch is issued.
+  Silently evicting different pods would cause data loss.
 * ``release_hosts`` full-release path: ``spec.replicas`` is patched to
   zero and the StatefulSet is deleted.
 * ``check_hosts_status`` reads both the pod list and the StatefulSet
@@ -31,6 +31,7 @@
 from orb.domain.request.value_objects import RequestId, RequestType
 from orb.domain.template.template_aggregate import Template
 from orb.providers.k8s.configuration.config import K8sProviderConfig
+from orb.providers.k8s.exceptions.k8s_errors import K8sError
 from orb.providers.k8s.infrastructure.handlers.statefulset_handler import K8sStatefulSetHandler
 
 # ---------------------------------------------------------------------------
@@ -246,11 +247,15 @@ async def test_release_hosts_selective_highest_ordinal_no_warning() -> None:
 
 
 @pytest.mark.asyncio
-async def test_release_hosts_selective_non_highest_ordinal_warns_and_still_scales() -> None:
-    """When the victims include non-highest ordinals, a WARNING is logged
-    and the controller still scales down by the victim count (it will
-    pick the highest-ordinal pods regardless of what the caller asked
-    for)."""
+async def test_release_hosts_selective_non_highest_ordinal_refused() -> None:
+    """When the victims are not the top-of-stack ordinals, the handler
+    raises K8sError and issues no scale patch.
+
+    Commit b4a77b1a deliberately changed this behaviour: silently evicting
+    different pods (the controller always picks the highest-ordinal set)
+    would cause data loss, so the handler refuses the request and tells
+    the caller which victims to supply instead.
+    """
     core_v1 = MagicMock()
     apps_v1 = MagicMock()
     apps_v1.read_namespaced_stateful_set.return_value = _make_statefulset_status(spec_replicas=5)
@@ -266,35 +271,30 @@ async def test_release_hosts_selective_non_highest_ordinal_warns_and_still_scale
     )
 
     # current=5, victims=[ordinal-1, ordinal-2] — non-highest ordinals.
-    # The controller will evict ordinals 3 and 4 instead.
-    await handler.release_hosts(["orb-deadbeef-1", "orb-deadbeef-2"], request.provider_data)
+    # The controller would evict ordinals 3 and 4 instead, so the handler
+    # refuses rather than silently releasing the wrong pods.
+    with pytest.raises(K8sError) as exc_info:
+        await handler.release_hosts(["orb-deadbeef-1", "orb-deadbeef-2"], request.provider_data)
 
-    # WARNING logged.
-    warning_messages = [
-        call.args[0]
-        for call in handler._logger.warning.call_args_list  # type: ignore[attr-defined]
-    ]
-    assert any("non-highest-ordinal" in msg for msg in warning_messages), warning_messages
+    # The error message must mention the correct top-of-stack victims.
+    assert "orb-deadbeef-3" in str(exc_info.value) or "orb-deadbeef-4" in str(exc_info.value)
 
-    # Replicas still patched down by the victim count (5 -> 3) — the
-    # caller asked to release 2 pods and 2 pods will be released; the
-    # only thing that changes is *which* ordinals.
-    apps_v1.patch_namespaced_stateful_set_scale.assert_called_once()
-    assert (
-        apps_v1.patch_namespaced_stateful_set_scale.call_args.kwargs["body"]["spec"]["replicas"]
-        == 3
-    )
+    # Refusal happens before any scale patch — the StatefulSet must be
+    # left untouched.
+    apps_v1.patch_namespaced_stateful_set_scale.assert_not_called()
 
     # Pods are NEVER deleted directly.
     core_v1.delete_namespaced_pod.assert_not_called()
     apps_v1.delete_namespaced_stateful_set.assert_not_called()
 
 
 @pytest.mark.asyncio
-async def test_release_hosts_selective_unparseable_victim_names_warns() -> None:
-    """Victim names that do not parse as ``<statefulset>-<ordinal>``
-    should also trigger the WARNING (since they cannot be the top-of-
-    stack ordinals)."""
+async def test_release_hosts_selective_unparseable_victim_names_refused() -> None:
+    """Victim names that do not parse as ``<statefulset>-<ordinal>`` are
+    refused with K8sError — an unparseable name cannot be the top-of-
+    stack ordinal, so honouring the request would silently evict the
+    wrong pods.
+    """
     core_v1 = MagicMock()
     apps_v1 = MagicMock()
     apps_v1.read_namespaced_stateful_set.return_value = _make_statefulset_status(spec_replicas=3)
@@ -310,19 +310,16 @@ async def test_release_hosts_selective_unparseable_victim_names_warns() -> None:
     )
 
     # Victim name does not match the StatefulSet's name prefix — its
-    # ordinal cannot be parsed.
-    await handler.release_hosts(["some-other-pod"], request.provider_data)
+    # ordinal cannot be parsed, so the handler refuses rather than
+    # silently releasing the wrong pod.
+    with pytest.raises(K8sError):
+        await handler.release_hosts(["some-other-pod"], request.provider_data)
 
-    warning_messages = [
-        call.args[0]
-        for call in handler._logger.warning.call_args_list  # type: ignore[attr-defined]
-    ]
-    assert any("non-highest-ordinal" in msg for msg in warning_messages)
-    # Still scales down by 1 victim (3 -> 2).
-    assert (
-        apps_v1.patch_namespaced_stateful_set_scale.call_args.kwargs["body"]["spec"]["replicas"]
-        == 2
-    )
+    # Refusal happens before any scale patch.
+    apps_v1.patch_namespaced_stateful_set_scale.assert_not_called()
+    # Pods are NEVER deleted directly.
+    core_v1.delete_namespaced_pod.assert_not_called()
+    apps_v1.delete_namespaced_stateful_set.assert_not_called()
 
 
 # ---------------------------------------------------------------------------

tests/providers/k8s/unit/test_coverage_gaps.py

@@ -488,12 +488,16 @@ def test_check_health_returns_unhealthy_when_kubernetes_package_absent(
 
 @pytest.mark.asyncio
 async def test_job_release_logs_selective_release_not_supported() -> None:
-    """release_hosts with a partial machine_ids list must log that selective release
-    is not supported.
+    """release_hosts with a partial machine_ids list must raise K8sError when
+    parallelism is known, refusing the subset release.
 
-    The Job handler deletes the whole Job regardless of ``machine_ids``.
-    The operator must be informed via an info-level log message that
-    includes "selective release not supported".
+    When ``parallelism`` is NOT recorded in ``provider_data`` (legacy or
+    pre-acquire state), the handler falls back to deleting the whole Job
+    and emits an info-level log recording the requested machine_ids.
+
+    This test exercises the fallback path (no ``parallelism`` in
+    ``provider_data``) and confirms the Job is still deleted and an
+    info-level log is emitted that identifies the release operation.
     """
     from orb.providers.k8s.infrastructure.handlers.job_handler import K8sJobHandler
 
@@ -513,25 +517,28 @@ async def test_job_release_logs_selective_release_not_supported() -> None:
         requested_count=3,
         provider_api="Job",
     )
-    # Override provider_data with a stable job_name so we don't need to
-    # call acquire first.
+    # Override provider_data with a stable job_name but NO parallelism key
+    # so we don't need to call acquire first and the K8sError refuse path
+    # is not triggered (parallelism defaults to 0).
     object.__setattr__(
         request,
         "provider_data",
         {"namespace": "orb-test", "job_name": "orb-testreq1"},
     )
 
     # Selective release: caller only passes one of the three pods.
+    # With no parallelism recorded, the handler deletes the whole Job.
     await handler.release_hosts(["orb-testreq1-pod0"], request.provider_data)
 
     # The whole Job must have been deleted.
     batch_v1.delete_namespaced_job.assert_called_once()
 
-    # An info-level message must mention selective release.
+    # An info-level message must record the job release operation with the
+    # requested machine_ids (the log reads "deleting whole Job").
     info_calls = logger.info.call_args_list
-    selective_log = any("selective release" in str(call).lower() for call in info_calls)
-    assert selective_log, (
-        "Expected an info-level log message about selective release not being supported; "
+    release_log = any("job release" in str(call).lower() for call in info_calls)
+    assert release_log, (
+        "Expected an info-level log message about the job release operation; "
         f"info calls: {info_calls}"
     )