lean-ctx-engine 3.12.2

// SPDX-License-Identifier: Apache-2.0

use super::*;

#[test]
fn test_processes_never_reach_a_remote_cloud_endpoint() {
    for remote in [
        "https://api.leanctx.com",
        "http://10.0.0.1:8080",
        "http://localhost.evil.com",
        "http://127.0.0.1@api.leanctx.com",
        "ftp://127.0.0.1",
        "127.0.0.1:9",
    ] {
        assert!(!cloud_endpoint_allowed(remote, true), "{remote}");
    }
    for loopback in [
        "http://127.0.0.1:9",
        "http://127.3.2.1",
        "http://localhost:8088/prefix",
        "https://LOCALHOST",
        "http://[::1]:9",
    ] {
        assert!(cloud_endpoint_allowed(loopback, true), "{loopback}");
    }
}

#[test]
fn an_installed_binary_keeps_its_configured_endpoint() {
    for configured in ["https://api.leanctx.com", "http://10.0.0.1:8080"] {
        assert!(cloud_endpoint_allowed(configured, false), "{configured}");
    }
}

#[test]
fn this_unit_test_harness_is_a_test_process() {
    assert!(crate::core::data_dir::sandboxed());
    assert!(test_process());
}

#[test]
fn every_workspace_package_counts_as_cargo_launched() {
    // Read the members from the manifests, so a crate split out later is
    // covered too, or fails here instead of reaching production.
    let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
    let workspace = std::fs::read_to_string(root.join("Cargo.toml")).unwrap();
    let members = workspace
        .split_once("members = [")
        .and_then(|(_, rest)| rest.split_once(']'))
        .map(|(list, _)| list)
        .unwrap();
    let names: Vec<String> = members
        .split(',')
        .map(|member| member.trim().trim_matches('"'))
        .filter(|member| !member.is_empty())
        .map(|member| {
            let manifest = std::fs::read_to_string(root.join(member).join("Cargo.toml")).unwrap();
            let package = manifest.split_once("[package]").unwrap().1;
            package
                .lines()
                .find_map(|line| line.trim().strip_prefix("name = "))
                .unwrap()
                .trim_matches('"')
                .to_string()
        })
        .collect();
    for expected in [
        "lean-ctx-engine",
        "lean-ctx",
        "lean-ctx-server",
        "lean-ctx-hive",
        "lean-ctx-hooks",
        "lean-ctx-proxy",
        "leanctx-native-storage",
    ] {
        assert!(names.iter().any(|name| name == expected), "{names:?}");
    }
    for name in &names {
        assert!(workspace_package(name), "{name}");
    }
    for foreign in ["", "serde", "my-lean-ctx", "lean_ctx", "cargo"] {
        assert!(!workspace_package(foreign), "{foreign}");
    }
}

#[test]
fn api_url_redirects_test_processes_away_from_production() {
    let _lock = crate::core::data_dir::test_env_lock();
    let previous = std::env::var_os("LEAN_CTX_API_URL");

    // Every cloud call (feedback, stats, wrapped, telemetry) builds on api_url,
    // so the default and any remote override must land on the loopback sink.
    crate::test_env::remove_var("LEAN_CTX_API_URL");
    assert_eq!(api_url(), "http://127.0.0.1:9");
    crate::test_env::set_var("LEAN_CTX_API_URL", "https://api.leanctx.com");
    assert_eq!(api_url(), "http://127.0.0.1:9");
    crate::test_env::set_var("LEAN_CTX_API_URL", "http://127.0.0.1:8088");
    assert_eq!(api_url(), "http://127.0.0.1:8088");

    match previous {
        Some(value) => crate::test_env::set_var("LEAN_CTX_API_URL", value),
        None => crate::test_env::remove_var("LEAN_CTX_API_URL"),
    }
}

#[test]
fn cargo_test_processes_are_recognised_as_cargo_launched() {
    // Children spawned by integration tests inherit this, which is what keeps
    // the release binary under test from reaching production telemetry.
    assert!(launched_by_cargo());
}

#[test]
fn telemetry_delete_response_requires_explicit_confirmation() {
    assert!(deletion_confirmed(200, r#"{"deleted":true}"#).unwrap());
    assert!(!deletion_confirmed(204, "").unwrap());
    assert!(!deletion_confirmed(200, r#"{"deleted":false}"#).unwrap());
    assert!(deletion_confirmed(200, "not-json").is_err());
}

struct TelemetryTestEnvironment(Vec<(&'static str, Option<std::ffi::OsString>)>);

impl TelemetryTestEnvironment {
    // Call only while isolated_data_dir holds the shared test environment lock.
    fn new(listener: &std::net::TcpListener) -> Self {
        let values = [
            "LEAN_CTX_API_URL",
            "DO_NOT_TRACK",
            "LEAN_CTX_TELEMETRY",
            "HTTP_PROXY",
            "HTTPS_PROXY",
            "ALL_PROXY",
            "http_proxy",
            "https_proxy",
            "all_proxy",
        ]
        .into_iter()
        .map(|key| {
            let previous = std::env::var_os(key);
            crate::test_env::remove_var(key);
            (key, previous)
        })
        .collect();
        let guard = Self(values);
        crate::test_env::set_var(
            "LEAN_CTX_API_URL",
            format!("http://{}", listener.local_addr().unwrap()),
        );
        guard
    }
}

impl Drop for TelemetryTestEnvironment {
    fn drop(&mut self) {
        for (key, value) in &self.0 {
            match value {
                Some(value) => crate::test_env::set_var(key, value),
                None => crate::test_env::remove_var(key),
            }
        }
    }
}

fn telemetry_test_config(contents: &str) {
    let path = crate::core::config::Config::path().unwrap();
    std::fs::create_dir_all(path.parent().unwrap()).unwrap();
    std::fs::write(path, contents).unwrap();
}

const TELEMETRY_ALLOWED: &str = "[telemetry]\nenabled = true\n";

fn telemetry_test_batch() -> crate::core::telemetry_v2::TelemetryBatchV2 {
    use crate::core::telemetry_v2::{ClientFamily, DistributionChannel};
    crate::core::telemetry_aggregate::build_daily_heartbeat(
        "550e8400-e29b-41d4-a716-446655440000".into(),
        "a".repeat(64),
        "2026-03-01".into(),
        DistributionChannel::Unknown,
        ClientFamily::Other,
    )
    .unwrap()
}

fn receive_telemetry(listener: &std::net::TcpListener) -> (std::net::TcpStream, serde_json::Value) {
    use std::io::{BufRead, Read};
    listener.set_nonblocking(true).unwrap();
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
    let mut stream = loop {
        match listener.accept() {
            Ok((stream, _)) => break stream,
            Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                assert!(std::time::Instant::now() < deadline, "request not received");
                std::thread::sleep(std::time::Duration::from_millis(5));
            }
            Err(error) => panic!("local accept failed: {error}"),
        }
    };
    stream.set_nonblocking(false).unwrap();
    stream
        .set_read_timeout(Some(std::time::Duration::from_secs(3)))
        .unwrap();
    let mut reader = std::io::BufReader::new(&mut stream);
    let mut line = String::new();
    reader.read_line(&mut line).unwrap();
    assert!(line.starts_with("POST /api/telemetry/v2/batch "));
    let mut length = None;
    loop {
        line.clear();
        assert!(reader.read_line(&mut line).unwrap() > 0);
        if line == "\r\n" {
            break;
        }
        if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") {
            length = Some(value.trim().parse::<usize>().unwrap());
        }
    }
    let mut body = vec![0; length.unwrap()];
    reader.read_exact(&mut body).unwrap();
    (stream, serde_json::from_slice(&body).unwrap())
}

#[test]
fn telemetry_send_rechecks_config_and_environment_before_any_connection() {
    let _iso = crate::core::data_dir::isolated_data_dir();
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();
    let _env = TelemetryTestEnvironment::new(&listener);
    let batch = telemetry_test_batch();
    telemetry_test_config(TELEMETRY_ALLOWED);
    let stale = crate::core::config::Config::try_load_global().unwrap();
    assert!(stale.telemetry.send_eligible(None, None));
    for config in [
        "[telemetry]\nenabled = false\n",
        "[telemetry]\nenabled = true\npreference = 'explicitly_disabled'\n",
    ] {
        telemetry_test_config(config);
        assert_eq!(
            telemetry_v2_batch(&batch).unwrap_err(),
            "Telemetry sending is disabled"
        );
    }
    telemetry_test_config(TELEMETRY_ALLOWED);
    for (key, value) in [("DO_NOT_TRACK", "1"), ("LEAN_CTX_TELEMETRY", "off")] {
        crate::test_env::set_var(key, value);
        assert_eq!(
            telemetry_v2_batch(&batch).unwrap_err(),
            "Telemetry sending is disabled"
        );
        crate::test_env::remove_var(key);
    }
    telemetry_test_config("[telemetry");
    assert_eq!(
        telemetry_v2_batch(&batch).unwrap_err(),
        "Telemetry configuration unavailable"
    );
    assert_eq!(
        listener.accept().unwrap_err().kind(),
        std::io::ErrorKind::WouldBlock
    );
}

#[test]
fn telemetry_send_posts_the_validated_batch_to_loopback() {
    use std::io::Write;
    let _iso = crate::core::data_dir::isolated_data_dir();
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let _env = TelemetryTestEnvironment::new(&listener);
    telemetry_test_config(TELEMETRY_ALLOWED);
    let batch = telemetry_test_batch();
    let server = std::thread::spawn(move || {
        let (mut stream, body) = receive_telemetry(&listener);
        let response = r#"{"message":"accepted"}"#;
        write!(
            stream,
            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}",
            response.len()
        )
        .unwrap();
        body
    });
    let result = telemetry_v2_batch(&batch);
    let captured = server.join().unwrap();
    assert_eq!(result.unwrap(), "accepted");
    assert_eq!(captured, serde_json::to_value(batch).unwrap());
}

#[test]
fn a_notice_in_the_acknowledgement_waits_for_the_terminal() {
    use std::io::Write;
    let _iso = crate::core::data_dir::isolated_data_dir();
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let _env = TelemetryTestEnvironment::new(&listener);
    telemetry_test_config(TELEMETRY_ALLOWED);
    let batch = telemetry_test_batch();
    let server = std::thread::spawn(move || {
        let (mut stream, _) = receive_telemetry(&listener);
        let response = r#"{"message":"accepted","notices":[
            {"id":"6f1c2b9e-3d4a-4b5c-8d6e-7f8091a2b3c4","message":"Fix is in 3.11.3","link":"https://leanctx.com/changelog"},
            {"id":"7f1c2b9e-3d4a-4b5c-8d6e-7f8091a2b3c4","message":"phish","link":"https://evil.example/"}]}"#;
        write!(
            stream,
            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}",
            response.len()
        )
        .unwrap();
    });
    assert_eq!(telemetry_v2_batch(&batch).unwrap(), "accepted");
    server.join().unwrap();
    let pending = crate::core::telemetry_notices::pending();
    assert_eq!(pending.len(), 1, "{pending:?}");
    assert_eq!(pending[0].message, "Fix is in 3.11.3");
}

#[test]
fn telemetry_send_times_out_when_the_gateway_stalls() {
    assert_telemetry_gateway_timeout(false);
}

#[test]
fn telemetry_send_times_out_after_receiving_response_headers() {
    assert_telemetry_gateway_timeout(true);
}

fn assert_telemetry_gateway_timeout(send_headers: bool) {
    use std::io::Write;
    let _iso = crate::core::data_dir::isolated_data_dir();
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let _env = TelemetryTestEnvironment::new(&listener);
    telemetry_test_config(TELEMETRY_ALLOWED);
    let (release, wait) = std::sync::mpsc::channel::<()>();
    let server = std::thread::spawn(move || {
        let (mut stream, body) = receive_telemetry(&listener);
        if send_headers {
            stream
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1024\r\nConnection: close\r\n\r\n{")
                .unwrap();
        }
        let _ = wait.recv_timeout(std::time::Duration::from_secs(3));
        body
    });
    let batch = telemetry_test_batch();
    let started = std::time::Instant::now();
    let result = telemetry_v2_batch_with_timeout(&batch, std::time::Duration::from_millis(300));
    let elapsed = started.elapsed();
    let _ = release.send(());
    let captured = server.join().unwrap();
    let error = result.unwrap_err().to_ascii_lowercase();
    assert!(
        error.contains("timeout") || error.contains("timed out"),
        "{error}"
    );
    assert!(elapsed < std::time::Duration::from_secs(2), "{elapsed:?}");
    assert_eq!(captured, serde_json::to_value(batch).unwrap());
}