Plan: canopy-test-lib world-class testing port (closes #436)

crates/canopy-contracts-{service}/
    Cargo.toml          # See template below
    src/
        lib.rs          # pub mod {operation_family}; pub mod paths;
        {family}.rs     # Request/Response/sub-DTOs for one operation family
        paths.rs        # pub const {OPERATION}: &str = "/v1/...";
    tests/
        roundtrip.rs    # proptest serde round-trip for every DTO

# SPDX-License-Identifier: AGPL-3.0-or-later

[package]
name = "canopy-contracts-{service}"
version.workspace = true
edition.workspace = true
license.workspace = true

[dependencies]
canopy-common = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }      # only if any DTO carries serde_json::Value
chrono = { workspace = true }          # only if any DTO carries DateTime / NaiveDate
rust_decimal = { workspace = true }    # only if any DTO carries Decimal
utoipa = { workspace = true }

[dev-dependencies]
proptest = { workspace = true }
serde_json = { workspace = true }      # round-trip tests serialise to/from JSON

[workspace]
members = [
    # … existing members …
    "crates/canopy-contracts-eligibility",   # A1
    "crates/canopy-contracts-rules",         # A2
    "crates/canopy-contracts-persons",       # A2
    "crates/canopy-contracts-applications",  # A2
    "crates/canopy-contracts-verification",  # A2
    "crates/canopy-contracts-snap",          # A3
    "crates/canopy-contracts-tanf",          # A3
    "crates/canopy-contracts-medicaid",      # A3
    "crates/canopy-contracts-caps",          # A3
    "crates/canopy-contracts-wic",           # A3
    "crates/canopy-contracts-enrollment",    # A4
    "crates/canopy-contracts-renewals",      # A4
    "crates/canopy-contracts-notices",       # A4
    "crates/canopy-contracts-appeals",       # A4
    "crates/canopy-contracts-reporting",     # A4
    "crates/canopy-contracts-security",      # A4
]

[workspace.dependencies]
# … existing deps …
canopy-contracts-eligibility = { path = "crates/canopy-contracts-eligibility" }
canopy-contracts-rules = { path = "crates/canopy-contracts-rules" }
# … one entry per contracts crate, added in the same MR that adds the crate …

# Cross-cutting dev-deps added per-MR:
# - MR A1 adds: proptest (used by Phase A round-trip tests)
# - MR B1 adds: insta (snapshot testing from Phase B onwards)
# - MR D adds: paste (only needed by the define_ephemeral_schema_for! macro)
proptest = "1"
insta = { version = "1", features = ["yaml", "redactions", "filters"] }
paste = "1"

let rows = sqlx::query_as::<_, ApplicationRow>("SELECT …").fetch_all(pool).await?;
Ok(rows.into_iter().map(Application::from).collect())

use canopy_contracts_eligibility::determine::DetermineRequest;
use proptest::prelude::*;

proptest! {
    #[test]
    fn determine_request_serde_roundtrip(req in arb_determine_request()) {
        let json = serde_json::to_string(&req).expect("serialize");
        let parsed: DetermineRequest = serde_json::from_str(&json).expect("deserialize");
        prop_assert_eq!(req, parsed);
    }
}

fn arb_determine_request() -> impl Strategy<Value = DetermineRequest> {
    (
        any::<[u8; 16]>().prop_map(|bytes| ApplicationId(Uuid::from_bytes(bytes))),
        any::<[u8; 16]>().prop_map(|bytes| HouseholdId(Uuid::from_bytes(bytes))),
        prop::collection::vec("snap|tanf|medicaid|caps|wic", 1..5),
        any::<String>(),
    ).prop_map(|(application_id, household_id, programs, requested_by)| DetermineRequest {
        application_id, household_id, programs, requested_by,
    })
}

fn arb_program_result() -> impl Strategy<Value = ProgramResult> {
    (
        "snap|tanf|medicaid|caps|wic",
        "approved|denied|pending",
        prop::option::of(any::<i64>().prop_map(|n| Decimal::from(n / 100))),
        prop::option::of(any::<String>()),
    ).prop_map(|(program, status, benefit_amount, basis)| ProgramResult {
        program: program.into(), status: status.into(), benefit_amount, basis,
    })
}

fn arb_determine_response() -> impl Strategy<Value = DetermineResponse> {
    (
        any::<[u8; 16]>().prop_map(|bytes| EligibilityRequestId(Uuid::from_bytes(bytes))),
        any::<[u8; 16]>().prop_map(|bytes| ApplicationId(Uuid::from_bytes(bytes))),
        prop::collection::vec(arb_program_result(), 0..5),
        prop::collection::vec(arb_program_result(), 0..5),
        prop::collection::vec(arb_program_result(), 0..5),
        any::<i64>().prop_map(|n| Decimal::from(n / 100)),
        // ISO-8601 timestamp; restrict to a stable range to avoid year-9999 nondeterminism
        ("[12][0-9]{3}-[01][0-9]-[0-3][0-9]T[0-2][0-9]:[0-5][0-9]:[0-5][0-9]Z".prop_map(String::from)),
    ).prop_map(/* … construct DetermineResponse … */)
}

// SPDX-License-Identifier: AGPL-3.0-or-later

use canopy_contracts_eligibility::determine::{DetermineRequest, DetermineResponse};
use canopy_contracts_eligibility::paths;
use crate::client::{TestClient, TestApiError};

pub struct EligibilityClient {
    inner: TestClient,
}

impl EligibilityClient {
    pub fn new(base_url: &str) -> Self { Self { inner: TestClient::new(base_url) } }
    pub fn with_token(mut self, token: String) -> Self { self.inner = self.inner.with_token(token); self }

    /// Canonical handler: services/canopy-eligibility/src/api/handlers.rs:determine
    pub async fn determine(
        &self,
        req: &DetermineRequest,
    ) -> Result<DetermineResponse, TestApiError> {
        self.inner.post_json(paths::DETERMINE, req).await.into_typed()
    }
}

#[derive(Debug)]
pub struct TestApiError {
    pub status: u16,
    pub body: String,
}

impl TestResponse {
    pub fn into_typed<T: serde::de::DeserializeOwned>(self) -> Result<T, TestApiError> {
        if (200..300).contains(&self.status) {
            serde_json::from_slice(&self.body).map_err(|e| TestApiError {
                status: self.status,
                body: format!("deserialise failed: {e} — body: {}", String::from_utf8_lossy(&self.body)),
            })
        } else {
            Err(TestApiError { status: self.status, body: String::from_utf8_lossy(&self.body).into_owned() })
        }
    }
}

let resp = client.determine(&DetermineRequest { /* … */ }).await.expect("determine");
assert_eq!(resp.programs_approved.len(), 1);   // typed structural assertion
insta::assert_yaml_snapshot!("determine_snap_approved", resp);  // shape-drift guard

//! Deterministic clock control for time-sensitive tests.
//!
//! Tests that exercise scheduler ticks, retry backoffs, certification
//! periods, JWT expiry, advance-notice math, etc. should pause real
//! time and advance it manually. Avoid `tokio::time::sleep` in tests.

pub async fn pause() { tokio::time::pause().await }
pub async fn advance(d: std::time::Duration) { tokio::time::advance(d).await }

/// Run a future to completion while time is paused, advancing the
/// clock each time the future yields. Useful for testing retry
/// loops without sleeping real time.
pub async fn run_paused_advancing<F: Future>(fut: F, step: std::time::Duration) -> F::Output {
    pause().await;
    tokio::pin!(fut);
    loop {
        tokio::select! {
            v = &mut fut => return v,
            _ = tokio::time::sleep(step) => advance(step).await,
        }
    }
}

use std::ops::Range;
use std::time::Duration;

pub struct EvilLayer { /* composition state */ }

impl EvilLayer {
    pub fn new() -> Self;
    pub fn with_latency_jitter(self, range: Range<Duration>) -> Self;
    pub fn with_failure_rate(self, p: f64) -> Self;     // probability of synthesized 503
    pub fn drop_connection_after(self, n: u32) -> Self;
    pub fn tamper_payload<F>(self, f: F) -> Self
        where F: Fn(&mut serde_json::Value) + Send + Sync + 'static;
}

pub struct EvilProxyHandle {
    pub url: String,    // 127.0.0.1:{dynamic_port}
    _task: tokio::task::JoinHandle<()>,
}

pub fn evil_proxy(target_url: &str, layer: EvilLayer) -> EvilProxyHandle;

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

/// Capture-and-assert helpers for tests verifying that production
/// code emits the tracing spans + metrics operators rely on.

pub struct SpanCapture { /* tracing-subscriber test layer */ }

impl SpanCapture {
    pub fn install() -> Self;
    pub fn assert_span_emitted(&self, name: &str);
    pub fn assert_span_field(&self, span_name: &str, field: &str, value: &str);
    pub fn drain(self) -> Vec<CapturedSpan>;
}

pub struct MetricCapture { /* opentelemetry-stdout sink */ }

impl MetricCapture {
    pub fn install() -> Self;
    pub fn assert_metric_recorded(&self, name: &str);
    pub fn assert_metric_value(&self, name: &str, expected: f64);
}

// crates/canopy-test-lib/src/observability.rs
use std::sync::{Arc, Mutex};
use tracing::{Event, Subscriber};
use tracing::subscriber::DefaultGuard;
use tracing_subscriber::layer::{Context, Layer};
use tracing_subscriber::prelude::*;

#[derive(Clone)]
pub struct SpanCapture {
    events: Arc<Mutex<Vec<CapturedEvent>>>,
}

#[derive(Debug, Clone)]
pub struct CapturedEvent {
    pub target: String,
    pub level: tracing::Level,
    pub message: String,
    pub fields: std::collections::HashMap<String, String>,
}

impl SpanCapture {
    /// Install a per-test tracing subscriber. Returns a guard the caller
    /// MUST hold for the test scope; dropping the guard restores the
    /// previous default. Use:
    ///
    ///     let (capture, _guard) = SpanCapture::install_scoped();
    ///     /* … run code under test … */
    ///     capture.assert_span_emitted("event_process");
    ///
    /// Tests MUST run with `#[tokio::test(flavor = "current_thread")]`.
    /// On a multi-threaded runtime, `set_default` is thread-local — the
    /// subscriber doesn't reach work-stealing tasks on other threads.
    pub fn install_scoped() -> (Self, DefaultGuard) {
        let events = Arc::new(Mutex::new(Vec::new()));
        let layer = SpanCaptureLayer { events: events.clone() };
        let subscriber = tracing_subscriber::Registry::default().with(layer);
        let guard = tracing::subscriber::set_default(subscriber);
        (Self { events }, guard)
    }

    pub fn assert_span_emitted(&self, name: &str) {
        let events = self.events.lock().unwrap();
        let found = events.iter().any(|e| e.target == name || e.message.contains(name));
        assert!(found, "expected span '{name}' not emitted; events: {events:#?}");
    }

    pub fn assert_field(&self, span_name: &str, field: &str, value: &str) { /* … */ }
}

struct SpanCaptureLayer { events: Arc<Mutex<Vec<CapturedEvent>>> }

impl<S: Subscriber> Layer<S> for SpanCaptureLayer {
    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
        let mut visitor = FieldCollector::default();
        event.record(&mut visitor);
        self.events.lock().unwrap().push(CapturedEvent {
            target: event.metadata().target().to_string(),
            level: *event.metadata().level(),
            message: visitor.message.unwrap_or_default(),
            fields: visitor.fields,
        });
    }
}

// MetricCapture follows the same pattern — opentelemetry-stdout's
// in-memory exporter wrapped in a per-test scope guard.

pub struct MultiReplicaFixture {
    service: String,
    replicas: Vec<ReplicaHandle>,
    queue_prefix: String,
}

pub struct ReplicaHandle {
    pub port: u16,
    pub replica_id: String,
    process: tokio::process::Child,
}

impl MultiReplicaFixture {
    pub async fn spawn(service: &str, n: usize) -> Result<Self, FixtureError>;
    pub fn client_for<C: TypedClient>(&self, idx: usize) -> C;
    pub async fn kill_replica(&mut self, idx: usize);
    pub async fn restart_replica(&mut self, idx: usize) -> Result<(), FixtureError>;
}

CANOPY_{SERVICE_UPPER}__PORT={free_port}   # canopy-common settings overlay (no source change)
CANOPY_MQ_REPLICA_ID={service}-replica-{idx}
CANOPY_MQ_QUEUE_PREFIX={random_prefix}

let target = std::env::var("CARGO_TARGET_DIR").unwrap_or_else(|_| "target".to_string());
let binary = format!("{target}/debug/{service}");

// crates/canopy-test-lib/src/db.rs

//! Per-test schema isolation. Each test creates a unique Postgres schema,
//! scopes connections via `SET search_path`, and drops the schema at end.

pub struct EphemeralSchema {
    pub name: String,            // e.g. "test_a3f9b2"
    pool: sqlx::PgPool,
    base_url: String,            // for DROP SCHEMA cleanup
}

impl EphemeralSchema {
    pub fn pool(&self) -> &sqlx::PgPool { &self.pool }
}

impl Drop for EphemeralSchema {
    fn drop(&mut self) {
        let name = self.name.clone();
        let base_url = self.base_url.clone();
        // Schema drop runs in a detached task using the admin connection;
        // we cannot await inside Drop. Best-effort cleanup; periodic devstack
        // refresh sweeps any survivors.
        tokio::spawn(async move {
            if let Ok(admin) = sqlx::PgPool::connect(&base_url).await {
                let _ = sqlx::query(&format!("DROP SCHEMA {name} CASCADE")).execute(&admin).await;
            }
        });
    }
}

// Internal constructor that the per-service macro expands to:
async fn create_schema(base_url: &str, name: &str) -> Result<sqlx::PgPool, sqlx::Error> {
    let admin = sqlx::PgPool::connect(base_url).await?;
    sqlx::query(&format!("CREATE SCHEMA {name}")).execute(&admin).await?;
    let url = format!("{base_url}?options=-c%20search_path%3D{name}");
    sqlx::PgPool::connect(&url).await
}

#[macro_export]
macro_rules! define_ephemeral_schema_for {
    ($service:ident, $migrations:expr) => {
        impl $crate::db::EphemeralSchema {
            paste::paste! {
                pub async fn [<new_for_ $service>](base_url: &str) -> Result<Self, sqlx::Error> {
                    let suffix = uuid::Uuid::now_v7().to_string().replace('-', "");
                    let name = format!("test_{}", &suffix[..12]);
                    let pool = $crate::db::create_schema(base_url, &name).await?;
                    sqlx::migrate!($migrations).run(&pool).await?;
                    Ok(Self {
                        name,
                        pool,
                        base_url: base_url.to_string(),
                    })
                }
            }
        }
    };
}

// One-line per service (lives in crates/canopy-test-lib/src/db_constructors.rs):
define_ephemeral_schema_for!(eligibility, "../../services/canopy-eligibility/migrations");
define_ephemeral_schema_for!(snap,        "../../services/canopy-snap/migrations");
// … 14 more …

let schema = EphemeralSchema::new_for_eligibility(&cfg.eligibility_db_url).await?;
let pool = schema.pool();   // pool scoped to the ephemeral schema
// run test work; schema drops on test scope exit

let resp = client.determine(&req).await?;
insta::with_settings!({
    filters => vec![
        // UUID: case-insensitive to match both serde's default (lowercase) and any
        // upstream service that emits uppercase. UUID v7 IDs in canopy serialize
        // as standard 8-4-4-4-12 hex per canopy-common/src/id.rs.
        (r"(?i)\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", "[uuid]"),
        (r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})", "[timestamp]"),
    ]
}, {
    insta::assert_yaml_snapshot!("determine_snap_approved", resp);
});

[workspace.dependencies]
insta = { version = "1", features = ["yaml", "redactions", "filters"] }

[behavior]
# Tests fail unmatched snapshots rather than auto-creating (CI safety).
auto_review = "no"
auto_accept_unseen = false

[diff]
# Use the standard diff format; reviewers see snapshot changes in MR diffs.

//! Byte-comparison goldenfile testing for deterministic outputs
//! (canopy-notices PDFs today; future use cases land here too).

pub fn assert_matches_golden(actual: &[u8], golden_path: &Path) {
    // Hard guard: in CI, UPDATE_GOLDEN must not be set. Catches a
    // developer's shell-rc leak that would silently accept stale goldens.
    if std::env::var("CI").is_ok() && std::env::var("UPDATE_GOLDEN").is_ok() {
        panic!(
            "UPDATE_GOLDEN must not be set in CI \
             (would silently accept template changes without review)"
        );
    }

    if std::env::var("UPDATE_GOLDEN").is_ok() {
        std::fs::write(golden_path, actual).expect("write golden");
        return;
    }
    let expected = std::fs::read(golden_path).expect("read golden");
    if actual != expected.as_slice() {
        let diff_path = golden_path.with_extension("actual");
        std::fs::write(&diff_path, actual).expect("write actual");
        panic!(
            "goldenfile mismatch: {golden_path:?}\nactual written to {diff_path:?}\nset UPDATE_GOLDEN=1 to accept"
        );
    }
}

pub async fn run(args: CoverageArgs) -> anyhow::Result<()> {
    let threshold = args.threshold.unwrap_or(58.0);  // baseline - 2%, set after A1 baseline measurement
    let status = tokio::process::Command::new("cargo")
        .args([
            "llvm-cov", "--workspace", "--summary-only",
            "--fail-under-lines", &threshold.to_string(),
        ])
        .status().await?;
    if !status.success() { anyhow::bail!("coverage below {threshold}% threshold"); }
    Ok(())
}

coverage:
  stage: test
  script:
    - cargo install cargo-llvm-cov --locked   # cache via GitLab job artifacts
    - cargo xtask coverage
  rules:
    - if: $CI_MERGE_REQUEST_IID
  coverage: '/^TOTAL.*\s(\d+\.\d+)%\s/'        # GitLab regex parses coverage from output