Spaces:
Running
Running
File size: 7,506 Bytes
e163512 | 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 163 164 165 166 167 168 169 | #!/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) |