cf-gears-toolkit 0.12.0

//! Full-stack end-to-end test for an `OoP` gear booted through
//! [`HostRuntime::run_oop_serving`] (`cpt-cf-component-oop-bootstrap`).
//!
//! Unlike the serve-edge test in `oop_serve_tests.rs` (which drives
//! `OopHttpServer` directly with a hand-built router), this test exercises the
//! *entire* lifecycle a real gear goes through:
//!
//! - gear discovery via a manually-built [`RegistryBuilder`],
//! - the full phase sequence (`pre_init` → `init` → proxy-wiring → `post_init`
//!   → `grpc` → `start`), matching the in-process order,
//! - [`HostRuntime::compose_oop_router`] building routes + `OpenAPI` from an
//!   actual [`RestApiCapability`] gear (not a synthetic router),
//! - a real bound HTTP server serving the gear's route,
//! - background self-registration with a live `DirectoryClient`,
//! - graceful shutdown draining and deregistration,
//! - the `stop` phase running after the server exits.

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

use async_trait::async_trait;
use axum::Router;
use axum::routing::get;
use parking_lot::Mutex;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use cf_system_sdks::directory::{
    DirectoryClient, RegisterInstanceInfo, ServiceEndpoint, ServiceInstanceInfo,
};

use super::{DbOptions, HostRuntime};
use crate::client_hub::ClientHub;
use crate::config::ConfigProvider;
use crate::context::GearCtx;
use crate::contracts::{Gear, OpenApiRegistry, RestApiCapability};
use crate::registry::RegistryBuilder;
use crate::runtime::OopServeOptions;

/// A minimal gear that serves a single REST route. Implements both [`Gear`]
/// (lifecycle) and [`RestApiCapability`] (routing) so it is composed into the
/// host-less router exactly like a production gear.
#[derive(Default)]
struct PingGear;

#[async_trait]
impl Gear for PingGear {
    async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
        Ok(())
    }
}

impl RestApiCapability for PingGear {
    fn register_rest(
        &self,
        _ctx: &GearCtx,
        router: Router,
        _openapi: &dyn OpenApiRegistry,
    ) -> anyhow::Result<Router> {
        Ok(router.route("/ping", get(|| async { "pong" })))
    }
}

/// In-memory `DirectoryClient` that records registration / deregistration and
/// captures the REST endpoint the gear advertised.
#[derive(Default)]
struct RecordingDirectory {
    register_calls: AtomicUsize,
    deregister_calls: AtomicUsize,
    registered_endpoint: Mutex<Option<String>>,
}

#[async_trait]
impl DirectoryClient for RecordingDirectory {
    async fn resolve_grpc_service(&self, _service: &str) -> anyhow::Result<ServiceEndpoint> {
        Ok(ServiceEndpoint::new("http://grpc"))
    }
    async fn resolve_rest_service(&self, gear: &str) -> anyhow::Result<ServiceEndpoint> {
        Ok(ServiceEndpoint::new(format!("http://{gear}:8080")))
    }
    async fn get_openapi_spec(&self, _gear: &str) -> anyhow::Result<String> {
        Ok(String::new())
    }
    async fn list_instances(&self, _gear: &str) -> anyhow::Result<Vec<ServiceInstanceInfo>> {
        Ok(vec![])
    }
    async fn list_all_instances(&self) -> anyhow::Result<Vec<ServiceInstanceInfo>> {
        Ok(vec![])
    }
    async fn register_instance(&self, info: RegisterInstanceInfo) -> anyhow::Result<()> {
        if let Some(ep) = &info.rest_endpoint {
            *self.registered_endpoint.lock() = Some(ep.uri.clone());
        }
        self.register_calls.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
    async fn deregister_instance(&self, _gear: &str, _instance: &str) -> anyhow::Result<()> {
        self.deregister_calls.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
    async fn send_heartbeat(&self, _gear: &str, _instance: &str) -> anyhow::Result<()> {
        Ok(())
    }
}

struct EmptyConfigProvider;
impl ConfigProvider for EmptyConfigProvider {
    fn get_gear_config(&self, _gear_name: &str) -> Option<&serde_json::Value> {
        None
    }
}

/// Reserve an ephemeral port and return it (closing the listener so the server
/// can rebind it).
fn free_addr() -> SocketAddr {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
    listener.local_addr().expect("local_addr")
}

/// Minimal raw HTTP/1.1 GET returning `(status, body)`. Uses `Connection: close`
/// so the read completes on EOF.
async fn http_get(addr: SocketAddr, path: &str) -> Option<(u16, String)> {
    let mut stream = tokio::net::TcpStream::connect(addr).await.ok()?;
    let req = format!("GET {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n");
    stream.write_all(req.as_bytes()).await.ok()?;
    let mut buf = Vec::new();
    stream.read_to_end(&mut buf).await.ok()?;
    let text = String::from_utf8_lossy(&buf);
    let status = text
        .lines()
        .next()
        .and_then(|l| l.split_whitespace().nth(1))
        .and_then(|s| s.parse().ok())?;
    let body = text
        .split_once("\r\n\r\n")
        .map_or(String::new(), |(_, b)| b.to_owned());
    Some((status, body))
}

/// Poll `path` until it returns `want` or the deadline elapses.
async fn poll_status(addr: SocketAddr, path: &str, want: u16, timeout: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        if let Some((status, _)) = http_get(addr, path).await
            && status == want
        {
            return true;
        }
        if tokio::time::Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
}

#[tokio::test]
async fn e2e_oop_gear_boots_serves_registers_and_shuts_down() {
    let addr = free_addr();
    let directory = Arc::new(RecordingDirectory::default());

    // 1. Build a registry with a single REST-capable gear, exactly as discovery
    //    would produce it.
    let gear = Arc::new(PingGear);
    let mut builder = RegistryBuilder::default();
    builder.register_core_with_meta("ping-gear", &[], gear.clone() as Arc<dyn Gear>);
    builder.register_rest_with_meta("ping-gear", gear as Arc<dyn RestApiCapability>);
    let registry = builder.build_topo_sorted().unwrap();

    let hub = Arc::new(ClientHub::new());
    let cancel = CancellationToken::new();
    let config: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

    let host = HostRuntime::new(
        registry,
        config,
        DbOptions::None,
        hub,
        cancel.clone(),
        Uuid::new_v4(),
        None,
    );

    let options = OopServeOptions {
        gear_name: "ping-gear".to_owned(),
        instance_id: "i-1".to_owned(),
        version: Some("1.2.3".to_owned()),
        advertise_uri: format!("http://{addr}"),
        listen_addr: addr,
        probe_bind_addr: None,
        drain_timeout: Duration::from_secs(5),
        heartbeat_interval: Duration::from_secs(30),
        healthcheck_timeout: Duration::from_millis(500),
        directory: Arc::clone(&directory) as Arc<dyn DirectoryClient>,
        bearer_authenticator: None,
        internal_authenticator: None,
        // No authenticator: the raw `/ping` route runs only in development mode.
        auth_disabled: true,
        labels: std::collections::BTreeMap::new(),
    };

    // 2. Drive the full OoP serving lifecycle on a background task.
    let server = tokio::spawn(host.run_oop_serving(options));

    // 3. Liveness comes up once phases complete and the server binds.
    assert!(
        poll_status(addr, "/healthz", 200, Duration::from_secs(5)).await,
        "/healthz should return 200 once the gear has booted"
    );

    // 4. The gear's OWN route is served — proves compose_oop_router wired a real
    //    RestApiCapability gear into the host-less router.
    let ping = http_get(addr, "/ping").await;
    assert_eq!(
        ping,
        Some((200, "pong".to_owned())),
        "the gear's REST route should serve its response body"
    );

    // 5. No external dependencies → /readyz becomes 200.
    assert!(
        poll_status(addr, "/readyz", 200, Duration::from_secs(3)).await,
        "/readyz should be 200 when the gear has no unresolved deps"
    );

    // 6. The gear's OpenAPI is published at the well-known path.
    let (oas_status, oas_body) = http_get(addr, "/.well-known/openapi.json")
        .await
        .expect("openapi endpoint should respond");
    assert_eq!(oas_status, 200);
    assert!(
        oas_body.contains("ping-gear") && oas_body.contains("1.2.3"),
        "OpenAPI document should carry the gear title + version, got: {oas_body}"
    );

    // 7. Self-registration ran through the full stack with the advertised URI.
    let registered = {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
        loop {
            if directory.register_calls.load(Ordering::SeqCst) >= 1 {
                break true;
            }
            if tokio::time::Instant::now() >= deadline {
                break false;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
    };
    assert!(
        registered,
        "the gear should self-register with the directory"
    );
    assert_eq!(
        directory.registered_endpoint.lock().clone(),
        Some(format!("http://{addr}")),
        "the advertised REST endpoint should be registered"
    );

    // 8. Graceful shutdown: cancel, the server drains + deregisters, the stop
    //    phase runs, and the lifecycle returns Ok.
    cancel.cancel();
    let result = tokio::time::timeout(Duration::from_secs(5), server)
        .await
        .expect("run_oop_serving should finish promptly after cancel")
        .expect("run_oop_serving task should not panic");
    assert!(
        result.is_ok(),
        "OoP serving lifecycle should end cleanly: {result:?}"
    );

    assert_eq!(
        directory.deregister_calls.load(Ordering::SeqCst),
        1,
        "the gear should deregister exactly once on graceful shutdown"
    );
}

const AXIS_SA: &str = "good-sa-token";

/// Rejects every bearer so tenant credentials cannot admit these routes.
struct RejectingBearer;

impl toolkit_security::BearerAuthenticator for RejectingBearer {
    async fn authenticate(
        &self,
        _token: &str,
    ) -> Result<toolkit_security::SecurityContext, toolkit_security::AuthNError> {
        Err(toolkit_security::AuthNError::InvalidToken)
    }
}

/// Platform authenticator: `good-sa-token` is a workload, the rest rejected.
struct AxisInternal;

impl toolkit_security::InternalAuthenticator for AxisInternal {
    async fn authenticate(
        &self,
        token: &str,
    ) -> Result<toolkit_security::PlatformIdentity, toolkit_security::InternalAuthNError> {
        if token == AXIS_SA {
            Ok(
                toolkit_security::PlatformIdentity::KubernetesServiceAccount {
                    namespace: "toolkit".to_owned(),
                    service_account: "axis".to_owned(),
                    pod: None,
                },
            )
        } else {
            Err(toolkit_security::InternalAuthNError::InvalidToken)
        }
    }
}

/// Uses `OperationBuilder` specs to test `compose_oop_router` auth derivation.
#[derive(Default)]
struct AxisGear;

#[async_trait]
impl Gear for AxisGear {
    async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
        Ok(())
    }
}

impl RestApiCapability for AxisGear {
    fn register_rest(
        &self,
        _ctx: &GearCtx,
        router: Router,
        openapi: &dyn OpenApiRegistry,
    ) -> anyhow::Result<Router> {
        use crate::api::OperationBuilder;
        let router = OperationBuilder::get("/anon")
            .operation_id("axis.anon")
            .anonymous()
            .handler(|| async { "anon" })
            .text_response(http::StatusCode::OK, "OK", "text/plain")
            .register(router, openapi);
        Ok(OperationBuilder::get("/platform")
            .operation_id("axis.platform")
            .platform_authenticated()
            .no_license_required()
            .handler(|| async { "platform" })
            .text_response(http::StatusCode::OK, "OK", "text/plain")
            .register(router, openapi))
    }
}

/// A raw HTTP/1.1 GET with extra headers, returning the status.
async fn http_status(addr: SocketAddr, path: &str, headers: &[(&str, &str)]) -> Option<u16> {
    let mut stream = tokio::net::TcpStream::connect(addr).await.ok()?;
    let mut req = format!("GET {path} HTTP/1.1\r\nHost: localhost\r\n");
    for (name, value) in headers {
        req.push_str(name);
        req.push_str(": ");
        req.push_str(value);
        req.push_str("\r\n");
    }
    req.push_str("Connection: close\r\n\r\n");
    stream.write_all(req.as_bytes()).await.ok()?;
    let mut buf = Vec::new();
    stream.read_to_end(&mut buf).await.ok()?;
    String::from_utf8_lossy(&buf)
        .lines()
        .next()
        .and_then(|l| l.split_whitespace().nth(1))
        .and_then(|s| s.parse().ok())
}

#[tokio::test]
async fn the_composed_router_enforces_each_routes_declared_auth() {
    let addr = free_addr();
    let directory = Arc::new(RecordingDirectory::default());
    let gear = Arc::new(AxisGear);
    let mut builder = RegistryBuilder::default();
    builder.register_core_with_meta("axis-gear", &[], gear.clone() as Arc<dyn Gear>);
    builder.register_rest_with_meta("axis-gear", gear as Arc<dyn RestApiCapability>);
    let registry = builder.build_topo_sorted().unwrap();
    let cancel = CancellationToken::new();
    let host = HostRuntime::new(
        registry,
        Arc::new(EmptyConfigProvider) as Arc<dyn ConfigProvider>,
        DbOptions::None,
        Arc::new(ClientHub::new()),
        cancel.clone(),
        Uuid::new_v4(),
        None,
    );
    let options = OopServeOptions {
        gear_name: "axis-gear".to_owned(),
        instance_id: "i-1".to_owned(),
        version: None,
        advertise_uri: format!("http://{addr}"),
        listen_addr: addr,
        probe_bind_addr: None,
        drain_timeout: Duration::from_secs(5),
        heartbeat_interval: Duration::from_secs(30),
        healthcheck_timeout: Duration::from_millis(500),
        directory: Arc::clone(&directory) as Arc<dyn DirectoryClient>,
        bearer_authenticator: Some(toolkit_security::DynBearerAuthenticator::new(
            RejectingBearer,
        )),
        internal_authenticator: Some(toolkit_security::DynInternalAuthenticator::new(
            AxisInternal,
        )),
        auth_disabled: false,
        labels: std::collections::BTreeMap::new(),
    };
    let server = tokio::spawn(host.run_oop_serving(options));
    assert!(
        poll_status(addr, "/healthz", 200, Duration::from_secs(5)).await,
        "the gear boots"
    );

    // Without the derived policy every route would demand a bearer.
    assert_eq!(
        http_status(addr, "/anon", &[]).await,
        Some(200),
        "an anonymous route needs no credential"
    );
    assert_eq!(
        http_status(addr, "/platform", &[("X-ToolKit-Internal-Token", AXIS_SA)]).await,
        Some(200),
        "a platform route admits a validated internal token"
    );
    assert_eq!(
        http_status(addr, "/platform", &[]).await,
        Some(401),
        "and refuses a caller without one"
    );

    cancel.cancel();
    let result = tokio::time::timeout(Duration::from_secs(5), server)
        .await
        .expect("finishes after cancel")
        .expect("does not panic");
    assert!(result.is_ok(), "{result:?}");
}

/// Serves one tenant operation, so it needs a tenant authenticator to start.
struct TenantGear;

#[async_trait]
impl Gear for TenantGear {
    async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
        Ok(())
    }
}

impl RestApiCapability for TenantGear {
    fn register_rest(
        &self,
        _ctx: &GearCtx,
        router: Router,
        openapi: &dyn OpenApiRegistry,
    ) -> anyhow::Result<Router> {
        use crate::api::OperationBuilder;
        Ok(OperationBuilder::get("/tenant")
            .operation_id("tenant.get")
            .authenticated()
            .no_license_required()
            .handler(|| async { "tenant" })
            .text_response(http::StatusCode::OK, "OK", "text/plain")
            .register(router, openapi))
    }
}

#[tokio::test]
async fn a_worker_without_a_tenant_authenticator_fails_before_it_is_advertised() {
    let addr = free_addr();
    let directory = Arc::new(RecordingDirectory::default());
    let gear = Arc::new(TenantGear);
    let mut builder = RegistryBuilder::default();
    builder.register_core_with_meta("tenant-gear", &[], Arc::clone(&gear) as Arc<dyn Gear>);
    builder.register_rest_with_meta("tenant-gear", gear as Arc<dyn RestApiCapability>);
    let registry = builder.build_topo_sorted().unwrap();
    let host = HostRuntime::new(
        registry,
        Arc::new(EmptyConfigProvider) as Arc<dyn ConfigProvider>,
        DbOptions::None,
        Arc::new(ClientHub::new()),
        CancellationToken::new(),
        Uuid::new_v4(),
        None,
    );
    let options = OopServeOptions {
        gear_name: "tenant-gear".to_owned(),
        instance_id: "i-1".to_owned(),
        version: None,
        advertise_uri: format!("http://{addr}"),
        listen_addr: addr,
        probe_bind_addr: None,
        drain_timeout: Duration::from_secs(5),
        heartbeat_interval: Duration::from_secs(30),
        healthcheck_timeout: Duration::from_millis(500),
        directory: Arc::clone(&directory) as Arc<dyn DirectoryClient>,
        bearer_authenticator: None,
        internal_authenticator: None,
        auth_disabled: false,
        labels: std::collections::BTreeMap::new(),
    };

    let result = tokio::time::timeout(Duration::from_secs(5), host.run_oop_serving(options))
        .await
        .expect("startup fails promptly");
    let error = result
        .expect_err("a tenant route without an authenticator")
        .to_string();
    assert!(error.contains("tenant authenticator"), "{error}");
    assert_eq!(
        directory.register_calls.load(Ordering::SeqCst),
        0,
        "the gear is never advertised"
    );
    let served = tokio::time::timeout(Duration::from_secs(5), http_get(addr, "/tenant"))
        .await
        .expect("the probe finishes promptly");
    assert_eq!(
        served, None,
        "the listener is torn down without serving the route"
    );
}

/// Accepts the bearer `tenant-token` as a tenant subject, rejects the rest.
struct TenantBearer;

impl toolkit_security::BearerAuthenticator for TenantBearer {
    async fn authenticate(
        &self,
        token: &str,
    ) -> Result<toolkit_security::SecurityContext, toolkit_security::AuthNError> {
        if token == "tenant-token" {
            toolkit_security::SecurityContext::builder()
                .subject_id(Uuid::from_u128(1))
                .subject_tenant_id(Uuid::from_u128(2))
                .build()
                .map_err(|_| toolkit_security::AuthNError::InvalidToken)
        } else {
            Err(toolkit_security::AuthNError::InvalidToken)
        }
    }
}

/// The startup check sees the tenant authenticator a linked gear registers in `ClientHub`
/// (as `authn-resolver` does): the worker starts and checks each tenant request.
#[tokio::test]
async fn a_worker_with_a_hub_tenant_authenticator_starts_and_checks_bearers() {
    let addr = free_addr();
    let directory = Arc::new(RecordingDirectory::default());
    let gear = Arc::new(TenantGear);
    let mut builder = RegistryBuilder::default();
    builder.register_core_with_meta("tenant-gear", &[], Arc::clone(&gear) as Arc<dyn Gear>);
    builder.register_rest_with_meta("tenant-gear", gear as Arc<dyn RestApiCapability>);
    let registry = builder.build_topo_sorted().unwrap();
    let hub = Arc::new(ClientHub::new());
    hub.register::<toolkit_security::DynBearerAuthenticator>(Arc::new(
        toolkit_security::DynBearerAuthenticator::new(TenantBearer),
    ));
    let cancel = CancellationToken::new();
    let host = HostRuntime::new(
        registry,
        Arc::new(EmptyConfigProvider) as Arc<dyn ConfigProvider>,
        DbOptions::None,
        hub,
        cancel.clone(),
        Uuid::new_v4(),
        None,
    );
    let options = OopServeOptions {
        gear_name: "tenant-gear".to_owned(),
        instance_id: "i-1".to_owned(),
        version: None,
        advertise_uri: format!("http://{addr}"),
        listen_addr: addr,
        probe_bind_addr: None,
        drain_timeout: Duration::from_secs(5),
        heartbeat_interval: Duration::from_secs(30),
        healthcheck_timeout: Duration::from_millis(500),
        directory: Arc::clone(&directory) as Arc<dyn DirectoryClient>,
        bearer_authenticator: None,
        internal_authenticator: None,
        auth_disabled: false,
        labels: std::collections::BTreeMap::new(),
    };
    let server = tokio::spawn(host.run_oop_serving(options));

    // Gear routes answer 503 until attached; poll until the tenant route serves.
    let bearer = [("Authorization", "Bearer tenant-token")];
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    let mut status = None;
    while tokio::time::Instant::now() < deadline {
        status = http_status(addr, "/tenant", &bearer).await;
        if status == Some(200) {
            break;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
    assert_eq!(status, Some(200), "a valid bearer is admitted");
    assert_eq!(
        http_status(addr, "/tenant", &[]).await,
        Some(401),
        "a missing bearer is refused"
    );

    cancel.cancel();
    let result = tokio::time::timeout(Duration::from_secs(5), server)
        .await
        .expect("finishes after cancel")
        .expect("does not panic");
    assert!(result.is_ok(), "{result:?}");
}