From ce6c93d3b53c601dc5ebece2e5c1e5a7d1ae3f95 Mon Sep 17 00:00:00 2001 From: rock Date: Mon, 14 Sep 2026 22:55:31 +0900 Subject: [PATCH] refactor: focus on K8s Job integration testing, remove random scripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- apply_migrations.sh | 124 --------- collect_prod_logs.sh | 114 -------- k8s/test/integration-test-job.yaml | 194 ++++---------- k8s/test/integration-test-secrets.enc.yaml | 0 run_production_test.sh | 81 ------ test_prod_ingest_real.sh | 197 -------------- tests/integration_ingest_with_gw.rs | 286 --------------------- tests/unit_ingest_logging.rs | 221 ---------------- 8 files changed, 52 insertions(+), 1165 deletions(-) delete mode 100755 apply_migrations.sh delete mode 100755 collect_prod_logs.sh create mode 100644 k8s/test/integration-test-secrets.enc.yaml delete mode 100755 run_production_test.sh delete mode 100755 test_prod_ingest_real.sh delete mode 100644 tests/integration_ingest_with_gw.rs delete mode 100644 tests/unit_ingest_logging.rs diff --git a/apply_migrations.sh b/apply_migrations.sh deleted file mode 100755 index 879db71..0000000 --- a/apply_migrations.sh +++ /dev/null @@ -1,124 +0,0 @@ -#!/bin/bash - -# Apply all database migrations to production PostgreSQL -# -# Usage: -# ./apply_migrations.sh -# -# Connects to: poimen namespace, memory-db-rw service - -set -e - -NAMESPACE="poimen" -DB_SERVICE="memory-db-rw" -DB_PORT="5432" -DB_USER="app" -DB_NAME="memory" -LOCAL_PORT="5433" - -echo "==========================================" -echo "Poimen Memory Database Migrations" -echo "==========================================" -echo "" - -# Start port-forward -echo "Starting port-forward to $DB_SERVICE..." -kubectl -n "$NAMESPACE" port-forward "svc/$DB_SERVICE" "$LOCAL_PORT:$DB_PORT" >/dev/null 2>&1 & -PF_PID=$! - -cleanup() { - if [ -n "$PF_PID" ]; then - kill $PF_PID 2>/dev/null || true - wait $PF_PID 2>/dev/null || true - fi -} -trap cleanup EXIT - -sleep 2 - -if ! kill -0 $PF_PID 2>/dev/null; then - echo "✗ Port-forward failed" - exit 1 -fi - -echo "✓ Port-forward active (PID $PF_PID)" -echo "" - -# Test connection -echo "Testing database connection..." -if ! PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -c "SELECT version();" >/dev/null 2>&1; then - echo "✗ Cannot connect to database" - echo " Host: localhost:$LOCAL_PORT" - echo " User: $DB_USER" - echo " Database: $DB_NAME" - exit 1 -fi -echo "✓ Database connected" -echo "" - -# Get migration files -MIGRATION_DIR="crates/mem-store/migrations" -if [ ! -d "$MIGRATION_DIR" ]; then - echo "✗ Migration directory not found: $MIGRATION_DIR" - exit 1 -fi - -MIGRATIONS=($(ls -1 "$MIGRATION_DIR"/*.sql | sort)) - -if [ ${#MIGRATIONS[@]} -eq 0 ]; then - echo "✗ No migrations found in $MIGRATION_DIR" - exit 1 -fi - -echo "Found ${#MIGRATIONS[@]} migration(s):" -for m in "${MIGRATIONS[@]}"; do - echo " - $(basename $m)" -done -echo "" - -# Run migrations -echo "==========================================" -echo "Running Migrations" -echo "==========================================" -echo "" - -success=0 -failed=0 - -for migration in "${MIGRATIONS[@]}"; do - name=$(basename "$migration") - echo -n "▶ $name ... " - - if PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$migration" >/dev/null 2>&1; then - echo "✓" - ((success++)) - else - echo "✗" - echo " Error output:" - PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -f "$migration" 2>&1 | sed 's/^/ /' - ((failed++)) - fi -done - -echo "" -echo "==========================================" -echo "Migration Summary" -echo "==========================================" -echo " Success: $success" -echo " Failed: $failed" -echo "" - -# Verify schema -echo "Verifying schema..." -echo "" -echo "Tables created:" -PGPASSWORD="$DB_PASSWORD" psql -h localhost -p "$LOCAL_PORT" -U "$DB_USER" -d "$DB_NAME" -c "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename;" | grep -v "^--" | tail -n+3 - -echo "" -if [ $failed -eq 0 ]; then - echo "✓ All migrations applied successfully" - exit 0 -else - echo "✗ Some migrations failed" - exit 1 -fi diff --git a/collect_prod_logs.sh b/collect_prod_logs.sh deleted file mode 100755 index b34dcfd..0000000 --- a/collect_prod_logs.sh +++ /dev/null @@ -1,114 +0,0 @@ -#!/bin/bash - -# Collect logs from production pods for debugging ingest errors -# -# Usage: -# ./collect_prod_logs.sh before # Capture baseline -# ./test_production_ingest.sh # Run test -# ./collect_prod_logs.sh after # Capture post-test logs -# ./collect_prod_logs.sh analyze # Show diff + errors - -set -e - -NAMESPACE="poimen" -TIMESTAMP=$(date +%Y%m%d-%H%M%S) -LOG_DIR="prod_logs_${TIMESTAMP}" - -case "${1:-all}" in - before) - echo "Collecting pre-test baseline logs..." - mkdir -p "$LOG_DIR/before" - - kubectl -n "$NAMESPACE" get pods > "$LOG_DIR/before/pods.txt" - - for pod in $(kubectl -n "$NAMESPACE" get pods -l app=memory-service -o jsonpath='{.items[*].metadata.name}'); do - echo " Collecting logs from $pod..." - kubectl -n "$NAMESPACE" logs "$pod" --all-containers=true > "$LOG_DIR/before/${pod}.log" 2>&1 || true - done - - echo "✓ Baseline logs saved to $LOG_DIR/before/" - ;; - - after) - echo "Collecting post-test logs..." - mkdir -p "$LOG_DIR/after" - - kubectl -n "$NAMESPACE" get pods > "$LOG_DIR/after/pods.txt" - - for pod in $(kubectl -n "$NAMESPACE" get pods -l app=memory-service -o jsonpath='{.items[*].metadata.name}'); do - echo " Collecting logs from $pod..." - kubectl -n "$NAMESPACE" logs "$pod" --all-containers=true > "$LOG_DIR/after/${pod}.log" 2>&1 || true - done - - echo "✓ Post-test logs saved to $LOG_DIR/after/" - ;; - - analyze) - if [ ! -d "$LOG_DIR/before" ] || [ ! -d "$LOG_DIR/after" ]; then - echo "✗ Before/after log directories not found" - echo "Run: ./collect_prod_logs.sh before && ./test_production_ingest.sh && ./collect_prod_logs.sh after" - exit 1 - fi - - echo "==========================================" - echo "Log Analysis" - echo "==========================================" - echo "" - - # Find errors - echo "ERRORS found in logs:" - echo "-----" - grep -h "error\|Error\|ERROR" "$LOG_DIR/after"/*.log 2>/dev/null | tail -20 || echo " (none)" - echo "" - - # Find warnings - echo "WARNINGS found in logs:" - echo "-----" - grep -h "warn\|Warn\|WARN" "$LOG_DIR/after"/*.log 2>/dev/null | tail -10 || echo " (none)" - echo "" - - # Find ingest events - echo "INGEST events:" - echo "-----" - grep -h "ingest\|Ingest" "$LOG_DIR/after"/*.log 2>/dev/null | tail -20 || echo " (none)" - echo "" - - # Find embedding calls - echo "EMBEDDING events:" - echo "-----" - grep -h "embed\|Embed" "$LOG_DIR/after"/*.log 2>/dev/null | tail -20 || echo " (none)" - echo "" - - # Show new logs (after only) - echo "NEW LOG ENTRIES (post-test only):" - echo "-----" - for before_file in "$LOG_DIR/before"/*.log; do - after_file="${before_file//\/before\//\/after\/}" - if [ -f "$after_file" ]; then - pod_name=$(basename "$before_file" .log) - before_lines=$(wc -l < "$before_file" 2>/dev/null || echo 0) - after_lines=$(wc -l < "$after_file" 2>/dev/null || echo 0) - new_lines=$((after_lines - before_lines)) - if [ $new_lines -gt 0 ]; then - echo "" - echo "Pod: $pod_name (new: $new_lines lines)" - tail -$new_lines "$after_file" | grep -E "error|warn|ingest|embed" || true - fi - fi - done - - echo "" - echo "Full logs in: $LOG_DIR/" - ;; - - *) - echo "Usage: $0 {before|after|analyze}" - echo "" - echo "Steps:" - echo " 1. ./collect_prod_logs.sh before" - echo " 2. ./test_production_ingest.sh" - echo " 3. ./collect_prod_logs.sh after" - echo " 4. ./collect_prod_logs.sh analyze" - exit 1 - ;; -esac diff --git a/k8s/test/integration-test-job.yaml b/k8s/test/integration-test-job.yaml index 9a3e2f8..8f9cd44 100644 --- a/k8s/test/integration-test-job.yaml +++ b/k8s/test/integration-test-job.yaml @@ -1,17 +1,16 @@ --- -# Integration Test Job +# Integration Test Job - Runs existing Rust integration tests in K8s # -# Runs after image build in CI/CD pipeline. -# Tests the new image SHA against actual K8s cluster. +# Two-stage execution: +# 1. migrate: Apply database migrations +# 2. test: Run existing integration tests (cargo test) # -# Usage: -# kubectl apply -f k8s/test/integration-test-job.yaml \ -# -n poimen \ -# --dry-run=client -o yaml | \ -# sed "s|IMAGE_SHA|sha256:abcd1234|g" | \ -# kubectl apply -f - -# -# Or via kustomize with image patch +# Tests executed: +# - it_phase3_phase4: Ingest + persistence tests +# - it_unified_query_4_6: Query endpoint tests +# - it_temporal_filtering_4_2_fixed: Temporal query tests +# - mem_ingest: Extraction pipeline tests +# - mem_cli::query: Query handler tests apiVersion: batch/v1 kind: Job @@ -23,14 +22,9 @@ metadata: test: integration component: ci-cd spec: - # Don't retry on failure - we want to see the actual error backoffLimit: 0 - - # Timeout after 10 minutes activeDeadlineSeconds: 600 - - # Keep the pod for debugging - ttlSecondsAfterFinished: 3600 # 1 hour + ttlSecondsAfterFinished: 3600 template: metadata: @@ -42,7 +36,7 @@ spec: restartPolicy: Never containers: - # Step 1: Run migrations + # Stage 1: Apply migrations - name: migrate image: forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:IMAGE_SHA imagePullPolicy: IfNotPresent @@ -53,16 +47,16 @@ spec: - | set -e - # Copy migrations script from image to working dir - cp /app/migrations/run_migrations.sh /tmp/run_migrations.sh - chmod +x /tmp/run_migrations.sh + echo "==========================================" + echo "Applying Database Migrations" + echo "==========================================" + echo "" - # Run migrations - /tmp/run_migrations.sh + # Run migrations script + /app/migrations/run_migrations.sh echo "" echo "✓ Migrations complete" - echo "Database ready for tests" env: - name: DB_HOST @@ -87,7 +81,7 @@ spec: memory: "512Mi" cpu: "500m" - # Step 2: Run integration tests + # Stage 2: Run integration tests - name: test image: forgejo.riotpiao.com/riotpiao-poimen/poimen-memory:IMAGE_SHA imagePullPolicy: IfNotPresent @@ -99,133 +93,51 @@ spec: set -e echo "==========================================" - echo "Integration Test: Ingest + Embedding" + echo "Running Integration Tests" echo "==========================================" echo "" - # Start HTTP server - echo "Starting memory-service..." - mem-cli serve --port 8080 & - SERVER_PID=$! - trap "kill $SERVER_PID 2>/dev/null || true" EXIT - - echo "Server PID: $SERVER_PID" - echo "Waiting for server to be ready..." - - # Wait for /health endpoint - for i in {1..30}; do - if curl -s http://localhost:8080/health >/dev/null 2>&1; then - echo "✓ Server ready" - break - fi - if [ $i -eq 30 ]; then - echo "✗ Server did not start" - exit 1 - fi - echo " Attempt $i/30..." - sleep 1 - done + # Run existing integration tests + echo "Test Suite 1: Ingest + Persistence (it_phase3_phase4)" + cargo test --test it_phase3_phase4 --lib 2>&1 | tail -50 || TEST_FAILED=1 echo "" - echo "Running E2E ingest test..." + echo "Test Suite 2: Unified Query (it_unified_query_4_6)" + cargo test --test it_unified_query_4_6 --lib 2>&1 | tail -50 || TEST_FAILED=1 + echo "" + echo "Test Suite 3: Temporal Filtering (it_temporal_filtering_4_2_fixed)" + cargo test --test it_temporal_filtering_4_2_fixed --lib 2>&1 | tail -50 || TEST_FAILED=1 - # Send ingest request - INGEST_ID="test-$(date +%s)" - RESPONSE=$(curl -s -X POST http://localhost:8080/memory/ingest \ - -H "Content-Type: application/json" \ - -H "Authorization: Bearer test-key" \ - -d "{ - \"project\": \"integration-test\", - \"source\": \"k8s-job-test\", - \"ingest_id\": \"$INGEST_ID\", - \"records\": [ - { - \"role\": \"user\", - \"text\": \"Kubernetes [[Docker]] [[Linux]] container platform\", - \"timestamp\": \"2026-09-14T13:00:00Z\", - \"source_position\": 0 - }, - { - \"role\": \"user\", - \"text\": \"Docker [[Container]] microservices architecture\", - \"timestamp\": \"2026-09-14T13:01:00Z\", - \"source_position\": 1 - } - ] - }") + echo "" + echo "Test Suite 4: Ingest Pipeline Unit Tests" + cargo test --lib mem_ingest 2>&1 | tail -100 || TEST_FAILED=1 - # Check response - STATUS=$(echo "$RESPONSE" | jq -r '.status // "error"') - ID=$(echo "$RESPONSE" | jq -r '.ingest_id // empty') + echo "" + echo "Test Suite 5: Query Handler Tests" + cargo test --lib mem_cli::query 2>&1 | tail -100 || TEST_FAILED=1 - if [ -z "$ID" ]; then - echo "✗ FAILED: No ingest_id in response" - echo "Response: $RESPONSE" + echo "" + echo "==========================================" + echo "Integration Tests Complete" + echo "==========================================" + + if [ -n "$TEST_FAILED" ]; then + echo "✗ Some tests failed" exit 1 fi - echo "Ingest ID: $ID" - echo "Status: $STATUS" - echo "" - echo "Polling for completion..." - - # Poll until done - for poll in {1..60}; do - RESP=$(curl -s http://localhost:8080/memory/ingest/$ID \ - -H "Authorization: Bearer test-key") - - STATE=$(echo "$RESP" | jq -r '.status // "unknown"') - - if [ "$STATE" = "done" ]; then - echo "Poll $poll: $STATE ✓" - echo "" - echo "✓ INGEST SUCCESSFUL" - break - elif [ "$STATE" = "failed" ] || [ "$STATE" = "error" ]; then - echo "Poll $poll: $STATE ✗" - echo "Response: $RESP" - echo "✗ INGEST FAILED" - exit 1 - fi - - echo "Poll $poll: $STATE" - sleep 2 - done - - echo "" - echo "Testing query endpoint..." - QUERY=$(curl -s "http://localhost:8080/memory/query?project=integration-test&question=what%20is%20docker" \ - -H "Authorization: Bearer test-key") - - ENTITY_COUNT=$(echo "$QUERY" | jq '.count.entities // 0') - echo "Entities returned: $ENTITY_COUNT" - - if [ "$ENTITY_COUNT" -gt 0 ]; then - echo "✓ QUERY SUCCESSFUL" - echo "" - echo "Entities:" - echo "$QUERY" | jq '.entities[].name' - else - echo "⚠ No entities returned (schema issue)" - echo "✗ Query test FAILED" - exit 1 - fi - - echo "" - echo "==========================================" - echo "✓ ALL TESTS PASSED" - echo "==========================================" + echo "✓ All tests passed" env: - name: DATABASE_URL value: "postgresql://app@memory-db-rw.poimen.svc.cluster.local:5432/memory" - name: RUST_LOG - value: "info,mem_cli=debug,mem_ingest=debug" + value: "info,mem_cli=debug,mem_ingest=debug,mem_store=debug" - name: MEM_AUTH_MODE value: "none" - - # Password via secret + - name: SQLX_OFFLINE + value: "true" - name: PGPASSWORD valueFrom: secretKeyRef: @@ -234,24 +146,22 @@ spec: resources: requests: - memory: "512Mi" - cpu: "200m" - limits: memory: "1Gi" - cpu: "1000m" + cpu: "500m" + limits: + memory: "2Gi" + cpu: "2000m" livenessProbe: exec: command: - /bin/sh - -c - - curl -s http://localhost:8080/health >/dev/null - initialDelaySeconds: 30 - periodSeconds: 10 - failureThreshold: 2 + - test -f /proc/self/status + initialDelaySeconds: 10 + periodSeconds: 30 --- -# ServiceAccount for integration test apiVersion: v1 kind: ServiceAccount metadata: diff --git a/k8s/test/integration-test-secrets.enc.yaml b/k8s/test/integration-test-secrets.enc.yaml new file mode 100644 index 0000000..e69de29 diff --git a/run_production_test.sh b/run_production_test.sh deleted file mode 100755 index 9eff08d..0000000 --- a/run_production_test.sh +++ /dev/null @@ -1,81 +0,0 @@ -#!/bin/bash - -# Master script: Run full production ingest test with logging -# -# Usage: -# ./run_production_test.sh -# ./run_production_test.sh [api-key] -# -# What it does: -# 1. Collect baseline logs -# 2. Run ingest test -# 3. Collect post-test logs -# 4. Analyze for errors -# 5. Display results - -set -e - -API_KEY="${1:-test-key}" -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" - -echo "" -echo "╔═════════════════════════════════════════╗" -echo "║ Production Ingest Test with api-gw ║" -echo "║ (Full root-cause error logging) ║" -echo "╚═════════════════════════════════════════╝" -echo "" - -# Verify scripts exist -for script in test_production_ingest.sh collect_prod_logs.sh; do - if [ ! -f "$SCRIPT_DIR/$script" ]; then - echo "✗ Missing: $script" - exit 1 - fi -done - -echo "Step 1: Collecting baseline logs..." -"$SCRIPT_DIR/collect_prod_logs.sh" before - -echo "" -echo "Step 2: Running ingest test..." -echo " (Sending records through embedding pipeline to api-gw)" -echo "" - -if MEM_API_KEY="$API_KEY" "$SCRIPT_DIR/test_production_ingest.sh"; then - echo "" - echo "✓ Test passed!" - test_status=0 -else - echo "" - echo "✗ Test failed!" - test_status=1 -fi - -echo "" -echo "Step 3: Collecting post-test logs..." -"$SCRIPT_DIR/collect_prod_logs.sh" after - -echo "" -echo "Step 4: Analyzing logs for errors..." -echo "" -"$SCRIPT_DIR/collect_prod_logs.sh" analyze - -echo "" -echo "════════════════════════════════════════" -if [ $test_status -eq 0 ]; then - echo "✓ INGEST TEST PASSED" -else - echo "✗ INGEST TEST FAILED" - echo "" - echo "Next steps:" - echo " 1. Check logs in prod_logs_*/ directory" - echo " 2. Look for errors in:" - echo " - /memory/ingest endpoint response" - echo " - Embedding service (LLM_ENDPOINT)" - echo " - api-gw gateway logs" - echo " - Database connection" -fi -echo "════════════════════════════════════════" -echo "" - -exit $test_status diff --git a/test_prod_ingest_real.sh b/test_prod_ingest_real.sh deleted file mode 100755 index db079d3..0000000 --- a/test_prod_ingest_real.sh +++ /dev/null @@ -1,197 +0,0 @@ -#!/bin/bash - -# Production real test: Full ingest with api-gw + embedding -# Sends records through the complete pipeline and logs all errors -# -# Usage: -# ./test_prod_ingest_real.sh [--verbose] - -set -e - -NAMESPACE="poimen" -SERVICE="poimen-memory" -LOCAL_PORT="9990" -TIMESTAMP=$(date +%Y%m%d-%H%M%S) -LOG_FILE="/tmp/ingest_test_${TIMESTAMP}.log" -VERBOSE="${1:-}" - -{ - echo "==========================================" - echo "Production Ingest Test: $(date)" - echo "==========================================" - echo "" - echo "Namespace: $NAMESPACE" - echo "Service: $SERVICE" - echo "Local Port: $LOCAL_PORT" - echo "Log: $LOG_FILE" - echo "" - - # Start port-forward - echo "Starting port-forward..." - kubectl -n "$NAMESPACE" port-forward "svc/$SERVICE" "$LOCAL_PORT:8080" >/dev/null 2>&1 & - PF_PID=$! - - cleanup() { - if [ -n "$PF_PID" ]; then - kill $PF_PID 2>/dev/null || true - wait $PF_PID 2>/dev/null || true - fi - } - trap cleanup EXIT - - sleep 2 - - if ! kill -0 $PF_PID 2>/dev/null; then - echo "✗ Port-forward failed" - exit 1 - fi - echo "✓ Port-forward running (PID $PF_PID)" - echo "" - - # Check health - echo "Checking /health endpoint..." - if ! curl -s "http://localhost:$LOCAL_PORT/health" >/dev/null 2>&1; then - echo "✗ Health check failed" - exit 1 - fi - echo "✓ Health check passed" - echo "" - - # Prepare ingest request - INGEST_ID="ingest-test-${TIMESTAMP}" - - PAYLOAD=$(cat <<'EOFPAYLOAD' -{ - "project": "production-real-test", - "source": "integration-test", - "ingest_id": "INGEST_ID_PLACEHOLDER", - "records": [ - { - "role": "user", - "text": "Kubernetes [[Docker]] [[Linux]] is an open-source container orchestration platform. It automates many manual processes involved in deploying, managing, and scaling containerized applications.", - "timestamp": "2026-09-14T13:00:00Z", - "source_position": 0 - }, - { - "role": "user", - "text": "Docker [[Container]] [[Go]] is a containerization platform that makes it easier to build, ship, and run applications. Docker achieves high efficiency through the use of operating system-level virtualization.", - "timestamp": "2026-09-14T13:01:00Z", - "source_position": 1 - }, - { - "role": "user", - "text": "Go [[Concurrency]] [[Static Typing]] is a programming language designed at Google. It is statically typed, compiled, and known for its simplicity, concurrent programming model, and efficient execution.", - "timestamp": "2026-09-14T13:02:00Z", - "source_position": 2 - } - ] -} -EOFPAYLOAD -) - - # Replace placeholder - PAYLOAD="${PAYLOAD//INGEST_ID_PLACEHOLDER/$INGEST_ID}" - - echo "Sending ingest request..." - if [ -n "$VERBOSE" ]; then - echo "Payload:" - echo "$PAYLOAD" | jq . 2>/dev/null || echo "$PAYLOAD" - echo "" - fi - - RESPONSE=$(curl -s -w "\n%{http_code}" -X POST \ - "http://localhost:$LOCAL_PORT/memory/ingest" \ - -H "Content-Type: application/json" \ - -H "Authorization: Bearer test-key" \ - -d "$PAYLOAD") - - HTTP_CODE=$(echo "$RESPONSE" | tail -1) - BODY=$(echo "$RESPONSE" | head -n-1) - - echo "HTTP Status: $HTTP_CODE" - if [ "$HTTP_CODE" != "202" ]; then - echo "✗ Unexpected HTTP status" - echo "Response: $BODY" - exit 1 - fi - echo "✓ Request accepted" - echo "" - - if [ -n "$VERBOSE" ]; then - echo "Response body:" - echo "$BODY" | jq . 2>/dev/null || echo "$BODY" - echo "" - fi - - # Extract ID - ID=$(echo "$BODY" | jq -r '.ingest_id // empty' 2>/dev/null) - if [ -z "$ID" ]; then - echo "✗ Missing ingest_id in response" - echo "Response: $BODY" - exit 1 - fi - - echo "Ingest ID: $ID" - echo "" - - # Poll status - echo "Polling job status..." - echo "==========================================" - - MAX_POLLS=120 # 10 minutes at 5s intervals - poll_count=0 - - while [ $poll_count -lt $MAX_POLLS ]; do - poll_count=$((poll_count+1)) - - STATUS_RESP=$(curl -s "http://localhost:$LOCAL_PORT/memory/ingest/$ID" \ - -H "Authorization: Bearer test-key") - - STATUS=$(echo "$STATUS_RESP" | jq -r '.status // "unknown"' 2>/dev/null) - - printf "[%3d] %-20s" "$poll_count" "$STATUS" - - case "$STATUS" in - done) - echo " ✓" - echo "==========================================" - echo "" - echo "✓ SUCCESS: Ingest completed" - - if [ -n "$VERBOSE" ]; then - echo "" - echo "Final response:" - echo "$STATUS_RESP" | jq . 2>/dev/null || echo "$STATUS_RESP" - fi - exit 0 - ;; - failed|error) - echo " ✗" - echo "==========================================" - echo "" - echo "✗ FAILED: Ingest did not complete" - echo "" - echo "Final response:" - echo "$STATUS_RESP" | jq . 2>/dev/null || echo "$STATUS_RESP" - exit 1 - ;; - processing|queued|pending) - echo "" - sleep 5 - ;; - *) - echo " (unknown)" - sleep 5 - ;; - esac - done - - echo "==========================================" - echo "" - echo "✗ TIMEOUT: Ingest did not complete after ${MAX_POLLS} polls (${poll_count}m)" - exit 1 - -} 2>&1 | tee "$LOG_FILE" - -echo "" -echo "Full log saved to: $LOG_FILE" diff --git a/tests/integration_ingest_with_gw.rs b/tests/integration_ingest_with_gw.rs deleted file mode 100644 index 30c5756..0000000 --- a/tests/integration_ingest_with_gw.rs +++ /dev/null @@ -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'"); -} diff --git a/tests/unit_ingest_logging.rs b/tests/unit_ingest_logging.rs deleted file mode 100644 index 8f89425..0000000 --- a/tests/unit_ingest_logging.rs +++ /dev/null @@ -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" - ); - } -}