`dispatcher_e2e` failed about one run in three, on a different test each time, and never under `--test-threads=1`. Root cause, not timing: Each test calls `build_app`, which spawns a REAL dispatcher, and they all shared one database — so they shared one `outbox`. `OutboxRepo::claim_due` is deliberately not app-scoped (in production one dispatcher serves the whole instance, so claiming any due row is correct). Test A's dispatcher would therefore claim test B's row, test A would finish, its server would be dropped mid-dispatch, and the claim was stranded. Test B polled for its handler's side effect until it timed out. The harness was modelling something production never does: N independent instances sharing one database. So the fix belongs in the harness. Weakening the dispatcher (app-scoping `claim_due`) would be wrong, and shortening `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` to fit a 10s test window would make production abandon the live claims of scripts that may legitimately run 300s. `tests/common` now hands each test its own database. Replaying 73 migrations per test is too slow, so the first caller builds a migrated TEMPLATE database once and every test clones it (`CREATE DATABASE ... TEMPLATE` is a file copy), guarded by a cluster-wide advisory lock because Postgres will not clone a template that has an open connection. Names are derived from (suite, test), so a test reclaims its own database on entry: a crashed run leaves nothing to clean up. Only the five local `pool_or_skip` wrappers change — all 33 call sites are untouched, and the skip-when-`DATABASE_URL`-is-unset contract is preserved. (`#[sqlx::test]` already does this and is what `api.rs`/`authz.rs` use, but it requires `DATABASE_URL` and would force `#[ignore]`, silently dropping these tests from the default `cargo test --workspace` gate.) This also stops the e2e tests leaving stranded claims behind in the shared dev database, which is the likely source of the occasional lone journey failure. Before: dispatcher_e2e red ~1 run in 3. After: workspace green twice over, journeys 157/157. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
267 lines
8.0 KiB
Rust
267 lines
8.0 KiB
Rust
//! v1.1.9 invoke end-to-end tests. invoke()/invoke_async() exercised
|
|
//! against the full all-in-one app via build_app, so the invoke service
|
|
//! + Engine back-reference + InvokeServiceImpl all participate.
|
|
//!
|
|
//! The caller scripts run via the admin bypass `POST /api/v1/execute/{id}`
|
|
//! (same pattern dispatcher_e2e uses) — sidesteps the per-app domain /
|
|
//! route matcher entirely.
|
|
//!
|
|
//! Skips when DATABASE_URL is unset.
|
|
|
|
#![allow(clippy::needless_pass_by_value)]
|
|
|
|
use std::time::Duration;
|
|
|
|
use axum_test::TestServer;
|
|
use serde_json::{json, Value};
|
|
use sqlx::PgPool;
|
|
use uuid::Uuid;
|
|
|
|
mod common;
|
|
|
|
async fn pool_or_skip() -> Option<PgPool> {
|
|
common::test_pool("invoke_e2e").await
|
|
}
|
|
|
|
async fn server_for(pool: PgPool, suffix: &str) -> (TestServer, String) {
|
|
use picloud_manager_core::auth::hash_password;
|
|
use picloud_shared::InstanceRole;
|
|
|
|
let unique = format!("{suffix}-{}", Uuid::new_v4().simple());
|
|
let auth = picloud::AuthDeps::from_pool(pool.clone());
|
|
let username = format!("e2e-{unique}");
|
|
let hash = hash_password("pw").expect("hash");
|
|
auth.users
|
|
.create(&username, &hash, InstanceRole::Owner, None)
|
|
.await
|
|
.expect("seed admin");
|
|
|
|
let app = picloud::build_app(
|
|
pool,
|
|
auth,
|
|
picloud_shared::MasterKey::from_bytes([0x42u8; 32]),
|
|
)
|
|
.await
|
|
.expect("build_app");
|
|
let mut server = TestServer::new(app).expect("TestServer");
|
|
let resp = server
|
|
.post("/api/v1/admin/auth/login")
|
|
.json(&json!({ "username": username, "password": "pw" }))
|
|
.await;
|
|
resp.assert_status_ok();
|
|
let token = resp.json::<Value>()["token"]
|
|
.as_str()
|
|
.expect("login token")
|
|
.to_string();
|
|
server.add_header("authorization", format!("Bearer {token}"));
|
|
|
|
let slug = format!("e2e-{unique}");
|
|
let created: Value = server
|
|
.post("/api/v1/admin/apps")
|
|
.json(&json!({ "slug": slug, "name": slug }))
|
|
.await
|
|
.json();
|
|
let app_id = created["id"].as_str().expect("app id").to_string();
|
|
(server, app_id)
|
|
}
|
|
|
|
async fn create_script(server: &TestServer, app_id: &str, name: &str, source: &str) -> String {
|
|
let created: Value = server
|
|
.post("/api/v1/admin/scripts")
|
|
.json(&json!({ "app_id": app_id, "name": name, "source": source }))
|
|
.await
|
|
.json();
|
|
created["id"].as_str().expect("script id").to_string()
|
|
}
|
|
|
|
/// Direct executor bypass — same as dispatcher_e2e::execute. The script
|
|
/// runs sync with an empty body; sidesteps the route/host matcher.
|
|
async fn execute(server: &TestServer, script_id: &str) -> Value {
|
|
server
|
|
.post(&format!("/api/v1/execute/{script_id}"))
|
|
.json(&json!({}))
|
|
.await
|
|
.json()
|
|
}
|
|
|
|
async fn poll_marker(pool: &PgPool, app_id: &str) -> Option<Value> {
|
|
for _ in 0..150 {
|
|
let row: Option<(Value,)> = sqlx::query_as(
|
|
"SELECT value FROM kv_entries WHERE app_id = $1 \
|
|
AND collection = 'e2e_markers' AND key = 'marker'",
|
|
)
|
|
.bind(Uuid::parse_str(app_id).expect("uuid"))
|
|
.fetch_optional(pool)
|
|
.await
|
|
.expect("kv read");
|
|
if let Some((v,)) = row {
|
|
return Some(v);
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
}
|
|
None
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn invoke_by_name_same_app_returns_value() {
|
|
let Some(pool) = pool_or_skip().await else {
|
|
return;
|
|
};
|
|
let (server, app_id) = server_for(pool.clone(), "invoke-name").await;
|
|
|
|
// Callee returns 42 — the simplest possible round-trip.
|
|
create_script(
|
|
&server,
|
|
&app_id,
|
|
"callee",
|
|
r"#{ statusCode: 200, body: 42 }",
|
|
)
|
|
.await;
|
|
|
|
// Caller invokes by name and surfaces the result via the response body.
|
|
let caller = create_script(
|
|
&server,
|
|
&app_id,
|
|
"caller",
|
|
r#"
|
|
let r = invoke("callee", #{});
|
|
#{ statusCode: 200, body: r }
|
|
"#,
|
|
)
|
|
.await;
|
|
|
|
let body = execute(&server, &caller).await;
|
|
assert_eq!(body, json!(42), "callee return value should round-trip");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn invoke_cross_app_rejects() {
|
|
let Some(pool) = pool_or_skip().await else {
|
|
return;
|
|
};
|
|
let (server, app_a) = server_for(pool.clone(), "invoke-xa").await;
|
|
// Create a second app via the SAME server / admin (Owner can manage
|
|
// every app) so the script-create calls don't run into role gates.
|
|
let slug_b = format!("e2e-invoke-xb-{}", Uuid::new_v4().simple());
|
|
let created_b: Value = server
|
|
.post("/api/v1/admin/apps")
|
|
.json(&json!({ "slug": slug_b, "name": slug_b }))
|
|
.await
|
|
.json();
|
|
let app_b = created_b["id"].as_str().expect("app_b id").to_string();
|
|
|
|
// Callee lives in app_b — caller in app_a must not be able to reach it.
|
|
let callee_b = create_script(
|
|
&server,
|
|
&app_b,
|
|
"callee_b",
|
|
r"#{ statusCode: 200, body: 0 }",
|
|
)
|
|
.await;
|
|
|
|
// Caller in app_a tries to invoke callee_b by id; the error must
|
|
// surface to the response body.
|
|
let caller_src = format!(
|
|
r#"
|
|
let err = "no error";
|
|
try {{
|
|
invoke("{callee_b}", #{{}});
|
|
}} catch(e) {{
|
|
err = e;
|
|
}}
|
|
#{{ statusCode: 200, body: err }}
|
|
"#
|
|
);
|
|
let caller_a = create_script(&server, &app_a, "caller_a", &caller_src).await;
|
|
let body = execute(&server, &caller_a).await;
|
|
let err_str = body.as_str().unwrap_or_default();
|
|
assert!(
|
|
err_str.contains("different app")
|
|
|| err_str.contains("CrossApp")
|
|
|| err_str.contains("not found"),
|
|
"expected cross-app or not-found rejection, got: {err_str}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn invoke_depth_limit_exceeds_cleanly() {
|
|
let Some(pool) = pool_or_skip().await else {
|
|
return;
|
|
};
|
|
let (server, app_id) = server_for(pool.clone(), "invoke-depth").await;
|
|
|
|
// The recurser propagates any throw (no try-catch). At the depth
|
|
// ceiling Limits::trigger_depth_max=8 the innermost invoke() throws
|
|
// "invoke: depth limit exceeded (max 8)"; each nested invoke() then
|
|
// re-throws up the stack until the OUTER caller's try-catch surfaces
|
|
// it.
|
|
create_script(
|
|
&server,
|
|
&app_id,
|
|
"recurser",
|
|
r#"invoke("recurser", #{}); #{ statusCode: 200 }"#,
|
|
)
|
|
.await;
|
|
|
|
let caller = create_script(
|
|
&server,
|
|
&app_id,
|
|
"depth_caller",
|
|
r#"
|
|
let err = "ok";
|
|
try {
|
|
invoke("recurser", #{});
|
|
} catch(e) {
|
|
err = e;
|
|
}
|
|
#{ statusCode: 200, body: err }
|
|
"#,
|
|
)
|
|
.await;
|
|
|
|
let body = execute(&server, &caller).await;
|
|
let err = body.as_str().unwrap_or_default();
|
|
assert!(
|
|
err.contains("depth"),
|
|
"expected depth-limit error, got: {err}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn invoke_async_enqueues_outbox_row() {
|
|
let Some(pool) = pool_or_skip().await else {
|
|
return;
|
|
};
|
|
let (server, app_id) = server_for(pool.clone(), "invoke-async").await;
|
|
|
|
// Callee writes a marker the test polls — the dispatcher must fire
|
|
// the OutboxSourceKind::Invoke row.
|
|
create_script(
|
|
&server,
|
|
&app_id,
|
|
"async_callee",
|
|
r#"
|
|
kv::collection("e2e_markers").set("marker", #{ from: "async" });
|
|
#{ statusCode: 200 }
|
|
"#,
|
|
)
|
|
.await;
|
|
|
|
let caller = create_script(
|
|
&server,
|
|
&app_id,
|
|
"async_caller",
|
|
r#"
|
|
let id = invoke_async("async_callee", #{});
|
|
#{ statusCode: 200, body: id }
|
|
"#,
|
|
)
|
|
.await;
|
|
execute(&server, &caller).await;
|
|
|
|
let marker = poll_marker(&pool, &app_id)
|
|
.await
|
|
.expect("dispatcher should have fired the invoke_async row");
|
|
assert_eq!(marker["from"], "async");
|
|
}
|