fix(planner/tests): scope test_advisory_mode stubs so they don't leak (#8418)

Signed-off-by: Dmitry Tokarev <dtokarev@nvidia.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.github/workflows/post-merge-ci.yml

@@ -36,7 +36,7 @@ jobs:
       run_cpu_only_tests: true
       cpu_only_test_markers: '(pre_merge or post_merge) and planner and gpu_0'
       cpu_only_test_timeout_minutes: 30
-      cpu_parallel_mode: '0'
+      cpu_parallel_mode: '2'
       run_single_gpu_tests: false
       run_multi_gpu_tests: false
       copy_to_acr: false

.github/workflows/pr.yaml

@@ -367,7 +367,7 @@ jobs:
       run_cpu_only_tests: true
       cpu_only_test_markers: 'pre_merge and planner and gpu_0'
       cpu_only_test_timeout_minutes: 30
-      cpu_parallel_mode: '0'
+      cpu_parallel_mode: '2'
       run_gpu_tests: false
     secrets: inherit
 

components/src/dynamo/planner/tests/unit/test_advisory_mode.py

@@ -3,17 +3,28 @@
 
 """Unit tests for advisory mode and decision summary logging."""
 
-import sys
-import types
-from unittest.mock import MagicMock
-
 import pytest
 
-# ---------------------------------------------------------------
-# Stub native Rust modules so planner config can be imported
-# without building dynamo._core
-# ---------------------------------------------------------------
-_stubs = {
+from dynamo.planner.config.defaults import SLAPlannerDefaults
+
+pytestmark = [
+    pytest.mark.gpu_0,
+    pytest.mark.pre_merge,
+    pytest.mark.unit,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _stub_heavy_deps():
+    # Stub optional Rust/IO-heavy dynamo modules when absent so planner.config
+    # imports used below can resolve. Scoped to this module and torn down after
+    # so sibling test modules see the real modules (regression: #8244 left
+    # these stubs in sys.modules and caused later tests to skip).
+    import sys
+    import types
+    from unittest.mock import MagicMock
+
+    stubs = {
         "dynamo._core": {
             "Client": MagicMock,
             "DistributedRuntime": MagicMock,
@@ -34,20 +45,17 @@ _stubs = {
             "ForwardPassMetrics": MagicMock,
             "ScheduledRequestMetrics": MagicMock,
         },
-}
-for _mod_name, _attrs in _stubs.items():
-    _m = types.ModuleType(_mod_name)
-    for _k, _v in _attrs.items():
-        setattr(_m, _k, _v)
-    sys.modules.setdefault(_mod_name, _m)
-
-from dynamo.planner.config.defaults import SLAPlannerDefaults  # noqa: E402
-
-pytestmark = [
-    pytest.mark.gpu_0,
-    pytest.mark.pre_merge,
-    pytest.mark.unit,
-]
+    }
+    mp = pytest.MonkeyPatch()
+    for name, attrs in stubs.items():
+        if name in sys.modules:
+            continue
+        mod = types.ModuleType(name)
+        for k, v in attrs.items():
+            setattr(mod, k, v)
+        mp.setitem(sys.modules, name, mod)
+    yield
+    mp.undo()
 
 
 class TestAdvisoryDefaults:

components/src/dynamo/planner/tests/unit/test_load_based_scaling.py

@@ -9,6 +9,7 @@ DecodeRegressionModel, AggRegressionModel) without any planner adapter.
 FPM-driven scaling integration tests live in test_state_machine.py.
 """
 
+import os
 from unittest.mock import Mock, patch
 
 import pytest
@@ -671,8 +672,17 @@ class TestRefreshWorkerInfoFromConnector:
 
     def _make_planner(self, require_prefill=False, require_decode=True):
         """Build a minimal NativePlannerBase with no_operation=True."""
-        with patch("dynamo.planner.monitoring.planner_metrics.Gauge") as mock_gauge:
-            mock_gauge.return_value = Mock()
+        # Bypass Prometheus registration (Gauge+Enum double-register across
+        # tests). KubernetesConnector.__init__ loads ~/.kube/config and reads
+        # DYN_PARENT_DGD_K8S_NAME; stub both so this runs in plain pytest envs.
+        with patch(
+            "dynamo.planner.core.base.PlannerPrometheusMetrics"
+        ) as mock_metrics, patch(
+            "dynamo.planner.connectors.kubernetes.KubernetesAPI"
+        ), patch.dict(
+            os.environ, {"DYN_PARENT_DGD_K8S_NAME": "test-graph"}
+        ):
+            mock_metrics.return_value = Mock()
             config = PlannerConfig.model_construct(
                 throughput_adjustment_interval=60,
                 prefill_engine_num_gpu=1,