riotpiao-poimen/poimen-memory
2
0
Fork 0
Code Issues 11 Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

refactor: focus on K8s Job integration testing, remove random scripts
CI / CI (pull_request) Successful in 15m38s
Details

Browse Source 
Remove unfocused shell scripts - rely on existing integration tests instead:
  - ✓ tests/it_unified_query_4_6.rs (query tests)
  - ✓ tests/it_temporal_filtering_4_2_fixed.rs (temporal query)
  - ✓ tests/it_phase3_phase4.rs (ingest tests)
  - ✓ tests/it_authorized_pipeline.rs (auth + ingest)

Removed:
  - apply_migrations.sh (use migrations/ runner script)
  - collect_prod_logs.sh (k8s logs available)
  - run_production_test.sh (use cargo test)
  - test_prod_ingest_real.sh (existing it_phase3_phase4.rs)
  - tests/integration_ingest_with_gw.rs (duplicate)
  - tests/unit_ingest_logging.rs (duplicate)

Keep:
  - migrations/run_migrations.sh (K8s Job requirement)
  - k8s/test/integration-test-job.yaml (CI/CD integration)
  - .gitea/workflows/integration-test.yaml (CI orchestration)
  - k8s/test/db-credentials.enc.yaml (SOPS encrypted secrets)

Proper approach: K8s Job runs existing integration tests via 'cargo test'
                 ArgoCD+KSOPS decrypts secrets
                 Tests execute against new image SHA
This commit is contained in:
rock
2026-09-14 22:55:58 +09:00
parent 1ce9458347
commit ce6c93d3b5
8 changed files with 52 additions and 1165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration_ingest_with_gw.rs
-286
View File
@@ -1,286 +0,0 @@
//! 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'");
}
tests/unit_ingest_logging.rs
-221
View File
@@ -1,221 +0,0 @@
//! Unit test: Ingest pipeline with detailed error logging
//!
//! Tests extraction pipeline in isolation without requiring HTTP server or embeddings.
//! Useful for debugging extraction errors.
//!
//! Usage:
//! ```
//! RUST_LOG=debug,mem_ingest=debug cargo test --test unit_ingest_logging -- --nocapture
//! ```
#[cfg(test)]
mod tests {
use mem_ingest::ingest_pipeline::{IngestPipeline, Episode};
use mem_ingest::entity_extractor::WikiLinkFallbackExtractor;
use mem_ingest::fact_extractor::SimpleFactExtractor;
use mem_ingest::contradiction_detector::ContradictionHandler;
use std::sync::Arc;
fn init_logging() {
let _ = tracing_subscriber::fmt()
.with_max_level(tracing::Level::DEBUG)
.with_writer(std::io::stderr)
.try_init();
}
#[tokio::test]
async fn test_wiki_link_extraction() {
init_logging();
println!("\n[TEST] Wiki link extraction with logging\n");
let entity_extractor = Arc::new(WikiLinkFallbackExtractor);
let fact_extractor = Arc::new(SimpleFactExtractor);
let contradiction_detector = Arc::new(ContradictionHandler::default());
let pipeline = IngestPipeline::new(
entity_extractor,
fact_extractor,
contradiction_detector,
);
let episode = Episode {
id: "test-1".to_string(),
project_id: "test-project".to_string(),
text: "Kubernetes [[Docker]] is a [[Container]] orchestration platform. It works with [[Go]] programs."
.to_string(),
wiki_links: vec!["Docker".to_string(), "Container".to_string(), "Go".to_string()],
};
tracing::info!(
target: "test",
episode_id = %episode.id,
wiki_links = ?episode.wiki_links,
"Starting pipeline ingest"
);
match pipeline.ingest(&episode).await {
Ok(result) => {
tracing::info!(
target: "test",
entities = result.entities.len(),
edges = result.edges.len(),
reviews = result.reviews.len(),
"Pipeline succeeded"
);
println!("✓ Extracted {} entities", result.entities.len());
for entity in &result.entities {
println!(" - {} ({}): {}", entity.name, entity.entity_type.as_str(), entity.summary.as_deref().unwrap_or(""));
}
println!("✓ Extracted {} edges", result.edges.len());
for edge in &result.edges {
println!(" - {} --[{}]--> {}", edge.source_entity_id, edge.relation_type, edge.target_entity_id);
}
assert!(result.entities.len() > 0, "Should extract entities");
}
Err(e) => {
tracing::error!(
target: "test",
error = %e,
"Pipeline failed"
);
panic!("Pipeline failed: {}", e);
}
}
}
#[tokio::test]
async fn test_extraction_error_logging() {
init_logging();
println!("\n[TEST] Pipeline error handling with logging\n");
let entity_extractor = Arc::new(WikiLinkFallbackExtractor);
let fact_extractor = Arc::new(SimpleFactExtractor);
let contradiction_detector = Arc::new(ContradictionHandler::default());
let pipeline = IngestPipeline::new(
entity_extractor,
fact_extractor,
contradiction_detector,
);
// Episode with problematic content (empty, or only whitespace)
let episode = Episode {
id: "test-empty".to_string(),
project_id: "test-project".to_string(),
text: "".to_string(),
wiki_links: vec![],
};
tracing::info!(
target: "test",
episode_id = %episode.id,
text_len = episode.text.len(),
"Processing empty episode"
);
match pipeline.ingest(&episode).await {
Ok(result) => {
tracing::info!(
target: "test",
entities = result.entities.len(),
edges = result.edges.len(),
"Empty episode processed (no error expected)"
);
println!("✓ Empty episode handled gracefully");
}
Err(e) => {
tracing::error!(
target: "test",
error = %e,
"Empty episode caused error"
);
// Empty is OK for some extractors
println!("⚠ Empty episode error (may be expected): {}", e);
}
}
}
#[tokio::test]
async fn test_multiple_records_error_accumulation() {
init_logging();
println!("\n[TEST] Processing multiple records and logging errors\n");
let entity_extractor = Arc::new(WikiLinkFallbackExtractor);
let fact_extractor = Arc::new(SimpleFactExtractor);
let contradiction_detector = Arc::new(ContradictionHandler::default());
let pipeline = IngestPipeline::new(
entity_extractor,
fact_extractor,
contradiction_detector,
);
let records = vec![
("Kubernetes [[Docker]] is a container orchestrator", "wiki/k8s"),
("Docker [[Linux]] containers enable microservices", "wiki/docker"),
("", "wiki/empty"),
("Go [[Concurrency]] is powerful for backend services", "wiki/go"),
];
let mut success_count = 0;
let mut error_count = 0;
for (idx, (text, source)) in records.iter().enumerate() {
let episode = Episode {
id: format!("record-{}", idx),
project_id: "test-project".to_string(),
text: text.to_string(),
wiki_links: vec![],
};
tracing::info!(
target: "test",
record_idx = idx,
source = source,
text_len = text.len(),
"Processing record"
);
match pipeline.ingest(&episode).await {
Ok(result) => {
tracing::debug!(
target: "test",
record_idx = idx,
entities = result.entities.len(),
edges = result.edges.len(),
"Record succeeded"
);
println!(" ✓ Record {}: {} entities, {} edges", idx, result.entities.len(), result.edges.len());
success_count += 1;
}
Err(e) => {
tracing::warn!(
target: "test",
record_idx = idx,
error = %e,
source = source,
"Record failed"
);
println!(" ✗ Record {}: {}", idx, e);
error_count += 1;
}
}
}
println!("\nSummary: {} success, {} errors", success_count, error_count);
tracing::info!(
target: "test",
total_records = records.len(),
success = success_count,
errors = error_count,
"Batch processing complete"
);
}
}