Spaces:
Running
Running
File size: 2,140 Bytes
8c9362b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
import asyncio
import sys
import os
from datetime import datetime
# Add project root to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.ai.agent.graph import get_aida_graph
from app.ai.agent.schemas import ListingDraft
async def verify_publish():
print("🚀 Verifying Publish Flow...")
graph = get_aida_graph()
session_id = f"verify_pub_{int(datetime.now().timestamp())}"
user_id = "tester"
# Manually inject state that is ready to publish
# We can't easily inject state into LangGraph memory without running the flow
# So we'll run the flow quickly
# 1. Start Listing
print("\n1. creating listing...")
state_1 = await graph.ainvoke({
"user_id": user_id,
"session_id": session_id,
"user_role": "landlord",
"last_user_message": "List a 2-bed in Lagos for 50k",
"start_new_session": True
}, config={"configurable": {"thread_id": session_id}, "recursion_limit": 50})
if state_1.get("listing_draft"):
print(" Draft created successfully")
else:
print("❌ Draft creation failed")
return
# 2. Publish
print("\n2. Sending 'publish' command...")
state_2 = await graph.ainvoke({
"user_id": user_id,
"session_id": session_id,
"user_role": "landlord",
"last_user_message": "publish"
}, config={"configurable": {"thread_id": session_id}, "recursion_limit": 50})
temp_data = state_2.get("temp_data", {})
action = temp_data.get("action")
replace = temp_data.get("replace_last_message")
draft_ui = temp_data.get("draft_ui", {})
print(f" Action: {action} (Expected: published)")
print(f" Replace Flag: {replace} (Expected: True)")
print(f" Draft UI Status: {draft_ui.get('status')} (Expected: published)")
if action == "published" and replace is True and draft_ui.get("status") == "published":
print("\n✅ PUBLISH VERIFICATION PASSED!")
else:
print("\n❌ PUBLISH VERIFICATION FAILED")
if __name__ == "__main__":
asyncio.run(verify_publish())
|