Add integration and unit tests for monitoring JSON API endpoints

- Add typed response structs for JSON API parsing (minimal fields only)
- Add prometheus_metrics_assertions module with polling helpers
- Add #[track_caller] to assert helpers for better debugging
- Add shutdown calls to all integration tests
- Use POLL_TIMEOUT constant instead of magic numbers
- Add unit tests for http_server.rs endpoints

Author: Eric Price

integration-tests/Cargo.lock

@@ -27,6 +27,8 @@ tracing = { version = "0.1.41", default-features = false }
 tracing-subscriber = { version = "0.3.19", default-features = false }
 hex = "0.4.3"
 clap = { version = "^4.5.4", features = ["derive"] }
+serde_json = "1"
+serde = { version = "1", features = ["derive"] }
 
 # Direct dependencies kept only for the embedded `mining_device` module.
 # Remove this block when removing:

integration-tests/Cargo.toml

@@ -1,7 +1,104 @@
 //! Helpers for querying and asserting on Prometheus metrics and JSON API endpoints
 //! exposed by SV2 components during integration tests.
 
+use serde::Deserialize;
 use std::net::SocketAddr;
+use std::time::Duration;
+
+// ── Typed response structs for JSON API endpoints ─────────────────────────────
+
+/// Response from `/api/v1/global`
+#[derive(Debug, Deserialize)]
+pub struct GlobalResponse {
+    pub uptime_secs: u64,
+    pub server: Option<ServerSummary>,
+    pub sv2_clients: Option<Sv2ClientsSummary>,
+    pub sv1_clients: Option<Sv1ClientsSummary>,
+}
+
+/// Server summary in global response
+#[derive(Debug, Deserialize)]
+pub struct ServerSummary {
+    pub extended_channels: u64,
+    pub standard_channels: u64,
+}
+
+/// SV2 clients summary in global response
+#[derive(Debug, Deserialize)]
+pub struct Sv2ClientsSummary {
+    pub total_clients: u64,
+}
+
+/// SV1 clients summary in global response
+#[derive(Debug, Deserialize)]
+pub struct Sv1ClientsSummary {
+    pub total_clients: u64,
+}
+
+/// Response from `/api/v1/server`
+#[derive(Debug, Deserialize)]
+pub struct ServerResponse {
+    pub extended_channels_count: usize,
+}
+
+/// Response from `/api/v1/server/channels`
+#[derive(Debug, Deserialize)]
+pub struct ServerChannelsResponse {
+    pub total_extended: u64,
+    pub extended_channels: Vec<ServerExtendedChannel>,
+}
+
+/// Extended channel info from server perspective
+#[derive(Debug, Deserialize)]
+pub struct ServerExtendedChannel {
+    pub shares_acknowledged: Option<u64>,
+}
+
+/// Generic paginated response
+#[derive(Debug, Deserialize)]
+pub struct PaginatedResponse<T> {
+    pub total: u64,
+    pub items: Vec<T>,
+}
+
+/// Client metadata in paginated clients list
+#[derive(Debug, Deserialize)]
+pub struct ClientMetadata {
+    pub client_id: u64,
+}
+
+/// Response from `/api/v1/clients/{id}/channels`
+#[derive(Debug, Deserialize)]
+pub struct ClientChannelsResponse {
+    pub client_id: u64,
+    pub total_extended: u64,
+    pub total_standard: u64,
+}
+
+/// SV1 client info
+#[derive(Debug, Deserialize)]
+pub struct Sv1Client {
+    pub client_id: u64,
+    pub authorized_worker_name: String,
+}
+
+/// Error response body
+#[derive(Debug, Deserialize)]
+pub struct ErrorResponse {
+    pub error: String,
+}
+
+/// Root endpoint response
+#[derive(Debug, Deserialize)]
+pub struct RootResponse {
+    pub service: String,
+    pub endpoints: serde_json::Value,
+}
+
+/// Default timeout used when polling for eventually-consistent data.
+/// Needs to be generous enough for the monitoring snapshot cache (1s refresh) to populate
+/// under CI load, where components may take several seconds to complete handshakes.
+pub const POLL_TIMEOUT: Duration = Duration::from_secs(15);
 
 /// Fetch the raw Prometheus text-format metrics from a component's `/metrics` endpoint.
 /// Uses `spawn_blocking` to avoid blocking the tokio runtime with synchronous HTTP calls.
@@ -27,6 +124,98 @@ pub async fn fetch_api(monitoring_addr: SocketAddr, path: &str) -> String {
     .expect("spawn_blocking for fetch_api panicked")
 }
 
+/// Fetch a JSON API endpoint and parse the response into a `serde_json::Value`.
+pub async fn fetch_api_json(monitoring_addr: SocketAddr, path: &str) -> serde_json::Value {
+    let body = fetch_api(monitoring_addr, path).await;
+    serde_json::from_str(&body).unwrap_or_else(|e| {
+        panic!(
+            "Failed to parse JSON from {} response: {}\nBody: {}",
+            path, e, body
+        )
+    })
+}
+
+/// Fetch a JSON API endpoint and parse the response into a typed struct.
+pub async fn fetch_api_typed<T: serde::de::DeserializeOwned>(
+    monitoring_addr: SocketAddr,
+    path: &str,
+) -> T {
+    let body = fetch_api(monitoring_addr, path).await;
+    serde_json::from_str(&body).unwrap_or_else(|e| {
+        panic!(
+            "Failed to parse JSON from {} into {}: {}\nBody: {}",
+            path,
+            std::any::type_name::<T>(),
+            e,
+            body
+        )
+    })
+}
+
+/// Fetch a JSON API endpoint returning both the HTTP status code and parsed JSON body.
+/// Unlike `fetch_api_json`, this does **not** panic on non-2xx responses, so it can be
+/// used to test error endpoints (e.g. 404).
+pub async fn fetch_api_with_status(
+    monitoring_addr: SocketAddr,
+    path: &str,
+) -> (i32, serde_json::Value) {
+    let url = format!("http://{}{}", monitoring_addr, path);
+    tokio::task::spawn_blocking(move || {
+        let (status, bytes) = crate::utils::http::make_get_request_with_status(&url, 5);
+        let body = String::from_utf8(bytes).expect("api response should be valid UTF-8");
+        let json: serde_json::Value = serde_json::from_str(&body).unwrap_or_else(|e| {
+            panic!(
+                "Failed to parse JSON from {} (status {}): {}\nBody: {}",
+                url, status, e, body
+            )
+        });
+        (status, json)
+    })
+    .await
+    .expect("spawn_blocking for fetch_api_with_status panicked")
+}
+
+/// Poll a JSON API endpoint until a numeric field at `json_pointer` (RFC 6901, e.g.
+/// `"/sv2_clients/total_clients"`) reaches `>= min`. Returns the full JSON value once
+/// satisfied. Panics if the condition is not met within `timeout`.
+///
+/// This is the JSON equivalent of `poll_until_metric_gte` — use it for endpoints whose
+/// data only appears after the monitoring snapshot cache has refreshed.
+pub async fn poll_until_api_field_gte(
+    monitoring_addr: SocketAddr,
+    path: &str,
+    json_pointer: &str,
+    min: f64,
+    timeout: std::time::Duration,
+) -> serde_json::Value {
+    let deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        // Use fetch_api_with_status so that transient non-2xx responses (e.g. 404
+        // before the snapshot cache has populated) are retried instead of panicking.
+        let (status, json) = fetch_api_with_status(monitoring_addr, path).await;
+        if (200..300).contains(&status) {
+            if let Some(val) = json.pointer(json_pointer) {
+                let num = val.as_f64().unwrap_or(0.0);
+                if num >= min {
+                    return json;
+                }
+            }
+        }
+        if tokio::time::Instant::now() >= deadline {
+            panic!(
+                "JSON field '{}' at {} never reached >= {} within {:?}. Last status: {}. Last response:\n{}",
+                json_pointer,
+                path,
+                min,
+                timeout,
+                status,
+                serde_json::to_string_pretty(&json).unwrap_or_default()
+            );
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    }
+}
+
 /// Parse a specific metric value from Prometheus text format.
 /// Returns `None` if the metric line is not found.
 ///
@@ -53,6 +242,7 @@ pub(crate) fn parse_metric_value(metrics_text: &str, metric_name: &str) -> Optio
 }
 
 /// Assert that a metric is present and its value satisfies the given predicate.
+#[track_caller]
 pub(crate) fn assert_metric<F: Fn(f64) -> bool>(
     metrics_text: &str,
     metric_name: &str,
@@ -80,6 +270,7 @@ pub(crate) fn assert_metric<F: Fn(f64) -> bool>(
 }
 
 /// Assert that a metric is present with a value >= the given minimum.
+#[track_caller]
 pub fn assert_metric_gte(metrics_text: &str, metric_name: &str, min: f64) {
     assert_metric(
         metrics_text,
@@ -90,6 +281,7 @@ pub fn assert_metric_gte(metrics_text: &str, metric_name: &str, min: f64) {
 }
 
 /// Assert that a metric is present with the exact given value.
+#[track_caller]
 pub fn assert_metric_eq(metrics_text: &str, metric_name: &str, expected: f64) {
     assert_metric(
         metrics_text,
@@ -100,6 +292,7 @@ pub fn assert_metric_eq(metrics_text: &str, metric_name: &str, expected: f64) {
 }
 
 /// Assert that a metric name does NOT appear in the metrics output at all.
+#[track_caller]
 pub fn assert_metric_not_present(metrics_text: &str, metric_name: &str) {
     for line in metrics_text.lines() {
         if line.starts_with('#') {
@@ -118,6 +311,7 @@ pub fn assert_metric_not_present(metrics_text: &str, metric_name: &str) {
 }
 
 /// Assert that a metric name appears at least once in the metrics output (with any label/value).
+#[track_caller]
 pub fn assert_metric_present(metrics_text: &str, metric_name: &str) {
     for line in metrics_text.lines() {
         if line.starts_with('#') {
@@ -199,6 +393,7 @@ pub async fn assert_api_health(monitoring_addr: SocketAddr) {
 }
 
 /// Assert that the uptime metric is present and positive.
+#[track_caller]
 pub fn assert_uptime(metrics_text: &str) {
     assert_metric(
         metrics_text,

integration-tests/lib/prometheus_metrics_assertions.rs

@@ -407,6 +407,43 @@ pub fn into_static(m: AnyMessage<'_>) -> AnyMessage<'static> {
 }
 
 pub mod http {
+    /// Make a GET request that returns both the HTTP status code and the response body.
+    /// Unlike `make_get_request`, this does NOT panic on non-2xx status codes (e.g. 404),
+    /// making it suitable for testing API error responses.
+    /// Only retries on 5xx errors or connection failures.
+    pub fn make_get_request_with_status(url: &str, retries: usize) -> (i32, Vec<u8>) {
+        for attempt in 1..=retries {
+            let response = minreq::get(url).send();
+            match response {
+                Ok(res) => {
+                    let status_code = res.status_code;
+                    if (500..600).contains(&status_code) {
+                        eprintln!(
+                            "Attempt {attempt}: URL {url} returned a server error code {status_code}"
+                        );
+                    } else {
+                        return (status_code, res.as_bytes().to_vec());
+                    }
+                }
+                Err(err) => {
+                    eprintln!(
+                        "Attempt {}: Failed to fetch URL {}: {:?}",
+                        attempt + 1,
+                        url,
+                        err
+                    );
+                }
+            }
+
+            if attempt < retries {
+                let delay = 1u64 << (attempt - 1);
+                eprintln!("Retrying in {delay} seconds (exponential backoff)...");
+                std::thread::sleep(std::time::Duration::from_secs(delay));
+            }
+        }
+        panic!("Cannot reach URL {url} after {retries} attempts");
+    }
+
     pub fn make_get_request(download_url: &str, retries: usize) -> Vec<u8> {
         for attempt in 1..=retries {
             let response = minreq::get(download_url).send();