#!/usr/bin/env python3 """创作者关系真实 API / SpacetimeDB 验证，仅允许显式指定的隔离本地库。 先用仓库 dev 启动器启动全新 creator-hub-validation 库和 API，启用开发密码注册。 运行：python3 scripts/test-creator-runtime.py --database creator-hub-validation 测试凭据只保存在内存或隔离 data dir，不输出令牌，不触发外部媒体服务。 """ import argparse import concurrent.futures import hashlib import json import os from pathlib import Path import time import urllib.error import urllib.parse import urllib.request def request(url, method="GET", body=None, token=None, expected=200): headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = "Bearer " + token data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(url, method=method, data=data, headers=headers) try: response = urllib.request.urlopen(req, timeout=30) except urllib.error.HTTPError as error: response = error assert response.status == expected, f"{method} {urllib.parse.urlsplit(url).path}: {response.status} != {expected}" raw = response.read() return (json.loads(raw) if raw else None), response.headers def option(raw): return raw[1] if raw[0] == 0 else None def main(): parser = argparse.ArgumentParser() parser.add_argument("--database", required=True) parser.add_argument("--dev-stack", default=".app/dev-stack.json") args = parser.parse_args() state = json.loads(Path(args.dev_stack).read_text()) assert args.database == state["database"] == "creator-hub-validation", "只允许专用隔离库" api = state["services"]["api-server"]["url"] server = state["services"]["spacetime"]["url"] for url in [api, server]: parsed = urllib.parse.urlsplit(url) assert parsed.scheme == "http" and parsed.hostname in ["127.0.0.1", "localhost", "::1"] data_dir = Path(state["spacetimeDataDir"]) service = json.loads((data_dir / "dev-api-identities/local-node.json").read_text()) operator_path = data_dir / "creator-test-operator.json" if operator_path.exists(): operator = json.loads(operator_path.read_text()) else: operator, _ = request(server + "/v1/identity", "POST", {}) fd = os.open(operator_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as file: json.dump(operator, file) scope = hashlib.sha256((server + "\n" + args.database).encode()).hexdigest() secret = json.loads((data_dir / "dev-runtime-service-bootstrap-secrets" / (scope + ".json")).read_text())["secret"] result, _ = request(server + "/v1/database/" + args.database + "/call/authorize_database_migration_operator", "POST", [{"bootstrap_secret": secret, "operator_identity_hex": operator["identity"], "note": "创作者隔离测试夹具"}], operator["token"]) assert result[0], "测试迁移身份授权失败" def call(name, value, token=operator["token"]): return request(server + "/v1/database/" + args.database + "/call/" + name, "POST", [value], token)[0] def export(table): result = call("export_database_migration_to_file", {"include_tables": [table]}) assert result[0], option(result[-1]) return json.loads(option(result[2]))["tables"][0]["rows"] def fixtures(tables): result = call("import_database_migration_incremental_from_file", {"migration_json": json.dumps({"schema_version": 1, "exported_at_micros": int(time.time() * 1e6), "tables": [{"name": k, "rows": v} for k, v in tables.items()]}), "include_tables": list(tables), "replace_existing": False, "dry_run": False}) assert result[0], option(result[-1]) def login(number): value, _ = request(api + "/api/auth/entry", "POST", {"purePhoneNumber": str(13900006000 + number), "password": "Creator-test-2026"}) return value["user"]["id"], value["token"] def relation(viewer, target): return request(api + "/api/creators/" + target + "/relationship", token=viewer[1])[0]["relationship"] def write(viewer, target, method="PUT", remove=False, expected=200): path = "/api/creators/me/followers/" + target if remove else "/api/creators/" + target + "/follow" return request(api + path, method, token=viewer[1], expected=expected)[0] def profile(user): value, _ = request(api + "/api/creators/" + user) assert set(value["user"]) == {"id", "publicUserCode", "displayName", "avatarUrl"} return value request(api + "/healthz") a, b, c = [login(n) for n in [1, 2, 3]] # 重跑时先清理测试账号的相互出边；不清表或删除已有业务数据。 for actor in [a, b, c]: for target in [a, b, c]: if actor != target: write(actor, target[0], "DELETE") assert relation(a, a[0]) == {"isSelf": True, "isFollowing": False, "isFollowedBy": False} for kind in ["following", "followers"]: empty, _ = request(api + f"/api/creators/{c[0]}/{kind}") assert empty == {"items": [], "nextCursor": None, "total": 0} write(a, b[0]) first = next(row for row in export("user_follow") if row["follower_user_id"] == a[0] and row["followee_user_id"] == b[0]) write(a, b[0]) second = next(row for row in export("user_follow") if row["relationship_id"] == first["relationship_id"]) assert first == second, "重复关注不能改时间" write(b, a[0]) assert relation(a, b[0]) == {"isSelf": False, "isFollowing": True, "isFollowedBy": True} write(a, b[0], "DELETE", remove=True) write(a, b[0], "DELETE", remove=True) assert relation(a, b[0]) == {"isSelf": False, "isFollowing": True, "isFollowedBy": False} write(b, a[0]) # 移除不是拉黑。 write(a, b[0], "DELETE") assert not relation(a, b[0])["isFollowing"] write(a, b[0]) third = next(row for row in export("user_follow") if row["relationship_id"] == first["relationship_id"]) assert third["created_at"] != first["created_at"] write(a, a[0], expected=400) write(a, a[0], "DELETE", expected=400) write(a, a[0], "DELETE", remove=True, expected=400) write(a, "missing-creator", expected=404) request(api + "/api/creators/" + b[0] + "/follow", "PUT", expected=401) request(api + "/api/creators/" + b[0] + "/followers", token="invalid", expected=401) request(api + "/api/creators/" + a[0] + "/followers/" + b[0], "DELETE", token=c[1], expected=404) write(c, b[0], "DELETE", remove=True) assert relation(a, b[0])["isFollowedBy"], "第三方移除不能改变他人入边" stranger, _ = request(server + "/v1/identity", "POST", {}) denied = call("follow_creator_and_return", {"actor_id": a[0], "target_id": b[0]}, stranger["token"]) assert option(denied[0]) is None and option(denied[1]) == "forbidden" denied = call("list_creator_connections_and_return", {"owner_id": a[0], "viewer_id": [0, a[0]], "kind": "following", "limit": 20, "cursor": None}, stranger["token"]) assert option(denied[0]) is None and option(denied[1]) == "forbidden" print("PASS 双向关系、幂等、重新关注、移除粉丝、自操作、认证和越权", flush=True) peers = [login(n) for n in range(10, 65)] stamp = {"__timestamp_micros_since_unix_epoch__": 1700000000000000} rows = [] for peer, _ in peers: for follower, followee in [(a[0], peer), (peer, a[0])]: rows.append({"relationship_id": f"{len(follower.encode())}:{follower}{followee}", "follower_user_id": follower, "followee_user_id": followee, "created_at": stamp}) # 不存在的对端必须同时从计数和列表排除。 rows.extend([{"relationship_id": f"{len(x.encode())}:{x}{y}", "follower_user_id": x, "followee_user_id": y, "created_at": stamp} for x, y in [(a[0], "missing-peer"), ("missing-peer", a[0])]]) fixtures({"user_follow": rows}) for kind, count_key in [("following", "followingCount"), ("followers", "followerCount")]: for viewer in [None, a[1], b[1], c[1]]: seen, cursor, total = [], None, None while True: path = f"/api/creators/{a[0]}/{kind}?limit=20" if cursor: path += "&cursor=" + urllib.parse.quote(cursor) result, headers = request(api + path, token=viewer) total = result["total"] assert headers["Cache-Control"] == "private, no-store" for row in result["items"]: assert (row["relationship"] is not None) == (viewer == a[1]) assert set(row["user"]) == {"id", "publicUserCode", "displayName", "avatarUrl"} seen.append(row["user"]["id"]) cursor = result["nextCursor"] if not cursor: break request(api + f"/api/creators/{b[0]}/{kind}?cursor=" + cursor, expected=400) other = "followers" if kind == "following" else "following" request(api + f"/api/creators/{a[0]}/{other}?cursor=" + cursor, expected=400) assert len(seen) == len(set(seen)) == total == profile(a[0])[count_key] assert total == 56 and "missing-peer" not in seen before = export("user_follow") a = login(1) # 触发认证快照刷新和持久化。 assert export("user_follow") == before with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: list(pool.map(lambda _: write(c, b[0]), range(8))) assert len([row for row in export("user_follow") if row["follower_user_id"] == c[0] and row["followee_user_id"] == b[0]]) == 1 with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: list(pool.map(lambda n: write(c, b[0], "PUT" if n % 2 else "DELETE"), range(8))) db_has = any(row["follower_user_id"] == c[0] and row["followee_user_id"] == b[0] for row in export("user_follow")) assert relation(c, b[0])["isFollowing"] == db_has # 丢弃成功回执，模拟调用方不知道写入结果；正式读取决定下一动作。 write(c, b[0]) assert relation(c, b[0])["isFollowing"] write(c, b[0], "DELETE") assert not relation(c, b[0])["isFollowing"] print("PASS 同时间多页、公开只读、失效用户过滤、认证快照保留、并发及未知结果回读", flush=True) none = {"none": []} some = lambda value: {"some": value} games, versions = [], [] # 其它作者的更新作品占满旧广场前 48 项，目标作者仍应得到自己的 48 项。 for prefix, owner, count in [("target", a[0], 55), ("other", b[0], 55), ("hidden", a[0], 6)]: for n in range(count): game_id = f"creator-web-fixture-{prefix}-{n:03}" version_id = game_id + "-v1" created = {"__timestamp_micros_since_unix_epoch__": 1800000000000000 + (1000 if prefix == "other" else 0) + n} game = dict(game_id=game_id, owner_user_id=owner, title=game_id, summary="测试", description="测试", category="adventure", tags_json="[]", cover_asset_id=none, author_name=none, author_avatar_url=none, device_support_desktop=True, device_support_mobile=True, device_support_touch=True, input_modes_json='["touch"]', orientation="any", publication_revision=1, active_version_id=some(version_id), visibility="published", play_count=0, created_at=created, updated_at=created, local_project_id=none, cover_object_key=none, screenshots_json=none, deleted_at=none) version = dict(version_id=version_id, game_id=game_id, owner_user_id=owner, version_number=1, package_sha256="0" * 64, package_bytes=1, package_file_count=1, package_entry_path="index.html", status="published", review_reason=none, created_at=created, updated_at=created, local_project_id=none, package_object_key=none, package_manifest_json=none, entry_url=some("/games/game_" + hashlib.sha256(game_id.encode()).hexdigest()[:32] + "/"), reviewed_by_user_id=none, uploaded_at=none, submitted_at=none, reviewed_at=none, published_at=some(created), revoked_at=none, last_error_code=none, last_error_message=none, metadata_json=none) if prefix == "hidden": if n == 0: game["visibility"] = "unpublished" if n == 1: game["visibility"] = "suspended" if n == 2: game["deleted_at"] = some(created) if n == 3: game["active_version_id"] = none if n == 4: version["status"] = "pending_review" if n == 5: version["game_id"] = "wrong-game" games.append(game) versions.append(version) fixtures({"game_distribution_game": games, "game_distribution_version": versions}) target, _ = request(api + "/api/game-distribution/games?authorId=" + a[0]) assert len(target["games"]) == 48 and target["nextCursor"] is None assert all(game["author"]["id"] == a[0] and game["id"].startswith("creator-web-fixture-target-") for game in target["games"]) assert [game["id"] for game in target["games"]] == [f"creator-web-fixture-target-{n:03}" for n in range(54, 6, -1)] gallery, _ = request(api + "/api/game-distribution/games") assert len(gallery["games"]) == 48 and gallery["nextCursor"] is None assert all(game["author"]["id"] == b[0] for game in gallery["games"]) request(api + "/api/game-distribution/games?authorId=", expected=400) empty, _ = request(api + "/api/game-distribution/games?authorId=" + c[0]) assert empty["games"] == [] print("PASS 作者过滤先于 48 项截取、公开版本隔离、旧广场上限与空作者校验", flush=True) print("PASS 创作者后端隔离运行时验收", flush=True) if __name__ == "__main__": main()