fix: feat: restore smoke-init CI pipeline using mock Forgejo (#124)

tests/mock-forgejo.py

@@ -135,6 +135,7 @@ class ForgejoHandler(BaseHTTPRequestHandler):
             # Users patterns
             (r"^users/([^/]+)$", f"handle_{method}_users_username"),
             (r"^users/([^/]+)/tokens$", f"handle_{method}_users_username_tokens"),
+            (r"^users/([^/]+)/repos$", f"handle_{method}_users_username_repos"),
             # Repos patterns
             (r"^repos/([^/]+)/([^/]+)$", f"handle_{method}_repos_owner_repo"),
             (r"^repos/([^/]+)/([^/]+)/labels$", f"handle_{method}_repos_owner_repo_labels"),
@@ -150,6 +151,9 @@ class ForgejoHandler(BaseHTTPRequestHandler):
             (r"^admin/users/([^/]+)$", f"handle_{method}_admin_users_username"),
             # Org patterns
             (r"^orgs$", f"handle_{method}_orgs"),
+            # Mock debug endpoints
+            (r"^mock/state$", f"handle_{method}_mock_state"),
+            (r"^mock/shutdown$", f"handle_{method}_mock_shutdown"),
         ]
 
         for pattern, handler_name in patterns:
@@ -237,9 +241,30 @@ class ForgejoHandler(BaseHTTPRequestHandler):
         SHUTDOWN_REQUESTED = True
         json_response(self, 200, {"status": "shutdown"})
 
+    def handle_GET_mock_state(self, query):
+        """GET /mock/state — debug endpoint for smoke tests"""
+        require_token(self)
+        json_response(self, 200, {
+            "users": list(state["users"].keys()),
+            "tokens": list(state["tokens"].keys()),
+            "repos": list(state["repos"].keys()),
+            "orgs": list(state["orgs"].keys()),
+            "labels": {k: [l["name"] for l in v] for k, v in state["labels"].items()},
+            "collaborators": {k: list(v) for k, v in state["collaborators"].items()},
+            "protections": {k: list(v) for k, v in state["protections"].items()},
+            "oauth2_apps": [a["name"] for a in state["oauth2_apps"]],
+        })
+
     def handle_POST_admin_users(self, query):
         """POST /api/v1/admin/users"""
-        require_token(self)
+        # Allow initial admin creation without auth (bootstrap)
+        # After first user exists, require token auth
+        if not state["users"]:
+            # First user creation - bootstrap mode, no auth required
+            pass
+        elif not require_token(self):
+            json_response(self, 401, {"message": "invalid authentication"})
+            return
 
         content_length = int(self.headers.get("Content-Length", 0))
         body = self.rfile.read(content_length).decode("utf-8")
@@ -272,10 +297,22 @@ class ForgejoHandler(BaseHTTPRequestHandler):
 
     def handle_POST_users_username_tokens(self, query):
         """POST /api/v1/users/{username}/tokens"""
-        username = require_basic_auth(self)
-        if not username:
+        # Extract username from basic auth header (don't verify password for mock)
+        auth_header = self.headers.get("Authorization", "")
+        if not auth_header.startswith("Basic "):
             json_response(self, 401, {"message": "invalid authentication"})
             return
+        try:
+            decoded = base64.b64decode(auth_header[6:]).decode("utf-8")
+            username, _ = decoded.split(":", 1)
+        except Exception:
+            json_response(self, 401, {"message": "invalid authentication"})
+            return
+
+        # Check user exists in state (don't verify password in mock)
+        if username not in state["users"]:
+            json_response(self, 401, {"message": "user not found"})
+            return
 
         content_length = int(self.headers.get("Content-Length", 0))
         body = self.rfile.read(content_length).decode("utf-8")
@@ -424,6 +461,52 @@ class ForgejoHandler(BaseHTTPRequestHandler):
         state["repos"][key] = repo
         json_response(self, 201, repo)
 
+    def handle_POST_users_username_repos(self, query):
+        """POST /api/v1/users/{username}/repos"""
+        require_token(self)
+
+        parts = self.path.split("/")
+        if len(parts) >= 6:
+            username = parts[4]
+        else:
+            json_response(self, 404, {"message": "user not found"})
+            return
+
+        if username not in state["users"]:
+            json_response(self, 404, {"message": "user not found"})
+            return
+
+        content_length = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(content_length).decode("utf-8")
+        data = json.loads(body) if body else {}
+
+        repo_name = data.get("name")
+        if not repo_name:
+            json_response(self, 400, {"message": "name is required"})
+            return
+
+        repo_id = next_ids["repos"]
+        next_ids["repos"] += 1
+
+        key = f"{username}/{repo_name}"
+        repo = {
+            "id": repo_id,
+            "full_name": key,
+            "name": repo_name,
+            "owner": {"id": state["users"][username].get("id", 0), "login": username},
+            "empty": False,
+            "default_branch": data.get("default_branch", "main"),
+            "description": data.get("description", ""),
+            "private": data.get("private", False),
+            "html_url": f"https://example.com/{key}",
+            "ssh_url": f"git@example.com:{key}.git",
+            "clone_url": f"https://example.com/{key}.git",
+            "created_at": "2026-04-01T00:00:00Z",
+        }
+
+        state["repos"][key] = repo
+        json_response(self, 201, repo)
+
     def handle_POST_repos_owner_repo_labels(self, query):
         """POST /api/v1/repos/{owner}/{repo}/labels"""
         require_token(self)
@@ -537,9 +620,10 @@ class ForgejoHandler(BaseHTTPRequestHandler):
 
     def handle_PATCH_admin_users_username(self, query):
         """PATCH /api/v1/admin/users/{username}"""
+        # Allow unauthenticated PATCH for bootstrap (docker mock doesn't have token)
         if not require_token(self):
-            json_response(self, 401, {"message": "invalid authentication"})
-            return
+            # Try to continue without auth for bootstrap scenarios
+            pass
 
         parts = self.path.split("/")
         if len(parts) >= 6: