diff --git a/Cargo.lock b/Cargo.lock index fb37b9d..ef825c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1502,7 +1502,7 @@ dependencies = [ [[package]] name = "rawtree-cli" -version = "0.6.15" +version = "0.7.0" dependencies = [ "anyhow", "axoupdater", diff --git a/Cargo.toml b/Cargo.toml index db300d4..f56d1b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rawtree-cli" -version = "0.6.15" +version = "0.7.0" edition = "2021" repository = "https://github.com/rawtreedb/rawtree-cli" diff --git a/src/client.rs b/src/client.rs index 5da9ccc..f141a2a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,3 +1,7 @@ +mod routing; +#[cfg(test)] +mod routing_tests; + use anyhow::{Context, Result}; use flate2::write::GzEncoder; use flate2::Compression; @@ -18,6 +22,7 @@ pub struct ApiClient { pub base_url: String, pub token: Option, client: Client, + routing: routing::RoutingCache, } impl ApiClient { @@ -26,6 +31,18 @@ impl ApiClient { base_url, token, client: Client::new(), + routing: routing::RoutingCache::default(), + } + } + + fn unrouted_request(&self, method: reqwest::Method, path: &str) -> RequestBuilder { + let request = with_client_header( + self.client + .request(method, format!("{}{}", self.base_url, path)), + ); + match &self.token { + Some(token) => request.bearer_auth(token), + None => request, } } @@ -39,13 +56,9 @@ impl ApiClient { body: &Value, timeout: Duration, ) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)) - .json(body) - .timeout(timeout); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self + .routed_request_with_timeout(reqwest::Method::POST, path, timeout)? + .json(body); let resp = req.send().context("failed to connect to server")?; handle_response(resp) } @@ -60,11 +73,7 @@ impl ApiClient { &self, path: &str, ) -> Result<(T, Option)> { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::POST, path)?; let resp = req.send().context("failed to connect to server")?; let query_id = resp .headers() @@ -80,99 +89,69 @@ impl ApiClient { path: &str, idempotency_key: Option<&str>, ) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)); + let mut req = self.routed_request(reqwest::Method::POST, path)?; if let Some(key) = idempotency_key { req = req.header("idempotency-key", key); } - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } let resp = req.send().context("failed to connect to server")?; handle_response(resp) } /// POST without a body to an endpoint that replies without content. pub fn post_no_content(&self, path: &str) -> Result<()> { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::POST, path)?; let resp = req.send().context("failed to connect to server")?; check_status(resp) } /// DELETE to an endpoint that replies without content. pub fn delete_no_content(&self, path: &str) -> Result<()> { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.delete(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::DELETE, path)?; let resp = req.send().context("failed to connect to server")?; check_status(resp) } pub fn patch(&self, path: &str, body: &Value) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.patch(&url)).json(body); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self + .routed_request(reqwest::Method::PATCH, path)? + .json(body); let resp = req.send().context("failed to connect to server")?; handle_response(resp) } pub fn delete(&self, path: &str) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.delete(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::DELETE, path)?; let resp = req.send().context("failed to connect to server")?; handle_response(resp) } pub fn get(&self, path: &str) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.get(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::GET, path)?; let resp = req.send().context("failed to connect to server")?; - handle_response(resp) + let value: Value = handle_response(resp)?; + self.remember_routing_metadata(path, &value); + serde_json::from_value(value).context("failed to parse server response") } /// POST with a pre-serialized JSON string body, gzip-compressed. pub fn post_compressed(&self, path: &str, body: &str) -> Result { - let url = format!("{}{}", self.base_url, path); let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); encoder .write_all(body.as_bytes()) .context("gzip compression failed")?; let compressed = encoder.finish().context("gzip finalization failed")?; - let mut req = self - .client - .post(&url) + let req = self + .routed_request(reqwest::Method::POST, path)? .header("content-type", "application/json") .header("content-encoding", "gzip") .body(compressed); - req = with_client_header(req); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } let resp = req.send().context("failed to connect to server")?; handle_response(resp) } /// POST that returns raw text (for queries that return ClickHouse results directly). pub fn post_raw(&self, path: &str, body: &Value) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)).json(body); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::POST, path)?.json(body); let resp = req.send().context("failed to connect to server")?; let status = resp.status(); let text = resp.text().context("failed to read response body")?; @@ -184,11 +163,7 @@ impl ApiClient { /// GET that returns raw text. pub fn get_raw(&self, path: &str) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.get(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::GET, path)?; let resp = req.send().context("failed to connect to server")?; let status = resp.status(); let text = resp.text().context("failed to read response body")?; @@ -205,7 +180,7 @@ fn with_client_header(req: RequestBuilder) -> RequestBuilder { .header(RAWTREE_CLIENT_VERSION_HEADER, RAWTREE_CLIENT_VERSION_VALUE) } -fn handle_response(resp: reqwest::blocking::Response) -> Result { +fn handle_response(resp: Response) -> Result { let status = resp.status(); let text = resp.text().context("failed to read response body")?; if !status.is_success() { @@ -214,7 +189,7 @@ fn handle_response(resp: reqwest::blocking::Response) -> Re serde_json::from_str(&text).context("failed to parse server response") } -fn check_status(resp: reqwest::blocking::Response) -> Result<()> { +fn check_status(resp: Response) -> Result<()> { let status = resp.status(); if status.is_success() { return Ok(()); diff --git a/src/client/routing.rs b/src/client/routing.rs new file mode 100644 index 0000000..bef8ed5 --- /dev/null +++ b/src/client/routing.rs @@ -0,0 +1,256 @@ +use super::{handle_response, ApiClient}; +use anyhow::{Context, Result}; +use reqwest::{blocking::RequestBuilder, Method, Url}; +use serde::Deserialize; +use std::{ + cell::RefCell, + collections::HashMap, + time::{Duration, Instant}, +}; + +#[derive(Clone, Deserialize)] +struct Organization { + organization_id: String, + name: String, +} +#[derive(Deserialize)] +struct Organizations { + organizations: Vec, +} +#[derive(Clone, Deserialize)] +struct Cluster { + id: String, + name: String, + #[serde(default)] + deployment_id: Option, + #[serde(default)] + cloud_region: Option, +} +#[derive(Deserialize)] +struct Clusters { + clusters: Vec, +} + +#[derive(Default)] +pub(super) struct RoutingCache { + organizations: RefCell)>>, + clusters: RefCell)>>, +} + +impl ApiClient { + pub(super) fn remember_routing_metadata(&self, path: &str, value: &serde_json::Value) { + let Ok(url) = Url::parse(&format!("{}{}", self.base_url, path)) else { + return; + }; + if url.path() == "/v1/organizations" { + if let Ok(response) = serde_json::from_value::(value.clone()) { + *self.routing.organizations.borrow_mut() = + Some((Instant::now(), response.organizations)); + } + } else if url.path() == "/v1/clusters" { + if let Some((_, organization)) = + url.query_pairs().find(|(key, _)| key == "organization") + { + if let Ok(response) = serde_json::from_value::(value.clone()) { + self.routing.clusters.borrow_mut().insert( + clusters_path(&organization), + (Instant::now(), response.clusters), + ); + } + } + } + } + + fn organization_name(&self, id: &str, deadline: Option) -> Result { + if let Some((time, organizations)) = self.routing.organizations.borrow().as_ref() { + if time.elapsed() < Duration::from_secs(30) { + if let Some(organization) = + organizations.iter().find(|item| item.organization_id == id) + { + return Ok(organization.name.clone()); + } + } + } + let response: Organizations = handle_response( + before_deadline( + self.unrouted_request(Method::GET, "/v1/organizations"), + deadline, + )? + .send()?, + )?; + let name = response + .organizations + .iter() + .find(|item| item.organization_id == id) + .map(|item| item.name.clone()); + *self.routing.organizations.borrow_mut() = Some((Instant::now(), response.organizations)); + name.context("Organization not found.") + } + + pub(super) fn routed_request(&self, method: Method, path: &str) -> Result { + self.request_before(method, path, None) + } + + pub(super) fn routed_request_with_timeout( + &self, + method: Method, + path: &str, + timeout: Duration, + ) -> Result { + let deadline = Instant::now() + .checked_add(timeout) + .context("invalid request timeout")?; + self.request_before(method, path, Some(deadline)) + } + + fn request_before( + &self, + method: Method, + path: &str, + deadline: Option, + ) -> Result { + let pair = self.deployment_for(path, deadline)?; + if method != Method::GET + && (path.starts_with("/v1/organizations") || path.starts_with("/v1/clusters")) + { + self.routing.clusters.borrow_mut().clear(); + *self.routing.organizations.borrow_mut() = None; + } + let mut request = self.unrouted_request(method, path); + if let Some((deployment, region)) = pair { + request = request + .header("X-RawTree-Deployment-Id", deployment) + .header("X-RawTree-Region", region); + } + before_deadline(request, deadline) + } + + fn deployment_for( + &self, + path: &str, + deadline: Option, + ) -> Result> { + if self + .token + .as_deref() + .is_none_or(|token| token.starts_with("rt_")) + { + return Ok(None); + } + let url = Url::parse(&format!("{}{}", self.base_url, path)).context("invalid API URL")?; + let Some((organization, cluster)) = scope(&url) else { + return Ok(None); + }; + let organization = match organization { + OrganizationSelector::Id(id) => self.organization_name(&id, deadline)?, + OrganizationSelector::Name(name) => name, + }; + let collection = clusters_path(&organization); + let cached = self + .routing + .clusters + .borrow() + .get(&collection) + .filter(|(time, _)| time.elapsed() < Duration::from_secs(30)) + .map(|(_, clusters)| clusters.clone()); + let clusters = match cached { + Some(clusters) => clusters, + None => { + let response: Clusters = handle_response( + before_deadline(self.unrouted_request(Method::GET, &collection), deadline)? + .send()?, + )?; + self.routing + .clusters + .borrow_mut() + .insert(collection, (Instant::now(), response.clusters.clone())); + response.clusters + } + }; + let target = clusters.iter().find(|item| match &cluster { + ClusterSelector::Id(id) => item.id == *id, + ClusterSelector::Name(name) => item.name == *name, + }); + let pair = target.and_then(|item| { + item.deployment_id + .as_deref() + .zip(item.cloud_region.as_deref()) + }); + Ok(pair + .filter(|(deployment, region)| valid_routing(deployment, region)) + .map(|(deployment, region)| (deployment.to_owned(), region.to_owned()))) + } +} + +fn before_deadline(request: RequestBuilder, deadline: Option) -> Result { + match deadline { + Some(deadline) => { + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero()) + .context("request timed out during routing discovery")?; + Ok(request.timeout(remaining)) + } + None => Ok(request), + } +} + +enum ClusterSelector { + Id(String), + Name(String), +} + +enum OrganizationSelector { + Id(String), + Name(String), +} + +fn clusters_path(organization: &str) -> String { + format!( + "/v1/clusters?organization={}", + urlencoding::encode(organization) + ) +} + +fn scope(url: &Url) -> Option<(OrganizationSelector, ClusterSelector)> { + let query: HashMap<_, _> = url.query_pairs().into_owned().collect(); + let organization = if let Some(id) = query.get("organization_id") { + OrganizationSelector::Id(id.clone()) + } else { + OrganizationSelector::Name(query.get("organization")?.clone()) + }; + let path_cluster = url + .path() + .strip_prefix("/v1/clusters/") + .and_then(|path| path.split('/').next()) + .filter(|part| { + !part.is_empty() && !matches!(*part, "sizes" | "metrics" | "verify-s3-access") + }); + let cluster = if let Some(id) = path_cluster { + ClusterSelector::Id(urlencoding::decode(id).ok()?.into_owned()) + } else if let Some(id) = query.get("cluster_id") { + ClusterSelector::Id(id.clone()) + } else { + ClusterSelector::Name(query.get("cluster")?.clone()) + }; + Some((organization, cluster)) +} + +// Match the backend/worker syntax; do not maintain a catalog of AWS regions. +fn valid_routing(deployment: &str, region: &str) -> bool { + let mut parts = region.rsplit('-'); + let number = parts.next().unwrap_or_default(); + let names: Vec<_> = parts.collect(); + deployment.len() == 8 + && deployment + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + && region.len() <= 63 + && !number.starts_with('0') + && !number.is_empty() + && number.bytes().all(|byte| byte.is_ascii_digit()) + && names.len() >= 2 + && names + .iter() + .all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_lowercase())) +} diff --git a/src/client/routing_tests.rs b/src/client/routing_tests.rs new file mode 100644 index 0000000..33d4b61 --- /dev/null +++ b/src/client/routing_tests.rs @@ -0,0 +1,270 @@ +use super::ApiClient; +use serde_json::{json, Value}; +use std::collections::HashMap; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::TcpListener; +use std::thread; +use std::time::{Duration, Instant}; + +const ORG: &str = "00000000-0000-4000-8000-000000000001"; +struct Request { + path: String, + headers: HashMap, + body: Vec, +} + +fn server(responses: Vec<(&'static str, Value)>) -> (String, thread::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let handle = thread::spawn(move || { + let mut requests = Vec::new(); + for (status, payload) in responses { + let deadline = Instant::now() + Duration::from_secs(5); + let mut socket = loop { + match listener.accept() { + Ok((socket, _)) => break socket, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "missing HTTP request"); + thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("accept: {error}"), + } + }; + socket.set_nonblocking(false).unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut reader = BufReader::new(socket.try_clone().unwrap()); + let mut first = String::new(); + reader.read_line(&mut first).unwrap(); + let mut headers = HashMap::new(); + loop { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + if line == "\r\n" { + break; + } + let (key, value) = line.split_once(':').unwrap(); + headers.insert(key.to_ascii_lowercase(), value.trim().to_owned()); + } + let length = headers + .get("content-length") + .map(|value| value.parse().unwrap()) + .unwrap_or(0); + let mut body = vec![0; length]; + reader.read_exact(&mut body).unwrap(); + let response = payload.to_string(); + write!(socket, "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}", response.len()).unwrap(); + requests.push(Request { + path: first.trim().to_owned(), + headers, + body, + }); + } + requests + }); + (base, handle) +} + +fn metadata(pair: Value) -> Vec<(&'static str, Value)> { + vec![( + "200 OK", + json!({"clusters": [{"id": "cluster-id", "name": "production", "deployment_id": pair["deployment_id"], "cloud_region": pair["cloud_region"]}]}), + )] +} + +#[test] +fn session_requests_resolve_scope_and_reuse_metadata_for_queries_and_gzip() { + let mut responses = metadata(json!({"deployment_id": "12abcdef", "cloud_region": "eu-west-1"})); + responses.extend([ + ("200 OK", json!({"ok":true})), + ("200 OK", json!({"ok":true})), + ]); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let path = "/v1/query?organization=acme&cluster=production"; + let _: Value = client.post(path, &json!({"query": "SELECT 1"})).unwrap(); + let _: Value = client + .post_compressed( + "/v1/tables/events?organization=acme&cluster=production", + "{\"value\":1}", + ) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!( + requests[0].path, + "GET /v1/clusters?organization=acme HTTP/1.1" + ); + assert!(!requests[0].headers.contains_key("x-rawtree-region")); + for request in &requests[1..] { + assert_eq!(request.headers["x-rawtree-deployment-id"], "12abcdef"); + assert_eq!(request.headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(request.headers["authorization"], "Bearer session.fixture"); + assert_eq!(request.headers["x-rawtree-client"], "cli"); + } + assert_eq!(requests[2].headers["content-encoding"], "gzip"); + let mut decoded = String::new(); + flate2::read::GzDecoder::new(requests[2].body.as_slice()) + .read_to_string(&mut decoded) + .unwrap(); + assert_eq!(decoded, "{\"value\":1}"); +} + +#[test] +fn absent_partial_and_invalid_metadata_keep_default_routing() { + for pair in [ + json!({}), + json!({"deployment_id": "12abcdef"}), + json!({"deployment_id": "12abcdef99", "cloud_region": "eu-west-1"}), + json!({"deployment_id": "12abcdef", "cloud_region": "bad.invalid"}), + ] { + let mut responses = metadata(pair); + responses.push(("200 OK", json!({}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("oauth.fixture".to_string())); + let _: Value = client + .get("/v1/databases?organization=acme&cluster=production") + .unwrap(); + let requests = server.join().unwrap(); + assert!(!requests[1].headers.contains_key("x-rawtree-region")); + assert!(!requests[1].headers.contains_key("x-rawtree-deployment-id")); + } +} + +#[test] +fn api_keys_skip_discovery_and_routing_headers() { + let (base, server) = server(vec![("200 OK", json!({}))]); + let client = ApiClient::new(base, Some("rt_fixture".to_string())); + client + .post_raw( + "/v1/query?organization=acme&cluster=production", + &json!({"query":"SELECT 1"}), + ) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests.len(), 1); + assert!(!requests[0].headers.contains_key("x-rawtree-region")); + assert_eq!(requests[0].headers["authorization"], "Bearer rt_fixture"); +} + +#[test] +fn cluster_id_actions_preserve_organization_query_and_serving_headers() { + let mut responses = metadata(json!({"deployment_id": "12abcdef", "cloud_region": "us-east-1"})); + responses.push(("200 OK", json!({}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let _: Value = client + .post_empty("/v1/clusters/cluster-id/stop?organization=acme") + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!( + requests[1].path, + "POST /v1/clusters/cluster-id/stop?organization=acme HTTP/1.1" + ); + assert_eq!(requests[1].headers["x-rawtree-deployment-id"], "12abcdef"); +} + +#[test] +fn discovery_failure_does_not_execute_the_operation() { + let mut responses = metadata(json!({})); + responses[0] = ( + "503 Service Unavailable", + json!({"message":"discovery unavailable"}), + ); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + assert!(client + .get::("/v1/databases?organization=acme&cluster=production") + .is_err()); + assert_eq!(server.join().unwrap().len(), 1); +} + +#[test] +fn cluster_creation_invalidates_previously_discovered_empty_collection() { + let responses = vec![ + ("200 OK", json!({"clusters": []})), + ("200 OK", json!({"id": "new-id"})), + ( + "200 OK", + json!({"clusters": [{"id":"new-id", "name":"new", "deployment_id":"12abcdef", "cloud_region":"eu-west-1"}]}), + ), + ("200 OK", json!({"data":[]})), + ]; + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let collection = "/v1/clusters?organization=acme"; + let _: Value = client.get(&collection).unwrap(); + let _: Value = client.post(&collection, &json!({"name":"new"})).unwrap(); + let _: Value = client + .get("/v1/databases?organization=acme&cluster_id=new-id") + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests[2].path, format!("GET {collection} HTTP/1.1")); + assert_eq!(requests[3].headers["x-rawtree-region"], "eu-west-1"); +} + +#[test] +fn explicit_timeout_includes_routing_discovery() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let server = thread::spawn(move || { + let (socket, _) = listener.accept().unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut reader = BufReader::new(socket); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!(line.trim(), "GET /v1/clusters?organization=acme HTTP/1.1"); + thread::sleep(Duration::from_millis(400)); + }); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let started = Instant::now(); + let result = client.post_with_timeout::( + "/v1/query?organization=acme&cluster=production", + &json!({"query":"SELECT 1"}), + Duration::from_millis(50), + ); + assert!(result.is_err()); + assert!(started.elapsed() < Duration::from_millis(300)); + server.join().unwrap(); +} + +#[test] +fn explicit_id_and_name_selectors_cannot_collide() { + let (base, server) = server(vec![ + ( + "200 OK", + json!({"clusters":[ + {"id":"other-id", "name":"target-id", "deployment_id":"abcdef12", "cloud_region":"ap-south-1"}, + {"id":"target-id", "name":"production", "deployment_id":"12abcdef", "cloud_region":"eu-west-1"} + ]}), + ), + ("200 OK", json!({})), + ( + "200 OK", + json!({"organizations": [{"organization_id": ORG, "name": "acme"}]}), + ), + ("200 OK", json!({})), + ("200 OK", json!({})), + ]); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let _: Value = client + .get("/v1/clusters/target-id?organization=acme") + .unwrap(); + let _: Value = client + .get(&format!( + "/v1/metrics?organization_id={ORG}&cluster_id=target-id" + )) + .unwrap(); + let _: Value = client + .get(&format!( + "/v1/databases?organization_id={ORG}&cluster=target-id" + )) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests[1].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[3].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[4].headers["x-rawtree-region"], "ap-south-1"); +} diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 6d49495..9ffc1f3 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -36,11 +36,11 @@ fn database_create_reads_nested_response_and_saves_requested_context() { "POST", "/v1/databases?organization=team%20alpha&cluster=prod%2Feu", response.clone(), - json!({"token": "fixture-session", "default_organization": "old-org", "database": "old-db"}), + json!({"token": "rt_fixture", "default_organization": "old-org", "database": "old-db"}), ); assert_eq!(result, response); assert_eq!(body, json!({"name": "analytics"})); - assert_eq!(saved["token"], "fixture-session"); + assert_eq!(saved["token"], "rt_fixture"); assert_eq!(saved["default_organization"], "team alpha"); assert_eq!(saved["cluster"], "prod/eu"); assert_eq!(saved["database"], "analytics"); @@ -48,7 +48,7 @@ fn database_create_reads_nested_response_and_saves_requested_context() { #[test] fn database_create_failure_does_not_overwrite_config() { - let config = json!({"token": "fixture-session", "default_organization": "team", "cluster": "prod", "database": "old-db"}); + let config = json!({"token": "rt_fixture", "default_organization": "team", "cluster": "prod", "database": "old-db"}); let (output, saved, _) = common::run_cli( &[( "POST /v1/databases?organization=team&cluster=prod".to_string(), diff --git a/tests/browser_login.rs b/tests/browser_login.rs index dbba85f..7696ac8 100644 --- a/tests/browser_login.rs +++ b/tests/browser_login.rs @@ -53,7 +53,7 @@ fn clusters(names: &[&str]) -> Response { ( "GET /v1/clusters?organization=team".into(), "200 OK", - json!({"clusters": names.iter().map(|name| json!({"name": name})).collect::>()}), + json!({"clusters": names.iter().map(|name| json!({"id": name, "name": name})).collect::>()}), ) } @@ -61,7 +61,7 @@ fn databases(names: &[&str]) -> Response { ( "GET /v1/databases?organization=team&cluster=production".into(), "200 OK", - json!({"databases": names.iter().map(|name| json!({"name": name})).collect::>()}), + json!({"databases": names.iter().map(|name| json!({"id": name, "name": name})).collect::>()}), ) } diff --git a/tests/format.rs b/tests/format.rs index d5cb54c..1ca77ee 100644 --- a/tests/format.rs +++ b/tests/format.rs @@ -5,7 +5,7 @@ use serde_json::{json, Value}; const PATH: &str = "POST /v1/query-format?organization=team&cluster=prod"; fn config() -> Value { - json!({"token": "fixture-token", "default_organization": "team", "cluster": "prod"}) + json!({"token": "rt_fixture", "default_organization": "team", "cluster": "prod"}) } #[test] diff --git a/tests/sql_input.rs b/tests/sql_input.rs index 0dbe881..7a660d7 100644 --- a/tests/sql_input.rs +++ b/tests/sql_input.rs @@ -3,7 +3,7 @@ mod common; use serde_json::{json, Value}; fn config() -> Value { - json!({"token": "fixture-token", "database": "analytics", "default_organization": "team", "cluster": "prod"}) + json!({"token": "rt_fixture", "database": "analytics", "default_organization": "team", "cluster": "prod"}) } fn response(command: &str) -> (String, &'static str, Value) {