poimen-memory/tests/unit_ingest_logging.rs

//! Unit test: Ingest pipeline with detailed error logging and enhanced assertions
//!
//! Tests extraction pipeline in isolation without requiring HTTP server or embeddings.
//! Useful for debugging extraction errors.
//!
//! Features:
//! - Verify extracted entity names (not just count)
//! - Verify edge connections between entities
//! - Detailed error logging
//!
//! 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_with_entity_verification() {
        init_logging();

        println!("\n[TEST] Wiki link extraction with entity name verification\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);
                }

                // ENHANCED: Verify extracted entity names (not just count)
                assert!(!result.entities.is_empty(), "Should extract at least one entity");
                
                let entity_names: Vec<&str> = result.entities.iter().map(|e| e.name.as_str()).collect();
                println!("\nEntity names extracted: {:?}", entity_names);
                
                assert!(
                    entity_names.iter().any(|&name| name.contains("Docker") || name.contains("docker")),
                    "Should extract Docker entity"
                );
                assert!(
                    entity_names.iter().any(|&name| name.contains("Container") || name.contains("container")),
                    "Should extract Container entity"
                );
                assert!(
                    entity_names.iter().any(|&name| name.contains("Go") || name.contains("go")),
                    "Should extract Go entity"
                );

                // ENHANCED: Verify edges connect correct entity pairs
                if !result.edges.is_empty() {
                    println!("\nEdge connections:");
                    for edge in &result.edges {
                        println!("  {} → {}", edge.source_entity_id, edge.target_entity_id);
                        
                        // Verify both endpoints exist in entities
                        let source_exists = result.entities.iter().any(|e| e.id == edge.source_entity_id);
                        let target_exists = result.entities.iter().any(|e| e.id == edge.target_entity_id);
                        
                        assert!(source_exists, "Edge source entity {} must exist in extracted entities", edge.source_entity_id);
                        assert!(target_exists, "Edge target entity {} must exist in extracted entities", edge.target_entity_id);
                    }
                }
            }
            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_with_entity_verification() {
        init_logging();

        println!("\n[TEST] Processing multiple records with entity name verification\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;
        let mut all_extracted_entities = Vec::new();

        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());
                    
                    // Collect entity names for batch verification
                    for entity in &result.entities {
                        all_extracted_entities.push(entity.name.clone());
                    }
                    
                    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);
        println!("All extracted entities: {:?}", all_extracted_entities);

        tracing::info!(
            target: "test",
            total_records = records.len(),
            success = success_count,
            errors = error_count,
            "Batch processing complete"
        );

        // ENHANCED: Verify that expected entities were extracted across records
        assert!(success_count > 0, "At least some records should succeed");
        assert!(
            all_extracted_entities.iter().any(|name| name.contains("Docker") || name.contains("docker")),
            "Docker entity should be extracted from at least one record"
        );
        assert!(
            all_extracted_entities.iter().any(|name| name.contains("Linux") || name.contains("linux")),
            "Linux entity should be extracted from at least one record"
        );
        assert!(
            all_extracted_entities.iter().any(|name| name.contains("Concurrency") || name.contains("concurrency")),
            "Concurrency entity should be extracted from at least one record (via [[Concurrency]] wiki link)"
        );
    }

    #[tokio::test]
    async fn test_entity_deduplication() {
        init_logging();

        println!("\n[TEST] Entity deduplication (same entity from multiple records)\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,
        );

        // Two records with overlapping entity references
        let episode1 = Episode {
            id: "record-1".to_string(),
            project_id: "test-project".to_string(),
            text: "Kubernetes uses [[Docker]] containers".to_string(),
            wiki_links: vec!["Docker".to_string()],
        };

        let episode2 = Episode {
            id: "record-2".to_string(),
            project_id: "test-project".to_string(),
            text: "Docker is used by [[Kubernetes]]".to_string(),
            wiki_links: vec!["Kubernetes".to_string()],
        };

        let mut all_entities = Vec::new();

        for episode in &[episode1, episode2] {
            match pipeline.ingest(episode).await {
                Ok(result) => {
                    all_entities.extend(result.entities);
                }
                Err(e) => {
                    tracing::error!(target: "test", error = %e, "Failed to ingest");
                }
            }
        }

        println!("Total entities extracted: {}", all_entities.len());
        for entity in &all_entities {
            println!("  - {}", entity.name);
        }

        // Verify both Docker and Kubernetes were extracted
        assert!(
            all_entities.iter().any(|e| e.name.contains("Docker") || e.name.contains("docker")),
            "Docker should be extracted"
        );
        assert!(
            all_entities.iter().any(|e| e.name.contains("Kubernetes") || e.name.contains("kubernetes")),
            "Kubernetes should be extracted"
        );
    }
}