mirror of
https://github.com/arkorty/B.Tech-Project-III.git
synced 2026-04-19 12:41:48 +00:00
413 lines
20 KiB
Python
413 lines
20 KiB
Python
"""
|
|
Milestone 5 — Full Feature Test Suite
|
|
Run from backend/ directory:
|
|
cd backend && python ../test/test_milestone5_all.py
|
|
|
|
Tests all 8 feature modules locally (no Telegram) against live Gemini + Tavily APIs.
|
|
"""
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/backend")
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
PASS = "✅"
|
|
FAIL = "❌"
|
|
SKIP = "⏭"
|
|
|
|
results = {}
|
|
|
|
|
|
# ─── Test helpers ─────────────────────────────────────────────────────────────
|
|
|
|
def check(name: str, condition: bool, detail: str = ""):
|
|
icon = PASS if condition else FAIL
|
|
results[name] = condition
|
|
print(f" {icon} {name}" + (f": {detail}" if detail else ""))
|
|
|
|
|
|
# ─── 0. Database init ─────────────────────────────────────────────────────────
|
|
|
|
async def test_database():
|
|
print("\n── Test 0: Database Init ──")
|
|
from database import init_db
|
|
try:
|
|
await init_db()
|
|
check("init_db runs", True)
|
|
except Exception as e:
|
|
check("init_db runs", False, str(e))
|
|
|
|
from database import (
|
|
create_user, get_user, create_negotiation,
|
|
add_participant, get_rounds, store_analytics, get_analytics
|
|
)
|
|
await create_user(99999, "testuser", "Test User")
|
|
user = await get_user(99999)
|
|
check("create_user + get_user", user is not None)
|
|
|
|
neg_id = await create_negotiation("expenses", 99999)
|
|
check("create_negotiation", len(neg_id) > 0, f"id={neg_id}")
|
|
|
|
await add_participant(neg_id, 99999, {"goal": "test"}, "balanced")
|
|
check("add_participant", True)
|
|
|
|
rows = await get_rounds(neg_id)
|
|
check("get_rounds returns list", isinstance(rows, list))
|
|
|
|
await store_analytics({
|
|
"negotiation_id": neg_id,
|
|
"satisfaction_timeline": "[]",
|
|
"concession_log": "[]",
|
|
"fairness_score": 80.0,
|
|
"total_concessions_a": 2,
|
|
"total_concessions_b": 1,
|
|
})
|
|
analytics = await get_analytics(neg_id)
|
|
check("store + get analytics", analytics is not None and analytics.get("fairness_score") == 80.0)
|
|
|
|
|
|
# ─── 1. Feature context fetching ─────────────────────────────────────────────
|
|
|
|
async def test_feature_contexts():
|
|
print("\n── Test 1: Feature Context Fetching ──")
|
|
|
|
from features.base_feature import get_feature
|
|
|
|
PREFS = {
|
|
"scheduling": (
|
|
{"raw_details": {"available_windows": ["Monday 10am-12pm", "Wednesday 3-5pm"], "location": "Bandra"}, "goal": "Schedule a coffee meeting", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
{"raw_details": {"available_windows": ["Monday 11am-1pm", "Friday 2-4pm"]}, "goal": "Find a time to meet", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
"expenses": (
|
|
{"raw_details": {"expenses": [{"name": "Hotel", "amount": 12000}, {"name": "Fuel", "amount": 3000}], "upi_id": "rahul@paytm"}, "goal": "Split trip expenses", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
{"raw_details": {"expenses": [{"name": "Dinner", "amount": 2000}]}, "goal": "Fair expense split", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
"collaborative": (
|
|
{"raw_details": {"cuisine": "Italian", "location": "Bandra", "budget": 800}, "goal": "Pick dinner spot", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
{"raw_details": {"cuisine": "Chinese", "location": "Bandra", "budget": 600}, "goal": "Restaurant for tonight", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
"marketplace": (
|
|
{"raw_details": {"item": "PS5 with 2 controllers", "asking_price": 35000, "minimum_price": 30000, "role": "seller", "upi_id": "seller@upi"}, "goal": "Sell PS5", "constraints": [], "preferences": [], "tone": "firm"},
|
|
{"raw_details": {"item": "PS5", "budget": 28000, "role": "buyer"}, "goal": "Buy PS5", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
"freelance": (
|
|
{"raw_details": {"skill": "React developer", "rate": 1500, "hours": 40, "upfront_minimum": "50"}, "goal": "Freelance React project", "constraints": [{"value": "minimum rate ₹1500/hr", "hard": True}], "preferences": [], "tone": "professional"},
|
|
{"raw_details": {"budget": 40000, "project_type": "web app", "required_features": ["auth", "dashboard", "API"], "role": "client"}, "goal": "Build web app in budget", "constraints": [{"value": "budget max ₹40000", "hard": True}], "preferences": [], "tone": "professional"},
|
|
),
|
|
"roommate": (
|
|
{"raw_details": {"decision_type": "wifi plan", "city": "Mumbai", "budget": 600}, "goal": "Pick WiFi plan", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
{"raw_details": {"decision_type": "wifi plan", "city": "Mumbai", "budget": 700}, "goal": "Fast internet within budget", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
"trip": (
|
|
{"raw_details": {"available_dates": ["March 15-17", "March 22-24"], "budget": 5000, "destination_preference": "beach", "origin": "Mumbai"}, "goal": "Weekend beach trip", "constraints": [], "preferences": [], "tone": "excited"},
|
|
{"raw_details": {"available_dates": ["March 15-17", "April 5-7"], "budget": 4000, "destination_preference": "hills", "origin": "Mumbai"}, "goal": "Weekend getaway", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
"conflict": (
|
|
{"raw_details": {"conflict_type": "parking spot", "position": "I need the spot Mon-Wed", "relationship_importance": "high"}, "goal": "Resolve parking dispute", "constraints": [], "preferences": [], "tone": "firm"},
|
|
{"raw_details": {"conflict_type": "parking spot", "position": "I need the spot Tue-Thu", "relationship_importance": "high"}, "goal": "Fair parking arrangement", "constraints": [], "preferences": [], "tone": "friendly"},
|
|
),
|
|
}
|
|
|
|
for feature_type, (prefs_a, prefs_b) in PREFS.items():
|
|
try:
|
|
feat = get_feature(feature_type)
|
|
ctx = await feat.get_context(prefs_a, prefs_b)
|
|
check(f"{feature_type}.get_context", len(ctx) > 50, f"{len(ctx)} chars")
|
|
except Exception as e:
|
|
check(f"{feature_type}.get_context", False, str(e))
|
|
|
|
|
|
# ─── 2. Full negotiation for each feature (live Gemini) ──────────────────────
|
|
|
|
async def test_negotiation_feature(feature_type: str, prefs_a: dict, prefs_b: dict, check_fn=None):
|
|
"""Run a full negotiation for one feature and verify the result."""
|
|
from agents.negotiation import run_negotiation
|
|
from features.base_feature import get_feature
|
|
import database as db
|
|
|
|
feat = get_feature(feature_type)
|
|
try:
|
|
feature_context = await feat.get_context(prefs_a, prefs_b)
|
|
except Exception:
|
|
feature_context = ""
|
|
|
|
neg_id = await db.create_negotiation(feature_type, 99901)
|
|
await db.add_participant(neg_id, 99901, prefs_a)
|
|
await db.add_participant(neg_id, 99902, prefs_b)
|
|
|
|
resolution = await run_negotiation(
|
|
negotiation_id=neg_id,
|
|
preferences_a=prefs_a,
|
|
preferences_b=prefs_b,
|
|
user_a_id=99901,
|
|
user_b_id=99902,
|
|
feature_type=feature_type,
|
|
personality_a="balanced",
|
|
personality_b="balanced",
|
|
feature_context=feature_context,
|
|
)
|
|
|
|
check(f"{feature_type} negotiation completes", resolution is not None)
|
|
check(f"{feature_type} has status", resolution.get("status") in ("resolved", "escalated"),
|
|
resolution.get("status"))
|
|
check(f"{feature_type} has rounds_taken", isinstance(resolution.get("rounds_taken"), int),
|
|
str(resolution.get("rounds_taken")))
|
|
check(f"{feature_type} has final_proposal", isinstance(resolution.get("final_proposal"), dict))
|
|
|
|
# Feature-specific checks
|
|
if check_fn:
|
|
check_fn(resolution, prefs_a, prefs_b)
|
|
|
|
# Verify format_resolution produces non-empty string
|
|
try:
|
|
formatted = feat.format_resolution(resolution, prefs_a, prefs_b)
|
|
check(f"{feature_type} format_resolution non-empty", len(formatted) > 30, f"{len(formatted)} chars")
|
|
print(f"\n 📄 Formatted resolution preview:\n {formatted[:200].replace(chr(10), chr(10)+' ')}\n")
|
|
except Exception as e:
|
|
check(f"{feature_type} format_resolution", False, str(e))
|
|
|
|
return resolution
|
|
|
|
|
|
async def test_all_negotiations():
|
|
print("\n── Test 2: Full Negotiations (live Gemini + Tavily) ──")
|
|
print(" (This may take 2-4 minutes due to API rate limits)\n")
|
|
|
|
from database import init_db
|
|
await init_db()
|
|
|
|
# ── Feature 1: Scheduling ──
|
|
print(" [1/8] Scheduling...")
|
|
await test_negotiation_feature(
|
|
"scheduling",
|
|
{"goal": "Schedule a coffee meeting", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "scheduling",
|
|
"raw_details": {"available_windows": ["Monday 10am-12pm", "Wednesday 3-5pm"], "location": "Bandra", "duration": "1 hour"}},
|
|
{"goal": "Coffee meeting next week", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "scheduling",
|
|
"raw_details": {"available_windows": ["Monday 11am-1pm", "Wednesday 4-6pm"]}},
|
|
)
|
|
|
|
# ── Feature 2: Expenses ──
|
|
print(" [2/8] Expenses...")
|
|
|
|
def check_expense_math(resolution, prefs_a, prefs_b):
|
|
# Verify the resolution doesn't hallucinate wrong math
|
|
final = resolution.get("final_proposal", {})
|
|
details = final.get("details", {})
|
|
# At minimum, the proposal should exist
|
|
check("expenses has details", bool(details), str(details)[:80])
|
|
|
|
await test_negotiation_feature(
|
|
"expenses",
|
|
{"goal": "Split Goa trip expenses fairly", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "expenses",
|
|
"raw_details": {"expenses": [{"name": "Hotel", "amount": 12000}, {"name": "Fuel", "amount": 3000}], "upi_id": "rahul@paytm"}},
|
|
{"goal": "Fair expense split", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "expenses",
|
|
"raw_details": {"expenses": [{"name": "Dinner", "amount": 2000}]}},
|
|
check_fn=check_expense_math,
|
|
)
|
|
|
|
# ── Feature 3: Collaborative ──
|
|
print(" [3/8] Collaborative (uses Tavily)...")
|
|
|
|
def check_collaborative(resolution, prefs_a, prefs_b):
|
|
final = resolution.get("final_proposal", {})
|
|
summary = final.get("summary", "")
|
|
check("collaborative has venue recommendation", bool(summary), summary[:60])
|
|
|
|
await test_negotiation_feature(
|
|
"collaborative",
|
|
{"goal": "Pick a restaurant for dinner", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "collaborative",
|
|
"raw_details": {"cuisine": "Italian", "location": "Bandra Mumbai", "budget": 800}},
|
|
{"goal": "Dinner somewhere nice", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "collaborative",
|
|
"raw_details": {"cuisine": "Chinese", "location": "Bandra Mumbai", "budget": 600}},
|
|
check_fn=check_collaborative,
|
|
)
|
|
|
|
# ── Feature 4: Marketplace ──
|
|
print(" [4/8] Marketplace (uses Tavily)...")
|
|
|
|
def check_marketplace(resolution, prefs_a, prefs_b):
|
|
final = resolution.get("final_proposal", {})
|
|
details = final.get("details", {})
|
|
price = details.get("agreed_price") or details.get("price") or details.get("final_price") or ""
|
|
reasoning = resolution.get("summary", "")
|
|
check("marketplace has price", bool(price) or bool(reasoning), f"price={price}")
|
|
|
|
await test_negotiation_feature(
|
|
"marketplace",
|
|
{"goal": "Sell my PS5", "constraints": [{"value": "minimum ₹30000", "hard": True}],
|
|
"preferences": [], "tone": "firm", "feature_type": "marketplace",
|
|
"raw_details": {"item": "PS5 with 2 controllers", "asking_price": 35000, "minimum_price": 30000, "role": "seller", "upi_id": "seller@upi"}},
|
|
{"goal": "Buy a PS5 in budget", "constraints": [{"value": "budget max ₹28000", "hard": True}],
|
|
"preferences": [], "tone": "friendly", "feature_type": "marketplace",
|
|
"raw_details": {"item": "PS5", "budget": 28000, "role": "buyer", "max_budget": 28000}},
|
|
check_fn=check_marketplace,
|
|
)
|
|
|
|
# ── Feature 5: Freelance ──
|
|
print(" [5/8] Freelance (uses Tavily + Calculator)...")
|
|
|
|
def check_freelance(resolution, prefs_a, prefs_b):
|
|
final = resolution.get("final_proposal", {})
|
|
details = final.get("details", {})
|
|
budget = details.get("budget") or details.get("agreed_budget") or details.get("price") or ""
|
|
check("freelance has budget in proposal", bool(budget) or bool(details), f"budget={budget}")
|
|
|
|
await test_negotiation_feature(
|
|
"freelance",
|
|
{"goal": "Get paid fairly for React project", "constraints": [{"value": "min rate ₹1500/hr", "hard": True}],
|
|
"preferences": [], "tone": "professional", "feature_type": "freelance",
|
|
"raw_details": {"skill": "React developer", "rate": 1500, "hours": 40, "upfront_minimum": "50"}},
|
|
{"goal": "Build web app within ₹40k budget", "constraints": [{"value": "budget max ₹40000", "hard": True}],
|
|
"preferences": [], "tone": "professional", "feature_type": "freelance",
|
|
"raw_details": {"budget": 40000, "project_type": "web app", "required_features": ["auth", "dashboard", "API"], "role": "client"}},
|
|
check_fn=check_freelance,
|
|
)
|
|
|
|
# ── Feature 6: Roommate ──
|
|
print(" [6/8] Roommate (uses Tavily for real plans)...")
|
|
|
|
def check_roommate(resolution, prefs_a, prefs_b):
|
|
final = resolution.get("final_proposal", {})
|
|
summary = final.get("summary", "")
|
|
check("roommate has decision", bool(summary), summary[:60])
|
|
|
|
await test_negotiation_feature(
|
|
"roommate",
|
|
{"goal": "Pick a shared WiFi plan", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "roommate",
|
|
"raw_details": {"decision_type": "wifi plan", "city": "Mumbai", "budget": 600}},
|
|
{"goal": "Fast internet within budget", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "roommate",
|
|
"raw_details": {"decision_type": "wifi plan", "city": "Mumbai", "budget": 700}},
|
|
check_fn=check_roommate,
|
|
)
|
|
|
|
# ── Feature 7: Conflict ──
|
|
print(" [7/8] Conflict Resolution...")
|
|
|
|
await test_negotiation_feature(
|
|
"conflict",
|
|
{"goal": "Resolve parking spot dispute", "constraints": [], "preferences": [],
|
|
"tone": "firm", "feature_type": "conflict",
|
|
"raw_details": {"conflict_type": "parking spot", "position": "I need Mon-Wed", "relationship_importance": "high"}},
|
|
{"goal": "Fair parking arrangement", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "conflict",
|
|
"raw_details": {"conflict_type": "parking spot", "position": "I need Tue-Thu", "relationship_importance": "high"}},
|
|
)
|
|
|
|
# ── Feature 8: Trip (Group Negotiation) ──
|
|
print(" [8/8] Trip Planning (group negotiation with 3 preferences)...")
|
|
from features.trip import run_group_negotiation
|
|
from database import init_db as _init
|
|
|
|
all_prefs = [
|
|
{"goal": "Beach weekend trip", "constraints": [], "preferences": [],
|
|
"tone": "excited", "feature_type": "trip",
|
|
"raw_details": {"available_dates": ["March 15-17", "March 22-24"], "budget": 5000, "destination_preference": "beach", "origin": "Mumbai"}},
|
|
{"goal": "Weekend trip with friends", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "trip",
|
|
"raw_details": {"available_dates": ["March 15-17", "April 5-7"], "budget": 4000, "destination_preference": "hills", "origin": "Mumbai"}},
|
|
{"goal": "Budget getaway", "constraints": [], "preferences": [],
|
|
"tone": "friendly", "feature_type": "trip",
|
|
"raw_details": {"available_dates": ["March 15-17", "March 29-31"], "budget": 3500, "destination_preference": "any", "origin": "Mumbai"}},
|
|
]
|
|
from database import create_negotiation, add_participant
|
|
trip_neg_id = await create_negotiation("trip", 99901)
|
|
for i, p in enumerate(all_prefs):
|
|
await add_participant(trip_neg_id, 99901 + i, p)
|
|
|
|
trip_resolution = {"status": None}
|
|
async def on_trip_round(data):
|
|
print(f" Round {data['round_number']}: {data['action']} | avg_sat={data['satisfaction_score']:.0f}")
|
|
|
|
async def on_trip_resolve(data):
|
|
trip_resolution["status"] = data.get("status")
|
|
trip_resolution["summary"] = data.get("summary", "")
|
|
trip_resolution["data"] = data
|
|
|
|
await run_group_negotiation(
|
|
negotiation_id=trip_neg_id,
|
|
all_preferences=all_prefs,
|
|
all_user_ids=[99901, 99902, 99903],
|
|
feature_type="trip",
|
|
personalities=["balanced", "empathetic", "balanced"],
|
|
on_round_update=on_trip_round,
|
|
on_resolution=on_trip_resolve,
|
|
)
|
|
check("trip group negotiation completes", trip_resolution["status"] in ("resolved", "escalated"),
|
|
trip_resolution.get("status"))
|
|
check("trip has summary", bool(trip_resolution.get("summary")))
|
|
|
|
from features.trip import TripFeature
|
|
if trip_resolution.get("data"):
|
|
try:
|
|
formatted = TripFeature().format_resolution(trip_resolution["data"], all_prefs[0], all_prefs[1])
|
|
check("trip format_resolution", len(formatted) > 30, f"{len(formatted)} chars")
|
|
except Exception as e:
|
|
check("trip format_resolution", False, str(e))
|
|
|
|
|
|
# ─── 3. Dispatcher test ───────────────────────────────────────────────────────
|
|
|
|
async def test_dispatcher():
|
|
print("\n── Test 3: Feature Dispatcher ──")
|
|
from features.base_feature import get_feature
|
|
from features.scheduling import SchedulingFeature
|
|
from features.expenses import ExpensesFeature
|
|
from features.collaborative import CollaborativeFeature
|
|
from features.marketplace import MarketplaceFeature
|
|
from features.freelance import FreelanceFeature
|
|
from features.roommate import RoommateFeature
|
|
from features.trip import TripFeature
|
|
from features.conflict import ConflictFeature
|
|
from features.generic import GenericFeature
|
|
|
|
for feature_type, expected_cls in [
|
|
("scheduling", SchedulingFeature), ("expenses", ExpensesFeature),
|
|
("collaborative", CollaborativeFeature), ("marketplace", MarketplaceFeature),
|
|
("freelance", FreelanceFeature), ("roommate", RoommateFeature),
|
|
("trip", TripFeature), ("conflict", ConflictFeature),
|
|
("generic", GenericFeature), ("unknown_xyz", GenericFeature),
|
|
]:
|
|
feat = get_feature(feature_type)
|
|
check(f"get_feature('{feature_type}')", isinstance(feat, expected_cls))
|
|
|
|
|
|
# ─── Summary ─────────────────────────────────────────────────────────────────
|
|
|
|
async def main():
|
|
print("=" * 60)
|
|
print(" negoT8 — Milestone 5 Test Suite")
|
|
print("=" * 60)
|
|
|
|
await test_database()
|
|
await test_dispatcher()
|
|
await test_feature_contexts()
|
|
# Full negotiations use live Gemini + Tavily — skip if you want a fast run
|
|
await test_all_negotiations()
|
|
|
|
print("\n" + "=" * 60)
|
|
total = len(results)
|
|
passed = sum(1 for v in results.values() if v)
|
|
failed = total - passed
|
|
print(f" RESULTS: {passed}/{total} passed" + (f" | {failed} FAILED" if failed else " ✅ All passed!"))
|
|
print("=" * 60)
|
|
|
|
if failed:
|
|
print("\nFailed tests:")
|
|
for name, status in results.items():
|
|
if not status:
|
|
print(f" ❌ {name}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|