Spaces:
Sleeping
Sleeping
File size: 5,689 Bytes
c50a873 | 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 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | import os
import sys
import json
from datetime import datetime
# Add app to path
sys.path.insert(0, os.path.abspath('.'))
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.core.database import Base, get_db
from app.main import limiter
# 1. Setup isolated database
TEST_DB_PATH = "./data/e2e_test.db"
if os.path.exists(TEST_DB_PATH):
try:
os.remove(TEST_DB_PATH)
except:
pass
TEST_DB = f"sqlite:///{TEST_DB_PATH}"
engine = create_engine(TEST_DB, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
from app.core.config import settings
# Force fallback generator to bypass rate limits
settings.GEMINI_API_KEYS = ""
app.dependency_overrides[get_db] = override_get_db
limiter._storage.reset() # Reset rate limits
client = TestClient(app)
print("\n--- Starting HALE E2E Simulation ---\n")
try:
# 2. Register
print("1. Registering new user...")
username = f"test_user_{int(datetime.now().timestamp())}"
res = client.post("/api/auth/register", json={
"user_id": username,
"name": "Bruce Wayne",
"password": "Batman_Password1"
})
assert res.status_code == 201, res.text
token = res.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
print(f"User registered: {username}")
# 3. Update Profile
print("\n2. Changing Coach Personality to 'drill_sergeant'...")
res = client.patch("/api/auth/profile", headers=headers, json={
"coach_personality": "drill_sergeant"
})
assert res.status_code == 200, res.text
print(f"Profile updated. Coach is now: {res.json()['coach_personality']}")
# 4. Create Goal
print("\n3. Creating a new goal: 'Learn Neuroscience in 7 days'")
print("... Waiting for AI Curriculum Generator (might take a few seconds)...")
res = client.post("/api/goals/", headers=headers, json={
"goal": "Learn Neuroscience basics",
"category": "education",
"duration_days": 10
})
assert res.status_code == 201, res.text
goal_data = res.json()
print(f"Goal Created! AI generated {len(goal_data['plan']['modules'])} modules.")
print(f" Sample module: {goal_data['plan']['modules'][0]['title']}")
# 5. Get Initial Daily Plan
print("\n4. Fetching today's personalized plan (Normal State)...")
res = client.get("/api/plan", headers=headers)
assert res.status_code == 200, res.text
plan_data = res.json()
print(f"Plan fetched:")
print(f" Topic: {plan_data['topic']}")
print(f" Behavior State: {plan_data['behavior']['action_label']} (Intensity: {plan_data['behavior']['intensity']})")
print(" Tasks:")
for t in plan_data['tasks']:
print(f" - {t}")
print(f" Coach Message (Drill Sergeant): \"{plan_data['coach_message']}\"")
# 6. Update State (Simulate high fatigue / sickness)
print("\n5. Oh no! User is extremely tired and burned out. Updating state...")
res = client.post("/api/update-state", headers=headers, json={
"fatigue": 0.9,
"mood": 0.2,
"stress": 0.8
})
assert res.status_code == 200, res.text
print("State updated successfully.")
# 7. Get Adapted Daily Plan
print("\n6. Fetching daily plan again to check AI adaptation...")
res = client.get("/api/plan", headers=headers)
assert res.status_code == 200, res.text
adapted_plan = res.json()
print(f"Adapted Plan fetched:")
print(f" Behavior State: {adapted_plan['behavior']['action_label']} (Intensity: {adapted_plan['behavior']['intensity']})")
print(" Adapted Tasks:")
for t in adapted_plan['tasks']:
print(f" - {t}")
print(" Notice how the tasks changed dynamically based on fatigue!")
# 8. Complete Task
print("\n7. Completing the tasks...")
today = datetime.today().strftime('%Y-%m-%d')
res = client.post("/api/complete-task", headers=headers, json={
"date": today,
"task_index": 0,
"actual_duration_min": 15,
"notes": "Was tired but pushed through a quick review."
})
assert res.status_code == 200, res.text
print(f"Task Completed! Result: {json.dumps(res.json(), ensure_ascii=True)}")
# 9. Check Complete Gamification Stats
print("\n8. Checking Gamification Stats...")
res = client.get("/api/stats", headers=headers)
assert res.status_code == 200, res.text
stats = res.json()
print("Current Stats:")
print(f" Level: {stats['level']} | XP: {stats['xp']}")
print(f" Current Streak: {stats['current_streak']}")
print(f" Tasks Completed: {stats['total_tasks_completed']}")
# 10. Generate Micro-Lesson
print("\n9. Asking AI to generate a quick lesson on 'Neurons'...")
res = client.get("/api/lesson/Neurons", headers=headers)
assert res.status_code == 200, res.text
print("Lesson generated successfully!")
print(f" Excerpt: {res.json()['content'][:150]}...\n")
print("--- E2E Simulation Completed Successfully! ---")
except Exception as e:
print(f"\nE2E Run Failed! Error:\n{e}")
finally:
# Cleanup DB
try:
if os.path.exists("./data/e2e_test.db"):
# Ensure engine is disposed to release the file handle
engine.dispose()
os.remove("./data/e2e_test.db")
except Exception as cleanup_err:
print(f"Cleanup warning: {cleanup_err}")
|