tensor-wasm-api 0.3.8

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Craton Software Company

//! Integration tests for PATH-TO-V1 v0.4 audit log.
//!
//! Verified behaviours:
//!
//! 1. A successful `POST /functions` produces a record with
//!    `action: "create_function"`, the right resource shape, and
//!    `outcome.status_code: 200`.
//! 2. `POST /functions/{id}/invoke` against an out-of-scope tenant
//!    produces a record with `outcome.status_code: 403` and
//!    `outcome.error_kind: "tenant_scope_denied"`.
//! 3. `GET /healthz` produces no audit record at all (read-only routes
//!    are suppressed at the route-filter stage).
//! 4. `TENSOR_WASM_API_AUDIT_LOG=none` semantics — the `NoopSink`
//!    swallows every record without panicking.
//! 5. Serialised records round-trip cleanly through `serde_json::Value`,
//!    confirming the public wire schema.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use axum::body::Body;
use axum::http::{Method, Request, StatusCode};
use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine;
use http_body_util::BodyExt;
use serde_json::{json, Value};
use tensor_wasm_api::{
    build_router_with_audit, AppState, AuditConfig, AuditRecord, AuditSink, AuthConfig, CorsConfig,
    NoopSink, RateLimitConfig, RateLimiter, TenantConfig, TokenScope, HEADER_TENANT,
};
use tensor_wasm_core::types::TenantId;
use tower::ServiceExt;

/// Test sink that buffers every emitted record so the test can assert
/// on the structured contents instead of inspecting stdout.
#[derive(Debug, Default)]
struct CapturingSink {
    records: Mutex<Vec<AuditRecord>>,
}

impl CapturingSink {
    fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    fn snapshot(&self) -> Vec<AuditRecord> {
        self.records.lock().expect("records mutex").clone()
    }
}

impl AuditSink for CapturingSink {
    fn emit(&self, record: &AuditRecord) {
        self.records
            .lock()
            .expect("records mutex")
            .push(record.clone());
    }
}

async fn body_json(body: Body) -> Value {
    let bytes = body.collect().await.expect("body").to_bytes().to_vec();
    serde_json::from_slice(&bytes).expect("body is JSON")
}

/// Build the production router with explicit scopes and a capturing
/// audit sink. Returns the router and a handle to the sink so the test
/// can read back what it emitted.
fn router_with_audit(scopes: &[(&str, TokenScope)]) -> (axum::Router, Arc<CapturingSink>) {
    let mut map: HashMap<String, TokenScope> = HashMap::new();
    for (k, v) in scopes {
        map.insert((*k).to_owned(), v.clone());
    }
    let auth = AuthConfig::from_scopes(map);
    let sink: Arc<CapturingSink> = CapturingSink::new();
    let audit_cfg = AuditConfig::from_sink(sink.clone() as Arc<dyn AuditSink>);
    let router = build_router_with_audit(
        Arc::new(AppState::default()),
        auth,
        TenantConfig::default(),
        RateLimiter::new(RateLimitConfig::disabled()),
        audit_cfg,
        CorsConfig::default(),
    );
    (router, sink)
}

async fn deploy_trivial_function(router: &axum::Router, bearer: &str, tenant: u64) -> String {
    let wasm_bytes = wat::parse_str(r#"(module (func (export "_start")))"#).expect("WAT parses");
    let wasm_b64 = BASE64.encode(&wasm_bytes);
    let req = Request::builder()
        .method(Method::POST)
        .uri("/functions")
        .header("authorization", format!("Bearer {bearer}"))
        .header("content-type", "application/json")
        .header(HEADER_TENANT, tenant.to_string())
        .body(Body::from(
            serde_json::to_vec(&json!({ "name": "t", "wasm_b64": wasm_b64 })).unwrap(),
        ))
        .unwrap();
    let resp = router.clone().oneshot(req).await.expect("deploy");
    assert_eq!(resp.status(), StatusCode::OK, "deploy must succeed");
    body_json(resp.into_body())
        .await
        .get("id")
        .and_then(Value::as_str)
        .map(str::to_owned)
        .expect("id")
}

#[tokio::test]
async fn create_function_emits_audit_record() {
    let (router, sink) = router_with_audit(&[("scoped", TokenScope::all())]);

    let _id = deploy_trivial_function(&router, "scoped", 0).await;
    let records = sink.snapshot();
    assert_eq!(records.len(), 1, "exactly one record per POST /functions");
    let rec = &records[0];

    // Serialise + re-parse so the assertions speak the wire shape.
    let v: Value = serde_json::to_value(rec).expect("record serialises");
    assert_eq!(v["action"], "create_function");
    assert_eq!(v["outcome"]["status_code"], 200);
    assert!(
        v["outcome"].get("error_kind").is_none(),
        "successful call must not carry error_kind: {v}",
    );
    assert_eq!(v["actor"]["kind"], "bearer");
    assert_eq!(v["actor"]["scope"]["kind"], "wildcard");
    // POST /functions has no function id in the URL — handler assigns
    // it. The record should reflect that absence.
    assert!(v["resource"].get("function_id").is_none(), "got {v}");
}

#[tokio::test]
async fn invoke_with_out_of_scope_tenant_emits_403_record() {
    // Token "scoped" may only address tenants 1 and 2.
    let (router, sink) = router_with_audit(&[(
        "scoped",
        TokenScope::from_tenants([TenantId(1), TenantId(2)]),
    )]);

    // Deploy under tenant 1 (in scope) so the deploy itself is authorized;
    // the invoke below targets out-of-scope tenant 3.
    let id = deploy_trivial_function(&router, "scoped", 1).await;
    // Reset the buffer so we focus on just the invoke record.
    sink.records.lock().unwrap().clear();

    // Invoke with tenant=3 → must 403 with kind = tenant_scope_denied.
    let req = Request::builder()
        .method(Method::POST)
        .uri(format!("/functions/{id}/invoke"))
        .header("authorization", "Bearer scoped")
        .header("content-type", "application/json")
        .header(HEADER_TENANT, "3")
        .body(Body::from(serde_json::to_vec(&json!({})).unwrap()))
        .unwrap();
    let resp = router.clone().oneshot(req).await.expect("invoke");
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);

    let records = sink.snapshot();
    assert_eq!(records.len(), 1, "exactly one record per invoke");
    let v: Value = serde_json::to_value(&records[0]).expect("serialises");
    assert_eq!(v["action"], "invoke_function");
    assert_eq!(v["outcome"]["status_code"], 403);
    assert_eq!(v["outcome"]["error_kind"], "tenant_scope_denied");
    assert_eq!(v["resource"]["tenant_id"], 3);
    // Function id was in the URL — record must carry it.
    let function_id = v["resource"]["function_id"]
        .as_str()
        .expect("function_id present");
    assert_eq!(function_id, id, "function id round-trips");
}

#[tokio::test]
async fn healthz_emits_no_audit_record() {
    let (router, sink) = router_with_audit(&[("scoped", TokenScope::all())]);

    for _ in 0..5 {
        let req = Request::builder()
            .method(Method::GET)
            .uri("/healthz")
            .header("authorization", "Bearer scoped")
            .body(Body::empty())
            .unwrap();
        let resp = router.clone().oneshot(req).await.expect("healthz");
        assert_eq!(resp.status(), StatusCode::OK);
    }
    assert!(
        sink.snapshot().is_empty(),
        "GET /healthz must never emit an audit record: got {:?}",
        sink.snapshot(),
    );
}

#[tokio::test]
async fn metrics_and_jobs_get_emit_no_audit_record() {
    // GET /metrics and GET /jobs/{id} are also read-only and must be
    // filtered out at the route-shape stage.
    let (router, sink) = router_with_audit(&[("scoped", TokenScope::all())]);

    let req = Request::builder()
        .method(Method::GET)
        .uri("/metrics")
        .header("authorization", "Bearer scoped")
        .body(Body::empty())
        .unwrap();
    let resp = router.clone().oneshot(req).await.expect("metrics");
    assert_eq!(resp.status(), StatusCode::OK);

    // A random job id will 404 — but it is still a GET, which must not
    // emit an audit record. The point of the assertion is that the
    // route-filter runs before the outcome is known.
    let req = Request::builder()
        .method(Method::GET)
        .uri("/jobs/00000000-0000-0000-0000-000000000000")
        .header("authorization", "Bearer scoped")
        .body(Body::empty())
        .unwrap();
    let _ = router.clone().oneshot(req).await.expect("jobs");

    assert!(sink.snapshot().is_empty(), "read-only routes are exempt");
}

#[tokio::test]
async fn audit_log_disabled_mode_swallows_records() {
    // TENSOR_WASM_API_AUDIT_LOG=none → NoopSink. The middleware still
    // runs (it stamps a request id, computes latency, classifies the
    // action) but nothing is observable. We can't easily *see* the
    // absence in stdout, so prove the property by giving the router
    // the explicit NoopSink and checking that the request still
    // succeeds and that the parallel CapturingSink-attached request
    // sees one record while this one sees nothing.
    let sink: Arc<CapturingSink> = CapturingSink::new();
    let audit_cfg = AuditConfig::from_sink(Arc::new(NoopSink::new()));
    let router = build_router_with_audit(
        Arc::new(AppState::default()),
        AuthConfig::from_tokens(["scoped"]),
        TenantConfig::default(),
        RateLimiter::new(RateLimitConfig::disabled()),
        audit_cfg,
        CorsConfig::default(),
    );

    // Drive a state-mutating call through the disabled router.
    let wasm_bytes = wat::parse_str(r#"(module (func (export "_start")))"#).unwrap();
    let wasm_b64 = BASE64.encode(&wasm_bytes);
    let req = Request::builder()
        .method(Method::POST)
        .uri("/functions")
        .header("authorization", "Bearer scoped")
        .header("content-type", "application/json")
        .body(Body::from(
            serde_json::to_vec(&json!({ "name": "t", "wasm_b64": wasm_b64 })).unwrap(),
        ))
        .unwrap();
    let resp = router.oneshot(req).await.expect("deploy");
    assert_eq!(resp.status(), StatusCode::OK, "deploy must still succeed");

    // The capturing sink we never attached must still be empty.
    assert!(
        sink.snapshot().is_empty(),
        "the unrelated capturing sink should never observe records",
    );
}

#[tokio::test]
async fn audit_record_round_trips_through_serde_json() {
    // The JSON shape is part of the public contract — round-tripping
    // proves the serialized form parses back into a structurally
    // equivalent shape.
    let (router, sink) = router_with_audit(&[("scoped", TokenScope::all())]);
    let _id = deploy_trivial_function(&router, "scoped", 0).await;
    let records = sink.snapshot();
    assert_eq!(records.len(), 1);
    let rec = &records[0];

    let line = serde_json::to_string(rec).expect("serialises");
    // Lines must not contain raw newlines — the on-wire contract is
    // JSONL (one record per line).
    assert!(
        !line.contains('\n'),
        "record contains an embedded newline: {line}",
    );
    let back: Value = serde_json::from_str(&line).expect("deserialises");
    assert_eq!(back["action"], "create_function");
    assert!(
        back["request_id"].as_str().is_some(),
        "request_id is a string",
    );
    assert!(back["ts_unix_ms"].as_u64().is_some(), "ts_unix_ms is a u64",);
    assert!(back["latency_ms"].as_u64().is_some(), "latency_ms is a u64",);
}

#[tokio::test]
async fn delete_function_emits_audit_record() {
    let (router, sink) = router_with_audit(&[("scoped", TokenScope::all())]);
    let id = deploy_trivial_function(&router, "scoped", 0).await;
    sink.records.lock().unwrap().clear();

    let req = Request::builder()
        .method(Method::DELETE)
        .uri(format!("/functions/{id}"))
        .header("authorization", "Bearer scoped")
        .body(Body::empty())
        .unwrap();
    let resp = router.clone().oneshot(req).await.expect("delete");
    assert_eq!(resp.status(), StatusCode::NO_CONTENT);
    let records = sink.snapshot();
    assert_eq!(records.len(), 1);
    let v: Value = serde_json::to_value(&records[0]).expect("serialises");
    assert_eq!(v["action"], "delete_function");
    assert_eq!(v["outcome"]["status_code"], 204);
    assert_eq!(
        v["resource"]["function_id"].as_str().map(str::to_owned),
        Some(id),
    );
}

#[tokio::test]
async fn invoke_async_emits_audit_record() {
    let (router, sink) = router_with_audit(&[("scoped", TokenScope::all())]);
    // Deploy under tenant 1 so the function is owned by tenant 1 and the
    // invoke-async below (also tenant 1) clears the per-resource owner check.
    let id = deploy_trivial_function(&router, "scoped", 1).await;
    sink.records.lock().unwrap().clear();

    let req = Request::builder()
        .method(Method::POST)
        .uri(format!("/functions/{id}/invoke-async"))
        .header("authorization", "Bearer scoped")
        .header("content-type", "application/json")
        .header(HEADER_TENANT, "1")
        .body(Body::from(serde_json::to_vec(&json!({})).unwrap()))
        .unwrap();
    let resp = router.clone().oneshot(req).await.expect("invoke-async");
    assert_eq!(resp.status(), StatusCode::ACCEPTED);
    let records = sink.snapshot();
    assert_eq!(records.len(), 1);
    let v: Value = serde_json::to_value(&records[0]).expect("serialises");
    assert_eq!(v["action"], "invoke_function_async");
    assert_eq!(v["outcome"]["status_code"], 202);
    assert_eq!(v["resource"]["tenant_id"], 1);
}