rws_k8s/k8s/
testing.rs

1//! Tests against a real API server: a client for a test cluster (kind,
2//! k3d, envtest's `kube-apiserver`) and a namespace of its own per test,
3//! deleted afterwards — what controllers and webhooks need, since a fake
4//! API server doesn't run admission, garbage collection or `status`
5//! subresources.
6//!
7//! ```rust,no_run
8//! use rws_k8s::k8s::testing::TestCluster;
9//!
10//! #[tokio::test]
11//! async fn my_controller_creates_the_deployment() {
12//!     let Some(cluster) = TestCluster::from_env().await else { return };   // skipped without a cluster
13//!     let ns = cluster.namespace("backup").await.unwrap();
14//!     // … create objects in ns.name(), run the reconcile, assert …
15//!     ns.delete().await;
16//! }
17//! ```
18//!
19//! The cluster comes from `RWS_TEST_KUBECONFIG` (a kubeconfig file). Without
20//! it the test is skipped, unless `RWS_TEST_REQUIRE_KUBERNETES` is set — then
21//! it fails, so CI can't pass by skipping. Never point it at a cluster that
22//! matters: the tests create and delete namespaces.
23
24use std::time::Duration;
25
26use k8s_openapi::api::core::v1::Namespace;
27use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
28use kube::api::{Api, DeleteParams, PostParams};
29
30use super::Result;
31
32/// A test cluster's client.
33#[derive(Clone)]
34pub struct TestCluster {
35    pub client: kube::Client,
36}
37
38impl TestCluster {
39    /// The cluster in `RWS_TEST_KUBECONFIG`, or `None` (skip the test).
40    ///
41    /// # Panics
42    ///
43    /// With `RWS_TEST_REQUIRE_KUBERNETES` set and no usable cluster.
44    pub async fn from_env() -> Option<TestCluster> {
45        let required = std::env::var("RWS_TEST_REQUIRE_KUBERNETES").is_ok_and(|v| !v.is_empty());
46        let Some(path) = std::env::var("RWS_TEST_KUBECONFIG").ok().filter(|p| !p.is_empty()) else {
47            assert!(!required, "RWS_TEST_REQUIRE_KUBERNETES is set but RWS_TEST_KUBECONFIG isn't");
48            eprintln!("skipped: no RWS_TEST_KUBECONFIG");
49            return None;
50        };
51        match super::client::from_kubeconfig(&path).await {
52            Ok(client) => Some(TestCluster { client }),
53            Err(e) => {
54                assert!(!required, "RWS_TEST_KUBECONFIG={path}: {e}");
55                eprintln!("skipped: RWS_TEST_KUBECONFIG={path}: {e}");
56                None
57            }
58        }
59    }
60
61    /// A new namespace `<prefix>-<random>`, labelled `rws-test=true`.
62    pub async fn namespace(&self, prefix: &str) -> Result<TestNamespace> {
63        let suffix = format!("{:08x}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.subsec_nanos() ^ (d.as_secs() as u32).rotate_left(7)).unwrap_or(0) ^ std::process::id());
64        let name = format!("{}-{suffix}", prefix.trim_matches('-').to_lowercase());
65        let ns = Namespace {
66            metadata: ObjectMeta { name: Some(name.clone()), labels: Some([("rws-test".to_string(), "true".to_string())].into()), ..Default::default() },
67            ..Default::default()
68        };
69        Api::<Namespace>::all(self.client.clone()).create(&PostParams::default(), &ns).await?;
70        Ok(TestNamespace { client: self.client.clone(), name })
71    }
72
73    /// Apply a CRD (server-side) and wait until the API server serves it.
74    pub async fn install_crd<K: kube::CustomResourceExt>(&self) -> Result<()> {
75        use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
76        use kube::api::{Patch, PatchParams};
77        let crd = K::crd();
78        let name = crd.metadata.name.clone().unwrap_or_default();
79        let api: Api<CustomResourceDefinition> = Api::all(self.client.clone());
80        api.patch(&name, &PatchParams::apply("rws-test").force(), &Patch::Apply(&crd)).await?;
81        let established = kube::runtime::wait::conditions::is_crd_established();
82        tokio::time::timeout(Duration::from_secs(30), kube::runtime::wait::await_condition(api, &name, established))
83            .await
84            .map_err(|_| super::Error::Other(format!("CRD {name} not established within 30 s")))?
85            .map_err(|e| super::Error::Other(e.to_string()))?;
86        Ok(())
87    }
88}
89
90/// A namespace of one test. [`delete`](Self::delete) it at the end (a
91/// failed test leaves it for inspection; they're labelled `rws-test=true`).
92pub struct TestNamespace {
93    client: kube::Client,
94    name: String,
95}
96
97impl TestNamespace {
98    pub fn name(&self) -> &str {
99        &self.name
100    }
101
102    /// An `Api` for `K` in this namespace.
103    pub fn api<K>(&self) -> Api<K>
104    where
105        K: kube::Resource<Scope = k8s_openapi::NamespaceResourceScope, DynamicType = ()>,
106    {
107        Api::namespaced(self.client.clone(), &self.name)
108    }
109
110    /// Delete it (in the background on the API server's side).
111    pub async fn delete(self) {
112        let _ = Api::<Namespace>::all(self.client.clone()).delete(&self.name, &DeleteParams::background()).await;
113    }
114}
115
116/// Wait until `check` answers `Some` (polled every 100 ms), at most
117/// `timeout`; for asserting what a controller eventually does.
118pub async fn eventually<T, F, Fut>(timeout: Duration, mut check: F) -> Option<T>
119where
120    F: FnMut() -> Fut,
121    Fut: std::future::Future<Output = Option<T>>,
122{
123    let deadline = tokio::time::Instant::now() + timeout;
124    loop {
125        if let Some(v) = check().await {
126            return Some(v);
127        }
128        if tokio::time::Instant::now() >= deadline {
129            return None;
130        }
131        tokio::time::sleep(Duration::from_millis(100)).await;
132    }
133}
rws_k8s/k8s/testing.rs

rws_k8s/k8s/
testing.rs
