#!/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)