tests/live_audit.py

import httpx
import asyncio
import json
import time

BASE_URL = "http://localhost:8004"

API_KEY = "GUVI_HACKATHON_V2"

async def test_live_analysis():
    print("--- Phase 1: Real API Call Execution ---")
    
    scam_message = "Hello, this is SBI Bank. Your KYC is expired. Please update it here: http://sbi-kyc-update.net. Transfer 5000 to upi: verify.kyc@ybl to prevent account block. Contact +91-9876543210."
    
    payload = {
        "message": scam_message,
        "sessionId": "live-test-session-001",
        "timestamp": int(time.time()),
        "conversationHistory": []
    }
    
    headers = {
        "x-api-key": API_KEY,
        "Content-Type": "application/json"
    }
    
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            start_time = time.time()
            resp = await client.post(f"{BASE_URL}/api/guvi/analyze", json=payload, headers=headers)
            duration = time.time() - start_time
            
            print(f"Status Code: {resp.status_code}")
            print(f"Response Time: {duration:.2f}s")
            
            if resp.status_code == 200:
                result = resp.json()
                print("--- Response Validation ---")
                print(f"Scam Detected: {result.get('scamDetected')}")
                print(f"Risk Level: {result.get('riskLevel')}")
                
                intel = result.get("extractedIntelligence", {})
                print(f"Extracted UPIs: {intel.get('upiIds')}")
                print(f"Extracted Links: {intel.get('phishingLinks')}")
                print(f"Extracted Phones: {intel.get('phoneNumbers')}")
                
                print("--- Agent Notes ---")
                print(result.get("agentNotes"))
                
                print("--- Honeypot Response ---")
                print(result.get("reply"))
                
                # Verification of AI Reasoning
                notes = result.get("agentNotes", "")
                if "[AI THOUGHT TRACE]" in notes:
                    print("\n[SUCCESS]: Real LLM Inference Verified via Thought Trace.")
                    # Extract and print snippet for visual audit
                    trace_snippet = notes.split("[AI THOUGHT TRACE]:")[-1].strip()
                    print(f"DEBUG: Trace Sample: {trace_snippet[:100]}...")
                else:
                    print("\n[WARNING]: No AI Thought trace found in agentNotes.")
                    print("   This might indicate a fallback to heuristic mode or a capture failure.")
                    
                # Check for Agentic Steps
                if result.get("agentNotes"):
                    print(f"INFO: Agent Notes: {notes[:150]}...")
            else:
                print(f"Error: {resp.text}")
                
        except Exception as e:
            print(f"Connection Failed: {e}")

async def test_threat_feeds():
    print("\n--- Phase 3: Real Threat Feed Ingestion ---")
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            resp = await client.get(f"{BASE_URL}/api/v1/intelligence/feed-stream-v2", headers={"x-api-key": API_KEY})
            if resp.status_code == 200:
                data = resp.json()
                feeds = data.get("feeds", [])
                print(f"Fetched {len(feeds)} live feed items.")
                
                if not feeds:
                    print(f"DEBUG: Full Response: {json.dumps(data, indent=2)}")
                
                sources = set(f.get("source") for f in feeds)
                print(f"Feed Sources: {sources}")
                
                if "Simulation_Fallback" in sources:
                    print("!!! FLAG: System fell back to Simulation for threat feeds.")
                else:
                    print("--- Verified Live Threat Feed Ingestion ---")
            else:
                print(f"Feed Error: {resp.text}")
        except Exception as e:
            print(f"Feed Connection Failed: {e}")

if __name__ == "__main__":
    asyncio.run(test_live_analysis())
    asyncio.run(test_threat_feeds())