Spaces:
Sleeping
Sleeping
File size: 5,776 Bytes
1e7a182 aac074c 1e7a182 | 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 | #!/usr/bin/env python3
import argparse
import os
import sys
from dataclasses import dataclass
import httpx
@dataclass
class Result:
name: str
method: str
path: str
status: str
ok: bool
detail: str | None = None
def _print_results(results: list[Result]) -> None:
for item in results:
status = "OK" if item.ok else "FAIL"
print(f"{status:4} {item.method:6} {item.path:50} {item.status}")
if item.detail:
print(f" {item.detail}")
def _request(
client: httpx.Client,
results: list[Result],
name: str,
method: str,
path: str,
headers: dict[str, str],
params: dict[str, str] | None = None,
) -> httpx.Response | None:
try:
response = client.request(method, path, headers=headers, params=params)
except Exception as exc: # pragma: no cover - runtime/network error
results.append(Result(name=name, method=method, path=path, status="ERR", ok=False, detail=str(exc)))
return None
ok = response.status_code == 200
detail = None
if not ok:
snippet = response.text[:200].replace("\n", " ")
detail = f"{response.status_code} {snippet}"
results.append(
Result(
name=name,
method=method,
path=path,
status=str(response.status_code),
ok=ok,
detail=detail,
)
)
return response
def main() -> int:
parser = argparse.ArgumentParser(description="Smoke test admin API endpoints.")
parser.add_argument("--base-url", default=os.getenv("BASE_URL", "http://127.0.0.1:7860"))
parser.add_argument("--email", default=os.getenv("ADMIN_EMAIL"))
parser.add_argument("--password", default=os.getenv("ADMIN_PASSWORD"))
parser.add_argument("--group-id", default=os.getenv("ADMIN_GROUP_ID"))
parser.add_argument("--timeout", type=float, default=10.0)
args = parser.parse_args()
if not args.email or not args.password:
print("Missing credentials. Provide --email/--password or set ADMIN_EMAIL/ADMIN_PASSWORD.")
return 1
base_url = args.base_url
results: list[Result] = []
with httpx.Client(base_url=base_url, timeout=args.timeout) as client:
login_res = client.post(
"/auth/login",
json={"email": args.email, "password": args.password},
)
if login_res.status_code != 200:
snippet = login_res.text[:200].replace("\n", " ")
print(f"Login failed: {login_res.status_code} {snippet}")
return 1
token = login_res.json().get("token")
if not token:
print("Login response missing token.")
return 1
headers = {"Authorization": f"Bearer {token}"}
groups_res = _request(client, results, "groups.list", "GET", "/groups", headers)
group_id = args.group_id
if not group_id:
if not groups_res:
_print_results(results)
return 1
try:
items = groups_res.json().get("items", [])
except Exception:
items = []
if items:
group_id = items[0].get("id")
if not group_id:
_print_results(results)
print("No group_id available. Create a group in /admin/groups or pass --group-id.")
return 2
gid = group_id
endpoints = [
("group.get", "GET", f"/groups/{gid}", None),
("config.get", "GET", f"/groups/{gid}/config", None),
("ad_filter.get", "GET", f"/groups/{gid}/ad-filter/config", None),
("ad_filter.reports", "GET", f"/groups/{gid}/ad-filter/reports", {"range": "day"}),
("spam.get", "GET", f"/groups/{gid}/spam/config", None),
("verification.get", "GET", f"/groups/{gid}/verification/config", None),
("auto_replies.list", "GET", f"/groups/{gid}/auto-replies", None),
("ban_rules.list", "GET", f"/groups/{gid}/ban-rules", None),
("delete_rules.list", "GET", f"/groups/{gid}/delete-rules", None),
("posts.list", "GET", f"/groups/{gid}/posts", None),
("posts.search", "GET", f"/groups/{gid}/posts/search", None),
("schedules.list", "GET", f"/groups/{gid}/schedules", None),
("broadcasts.list", "GET", f"/groups/{gid}/broadcasts", None),
("logs.actions", "GET", f"/groups/{gid}/logs/actions", None),
("logs.deletions", "GET", f"/groups/{gid}/logs/deletions", None),
("logs.admin", "GET", f"/groups/{gid}/logs/admin", None),
("stats.messages", "GET", f"/groups/{gid}/stats/messages", {"range": "day"}),
("stats.wordcloud", "GET", f"/groups/{gid}/stats/wordcloud", {"range": "day"}),
("invites.list", "GET", f"/groups/{gid}/invites", None),
("invites.rankings", "GET", f"/groups/{gid}/invites/rankings", {"range": "day"}),
("verified_users.list", "GET", f"/groups/{gid}/verified-users", None),
("verified_users.forms", "GET", f"/groups/{gid}/verified-users/forms", None),
("verified_users.reviews", "GET", f"/groups/{gid}/verified-users/reviews", None),
("verified_users.nearby", "GET", f"/groups/{gid}/verified-users/nearby", None),
("member_events.list", "GET", f"/groups/{gid}/member-events", None),
]
for name, method, path, params in endpoints:
_request(client, results, name, method, path, headers, params=params)
_print_results(results)
failed = [item for item in results if not item.ok]
if failed:
print(f"Failures: {len(failed)}")
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
|