Spaces:
Sleeping
Sleeping
| """One-command check that a Supabase project is set up for the Creighton Anatomy Learning Platform. | |
| Run it after creating the tables with supabase_schema.sql: | |
| export SUPABASE_URL="https://YOUR-PROJECT-REF.supabase.co" | |
| export SUPABASE_KEY="YOUR-SERVICE-ROLE-KEY" | |
| python3 verify_supabase.py | |
| It creates a throwaway account (verify-<timestamp>), exercises every operation | |
| the app performs (account creation, sign-in checks, chat history, session | |
| timing, click events), reads everything back, and deletes the account again. | |
| It touches no real user data. Exit code 0 means every check passed. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| import time | |
| import uuid | |
| from supabase_store import EXISTS, SupabaseStore, is_valid_email, is_valid_user_id | |
| PASS, FAIL = " PASS ", " FAIL " | |
| failures: list[str] = [] | |
| def check(name: str, ok: bool, detail: str = "") -> bool: | |
| print((PASS if ok else FAIL) + name + (f" [{detail}]" if detail and not ok else "")) | |
| if not ok: | |
| failures.append(name) | |
| return ok | |
| def main() -> int: | |
| store = SupabaseStore() | |
| print("\n1. Configuration") | |
| if not check("SUPABASE_URL and SUPABASE_KEY are set", store.enabled, store.config_hint()): | |
| print('\nSet them first:\n export SUPABASE_URL="https://YOUR-PROJECT-REF.supabase.co"\n export SUPABASE_KEY="YOUR-SERVICE-ROLE-KEY"') | |
| return 2 | |
| print(f" url = {store.url}") | |
| print(f" key = {store.key[:10]}...{store.key[-4:]} ({len(store.key)} chars)") | |
| print("\n2. Connectivity and schema") | |
| ok, error = store.ping() | |
| if not check("can read anatomy_users", ok, str(error)): | |
| print("\n -> 'relation ... does not exist': run supabase_schema.sql in Supabase Studio > SQL Editor." | |
| "\n -> 'permission denied' or 'JWT': use the service_role key (Project Settings > API), not the anon key.") | |
| return 1 | |
| for table in ("anatomy_messages", "anatomy_sessions", "anatomy_events"): | |
| ok, _, error = store._request("GET", table, params={"select": "*", "limit": "1"}) | |
| check(f"can read {table}", ok, str(error)) | |
| user_id = f"verify-{int(time.time())}" | |
| email = f"{user_id}@example.invalid" | |
| print(f"\n3. Accounts (throwaway user {user_id})") | |
| check("validators accept the test user", is_valid_user_id(user_id) and is_valid_email(email)) | |
| row, error = store.get_user(user_id) | |
| check("unknown user ID is reported as absent", row is None and error is None, str(error)) | |
| row, error = store.create_user(user_id, email) | |
| check("create_user stores the account", error is None and row is not None and row.get("email") == email, str(error)) | |
| _, error = store.create_user(user_id, "someone.else@example.invalid") | |
| check("create_user refuses a taken user ID", error == EXISTS, str(error)) | |
| row, error = store.get_user(user_id) | |
| check("get_user returns the account with its email", row is not None and row.get("email") == email, str(error)) | |
| ok, error = store.record_login(user_id) | |
| check("record_login updates last_login_at", ok, str(error)) | |
| session_id = uuid.uuid4().hex | |
| print("\n4. Time on site") | |
| ok, error = store.start_session(session_id, user_id) | |
| check("start_session inserts a session row", ok, str(error)) | |
| time.sleep(1.1) | |
| ok, error = store.heartbeat(session_id) | |
| check("heartbeat updates last_seen_at", ok, str(error)) | |
| ok, error = store.end_session(session_id) | |
| check("end_session sets ended_at", ok, str(error)) | |
| ok, data, error = store._request("GET", "anatomy_sessions", params={"select": "started_at,last_seen_at,ended_at", "session_id": "eq." + session_id}) | |
| session_row = data[0] if ok and data else {} | |
| check("session row has started_at < last_seen_at and ended_at", | |
| bool(session_row) and session_row["started_at"] < session_row["last_seen_at"] and session_row["ended_at"], | |
| str(error or session_row)) | |
| ok, data, error = store._request("GET", "anatomy_time_per_user", params={"select": "seconds_on_site,sessions", "user_id": "eq." + user_id}) | |
| check("anatomy_time_per_user view reports the session", ok and data and data[0]["sessions"] == 1 and data[0]["seconds_on_site"] >= 1, str(error or data)) | |
| print("\n5. Clicks") | |
| for event, detail in (("sign_in", {}), ("ask", {"chars": 12}), ("ask", {"chars": 30}), ("browse_slide", {"lecture": 7, "slide": 3})): | |
| ok, error = store.log_event(user_id, session_id, event, detail) | |
| check(f"log_event stores '{event}'", ok, str(error)) | |
| ok, data, error = store._request("GET", "anatomy_button_clicks", params={"select": "event,clicks", "user_id": "eq." + user_id, "order": "event.asc"}) | |
| counts = {r["event"]: r["clicks"] for r in (data or [])} if ok else {} | |
| check("anatomy_button_clicks view counts clicks per button", counts == {"ask": 2, "browse_slide": 1, "sign_in": 1}, str(error or counts)) | |
| print("\n6. Chat history") | |
| ok, error = store.append_messages(user_id, session_id, [ | |
| {"role": "user", "content": "Which nerve roots form the brachial plexus?", "mode": "ask", "topic": "brachial plexus", "source": None}, | |
| {"role": "assistant", "content": "C5 to T1 (Lecture 7, slide 3).\n\n---\n**Source:** Course lecture slides: Lecture 7 (Brachial Plexus), slide 3.", | |
| "mode": "ask", "topic": "brachial plexus", "source": "slides"}, | |
| ]) | |
| check("append_messages stores a question and its answer", ok, str(error)) | |
| ok, error = store.append_messages(user_id, session_id, [{"role": "user", "content": "Quiz me on: brachial plexus", "mode": "quiz", "topic": "brachial plexus", "source": None}]) | |
| check("append_messages stores a later message", ok, str(error)) | |
| rows, error = store.load_messages(user_id) | |
| check("load_messages returns all messages oldest first", | |
| error is None and [r["role"] for r in rows] == ["user", "assistant", "user"] and rows[1]["source"] == "slides", | |
| str(error or [r.get("role") for r in rows])) | |
| rows, error = store.load_messages(user_id, limit=2) | |
| check("load_messages(limit) keeps the most recent messages in order", | |
| error is None and [r["role"] for r in rows] == ["assistant", "user"], str(error or rows)) | |
| print("\n7. Background writer") | |
| store.enqueue("log_event", user_id, session_id, "sign_out", {}) | |
| check("queued write finishes", store.flush(10.0)) | |
| ok, data, error = store._request("GET", "anatomy_events", params={"select": "event", "user_id": "eq." + user_id, "event": "eq.sign_out"}) | |
| check("queued event is stored", ok and len(data or []) == 1, str(error or data)) | |
| print("\n8. Cleanup") | |
| ok, error = store.delete_user(user_id) | |
| check("delete_user removes the account and its data", ok, str(error)) | |
| row, error = store.get_user(user_id) | |
| check("account is gone", row is None and error is None, str(error)) | |
| print() | |
| if failures: | |
| print(f"{len(failures)} check(s) failed: " + ", ".join(failures)) | |
| return 1 | |
| print("All checks passed. Supabase is ready for the app.") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |