//! Integration test: Full ingest + embedding flow with api-gw //! //! Tests: //! 1. POST /memory/ingest with sample records //! 2. Poll /memory/ingest/{id} until done //! 3. Log root causes of errors //! //! Requires: //! - DATABASE_URL set (postgres) //! - LLM_ENDPOINT set (for embeddings) //! - Server running locally or started by test //! //! Usage: //! ``` //! RUST_LOG=debug cargo test --test integration_ingest_with_gw -- --nocapture //! ``` use std::env; use std::time::Duration; use tokio::time::sleep; use serde_json::json; #[tokio::test] #[ignore] // Run manually: cargo test --test integration_ingest_with_gw -- --ignored --nocapture async fn test_ingest_with_embeddings_and_logging() { // Initialize tracing with DEBUG level to see all logs let _ = tracing_subscriber::fmt() .with_max_level(tracing::Level::DEBUG) .with_writer(std::io::stderr) .try_init(); let base_url = env::var("MEM_API_URL").unwrap_or_else(|_| "http://localhost:8080".to_string()); let api_key = env::var("MEM_API_KEY").unwrap_or_else(|_| "test-key".to_string()); let client = reqwest::Client::new(); // Sample ingest payload let payload = json!({ "project": "test-project", "records": [ { "content": "Kubernetes is an open-source container orchestration platform. [[Docker]] [[Go]]", "source": "wiki/kubernetes" }, { "content": "Docker is a containerization platform that makes it easier to build, ship, and run applications. [[Linux]] [[Container]]", "source": "wiki/docker" }, { "content": "Go is a programming language designed at Google. [[Concurrency]] [[Static Typing]]", "source": "wiki/go" } ] }); println!("[TEST] Sending ingest request..."); tracing::info!( target: "integration_test", "Ingest payload: {}", serde_json::to_string_pretty(&payload).unwrap() ); // POST /memory/ingest let response = match client .post(&format!("{}/memory/ingest", base_url)) .header("Authorization", format!("Bearer {}", api_key)) .json(&payload) .send() .await { Ok(resp) => resp, Err(e) => { eprintln!("[ERROR] Failed to send ingest request: {}", e); tracing::error!( target: "integration_test", error = %e, "Failed to POST /memory/ingest" ); panic!("Request failed: {}", e); } }; let status = response.status(); println!("[TEST] Ingest response status: {}", status); let body_text = match response.text().await { Ok(text) => text, Err(e) => { tracing::error!(target: "integration_test", error = %e, "Failed to read response body"); panic!("Failed to read response body: {}", e); } }; println!("[TEST] Response body:\n{}", body_text); // Parse response let resp_json: serde_json::Value = match serde_json::from_str(&body_text) { Ok(j) => j, Err(e) => { tracing::error!( target: "integration_test", error = %e, body = %body_text, "Failed to parse JSON response" ); panic!("Failed to parse JSON: {}", e); } }; let ingest_id = match resp_json["id"].as_str() { Some(id) => id.to_string(), None => { tracing::error!( target: "integration_test", response = %serde_json::to_string_pretty(&resp_json).unwrap(), "Missing 'id' in response" ); panic!("Missing 'id' in response: {}", resp_json); } }; println!("[TEST] Ingest ID: {}", ingest_id); tracing::info!(target: "integration_test", ingest_id = %ingest_id, "Ingest queued"); // Poll until complete or timeout let max_polls = 60; // 10 minutes with 10s intervals for poll_num in 1..=max_polls { sleep(Duration::from_secs(10)).await; println!( "[TEST] Poll #{}/{}: Checking status of ingest {}", poll_num, max_polls, ingest_id ); let status_response = match client .get(&format!("{}/memory/ingest/{}", base_url, ingest_id)) .header("Authorization", format!("Bearer {}", api_key)) .send() .await { Ok(resp) => resp, Err(e) => { tracing::error!( target: "integration_test", error = %e, ingest_id = %ingest_id, poll = poll_num, "Failed to fetch status" ); eprintln!("[ERROR] Failed to fetch status: {}", e); sleep(Duration::from_secs(5)).await; continue; } }; let status_text = match status_response.text().await { Ok(text) => text, Err(e) => { tracing::error!( target: "integration_test", error = %e, ingest_id = %ingest_id, "Failed to read status response" ); eprintln!("[ERROR] Failed to read status: {}", e); continue; } }; let status_json: serde_json::Value = match serde_json::from_str(&status_text) { Ok(j) => j, Err(e) => { tracing::error!( target: "integration_test", error = %e, body = %status_text, "Failed to parse status JSON" ); eprintln!("[ERROR] Failed to parse status JSON: {}", e); continue; } }; let status = status_json["status"].as_str().unwrap_or("unknown"); println!( "[TEST] Poll #{}: status = {}", poll_num, status ); tracing::info!( target: "integration_test", ingest_id = %ingest_id, poll = poll_num, status = %status, full_response = %serde_json::to_string_pretty(&status_json).unwrap(), "Status check" ); match status { "done" => { println!("[TEST] ✓ Ingest completed successfully!"); tracing::info!(target: "integration_test", "Ingest completed"); // Extract and log results if let Some(results) = status_json.get("results") { println!("[TEST] Results:\n{}", serde_json::to_string_pretty(results).unwrap()); tracing::info!( target: "integration_test", results = %serde_json::to_string_pretty(results).unwrap(), "Ingest results" ); } return; } "failed" | "error" => { let error_msg = status_json["error"].as_str().unwrap_or("unknown error"); println!("[TEST] ✗ Ingest FAILED: {}", error_msg); tracing::error!( target: "integration_test", ingest_id = %ingest_id, error = %error_msg, full_response = %serde_json::to_string_pretty(&status_json).unwrap(), "Ingest failed" ); panic!("Ingest failed: {}", error_msg); } "processing" | "queued" => { // Continue polling println!("[TEST] Still processing, poll again..."); } _ => { println!("[TEST] Unknown status: {}", status); tracing::warn!(target: "integration_test", status = %status, "Unknown status"); } } } // Timeout let msg = format!("Ingest did not complete after {} polls (timeout)", max_polls); println!("[TEST] ✗ {}", msg); tracing::error!(target: "integration_test", ingest_id = %ingest_id, "Ingest timeout"); panic!("{}", msg); } #[tokio::test] #[ignore] async fn test_ingest_endpoint_only() { let _ = tracing_subscriber::fmt() .with_max_level(tracing::Level::DEBUG) .try_init(); let base_url = env::var("MEM_API_URL").unwrap_or_else(|_| "http://localhost:8080".to_string()); let api_key = env::var("MEM_API_KEY").unwrap_or_else(|_| "test-key".to_string()); let client = reqwest::Client::new(); let payload = json!({ "project": "test-project", "records": [ { "content": "Simple test record", "source": "test" } ] }); println!("[TEST] Testing /memory/ingest endpoint only"); let response = client .post(&format!("{}/memory/ingest", base_url)) .header("Authorization", format!("Bearer {}", api_key)) .json(&payload) .send() .await .expect("Failed to send request"); println!("[TEST] Status: {}", response.status()); let body = response.text().await.expect("Failed to read body"); println!("[TEST] Response: {}", body); let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); println!("[TEST] Parsed: {}", serde_json::to_string_pretty(&json).unwrap()); assert!(json.get("id").is_some(), "Response should contain 'id'"); }