api, cluster-controller: tidy the Direct-fixture test sort (tests only)

crates/libs/api/src/cluster_admission/mod.rs

@@ -31,7 +31,7 @@ use crate::controllers::substrate::{
 
 mod redis_bucket;
 #[cfg(test)]
-pub(crate) mod tests;
+mod tests;
 
 use redis_bucket::{ConsumeOutcome, RedisTokenBucket};
 

crates/libs/api/src/cluster_admission/tests.rs

@@ -14,16 +14,13 @@ use uuid::Uuid;
 
 use super::redis_bucket::{ConsumeOutcome, RedisTokenBucket};
 use super::*;
-use crate::controllers::substrate::{Substrate, direct::DirectSubstrate};
-
-use pplx_space_test_env::TestResult;
-
+use crate::controllers::substrate::Substrate;
 // Direct is only a placeholder capacity source for most of this module's tests:
 // they need zero or some eligible nodes. `capacity_probe_failure_fails_open`
 // needs a count that fails, which only Direct's can.
-pub(crate) fn direct_substrate(pool: &PgPool) -> Substrate {
-    Substrate::Direct(Box::new(DirectSubstrate::new(pool.clone())))
-}
+use crate::test_support::{direct_substrate, enabled_config};
+
+use pplx_space_test_env::TestResult;
 
 fn admission(
     substrate: Substrate,
@@ -50,15 +47,6 @@ fn admission_with_flags(
     }
 }
 
-pub(crate) fn enabled_config() -> ClusterSandboxAdmissionConfig {
-    ClusterSandboxAdmissionConfig {
-        enabled: true,
-        refill_rps: 6.0,
-        burst: 60.0,
-        redis_url: None,
-    }
-}
-
 /// A bucket pointed at a real Redis, plus an optional container handle that
 /// must be kept alive for the bucket's lifetime.
 ///

crates/libs/api/src/controllers/sandbox/lifecycle/characterization_tests.rs

@@ -20,8 +20,7 @@ use pplx_space_test_env::TestResult;
 
 use super::*;
 use crate::controllers::sandbox::wake_completion::wake_completion_redrive;
-use crate::controllers::substrate::Substrate;
-use crate::controllers::substrate::direct::DirectSubstrate;
+use crate::test_support::direct_substrate;
 
 /// Values `sandboxes.lifecycle` can hold that are not `SandboxLifecycle`
 /// variants (see `SandboxLifecycle::all`).
@@ -125,10 +124,6 @@ async fn tracked(pool: &PgPool, sandbox_id: SandboxId) -> TestResult<Sandbox> {
         .expect("sandbox row exists"))
 }
 
-fn direct_substrate(pool: &PgPool) -> Substrate {
-    Substrate::Direct(Box::new(DirectSubstrate::new(pool.clone())))
-}
-
 fn direct_controller(pool: &PgPool) -> SandboxController {
     SandboxController::new(direct_substrate(pool))
 }

crates/libs/api/src/controllers/substrate/direct/cluster_admission_tests.rs

@@ -6,28 +6,11 @@
 
 use pplx_space_common_core::types::NodeId;
 use pplx_space_db_test_fixtures::try_pg_test_db;
+use pplx_space_feature_flags::FeatureFlagClient;
 use pplx_space_test_env::TestResult;
 
-use crate::cluster_admission::tests::{direct_substrate, enabled_config};
-use crate::cluster_admission::{
-    ClusterActivationClass, ClusterSandboxAdmission, ClusterSandboxAdmissionConfig,
-};
-use crate::controllers::substrate::Substrate;
-
-/// Stands in for `cluster_admission::tests::admission`, whose bucket parameter
-/// is a type private to `cluster_admission`. The test passes no bucket, and
-/// `with_config` builds none from a config without a Redis URL.
-fn admission(
-    substrate: Substrate,
-    config: ClusterSandboxAdmissionConfig,
-    _bucket: Option<std::convert::Infallible>,
-) -> ClusterSandboxAdmission {
-    ClusterSandboxAdmission::with_config(
-        substrate,
-        config,
-        pplx_space_feature_flags::FeatureFlagClient::disabled(),
-    )
-}
+use crate::cluster_admission::{ClusterActivationClass, ClusterSandboxAdmission};
+use crate::test_support::{direct_substrate, enabled_config};
 
 #[tokio::test]
 async fn stale_heartbeat_direct_worker_is_not_empty_cluster() -> TestResult {
@@ -65,7 +48,11 @@ async fn stale_heartbeat_direct_worker_is_not_empty_cluster() -> TestResult {
 
     // No bucket → the only way this returns Ok is if the guard counted the
     // stale worker (fail open); a false empty_cluster would return EmptyCluster.
-    let admission = admission(direct_substrate(&db.pool), enabled_config(), None);
+    let admission = ClusterSandboxAdmission::with_config(
+        direct_substrate(&db.pool),
+        enabled_config(),
+        FeatureFlagClient::disabled(),
+    );
     admission
         .admit_activation(ClusterActivationClass::Cold)
         .await?;

crates/libs/api/src/controllers/substrate/kubernetes/tests.rs

@@ -9547,10 +9547,11 @@ mod kubernetes_tests {
             mock_kubernetes_client(Arc::clone(&captured_pod_creates)),
         );
 
-        // Empty nodes inventory → the admission capacity probe reports zero
-        // eligible nodes → EmptyCluster rejection (no Redis needed).
         let admission = std::sync::Arc::new(
             crate::cluster_admission::ClusterSandboxAdmission::with_config(
+                // Direct is only a placeholder capacity source: its empty nodes inventory reports
+                // zero eligible nodes, as a synced, empty Kubernetes node cache would, so admission
+                // rejects with EmptyCluster (no Redis needed).
                 Substrate::Direct(Box::new(
                     crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()),
                 )),

crates/libs/api/src/public_sandbox_operations/tests.rs

@@ -58,8 +58,11 @@ fn claimed_create_operation_for(
 }
 
 // Direct is only a placeholder for the existing-row tests, which never reach the
-// substrate. The admitted re-execution test's `sandbox_k8s_executions` check can
-// only fail on Kubernetes.
+// substrate. The admitted re-execution test differs on Kubernetes in two places:
+// its `sandbox_k8s_executions` check can only fail there, and its pool claim makes
+// the executor ask `admitted_pool_cleanup_pending`, which Direct answers with the
+// trait default `false` while Kubernetes reads the reserved slot, so a slot whose
+// cleanup is pending yields workflow progress instead of `Succeeded`.
 fn direct_sandbox_controller(pool: &PgPool) -> SandboxController {
     SandboxController::new(crate::controllers::substrate::Substrate::Direct(Box::new(
         crate::controllers::substrate::direct::DirectSubstrate::new(pool.clone()),

crates/libs/api/src/test_support.rs

@@ -2,6 +2,10 @@ use pplx_space_common_svc::bearer_token::{
     SPACE_INTERNAL_CONTROL_TOKEN_ENV, SPACE_INTERNAL_TOKEN_ENV,
     SPACE_NODE_API_INTERNAL_TOKEN_PATH_ENV, SPACE_NODE_API_TOKEN_PATH_ENV,
 };
+use sqlx::PgPool;
+
+use crate::cluster_admission::ClusterSandboxAdmissionConfig;
+use crate::controllers::substrate::{Substrate, direct::DirectSubstrate};
 
 const SIM_WORKER_TOKEN: &str = "sim-worker-test-token";
 const SIM_WORKER_CONTROL_TOKEN: &str = "sim-worker-test-control-token";
@@ -52,3 +56,19 @@ fn enter_sim_worker_process_with_token_paths(
         (SPACE_NODE_API_INTERNAL_TOKEN_PATH_ENV, node_api_internal),
     ])
 }
+
+/// A Direct substrate over the test database; it needs no apiserver.
+pub(crate) fn direct_substrate(pool: &PgPool) -> Substrate {
+    Substrate::Direct(Box::new(DirectSubstrate::new(pool.clone())))
+}
+
+/// An enabled cluster-admission config with no Redis URL, so the admission
+/// it configures has no token bucket.
+pub(crate) fn enabled_config() -> ClusterSandboxAdmissionConfig {
+    ClusterSandboxAdmissionConfig {
+        enabled: true,
+        refill_rps: 6.0,
+        burst: 60.0,
+        redis_url: None,
+    }
+}

crates/libs/cluster/controller/server/src/transient_lifecycle_sweeper/lifecycle_writer_tests.rs

@@ -47,7 +47,8 @@ fn age() -> PositiveDuration {
 }
 
 // Direct is only a placeholder: it supplies `managed_by`, and only the converge
-// CAS would decide differently for a Kubernetes label (see the module doc).
+// CAS and the failed-active repair CAS would decide differently for a
+// Kubernetes label (see the module doc).
 fn direct_sink(pool: &PgPool) -> SubstrateSweepSink {
     SubstrateSweepSink::new(
         pool.clone(),