""" 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())