//! Integration tests for the BigQuery source.
//!
//! Drives [`BigQuerySource`] against a wiremock server mocking the BigQuery
//! REST API. The same OAuth-token + service-account dance the sink tests
//! use lets us run the real `gcp_bigquery_client` client end-to-end without
//! reaching out to GCP — we just point both the auth and `v2` base URLs at
//! the local mock.
use std::collections::HashMap;
use faucet_core::Source;
use faucet_source_bigquery::{BigQueryCredentials, BigQuerySource, BigQuerySourceConfig};
use futures::StreamExt;
use gcp_bigquery_client::client_builder::ClientBuilder;
use serde::Serialize;
use serde_json::{Value, json};
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
const PROJECT_ID: &str = "test-project";
const AUTH_TOKEN_PATH: &str = "/:o/oauth2/token";
const AUTH_SCOPE_BASE: &str = "/auth/bigquery";
#[derive(Serialize)]
struct FakeToken {
access_token: &'static str,
token_type: &'static str,
expires_in: u32,
}
fn fake_token() -> FakeToken {
FakeToken {
access_token: "fake-token",
token_type: "bearer",
expires_in: 9_999_999,
}
}
fn dummy_service_account_json(oauth_server: &str) -> Value {
let token_uri = format!("{oauth_server}{AUTH_TOKEN_PATH}");
json!({
"type": "service_account",
"project_id": "dummy",
"private_key_id": "dummy",
"private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDNk6cKkWP/4NMu\nWb3s24YHfM639IXzPtTev06PUVVQnyHmT1bZgQ/XB6BvIRaReqAqnQd61PAGtX3e\n8XocTw+u/ZfiPJOf+jrXMkRBpiBh9mbyEIqBy8BC20OmsUc+O/YYh/qRccvRfPI7\n3XMabQ8eFWhI6z/t35oRpvEVFJnSIgyV4JR/L/cjtoKnxaFwjBzEnxPiwtdy4olU\nKO/1maklXexvlO7onC7CNmPAjuEZKzdMLzFszikCDnoKJC8k6+2GZh0/JDMAcAF4\nwxlKNQ89MpHVRXZ566uKZg0MqZqkq5RXPn6u7yvNHwZ0oahHT+8ixPPrAEjuPEKM\nUPzVRz71AgMBAAECggEAfdbVWLW5Befkvam3hea2+5xdmeN3n3elrJhkiXxbAhf3\nE1kbq9bCEHmdrokNnI34vz0SWBFCwIiWfUNJ4UxQKGkZcSZto270V8hwWdNMXUsM\npz6S2nMTxJkdp0s7dhAUS93o9uE2x4x5Z0XecJ2ztFGcXY6Lupu2XvnW93V9109h\nkY3uICLdbovJq7wS/fO/AL97QStfEVRWW2agIXGvoQG5jOwfPh86GZZRYP9b8VNw\ntkAUJe4qpzNbWs9AItXOzL+50/wsFkD/iWMGWFuU8DY5ZwsL434N+uzFlaD13wtZ\n63D+tNAxCSRBfZGQbd7WxJVFfZe/2vgjykKWsdyNAQKBgQDnEBgSI836HGSRk0Ub\nDwiEtdfh2TosV+z6xtyU7j/NwjugTOJEGj1VO/TMlZCEfpkYPLZt3ek2LdNL66n8\nDyxwzTT5Q3D/D0n5yE3mmxy13Qyya6qBYvqqyeWNwyotGM7hNNOix1v9lEMtH5Rd\nUT0gkThvJhtrV663bcAWCALmtQKBgQDjw2rYlMUp2TUIa2/E7904WOnSEG85d+nc\norhzthX8EWmPgw1Bbfo6NzH4HhebTw03j3NjZdW2a8TG/uEmZFWhK4eDvkx+rxAa\n6EwamS6cmQ4+vdep2Ac4QCSaTZj02YjHb06Be3gptvpFaFrotH2jnpXxggdiv8ul\n6x+ooCffQQKBgQCR3ykzGoOI6K/c75prELyR+7MEk/0TzZaAY1cSdq61GXBHLQKT\nd/VMgAN1vN51pu7DzGBnT/dRCvEgNvEjffjSZdqRmrAVdfN/y6LSeQ5RCfJgGXSV\nJoWVmMxhCNrxiX3h01Xgp/c9SYJ3VD54AzeR/dwg32/j/oEAsDraLciXGQKBgQDF\nMNc8k/DvfmJv27R06Ma6liA6AoiJVMxgfXD8nVUDW3/tBCVh1HmkFU1p54PArvxe\nchAQqoYQ3dUMBHeh6ZRJaYp2ATfxJlfnM99P1/eHFOxEXdBt996oUMBf53bZ5cyJ\n/lAVwnQSiZy8otCyUDHGivJ+mXkTgcIq8BoEwERFAQKBgQDmImBaFqoMSVihqHIf\nDa4WZqwM7ODqOx0JnBKrKO8UOc51J5e1vpwP/qRpNhUipoILvIWJzu4efZY7GN5C\nImF9sN3PP6Sy044fkVPyw4SYEisxbvp9tfw8Xmpj/pbmugkB2ut6lz5frmEBoJSN\n3osZlZTgx+pM3sO6ITV6U4ID2Q==\n-----END PRIVATE KEY-----\n",
"client_email": "dummy@developer.gserviceaccount.com",
"client_id": "dummy",
"auth_uri": format!("{oauth_server}/o/oauth2/auth"),
"token_uri": token_uri,
"auth_provider_x509_cert_url": format!("{oauth_server}/oauth2/v1/certs"),
"client_x509_cert_url": format!("{oauth_server}/robot/v1/metadata/x509/dummy"),
})
}
async fn mount_token(server: &MockServer) {
Mock::given(method("POST"))
.and(path(AUTH_TOKEN_PATH))
.respond_with(ResponseTemplate::new(200).set_body_json(fake_token()))
.mount(server)
.await;
}
async fn build_source(
server: &MockServer,
config: BigQuerySourceConfig,
) -> (BigQuerySource, tempfile::NamedTempFile) {
let sa_json = dummy_service_account_json(&server.uri());
let sa_file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(
sa_file.path(),
serde_json::to_string_pretty(&sa_json).unwrap(),
)
.unwrap();
let client = ClientBuilder::new()
.with_auth_base_url(format!("{}{AUTH_SCOPE_BASE}", server.uri()))
.with_v2_base_url(server.uri())
.build_from_service_account_key_file(sa_file.path().to_str().unwrap())
.await
.expect("build bigquery client against mock");
(BigQuerySource::from_parts(config, client), sa_file)
}
fn schema_two_cols() -> Value {
json!({
"fields": [
{"name": "id", "type": "INTEGER"},
{"name": "name", "type": "STRING"}
]
})
}
fn rows(start: i64, count: i64) -> Vec<Value> {
(start..start + count)
.map(|i| json!({"f": [{"v": i.to_string()}, {"v": format!("name-{i}")}]}))
.collect()
}
fn default_config() -> BigQuerySourceConfig {
BigQuerySourceConfig::new(
PROJECT_ID,
BigQueryCredentials::ApplicationDefault,
"SELECT id, name FROM events",
)
.with_batch_size(10)
}
#[tokio::test]
async fn fetch_all_returns_rows_when_first_response_is_complete() {
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-1"},
"rows": rows(0, 3),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let rows = src.fetch_all().await.unwrap();
assert_eq!(rows.len(), 3);
assert_eq!(rows[0], json!({"id": 0, "name": "name-0"}));
assert_eq!(rows[2], json!({"id": 2, "name": "name-2"}));
}
#[tokio::test]
async fn fetch_all_paginates_via_page_token() {
let server = MockServer::start().await;
mount_token(&server).await;
// jobs.query: first page (rows 0..5) + pageToken.
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-page"},
"rows": rows(0, 5),
"pageToken": "tok-1",
})))
.mount(&server)
.await;
// getQueryResults: second page (rows 5..10) — final page (no further token).
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-page")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"rows": rows(5, 5),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let rows = src.fetch_all().await.unwrap();
assert_eq!(rows.len(), 10);
assert_eq!(rows[0]["id"], 0);
assert_eq!(rows[9]["id"], 9);
}
#[tokio::test]
async fn stream_pages_chunks_by_batch_size() {
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-stream"},
"rows": rows(0, 13),
})))
.mount(&server)
.await;
// batch_size = 5 → 13 rows yield 3 pages of [5, 5, 3].
let (src, _f) = build_source(&server, default_config().with_batch_size(5)).await;
let ctx = HashMap::new();
let mut sizes = Vec::new();
let mut pages = src.stream_pages(&ctx, 0);
while let Some(page) = pages.next().await {
let page = page.unwrap();
assert!(page.bookmark.is_none());
sizes.push(page.records.len());
}
assert_eq!(sizes, vec![5, 5, 3]);
}
#[tokio::test]
async fn stream_pages_batch_size_zero_emits_one_page() {
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-flat"},
"rows": rows(0, 7),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config().with_batch_size(0)).await;
let ctx = HashMap::new();
let pages: Vec<_> = src.stream_pages(&ctx, 0).collect().await;
assert_eq!(pages.len(), 1);
assert_eq!(pages[0].as_ref().unwrap().records.len(), 7);
}
#[tokio::test]
async fn polls_until_job_complete_when_first_response_is_incomplete() {
let server = MockServer::start().await;
mount_token(&server).await;
// jobs.query returns jobComplete=false (statement still running).
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": false,
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-poll"},
})))
.mount(&server)
.await;
// First poll: still not complete.
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-poll")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": false,
})))
.up_to_n_times(1)
.mount(&server)
.await;
// Subsequent poll: complete with rows.
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-poll")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"rows": rows(0, 2),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let rows = src.fetch_all().await.unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0]["id"], 0);
}
#[tokio::test]
async fn submits_positional_bindings_for_params_and_context() {
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-binds"},
"rows": rows(0, 1),
})))
.mount(&server)
.await;
let cfg = default_config()
.with_params(vec![json!("us-east")])
.with_location("EU");
let mut cfg = cfg;
cfg.query = "SELECT * FROM events WHERE region = ? AND id > {parent.min_id}".into();
let (src, _f) = build_source(&server, cfg).await;
let mut ctx = HashMap::new();
ctx.insert("parent.min_id".to_string(), json!(100));
src.fetch_with_context(&ctx).await.unwrap();
let reqs = server.received_requests().await.unwrap();
let query_req = reqs
.iter()
.find(|r| r.url.path().ends_with("/queries"))
.expect("captured jobs.query request");
let body: Value = serde_json::from_slice(&query_req.body).unwrap();
assert_eq!(
body["query"],
"SELECT * FROM events WHERE region = ? AND id > ?"
);
assert_eq!(body["parameterMode"], "POSITIONAL");
let params = body["queryParameters"].as_array().unwrap();
assert_eq!(params.len(), 2);
assert_eq!(params[0]["parameterValue"]["value"], "us-east");
assert_eq!(params[1]["parameterValue"]["value"], "100");
assert_eq!(body["location"], "EU");
}
#[tokio::test]
async fn fetch_all_decodes_all_bigquery_types_to_json() {
// Type-aware decoding end-to-end through `jobs.query`: INTEGER, FLOAT,
// BOOL, STRING, TIMESTAMP, a nested RECORD, and a REPEATED column all
// round-trip into the expected JSON shapes.
let server = MockServer::start().await;
mount_token(&server).await;
let schema = json!({
"fields": [
{"name": "id", "type": "INTEGER"},
{"name": "ratio", "type": "FLOAT"},
{"name": "active", "type": "BOOLEAN"},
{"name": "label", "type": "STRING"},
{"name": "ts", "type": "TIMESTAMP"},
{"name": "tags", "type": "STRING", "mode": "REPEATED"},
{
"name": "owner",
"type": "RECORD",
"fields": [
{"name": "uid", "type": "INTEGER"},
{"name": "email", "type": "STRING"}
]
}
]
});
// A single row exercising every cell shape BigQuery returns.
let row = json!({"f": [
{"v": "42"},
{"v": "2.5"},
{"v": "true"},
{"v": "hello"},
{"v": "1.7e9"},
{"v": [{"v": "a"}, {"v": "b"}]},
{"v": {"f": [{"v": "7"}, {"v": "x@y.z"}]}}
]});
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema,
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-types"},
"rows": [row],
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let rows = src.fetch_all().await.unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(
rows[0],
json!({
"id": 42,
"ratio": 2.5,
"active": true,
"label": "hello",
"ts": "1.7e9",
"tags": ["a", "b"],
"owner": {"uid": 7, "email": "x@y.z"}
})
);
}
#[tokio::test]
async fn fetch_all_paginates_across_three_pages_via_get_query_results() {
// jobs.query → page 1 (+tok-1); getQueryResults → page 2 (+tok-2);
// getQueryResults → page 3 (no token, final). All three pages flow
// through the getQueryResults polling/paging loop.
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-3pages"},
"rows": rows(0, 4),
"pageToken": "tok-1",
})))
.mount(&server)
.await;
// Second page carries a further pageToken so the loop iterates again.
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-3pages")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"rows": rows(4, 4),
"pageToken": "tok-2",
})))
.up_to_n_times(1)
.mount(&server)
.await;
// Final page: no pageToken.
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-3pages")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"rows": rows(8, 2),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let rows = src.fetch_all().await.unwrap();
assert_eq!(rows.len(), 10);
assert_eq!(rows[0]["id"], 0);
assert_eq!(rows[4]["id"], 4);
assert_eq!(rows[9]["id"], 9);
assert_eq!(rows[9]["name"], "name-9");
}
#[tokio::test]
async fn stream_pages_initial_response_without_rows_then_paginates() {
// jobs.query returns a complete-but-rowless first response carrying only a
// pageToken (exercises `rows_from_response_owned`'s None branch); the
// actual rows arrive via getQueryResults.
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-norows"},
"pageToken": "tok-1",
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-norows")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"rows": rows(0, 3),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config().with_batch_size(0)).await;
let ctx = HashMap::new();
let pages: Vec<_> = src.stream_pages(&ctx, 0).collect().await;
assert_eq!(pages.len(), 1);
let page = pages[0].as_ref().unwrap();
assert_eq!(page.records.len(), 3);
assert_eq!(page.records[0], json!({"id": 0, "name": "name-0"}));
}
#[tokio::test]
async fn fetch_all_errors_when_job_query_returns_non_2xx() {
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(403).set_body_json(json!({
"error": {"code": 403, "message": "Access Denied"}
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
match src.fetch_all().await {
Err(faucet_core::FaucetError::Source(m)) => {
assert!(m.contains("jobs.query failed"), "got: {m}");
}
other => panic!("expected Source error, got: {other:?}"),
}
}
#[tokio::test]
async fn fetch_all_errors_when_get_query_results_returns_non_2xx() {
let server = MockServer::start().await;
mount_token(&server).await;
// First page completes with a token, forcing a getQueryResults call.
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-getfail"},
"rows": rows(0, 1),
"pageToken": "tok-1",
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path(format!("/projects/{PROJECT_ID}/queries/job-getfail")))
.respond_with(ResponseTemplate::new(500).set_body_string("boom"))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
match src.fetch_all().await {
Err(faucet_core::FaucetError::Source(m)) => {
assert!(m.contains("getQueryResults failed"), "got: {m}");
}
other => panic!("expected Source error, got: {other:?}"),
}
}
#[tokio::test]
async fn fetch_all_errors_when_job_reference_missing() {
// A complete 200 response with rows but no jobReference must surface a
// typed Source error rather than silently proceeding.
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"rows": rows(0, 1),
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
match src.fetch_all().await {
Err(faucet_core::FaucetError::Source(m)) => {
assert!(m.contains("missing jobReference"), "got: {m}");
}
other => panic!("expected Source error, got: {other:?}"),
}
}
#[tokio::test]
async fn submits_typed_bool_and_float_param_types() {
// `bq_param_type` infers BOOL / FLOAT64 / INT64 from the JSON value so a
// numeric/boolean bind isn't forced to STRING. A JSON null becomes a typed
// NULL with the value omitted.
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"schema": schema_two_cols(),
"jobReference": {"projectId": PROJECT_ID, "jobId": "job-typed"},
"rows": rows(0, 1),
})))
.mount(&server)
.await;
let cfg = default_config().with_params(vec![json!(true), json!(2.5), json!(7), json!(null)]);
let (src, _f) = build_source(&server, cfg).await;
src.fetch_all().await.unwrap();
let reqs = server.received_requests().await.unwrap();
let query_req = reqs
.iter()
.find(|r| r.url.path().ends_with("/queries"))
.expect("captured jobs.query request");
let body: Value = serde_json::from_slice(&query_req.body).unwrap();
let params = body["queryParameters"].as_array().unwrap();
assert_eq!(params.len(), 4);
assert_eq!(params[0]["parameterType"]["type"], "BOOL");
assert_eq!(params[0]["parameterValue"]["value"], "true");
assert_eq!(params[1]["parameterType"]["type"], "FLOAT64");
assert_eq!(params[1]["parameterValue"]["value"], "2.5");
assert_eq!(params[2]["parameterType"]["type"], "INT64");
assert_eq!(params[2]["parameterValue"]["value"], "7");
// JSON null → typed STRING NULL: value omitted from the value object.
assert_eq!(params[3]["parameterType"]["type"], "STRING");
assert!(
params[3]["parameterValue"]["value"].is_null(),
"null bind must omit the value: {:?}",
params[3]
);
}
#[tokio::test]
async fn connector_name_schema_and_dataset_uri_are_exposed() {
let server = MockServer::start().await;
mount_token(&server).await;
let mut cfg = default_config();
cfg.query = "SELECT 1".into();
let (src, _f) = build_source(&server, cfg).await;
assert_eq!(src.connector_name(), "bigquery");
assert_eq!(src.dataset_uri(), "bigquery://test-project?query=SELECT 1");
let schema = src.config_schema();
// schema_for! emits an object schema describing the config struct.
assert!(
schema.get("properties").is_some() || schema.get("$ref").is_some(),
"config_schema should be a JSON Schema object: {schema}"
);
}
#[tokio::test]
async fn check_probe_passes_on_successful_dry_run() {
use faucet_core::check::{CheckContext, ProbeStatus};
let server = MockServer::start().await;
mount_token(&server).await;
// The doctor probe submits the query with dryRun=true to the same
// jobs.query endpoint; a 200 → a passing probe.
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"jobComplete": true,
"jobReference": {"projectId": PROJECT_ID, "jobId": "dry"},
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let report = src.check(&CheckContext::default()).await.unwrap();
assert_eq!(report.probes.len(), 1);
assert_eq!(report.probes[0].name, "query");
assert!(
matches!(report.probes[0].status, ProbeStatus::Pass),
"expected Pass, got: {:?}",
report.probes[0].status
);
// And the submitted body really did carry dryRun=true.
let reqs = server.received_requests().await.unwrap();
let query_req = reqs
.iter()
.find(|r| r.url.path().ends_with("/queries"))
.expect("captured dry-run request");
let body: Value = serde_json::from_slice(&query_req.body).unwrap();
assert_eq!(body["dryRun"], true);
}
#[tokio::test]
async fn check_probe_fails_on_dry_run_error() {
use faucet_core::check::{CheckContext, ProbeStatus};
let server = MockServer::start().await;
mount_token(&server).await;
Mock::given(method("POST"))
.and(path(format!("/projects/{PROJECT_ID}/queries")))
.respond_with(ResponseTemplate::new(400).set_body_json(json!({
"error": {"code": 400, "message": "Syntax error"}
})))
.mount(&server)
.await;
let (src, _f) = build_source(&server, default_config()).await;
let report = src.check(&CheckContext::default()).await.unwrap();
assert_eq!(report.probes.len(), 1);
match &report.probes[0].status {
ProbeStatus::Fail { reason } => {
assert!(reason.contains("dry-run failed"), "got: {reason}");
}
other => panic!("expected Fail, got: {other:?}"),
}
assert!(
report.probes[0].hint.is_some(),
"a failing probe should carry a remediation hint"
);
}