Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Focused end-to-end checks for the social media feature set. | |
| Runs against a live server on 127.0.0.1:8899. Start it with a fresh writable | |
| DATABASE_URL before running (see README/test harness examples). | |
| """ | |
| import asyncio | |
| import json | |
| import os | |
| import sys | |
| import uuid | |
| import httpx | |
| BASE = "http://127.0.0.1:8899" | |
| failures = [] | |
| def check(cond, label, extra=""): | |
| if cond: | |
| print(f" \u2713 {label}") | |
| else: | |
| failures.append(label + (f" \u2014 {extra}" if extra else "")) | |
| print(f" \u2717 {label} {extra}") | |
| async def signup(client, username): | |
| r = await client.post(f"{BASE}/api/auth/signup", params={ | |
| "username": username, | |
| "password": "password123", | |
| "display_name": username.title(), | |
| }) | |
| return r | |
| async def upload_file(client, token, name, data, mime): | |
| params = { | |
| "chunk_index": 0, | |
| "total_chunks": 1, | |
| "file_name": name, | |
| "file_type": mime, | |
| "file_size": len(data), | |
| "upload_id": uuid.uuid4().hex, | |
| } | |
| r = await client.post( | |
| f"{BASE}/api/upload/chunk", | |
| params=params, | |
| files={"file": (name, data, mime)}, | |
| headers={"X-Auth-Token": token}, | |
| ) | |
| if r.status_code != 200: | |
| return None, r | |
| return r.json()["file_path"], r | |
| async def main(): | |
| async with httpx.AsyncClient(timeout=20) as http: | |
| # Unique usernames so the test can be run repeatedly. | |
| u1 = "soc_" + uuid.uuid4().hex[:8] | |
| u2 = "soc_" + uuid.uuid4().hex[:8] | |
| r = await signup(http, u1) | |
| check(r.status_code == 200 and r.json().get("token"), f"signup {u1}", r.text[:160]) | |
| t1 = r.json()["token"] | |
| h1 = {"X-Auth-Token": t1} | |
| r = await signup(http, u2) | |
| check(r.status_code == 200 and r.json().get("token"), f"signup {u2}", r.text[:160]) | |
| t2 = r.json()["token"] | |
| h2 = {"X-Auth-Token": t2} | |
| # --- image upload through the shared chunked endpoint + post create --- | |
| # Minimal 1x1 PNG. | |
| png = ( | |
| b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" | |
| b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\xcf\xc0" | |
| b"\x00\x00\x00\x03\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" | |
| b"\x00\x00\x00\x00IEND\xaeB`\x82" | |
| ) | |
| fp, up = await upload_file(http, t1, "pic.png", png, "image/png") | |
| check(up.status_code == 200 and fp, "chunked image upload returned a path", up.text[:160]) | |
| media = [{"file_path": fp, "file_name": "pic.png", "file_type": "image/png", "file_size": len(png)}] | |
| r = await http.post( | |
| f"{BASE}/api/social/posts", | |
| params={"body": "Hello social world", "media_json": json.dumps(media)}, | |
| headers=h1, | |
| ) | |
| check(r.status_code == 200 and len(r.json().get("post", {}).get("media", [])) == 1, | |
| "post with image created", r.text[:160]) | |
| post_id = r.json()["post"]["id"] | |
| # --- text attachments are allowed by the shared uploader but rejected on posts --- | |
| txt = b"just some notes" | |
| tfp, up = await upload_file(http, t1, "notes.txt", txt, "text/plain") | |
| check(up.status_code == 200 and tfp, "chunked text upload still works for chat", up.text[:160]) | |
| bad_media = [{"file_path": tfp, "file_name": "notes.txt", "file_type": "text/plain", "file_size": len(txt)}] | |
| r = await http.post( | |
| f"{BASE}/api/social/posts", | |
| params={"body": "should fail", "media_json": json.dumps(bad_media)}, | |
| headers=h1, | |
| ) | |
| check(r.status_code == 400 and "images and videos" in (r.json().get("detail") or ""), | |
| "text media rejected on social posts", r.text[:160]) | |
| # --- edit own post --- | |
| r = await http.patch(f"{BASE}/api/social/posts/{post_id}", params={"body": "Edited body"}, headers=h1) | |
| check(r.status_code == 200 and r.json()["post"]["body"] == "Edited body" | |
| and r.json()["post"].get("edited_at_ms"), | |
| "author can edit own post", r.text[:160]) | |
| # --- non-author cannot edit --- | |
| r = await http.patch(f"{BASE}/api/social/posts/{post_id}", params={"body": "nope"}, headers=h2) | |
| check(r.status_code == 403, "non-author cannot edit", r.text[:160]) | |
| # --- like / repost / bookmark toggles --- | |
| r = await http.post(f"{BASE}/api/social/posts/{post_id}/like", headers=h2) | |
| check(r.status_code == 200 and r.json()["liked"] and r.json()["like_count"] == 1, | |
| "second user can like", r.text[:160]) | |
| r = await http.post(f"{BASE}/api/social/posts/{post_id}/repost", headers=h2) | |
| check(r.status_code == 200 and r.json()["reposted"], "second user can repost", r.text[:160]) | |
| r = await http.get(f"{BASE}/api/social/feed", params={"feed": "home", "limit": 50}, headers=h2) | |
| repost = next((p for p in (r.json().get("posts") or []) | |
| if p["id"] == post_id and p.get("reposter_id")), None) | |
| check(repost is not None and repost.get("reposter", {}).get("username") == u2, | |
| "repost carries reposter info", r.text[:160]) | |
| r = await http.post(f"{BASE}/api/social/posts/{post_id}/bookmark", headers=h2) | |
| check(r.status_code == 200 and r.json()["bookmarked"], "second user can bookmark", r.text[:160]) | |
| # --- search shows follow state truthfully --- | |
| await http.post(f"{BASE}/api/social/users/{u1}/follow", headers=h2) | |
| r = await http.get(f"{BASE}/api/social/search", params={"q": u1}, headers=h2) | |
| hit = next((u for u in (r.json().get("users") or []) if u["username"] == u1), None) | |
| check(r.status_code == 200 and hit and hit["is_following"], | |
| "search reflects follow status", r.text[:160]) | |
| # --- delete own post --- | |
| r = await http.delete(f"{BASE}/api/social/posts/{post_id}", headers=h1) | |
| check(r.status_code == 200, "author can delete post", r.text[:160]) | |
| r = await http.get(f"{BASE}/api/social/posts/{post_id}", headers=h1) | |
| check(r.status_code == 404, "deleted post returns 404", r.text[:160]) | |
| # --- backup admin endpoints are admin-only --- | |
| admin_user = os.environ.get("ADMIN_USERNAME", "").strip() | |
| admin_pass = os.environ.get("ADMIN_PASSWORD", "") | |
| if admin_user and admin_pass: | |
| r = await http.post(f"{BASE}/api/admin/login", json={"username": admin_user, "password": admin_pass}) | |
| check(r.status_code == 200 and r.json().get("token"), "admin login succeeds", r.text[:160]) | |
| admin_headers = {"X-Admin-Token": r.json()["token"]} | |
| r = await http.get(f"{BASE}/api/admin/backups", headers=admin_headers) | |
| check(r.status_code == 200, "admin backup list reachable", r.text[:160]) | |
| r = await http.get(f"{BASE}/api/admin/backups", headers=h1) | |
| check(r.status_code == 401, "normal user cannot access admin backups", r.text[:160]) | |
| else: | |
| r = await http.get(f"{BASE}/api/admin/backups", headers=h1) | |
| check(r.status_code == 401, "backup admin requires dedicated admin auth", r.text[:160]) | |
| print(f"\nfinished. failures={len(failures)}") | |
| print("ALL SOCIAL CHECKS PASSED" if not failures else "SOCIAL CHECKS FAILED") | |
| return failures | |
| if __name__ == "__main__": | |
| fails = asyncio.run(main()) | |
| if fails: | |
| for f in fails: | |
| print(" -", f) | |
| sys.exit(1) | |
| sys.exit(0) |