use std::sync::Arc;
use axum::{
extract::{Path, Query, State},
http::StatusCode,
response::{IntoResponse, Response},
routing::{get, patch},
Json, Router,
};
use breathe_facade::{BreatheStore, DimensionId, StoreError};
use serde::Deserialize;
use serde_json::{json, Value};
pub type SharedStore = Arc<dyn BreatheStore>;
#[must_use]
pub fn router(store: SharedStore) -> Router {
let schema = graphql::schema(store.clone());
Router::new()
.route("/healthz", get(|| async { "ok" }))
.route("/api/v1/catalog", get(catalog))
.route("/api/v1/bands/:kind", get(list_bands))
.route("/api/v1/bands/:kind/:namespace/:name", get(get_band).patch(patch_band))
.route("/api/v1/bands/:kind/:namespace/:name/dry-run", patch(set_dry_run))
.route("/api/v1/bands/:kind/:namespace/:name/write-intent", patch(set_write_intent))
.route("/api/v1/bands/:kind/:namespace/:name/confirm", patch(confirm_band))
.route("/api/v1/nodepools", get(list_pools))
.route("/api/v1/nodepools/:name", get(get_pool))
.route("/api/v1/nodepools/:name/write-enabled", patch(set_write_enabled))
.route_service("/graphql", async_graphql_axum::GraphQL::new(schema))
.with_state(store)
}
fn respond(r: Result<Value, StoreError>) -> Response {
match r {
Ok(v) => (StatusCode::OK, Json(v)).into_response(),
Err(StoreError::BadRequest(m)) => (StatusCode::BAD_REQUEST, Json(json!({ "error": m }))).into_response(),
Err(e) => (StatusCode::BAD_GATEWAY, Json(json!({ "error": e.to_string() }))).into_response(),
}
}
fn kind_or_400(s: &str) -> Result<DimensionId, Response> {
DimensionId::parse(s).ok_or_else(|| {
let known: Vec<&str> = DimensionId::ALL.iter().map(|d| d.as_str()).collect();
(StatusCode::BAD_REQUEST, Json(json!({ "error": format!("unknown band kind '{s}'"), "known": known })))
.into_response()
})
}
#[derive(Deserialize)]
struct NsQuery {
namespace: Option<String>,
}
async fn catalog(State(store): State<SharedStore>) -> Response {
(StatusCode::OK, Json(store.catalog())).into_response()
}
async fn list_bands(State(store): State<SharedStore>, Path(kind): Path<String>, Query(q): Query<NsQuery>) -> Response {
match kind_or_400(&kind) {
Ok(k) => respond(store.list_bands(k, q.namespace).await),
Err(r) => r,
}
}
async fn get_band(State(store): State<SharedStore>, Path((kind, ns, name)): Path<(String, String, String)>) -> Response {
match kind_or_400(&kind) {
Ok(k) => respond(store.get_band(k, ns, name).await),
Err(r) => r,
}
}
async fn patch_band(
State(store): State<SharedStore>,
Path((kind, ns, name)): Path<(String, String, String)>,
Json(spec): Json<Value>,
) -> Response {
match kind_or_400(&kind) {
Ok(k) => respond(store.patch_band_spec(k, ns, name, spec).await),
Err(r) => r,
}
}
#[derive(Deserialize)]
struct DryRunBody {
#[serde(rename = "dryRun")]
dry_run: bool,
}
async fn set_dry_run(
State(store): State<SharedStore>,
Path((kind, ns, name)): Path<(String, String, String)>,
Json(b): Json<DryRunBody>,
) -> Response {
match kind_or_400(&kind) {
Ok(k) if !k.dry_run_is_honored() => (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "spec.dryRun has no effect on this band kind",
"kind": k.as_str(),
"retiredSince": "breathe@76924b0 (2026-06-19)",
"useInstead": "PATCH /api/v1/bands/{kind}/{namespace}/{name}/write-intent",
"honoredBy": ["host-param", "kube-param"],
"wroteNothing": true,
})),
)
.into_response(),
Ok(k) => respond(store.patch_band_spec(k, ns, name, json!({ "dryRun": b.dry_run })).await),
Err(r) => r,
}
}
#[derive(Deserialize)]
struct WriteIntentBody {
intent: String,
#[serde(rename = "confirmAfterSeconds", default)]
confirm_after_seconds: Option<u64>,
#[serde(rename = "authorizedBy", default)]
authorized_by: Option<String>,
}
async fn set_write_intent(
State(store): State<SharedStore>,
Path((kind, ns, name)): Path<(String, String, String)>,
Json(b): Json<WriteIntentBody>,
) -> Response {
let k = match kind_or_400(&kind) {
Ok(k) => k,
Err(r) => return r,
};
if b.intent == "write" && b.authorized_by.as_deref().map(str::trim).unwrap_or_default().is_empty() {
return (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "intent=write requires authorizedBy",
"why": "a live carve must name who authorized it; the witness is carried into \
status.effectiveGate so 'why is this band writing?' is answerable from the CR alone",
"wroteNothing": true,
})),
)
.into_response();
}
let mut intent = json!({ "intent": b.intent });
if let Some(secs) = b.confirm_after_seconds {
intent["confirmAfterSeconds"] = json!(secs);
}
if let Some(by) = b.authorized_by {
intent["authorizedBy"] = json!(by);
}
respond(store.patch_band_spec(k, ns, name, json!({ "writeIntent": intent })).await)
}
#[derive(Deserialize)]
struct ConfirmBody {
confirmed: bool,
}
async fn confirm_band(
State(store): State<SharedStore>,
Path((kind, ns, name)): Path<(String, String, String)>,
Json(b): Json<ConfirmBody>,
) -> Response {
match kind_or_400(&kind) {
Ok(k) => {
let v = if b.confirmed { json!("true") } else { Value::Null };
respond(store.annotate_band(k, ns, name, json!({ breathe_provider::CONFIRMED_ANNOTATION: v })).await)
}
Err(r) => r,
}
}
async fn list_pools(State(store): State<SharedStore>) -> Response {
respond(store.list_pools().await)
}
async fn get_pool(State(store): State<SharedStore>, Path(name): Path<String>) -> Response {
respond(store.get_pool(name).await)
}
#[derive(Deserialize)]
struct WriteEnabledBody {
#[serde(rename = "writeEnabled")]
write_enabled: bool,
}
async fn set_write_enabled(
State(store): State<SharedStore>,
Path(name): Path<String>,
Json(b): Json<WriteEnabledBody>,
) -> Response {
respond(store.patch_pool_spec(name, json!({ "writeEnabled": b.write_enabled })).await)
}
pub mod graphql {
use super::{DimensionId, SharedStore, StoreError};
use async_graphql::{Context, EmptySubscription, Json, Object, Schema};
use serde_json::Value;
fn gql(e: StoreError) -> async_graphql::Error {
async_graphql::Error::new(e.to_string())
}
fn parse_kind(s: &str) -> async_graphql::Result<DimensionId> {
DimensionId::parse(s).ok_or_else(|| {
let known: Vec<&str> = DimensionId::ALL.iter().map(|d| d.as_str()).collect();
async_graphql::Error::new(format!("unknown band kind '{s}' (known: {})", known.join(", ")))
})
}
fn store<'a>(ctx: &Context<'a>) -> async_graphql::Result<&'a SharedStore> {
ctx.data::<SharedStore>()
}
pub struct Query;
#[Object]
impl Query {
async fn catalog(&self, ctx: &Context<'_>) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.catalog()))
}
async fn bands(&self, ctx: &Context<'_>, kind: String, namespace: Option<String>) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.list_bands(parse_kind(&kind)?, namespace).await.map_err(gql)?))
}
async fn band(&self, ctx: &Context<'_>, kind: String, namespace: String, name: String) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.get_band(parse_kind(&kind)?, namespace, name).await.map_err(gql)?))
}
async fn nodepools(&self, ctx: &Context<'_>) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.list_pools().await.map_err(gql)?))
}
async fn nodepool(&self, ctx: &Context<'_>, name: String) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.get_pool(name).await.map_err(gql)?))
}
}
pub struct Mutation;
#[Object]
impl Mutation {
async fn patch_band(&self, ctx: &Context<'_>, kind: String, namespace: String, name: String, spec: Json<Value>) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.patch_band_spec(parse_kind(&kind)?, namespace, name, spec.0).await.map_err(gql)?))
}
async fn set_write_intent(
&self,
ctx: &Context<'_>,
kind: String,
namespace: String,
name: String,
intent: String,
confirm_after_seconds: Option<u64>,
authorized_by: Option<String>,
) -> async_graphql::Result<Json<Value>> {
if intent == "write" && authorized_by.as_deref().map(str::trim).unwrap_or_default().is_empty() {
return Err(async_graphql::Error::new(
"intent=write requires authorizedBy — a live carve must name who authorized it; nothing was written",
));
}
let mut body = serde_json::json!({ "intent": intent });
if let Some(secs) = confirm_after_seconds {
body["confirmAfterSeconds"] = serde_json::json!(secs);
}
if let Some(by) = authorized_by {
body["authorizedBy"] = serde_json::json!(by);
}
let spec = serde_json::json!({ "writeIntent": body });
Ok(Json(store(ctx)?.patch_band_spec(parse_kind(&kind)?, namespace, name, spec).await.map_err(gql)?))
}
async fn confirm_band(&self, ctx: &Context<'_>, kind: String, namespace: String, name: String, confirmed: bool) -> async_graphql::Result<Json<Value>> {
let v = if confirmed { serde_json::json!("true") } else { Value::Null };
let ann = serde_json::json!({ breathe_provider::CONFIRMED_ANNOTATION: v });
Ok(Json(store(ctx)?.annotate_band(parse_kind(&kind)?, namespace, name, ann).await.map_err(gql)?))
}
async fn set_dry_run(&self, ctx: &Context<'_>, kind: String, namespace: String, name: String, dry_run: bool) -> async_graphql::Result<Json<Value>> {
let k = parse_kind(&kind)?;
if !k.dry_run_is_honored() {
return Err(async_graphql::Error::new(format!(
"spec.dryRun has no effect on {kind} bands (retired breathe@76924b0, 2026-06-19); \
use setWriteIntent. Only host-param and kube-param read it. Nothing was written."
)));
}
Ok(Json(store(ctx)?.patch_band_spec(k, namespace, name, serde_json::json!({ "dryRun": dry_run })).await.map_err(gql)?))
}
async fn set_write_enabled(&self, ctx: &Context<'_>, name: String, write_enabled: bool) -> async_graphql::Result<Json<Value>> {
Ok(Json(store(ctx)?.patch_pool_spec(name, serde_json::json!({ "writeEnabled": write_enabled })).await.map_err(gql)?))
}
}
pub type BreatheSchema = Schema<Query, Mutation, EmptySubscription>;
#[must_use]
pub fn schema(store: SharedStore) -> BreatheSchema {
Schema::build(Query, Mutation, EmptySubscription).data(store).finish()
}
}
pub mod grpc {
use super::{DimensionId, SharedStore, StoreError};
use serde_json::Value;
use tonic::{Request, Response, Status};
pub mod pb {
tonic::include_proto!("breathe.v1");
include!(concat!(env!("OUT_DIR"), "/breathe.v1.serde.rs"));
}
fn st(e: StoreError) -> Status {
match e {
StoreError::BadRequest(m) => Status::invalid_argument(m),
other => Status::internal(other.to_string()),
}
}
fn typed<T: serde::de::DeserializeOwned>(v: Value) -> Result<T, Status> {
serde_json::from_value(v)
.map_err(|e| Status::internal(format!("response did not match the typed schema (spec drift?): {e}")))
}
fn opt_ns(s: String) -> Option<String> {
if s.is_empty() { None } else { Some(s) }
}
pub fn kind_of(k: i32) -> Result<DimensionId, Status> {
use pb::BandKind as P;
let p = P::try_from(k).map_err(|_| Status::invalid_argument("unknown band kind"))?;
match p {
P::Memory => Ok(DimensionId::Memory),
P::Cpu => Ok(DimensionId::Cpu),
P::Storage => Ok(DimensionId::Storage),
P::Replica => Ok(DimensionId::Replica),
P::Arc => Ok(DimensionId::Arc),
P::Cgroup => Ok(DimensionId::Cgroup),
P::CgroupCpu => Ok(DimensionId::CgroupCpu),
P::HostParam => Ok(DimensionId::HostParam),
P::KubeParam => Ok(DimensionId::KubeParam),
P::AppParam => Ok(DimensionId::AppParam),
P::Request => Ok(DimensionId::Request),
P::Unspecified => Err(Status::invalid_argument("band kind unspecified")),
}
}
pub struct GrpcService {
pub store: SharedStore,
}
#[tonic::async_trait]
impl pb::breathe_server::Breathe for GrpcService {
async fn band_list(&self, req: Request<pb::BandListRequest>) -> Result<Response<pb::BandListResponse>, Status> {
let r = req.into_inner();
let v = self.store.list_bands(kind_of(r.kind)?, opt_ns(r.namespace)).await.map_err(st)?;
Ok(Response::new(pb::BandListResponse { items: typed(v)? }))
}
async fn band_get(&self, req: Request<pb::BandGetRequest>) -> Result<Response<pb::Band>, Status> {
let r = req.into_inner();
let v = self.store.get_band(kind_of(r.kind)?, r.namespace, r.name).await.map_err(st)?;
Ok(Response::new(typed(v)?))
}
async fn band_patch(&self, req: Request<pb::BandPatchRequest>) -> Result<Response<pb::Band>, Status> {
let r = req.into_inner();
let spec = serde_json::to_value(r.body.unwrap_or_default()).map_err(|e| Status::internal(e.to_string()))?;
if spec["writeIntent"]["intent"] == "write"
&& spec["writeIntent"]["authorizedBy"].as_str().map(str::trim).unwrap_or_default().is_empty()
{
return Err(Status::invalid_argument(
"writeIntent.intent=write requires authorizedBy — a live carve must name who \
authorized it. Nothing was written.",
));
}
let v = self.store.patch_band_spec(kind_of(r.kind)?, r.namespace, r.name, spec).await.map_err(st)?;
Ok(Response::new(typed(v)?))
}
async fn band_set_dry_run(&self, req: Request<pb::BandSetDryRunRequest>) -> Result<Response<pb::Band>, Status> {
let r = req.into_inner();
let k = kind_of(r.kind)?;
if !k.dry_run_is_honored() {
return Err(Status::failed_precondition(
"spec.dryRun has no effect on this band kind (retired breathe@76924b0, 2026-06-19); \
set BandSpec.write_intent via BandPatch instead. Only host-param and kube-param \
read dryRun. Nothing was written.",
));
}
let v = self.store.patch_band_spec(k, r.namespace, r.name, serde_json::json!({ "dryRun": r.dry_run })).await.map_err(st)?;
Ok(Response::new(typed(v)?))
}
async fn catalog_list(&self, _req: Request<pb::CatalogListRequest>) -> Result<Response<pb::Catalog>, Status> {
Ok(Response::new(typed(self.store.catalog())?))
}
async fn nodepool_list(&self, _req: Request<pb::NodepoolListRequest>) -> Result<Response<pb::NodepoolListResponse>, Status> {
let v = self.store.list_pools().await.map_err(st)?;
Ok(Response::new(pb::NodepoolListResponse { items: typed(v)? }))
}
async fn nodepool_get(&self, req: Request<pb::NodepoolGetRequest>) -> Result<Response<pb::NodePool>, Status> {
let v = self.store.get_pool(req.into_inner().name).await.map_err(st)?;
Ok(Response::new(typed(v)?))
}
async fn nodepool_set_write_enabled(&self, req: Request<pb::NodepoolSetWriteEnabledRequest>) -> Result<Response<pb::NodePool>, Status> {
let r = req.into_inner();
let v = self.store.patch_pool_spec(r.name, serde_json::json!({ "writeEnabled": r.write_enabled })).await.map_err(st)?;
Ok(Response::new(typed(v)?))
}
async fn healthz(&self, _req: Request<pb::HealthzRequest>) -> Result<Response<::pbjson_types::Empty>, Status> {
Ok(Response::new(::pbjson_types::Empty {}))
}
}
#[must_use]
pub fn server(store: SharedStore) -> pb::breathe_server::BreatheServer<GrpcService> {
pb::breathe_server::BreatheServer::new(GrpcService { store })
}
}
#[cfg(test)]
mod tests {
use super::*;
use async_trait::async_trait;
use axum::body::Body;
use axum::http::Request;
use http_body_util::BodyExt;
use std::sync::Mutex;
use tower::ServiceExt;
#[derive(Default)]
struct MockStore {
patches: Mutex<Vec<(String, Value)>>,
}
#[async_trait]
impl BreatheStore for MockStore {
async fn list_bands(&self, kind: DimensionId, _ns: Option<String>) -> Result<Value, StoreError> {
Ok(json!([{ "kind": kind.as_str() }]))
}
async fn get_band(&self, _kind: DimensionId, ns: String, name: String) -> Result<Value, StoreError> {
Ok(json!({
"apiVersion": "breathe.pleme.io/v1",
"kind": "ArcBand",
"metadata": { "name": name, "namespace": ns, "resourceVersion": "42" },
"spec": { "setpoint": 0.8, "dryRun": false },
"status": { "phase": "Holding" }
}))
}
async fn patch_band_spec(&self, _k: DimensionId, _ns: String, name: String, spec: Value) -> Result<Value, StoreError> {
self.patches.lock().unwrap().push((name, spec.clone()));
Ok(json!({ "spec": spec }))
}
async fn annotate_band(&self, _k: DimensionId, _ns: String, name: String, ann: Value) -> Result<Value, StoreError> {
self.patches.lock().unwrap().push((name, ann.clone()));
Ok(json!({ "metadata": { "annotations": ann } }))
}
async fn list_pools(&self) -> Result<Value, StoreError> {
Ok(json!([{ "metadata": { "name": "rio" }, "spec": { "nodeName": "rio", "arcMaxGiB": 6 } }]))
}
async fn get_pool(&self, name: String) -> Result<Value, StoreError> {
Ok(json!({
"apiVersion": "breathe.pleme.io/v1",
"kind": "BreatheNodePool",
"metadata": { "name": name },
"spec": { "nodeName": name, "arcMaxGiB": 6, "writeEnabled": true },
"status": { "phase": "Active" }
}))
}
async fn patch_pool_spec(&self, name: String, spec: Value) -> Result<Value, StoreError> {
self.patches.lock().unwrap().push((name, spec.clone()));
Ok(json!({ "spec": spec }))
}
async fn list_postures(&self) -> Result<Value, StoreError> {
Ok(json!([{ "metadata": { "name": "platform-default" }, "spec": { "setpoint": 0.8 } }]))
}
async fn get_posture(&self, name: String) -> Result<Value, StoreError> {
Ok(json!({
"apiVersion": "breathe.pleme.io/v1",
"kind": "BreathePosture",
"metadata": { "name": name },
"spec": { "setpoint": 0.8, "growAbove": 0.85, "growFactor": 1.25, "shrinkBelow": 0.7, "shrinkFactor": 0.9, "cooldownSeconds": 600, "maxStalenessSeconds": 120, "disruptionPolicy": "restartFreeOnly" }
}))
}
async fn patch_posture_spec(&self, name: String, spec: Value) -> Result<Value, StoreError> {
self.patches.lock().unwrap().push((name, spec.clone()));
Ok(json!({ "spec": spec }))
}
fn catalog(&self) -> Value {
breathe_facade::catalog_json()
}
}
async fn body_json(resp: Response) -> Value {
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
serde_json::from_slice(&bytes).unwrap()
}
#[tokio::test]
async fn set_dry_run_route_refuses_where_the_field_is_inert() {
let mock = Arc::new(MockStore::default());
let app = router(mock.clone());
let resp = app
.oneshot(
Request::builder()
.method("PATCH")
.uri("/api/v1/bands/arc/pangea-system/rio-arc/dry-run")
.header("content-type", "application/json")
.body(Body::from(r#"{"dryRun":false}"#))
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
assert!(mock.patches.lock().unwrap().is_empty(), "a refused call must not reach the store");
}
#[tokio::test]
async fn set_dry_run_route_applies_on_a_param_kind() {
let mock = Arc::new(MockStore::default());
let app = router(mock.clone());
let resp = app
.oneshot(
Request::builder()
.method("PATCH")
.uri("/api/v1/bands/host-param/camelot/vm-dirty/dry-run")
.header("content-type", "application/json")
.body(Body::from(r#"{"dryRun":true}"#))
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ "dryRun": true }));
}
#[tokio::test]
async fn every_shipped_dimension_is_routable() {
for kind in DimensionId::ALL {
let app = router(Arc::new(MockStore::default()));
let uri = ["/api/v1/bands/", kind.as_str()].concat();
let resp = app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap()).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK, "{kind} must be routable");
}
}
#[tokio::test]
async fn write_intent_route_writes_the_intent_and_refuses_an_unattributed_go_live() {
let mock = Arc::new(MockStore::default());
let app = router(mock.clone());
let patch_intent = |app: Router, body: &'static str| async move {
app.oneshot(
Request::builder()
.method("PATCH")
.uri("/api/v1/bands/cpu/camelot/coredns/write-intent")
.header("content-type", "application/json")
.body(Body::from(body))
.unwrap(),
)
.await
.unwrap()
};
let resp = patch_intent(app.clone(), r#"{"intent":"write"}"#).await;
assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "an unattributed go-live is refused");
assert!(mock.patches.lock().unwrap().is_empty());
let resp = patch_intent(app, r#"{"intent":"write","authorizedBy":"drzzln 2026-07-26"}"#).await;
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
mock.patches.lock().unwrap()[0].1,
json!({ "writeIntent": { "intent": "write", "authorizedBy": "drzzln 2026-07-26" } })
);
}
#[tokio::test]
async fn confirm_route_sets_the_operator_annotation() {
let mock = Arc::new(MockStore::default());
let app = router(mock.clone());
let resp = app
.oneshot(
Request::builder()
.method("PATCH")
.uri("/api/v1/bands/memory/camelot/b/confirm")
.header("content-type", "application/json")
.body(Body::from(r#"{"confirmed":true}"#))
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ breathe_provider::CONFIRMED_ANNOTATION: "true" }));
}
#[tokio::test]
async fn unknown_band_kind_is_400() {
let app = router(Arc::new(MockStore::default()));
let resp = app
.oneshot(Request::builder().uri("/api/v1/bands/bogus").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn catalog_route_returns_all_dimensions() {
let app = router(Arc::new(MockStore::default()));
let resp = app
.oneshot(Request::builder().uri("/api/v1/catalog").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let v = body_json(resp).await;
assert_eq!(v["dimensions"].as_array().unwrap().len(), breathe_catalog::ALL_DIMENSIONS.len());
}
#[tokio::test]
async fn write_enabled_route_patches_the_pool() {
let mock = Arc::new(MockStore::default());
let app = router(mock.clone());
let resp = app
.oneshot(
Request::builder()
.method("PATCH")
.uri("/api/v1/nodepools/rio/write-enabled")
.header("content-type", "application/json")
.body(Body::from(r#"{"writeEnabled":true}"#))
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ "writeEnabled": true }));
}
#[tokio::test]
async fn graphql_set_dry_run_errors_where_the_field_is_inert() {
let mock = Arc::new(MockStore::default());
let schema = graphql::schema(mock.clone());
let resp = schema
.execute(r#"mutation { setDryRun(kind:"arc", namespace:"pangea-system", name:"rio-arc", dryRun:false) }"#)
.await;
assert!(!resp.errors.is_empty(), "an inert dryRun write must not report success");
assert!(mock.patches.lock().unwrap().is_empty());
}
#[tokio::test]
async fn graphql_set_write_intent_mutation_patches_the_band() {
let mock = Arc::new(MockStore::default());
let schema = graphql::schema(mock.clone());
let resp = schema
.execute(r#"mutation { setWriteIntent(kind:"cpu", namespace:"camelot", name:"coredns", intent:"observe") }"#)
.await;
assert!(resp.errors.is_empty(), "{:?}", resp.errors);
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ "writeIntent": { "intent": "observe" } }));
}
#[tokio::test]
async fn graphql_reaches_the_previously_invisible_kinds() {
let schema = graphql::schema(Arc::new(MockStore::default()));
for kind in ["cgroup-cpu", "host-param", "kube-param", "app-param", "replica"] {
let q = ["{ bands(kind:\"", kind, "\") }"].concat();
let resp = schema.execute(q).await;
assert!(resp.errors.is_empty(), "{kind}: {:?}", resp.errors);
}
}
#[tokio::test]
async fn graphql_catalog_query_returns_dimensions() {
let schema = graphql::schema(Arc::new(MockStore::default()));
let resp = schema.execute("{ catalog }").await;
assert!(resp.errors.is_empty(), "{:?}", resp.errors);
assert!(resp.data.to_string().contains("dimensions"));
}
#[tokio::test]
async fn grpc_set_write_enabled_returns_typed_nodepool() {
use grpc::pb::breathe_server::Breathe;
let mock = Arc::new(MockStore::default());
let svc = grpc::GrpcService { store: mock.clone() };
let resp = svc
.nodepool_set_write_enabled(tonic::Request::new(grpc::pb::NodepoolSetWriteEnabledRequest { name: "rio".into(), write_enabled: true }))
.await
.unwrap();
assert!(resp.into_inner().spec.unwrap().write_enabled);
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ "writeEnabled": true }));
}
#[tokio::test]
async fn grpc_band_get_returns_typed_band() {
use grpc::pb::breathe_server::Breathe;
let svc = grpc::GrpcService { store: Arc::new(MockStore::default()) };
let resp = svc
.band_get(tonic::Request::new(grpc::pb::BandGetRequest {
kind: grpc::pb::BandKind::Arc as i32,
namespace: "pangea-system".into(),
name: "rio-arc".into(),
}))
.await
.unwrap();
let band = resp.into_inner();
assert_eq!(band.api_version, "breathe.pleme.io/v1");
assert!((band.spec.unwrap().setpoint.unwrap() - 0.8).abs() < 1e-9);
assert_eq!(band.status.unwrap().phase, "Holding");
}
#[tokio::test]
async fn grpc_band_patch_transmits_zero_values_via_field_presence() {
use grpc::pb::breathe_server::Breathe;
let mock = Arc::new(MockStore::default());
let svc = grpc::GrpcService { store: mock.clone() };
let body = grpc::pb::BandSpec { dry_run: Some(false), ..Default::default() };
svc.band_patch(tonic::Request::new(grpc::pb::BandPatchRequest {
kind: grpc::pb::BandKind::Arc as i32,
namespace: "pangea-system".into(),
name: "rio-arc".into(),
body: Some(body),
}))
.await
.unwrap();
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ "dryRun": false }));
}
#[tokio::test]
async fn grpc_band_list_returns_typed_items() {
use grpc::pb::breathe_server::Breathe;
let svc = grpc::GrpcService { store: Arc::new(MockStore::default()) };
let resp = svc
.band_list(tonic::Request::new(grpc::pb::BandListRequest {
kind: grpc::pb::BandKind::Arc as i32,
namespace: String::new(),
}))
.await
.unwrap();
assert_eq!(resp.into_inner().items.len(), 1);
}
#[tokio::test]
async fn grpc_catalog_list_returns_typed_catalog() {
use grpc::pb::breathe_server::Breathe;
let svc = grpc::GrpcService { store: Arc::new(MockStore::default()) };
let resp = svc.catalog_list(tonic::Request::new(grpc::pb::CatalogListRequest {})).await.unwrap();
let cat = resp.into_inner();
assert_eq!(cat.dimensions.len(), breathe_catalog::ALL_DIMENSIONS.len());
assert!(cat.dimensions.iter().any(|d| d.id == "arc" && d.is_host));
assert!(cat.dimensions.iter().any(|d| d.id == "cgroup-cpu" && d.is_host));
assert!(cat.dimensions.iter().any(|d| d.id == "memory" && !d.is_host));
}
#[tokio::test]
async fn grpc_set_dry_run_refuses_where_inert_and_applies_where_honored() {
use grpc::pb::breathe_server::Breathe;
let mock = Arc::new(MockStore::default());
let svc = grpc::GrpcService { store: mock.clone() };
let call = |kind: grpc::pb::BandKind| {
tonic::Request::new(grpc::pb::BandSetDryRunRequest {
kind: kind as i32,
namespace: "pangea-system".into(),
name: "b".into(),
dry_run: false,
})
};
let err = svc.band_set_dry_run(call(grpc::pb::BandKind::Arc)).await.unwrap_err();
assert_eq!(err.code(), tonic::Code::FailedPrecondition);
assert!(mock.patches.lock().unwrap().is_empty());
let resp = svc.band_set_dry_run(call(grpc::pb::BandKind::HostParam)).await.unwrap();
assert_eq!(resp.into_inner().spec.unwrap().dry_run, Some(false));
assert_eq!(mock.patches.lock().unwrap()[0].1, json!({ "dryRun": false }));
}
#[tokio::test]
async fn grpc_band_patch_carries_write_intent_and_refuses_an_unattributed_go_live() {
use grpc::pb::breathe_server::Breathe;
let mock = Arc::new(MockStore::default());
let svc = grpc::GrpcService { store: mock.clone() };
let call = |wi: grpc::pb::WriteIntent| {
tonic::Request::new(grpc::pb::BandPatchRequest {
kind: grpc::pb::BandKind::Cpu as i32,
namespace: "camelot".into(),
name: "coredns".into(),
body: Some(grpc::pb::BandSpec { write_intent: Some(wi), ..Default::default() }),
})
};
let unattributed =
grpc::pb::WriteIntent { intent: "write".into(), confirm_after_seconds: None, authorized_by: None };
let err = svc.band_patch(call(unattributed)).await.unwrap_err();
assert_eq!(err.code(), tonic::Code::InvalidArgument);
assert!(mock.patches.lock().unwrap().is_empty(), "a refused go-live must not reach the store");
let observe =
grpc::pb::WriteIntent { intent: "observe".into(), confirm_after_seconds: None, authorized_by: None };
svc.band_patch(call(observe)).await.unwrap();
assert_eq!(mock.patches.lock().unwrap()[0].1["writeIntent"]["intent"], "observe");
}
#[test]
fn openapi_spec_band_kinds_match_the_code() {
let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../spec/breathe.openapi.yaml");
let raw = std::fs::read_to_string(path).expect("the spec is where the crate doc says it is");
let spec: serde_yaml::Value = serde_yaml::from_str(&raw).expect("spec/breathe.openapi.yaml must be valid YAML");
let declared: Vec<String> = spec["components"]["schemas"]["BandKind"]["enum"]
.as_sequence()
.expect("BandKind.enum is a sequence")
.iter()
.map(|v| v.as_str().expect("each enum value is a string").to_owned())
.collect();
let canonical: Vec<String> = DimensionId::ALL.iter().map(|d| d.as_str().to_owned()).collect();
assert_eq!(declared, canonical, "the OpenAPI BandKind enum has drifted from DimensionId::ALL");
let paths = spec["paths"].as_mapping().expect("paths is a mapping");
for route in [
"/api/v1/bands/{kind}/{namespace}/{name}/write-intent",
"/api/v1/bands/{kind}/{namespace}/{name}/confirm",
] {
assert!(
paths.contains_key(serde_yaml::Value::from(route)),
"{route} is served by the router but absent from the spec"
);
}
}
#[test]
fn grpc_band_kind_covers_every_dimension_without_renumbering() {
use grpc::pb::BandKind as P;
for (p, d) in [
(P::Memory, DimensionId::Memory),
(P::Cpu, DimensionId::Cpu),
(P::Storage, DimensionId::Storage),
(P::Replica, DimensionId::Replica),
(P::Arc, DimensionId::Arc),
(P::Cgroup, DimensionId::Cgroup),
(P::CgroupCpu, DimensionId::CgroupCpu),
(P::HostParam, DimensionId::HostParam),
(P::KubeParam, DimensionId::KubeParam),
(P::AppParam, DimensionId::AppParam),
(P::Request, DimensionId::Request),
] {
assert_eq!(grpc::kind_of(p as i32).unwrap(), d);
}
assert_eq!(P::Memory as i32, 1);
assert_eq!(P::Cpu as i32, 2);
assert_eq!(P::Storage as i32, 3);
assert_eq!(P::Arc as i32, 4);
assert_eq!(P::Cgroup as i32, 5);
assert_eq!(P::Request as i32, 11);
}
#[test]
fn every_dimension_has_a_wire_number() {
for d in DimensionId::ALL {
let found = (1..=64).any(|n| grpc::kind_of(n).ok() == Some(d));
assert!(found, "dimension {d} has no BandKind wire number — add one, never renumber");
}
}
}