"""Seed data for the all-tools sweep. :func:`seed` creates the resources the sweep's arguments point at (through the Gitea API, not through the tools under test) and returns the context dictionary. Everything created here is fake data in a disposable instance. """ from __future__ import annotations import base64 import secrets import time from typing import Any from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from tests.integration.harness import GiteaInstance, HarnessError from tests.integration.static_keys import GPG_EMAILS, GPG_PUBLIC_KEYS Ctx = dict[str, Any] # --------------------------------------------------------------------------- # Small helpers # --------------------------------------------------------------------------- def b64(text: str | bytes) -> str: """Base64 text of ``text``.""" raw = text.encode("utf-8") if isinstance(text, str) else text return base64.b64encode(raw).decode("ascii") def ssh_public_key() -> str: """A fresh ed25519 public key in OpenSSH format (throwaway).""" key = Ed25519PrivateKey.generate().public_key() raw = key.public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH) return raw.decode("ascii") # A 1x1 transparent PNG, for the avatar endpoints. PNG_1X1 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" # --------------------------------------------------------------------------- # Seeding # --------------------------------------------------------------------------- class Seeder: """Creates the sweep's seed data through the Gitea API (not through the tools).""" def __init__(self, gitea: GiteaInstance, user: str, user_token: str) -> None: self.gitea = gitea self.user = user self.user_token = user_token def call( self, method: str, path: str, *, token: str | None = None, ok: tuple[int, ...] = (200, 201, 202, 204), **kwargs: Any, ) -> Any: """Call the API, require a status in ``ok`` and return the JSON body (or None).""" response = self.gitea.api(method, path, token=token, **kwargs) if response.status_code not in ok: raise HarnessError( f"seed {method} {path} failed: HTTP {response.status_code}: {response.text[:200]}" ) if response.content and "json" in response.headers.get("content-type", ""): return response.json() return None def put_file(self, repo: str, path: str, text: str, branch: str = "main") -> dict[str, Any]: """Create a file through the contents API.""" owner = self.gitea.admin_login body = {"content": b64(text), "message": f"add {path}", "branch": branch} result: dict[str, Any] = self.call( "POST", f"/repos/{owner}/{repo}/contents/{path}", json=body ) return result WORKFLOW = """\ name: sweep on: [push, workflow_dispatch] jobs: hello: runs-on: aegis-sweep-nonexistent steps: - run: echo hello """ def seed(gitea: GiteaInstance, user: str, user_token: str) -> Ctx: """Create the sweep's seed data and return the context the arguments are built from.""" s = Seeder(gitea, user, user_token) admin = gitea.admin_login sfx = secrets.token_hex(3) c: Ctx = {"admin": admin, "user": user, "owner": admin, "sfx": sfx} c["admin_password"] = gitea.admin_password # A second user plus throwaway users for the destructive admin tools. c["user2"] = gitea.create_user(f"sw-u2-{sfx}")[0] c["user3"] = gitea.create_user(f"sw-u3-{sfx}")[0] c["del_user"] = gitea.create_user(f"sw-del-{sfx}")[0] c["rename_user"] = gitea.create_user(f"sw-ren-{sfx}")[0] # Organization with a team and a repository. c["org"] = org = f"sw-org-{sfx}" s.call("POST", "/orgs", json={"username": org, "visibility": "public"}) c["org_repo"] = f"sw-orgrepo-{sfx}" s.call("POST", f"/orgs/{org}/repos", json={"name": c["org_repo"], "auto_init": True}) team = s.call( "POST", f"/orgs/{org}/teams", json={ "name": "sweepers", "permission": "write", "includes_all_repositories": False, "units": ["repo.code", "repo.issues", "repo.pulls"], }, ) c["team_id"], c["team_name"] = team["id"], team["name"] team2 = s.call( "POST", f"/orgs/{org}/teams", json={"name": "sweepers2", "permission": "read", "units": ["repo.code"]}, ) c["team2_name"] = team2["name"] s.call("PUT", f"/teams/{c['team_id']}/members/{user}") s.call("PUT", f"/teams/{c['team_id']}/repos/{org}/{c['org_repo']}") s.call("PUT", f"/orgs/{org}/public_members/{user}") s.call("PUT", f"/orgs/{org}/blocks/{c['user2']}") c["org_label"] = s.call( "POST", f"/orgs/{org}/labels", json={"name": "org-bug", "color": "#ff0000"} )["id"] c["org_hook"] = s.call( "POST", f"/orgs/{org}/hooks", json={"type": "gitea", "config": {"url": "http://127.0.0.1:9/h", "content_type": "json"}}, )["id"] s.call("PUT", f"/orgs/{org}/actions/secrets/ORG_SECRET", json={"data": "org-secret-value"}) s.call("POST", f"/orgs/{org}/actions/variables/ORG_VAR", json={"value": "1"}) c["org_issue_id"] = s.call( "POST", f"/repos/{org}/{c['org_repo']}/issues", json={"title": "org issue"} )["id"] _project(s, c, "org", f"/orgs/{org}/projects") # Throwaway org and repository for the destructive tools. c["del_org"] = f"sw-delorg-{sfx}" s.call("POST", "/orgs", json={"username": c["del_org"]}) c["del_org2"] = f"sw-delorg2-{sfx}" s.call("POST", "/orgs", json={"username": c["del_org2"]}) c["rename_org"] = f"sw-renorg-{sfx}" s.call("POST", "/orgs", json={"username": c["rename_org"]}) c["del_repo"] = f"sw-delrepo-{sfx}" gitea.create_repo(admin, c["del_repo"]) c["xfer_repo"] = f"sw-xfer-{sfx}" gitea.create_repo(admin, c["xfer_repo"]) c["template_repo"] = f"sw-tpl-{sfx}" gitea.create_repo(admin, c["template_repo"]) s.call("PATCH", f"/repos/{admin}/{c['template_repo']}", json={"template": True}) # The main seed repository. c["repo"] = repo = f"sw-repo-{sfx}" created = gitea.create_repo(admin, repo) c["repo_id"] = created["id"] base = f"/repos/{admin}/{repo}" branch = s.call("GET", f"{base}/branches/main") c["sha"] = branch["commit"]["id"] s.put_file(repo, ".editorconfig", "root = true\n") s.put_file(repo, "docs/a.txt", "alpha\n") s.put_file(repo, ".gitea/workflows/sweep.yml", WORKFLOW) c["sha"] = s.call("GET", f"{base}/branches/main")["commit"]["id"] s.call("PUT", f"{base}/collaborators/{user}", json={"permission": "write"}) s.call("PUT", f"{base}/collaborators/{c['user2']}", json={"permission": "write"}) s.call("PUT", f"{base}/topics", json={"topics": ["sweep", "seed"]}) s.call("PUT", f"{base}/topics/extra") # A fork (for merge-upstream), a pull mirror and push-mirror targets, all from this instance. local = f"http://127.0.0.1:3000/{admin}" c["fork_repo"] = f"sw-fork-{sfx}" s.call("POST", f"{base}/forks", json={"organization": org, "name": c["fork_repo"]}) c["mirror_repo"] = f"sw-mirror-{sfx}" s.call( "POST", "/repos/migrate", json={ "clone_addr": f"{local}/{c['template_repo']}.git", "repo_name": c["mirror_repo"], "service": "git", "mirror": True, "mirror_interval": "8h0m0s", }, ) c["push_target"] = f"sw-ptarget-{sfx}" gitea.create_repo(admin, c["push_target"], auto_init=False) c["push_target2"] = f"sw-ptarget2-{sfx}" gitea.create_repo(admin, c["push_target2"], auto_init=False) c["push_mirror"] = s.call( "POST", f"{base}/push_mirrors", json={ "remote_address": f"{local}/{c['push_target']}.git", "remote_username": admin, "remote_password": gitea.admin_password, "interval": "8h0m0s", "sync_on_commit": False, }, )["remote_name"] s.call( "POST", f"{base}/statuses/{c['sha']}", json={"state": "success", "context": "ci/sweep"}, ) s.call("POST", f"{base}/tags", json={"tag_name": "v0.0.1", "target": "main"}) annotated = s.call( "POST", f"{base}/tags", json={"tag_name": "v0.0.3", "target": "main", "message": "note"} ) c["tag_sha"] = annotated["id"] for name in ("feat", "feat2", "feat3", "rn", "del-branch"): s.call( "POST", f"{base}/branches", json={"new_branch_name": name, "old_branch_name": "main"} ) s.put_file(repo, "feat.txt", "feature\n", branch="feat") s.put_file(repo, "feat2.txt", "feature two\n", branch="feat2") s.put_file(repo, "feat3.txt", "feature three\n", branch="feat3") s.call("POST", f"{base}/branch_protections", json={"rule_name": "release/*"}) c["protection"] = "release/*" s.call( "POST", f"{base}/tag_protections", json={"name_pattern": "v9*", "whitelist_usernames": [admin]}, ) c["tag_protection"] = s.call("GET", f"{base}/tag_protections")[0]["id"] s.put_file(repo, "to-delete.txt", "bye\n") s.put_file(repo, "to-update.txt", "before\n") c["delete_file_sha"] = s.call("GET", f"{base}/contents/to-delete.txt")["sha"] c["update_file_sha"] = s.call("GET", f"{base}/contents/to-update.txt")["sha"] c["blob_sha"] = s.call("GET", f"{base}/contents/README.md")["sha"] c["page_name"] = "Home" c["cron_task"] = "cleanup_hook_task_table" # Issues: 1 main, 2 dependency target, 3 pinned, 4/5 more targets. for number, title in enumerate(("main", "dependency", "pinned", "target four", "target five")): issue = s.call("POST", f"{base}/issues", json={"title": f"Sweep {title}", "body": "x"}) assert issue["number"] == number + 1 c["issue"], c["issue2"], c["issue3"], c["issue4"], c["issue5"] = 1, 2, 3, 4, 5 c["issue1_id"] = s.call("GET", f"{base}/issues/1")["id"] c["comment"] = s.call("POST", f"{base}/issues/1/comments", json={"body": "one"})["id"] c["comment2"] = s.call("POST", f"{base}/issues/1/comments", json={"body": "two"})["id"] c["label"] = s.call("POST", f"{base}/labels", json={"name": "bug", "color": "#d73a4a"})["id"] s.call("POST", f"{base}/issues/1/labels", json={"labels": [c["label"]]}) c["milestone"] = s.call("POST", f"{base}/milestones", json={"title": "m1"})["id"] c["milestone_name"] = "m1" s.call("POST", f"{base}/issues/1/assignees", json={"assignees": [admin]}) s.call("PUT", f"{base}/issues/1/subscriptions/{user}") s.call("POST", f"{base}/issues/1/reactions", json={"content": "+1"}) s.call("POST", f"{base}/issues/comments/{c['comment']}/reactions", json={"content": "+1"}) s.call("POST", f"{base}/issues/1/dependencies", json={"index": 2, "owner": admin, "repo": repo}) s.call("POST", f"{base}/issues/1/blocks", json={"index": 3, "owner": admin, "repo": repo}) # A comment by another user notifies the admin (author and assignee of issue 1). s.call("POST", f"{base}/issues/1/comments", token=user_token, json={"body": "ping"}) s.call("POST", f"{base}/issues/3/pin") s.call("POST", f"{base}/issues/2/stopwatch/start", token=user_token) time_entry = s.call("POST", f"{base}/issues/1/times", json={"time": 60}) c["time_id"] = time_entry["id"] attachment = s.call( "POST", f"{base}/issues/1/assets", files={"attachment": ("issue.txt", b"issue attachment", "text/plain")}, params={"name": "issue.txt"}, ) c["issue_attachment"] = attachment["id"] attachment = s.call( "POST", f"{base}/issues/comments/{c['comment']}/assets", files={"attachment": ("comment.txt", b"comment attachment", "text/plain")}, params={"name": "comment.txt"}, ) c["comment_attachment"] = attachment["id"] # Pull requests: 6 is the main one, 7 is merged by the sweep. pr = s.call("POST", f"{base}/pulls", json={"title": "Sweep PR", "base": "main", "head": "feat"}) c["pr"] = pr["number"] pr2 = s.call( "POST", f"{base}/pulls", json={"title": "Sweep PR two", "base": "main", "head": "feat2"} ) c["pr2"] = pr2["number"] pr_path = f"{base}/pulls/{c['pr']}" review = s.call( "POST", f"{pr_path}/reviews", token=user_token, json={ "event": "REQUEST_CHANGES", "body": "please change", "comments": [{"path": "feat.txt", "new_position": 1, "body": "line note"}], }, ) c["review"] = review["id"] c["review_comment"] = s.call("GET", f"{pr_path}/reviews/{review['id']}/comments")[0]["id"] c["review2"] = s.call( "POST", f"{pr_path}/reviews", token=user_token, json={"event": "COMMENT", "body": "fyi"} )["id"] c["pending_review"] = s.call( "POST", f"{pr_path}/reviews", token=user_token, json={"body": "draft"} )["id"] s.call( "POST", f"{pr_path}/requested_reviewers", json={"reviewers": [c["user2"]]}, ok=(201, 422) ) # Release with an asset (and a second release removed by tag). release = s.call( "POST", f"{base}/releases", json={"tag_name": "v1.0.0", "target_commitish": "main"} ) c["release"], c["release_tag"] = release["id"], "v1.0.0" asset = s.call( "POST", f"{base}/releases/{release['id']}/assets", files={"attachment": ("rel.bin", b"release bytes", "application/octet-stream")}, params={"name": "rel.bin"}, ) c["release_asset"] = asset["id"] s.call("POST", f"{base}/releases", json={"tag_name": "v2.0.0", "target_commitish": "main"}) c["release_tag2"] = "v2.0.0" # Hooks, keys, secrets, variables, wiki, projects. hook_body = {"type": "gitea", "config": {"url": "http://127.0.0.1:9/h", "content_type": "json"}} c["repo_hook"] = s.call("POST", f"{base}/hooks", json=hook_body)["id"] s.call( "PATCH", f"{base}/hooks/git/pre-receive", json={"content": "#!/bin/sh\nexit 0\n"}, ok=(200,), ) c["git_hook"] = "pre-receive" c["user_hook"] = s.call("POST", "/user/hooks", json=hook_body)["id"] c["admin_hook"] = s.call("POST", "/admin/hooks", json=hook_body)["id"] c["deploy_key"] = s.call( "POST", f"{base}/keys", json={"title": f"deploy-{sfx}", "key": ssh_public_key()} )["id"] c["user_key"] = s.call( "POST", "/user/keys", json={"title": f"mine-{sfx}", "key": ssh_public_key()} )["id"] c["admin_user_key"] = s.call( "POST", f"/admin/users/{user}/keys", json={"title": f"k-{sfx}", "key": ssh_public_key()} )["id"] s.call("PUT", f"{base}/actions/secrets/REPO_SECRET", json={"data": "repo-secret-value"}) c["user_secret"] = f"USER_SECRET_{sfx.upper()}" c["user_var"] = f"USER_VAR_{sfx.upper()}" s.call("PUT", f"/user/actions/secrets/{c['user_secret']}", json={"data": "user-secret-value"}) s.call("POST", f"{base}/actions/variables/REPO_VAR", json={"value": "1"}) s.call("POST", f"/user/actions/variables/{c['user_var']}", json={"value": "1"}) s.call( "POST", f"{base}/wiki/new", json={"title": "Home", "content_base64": b64("# home\n"), "message": "home"}, ) _project(s, c, "repo", f"{base}/projects") _project(s, c, "user", "/user/projects") c["oauth_app"] = s.call( "POST", "/user/applications/oauth2", json={"name": f"sweep-{sfx}", "redirect_uris": ["http://127.0.0.1:1/cb"]}, )["id"] # Current user: email, follow, star, watch, block. c["email"] = f"sweep-{sfx}@example.test" s.call("POST", "/user/emails", json={"emails": [c["email"], *GPG_EMAILS]}) # Gitea only accepts a GPG key whose identity matches an email of the account. c["gpg_key"] = s.call( "POST", "/user/gpg_keys", json={"armored_public_key": GPG_PUBLIC_KEYS[0]} )["id"] s.call("PUT", f"/user/following/{user}") s.call("PUT", f"/user/starred/{admin}/{repo}") s.call("PUT", f"{base}/subscription") s.call("PUT", f"/user/blocks/{c['user2']}") # A throwaway access token of the admin (basic auth), deleted by name. token_name = f"sweep-{sfx}" s.call( "POST", f"/users/{admin}/tokens", auth=(admin, gitea.admin_password), json={"name": token_name, "scopes": ["read:user"]}, ) c["throwaway_token"] = token_name # A generic package. c["pkg"] = pkg = f"sweep-pkg-{sfx}" upload = gitea.request( "PUT", f"/api/packages/{admin}/generic/{pkg}/1.0.0/payload.bin", content=b"package" ) if upload.status_code != 201: raise HarnessError(f"seed package upload failed: HTTP {upload.status_code}") c["pkg_version"] = "1.0.0" c["pkg2"] = f"sweep-pkg2-{sfx}" upload = gitea.request( "PUT", f"/api/packages/{admin}/generic/{c['pkg2']}/1.0.0/payload.bin", content=b"package" ) if upload.status_code != 201: raise HarnessError(f"seed package upload failed: HTTP {upload.status_code}") # The workflow run created by the push of the workflow file (queued, no runner). c["run"], c["job_id"] = _find_run(s, base) c["notification"] = _find_notification(s, admin) return c def _project(s: Seeder, c: Ctx, scope: str, path: str) -> None: """Create a project with two extra columns; store ids as ``_project/_column...``.""" project = s.call("POST", path, json={"title": f"{scope} project"})["id"] c[f"{scope}_project"] = project ids: dict[str, int] = {} for title in ("todo", "done"): ids[title] = s.call("POST", f"{path}/{project}/columns", json={"title": title})["id"] c[f"{scope}_column"], c[f"{scope}_column2"] = ids["todo"], ids["done"] columns = s.call("GET", f"{path}/{project}/columns") c[f"{scope}_columns"] = [column["id"] for column in columns] def _find_run(s: Seeder, base: str) -> tuple[Any, Any]: """Return ``(run id, job id)`` of the first workflow run; fail when none appears. A made-up id would only turn every dependent tool into a 404, which the sweep would accept as documented, so a missing run is a seeding failure. """ deadline = time.monotonic() + 30 while time.monotonic() < deadline: runs = s.call("GET", f"{base}/actions/runs") entries = runs.get("workflow_runs") or [] if entries: run_id = entries[0]["id"] jobs = s.call("GET", f"{base}/actions/runs/{run_id}/jobs") job_entries = jobs.get("jobs") or [] if job_entries: return run_id, job_entries[0]["id"] time.sleep(0.5) raise HarnessError("seed: the push of the workflow file did not create a workflow run") def _find_notification(s: Seeder, admin: str) -> Any: """Return the id of a notification of the admin; fail when none appears.""" deadline = time.monotonic() + 30 while time.monotonic() < deadline: listing = s.call("GET", "/notifications", params={"all": "true"}) if listing: return listing[0]["id"] time.sleep(0.5) raise HarnessError(f"seed: no notification arrived for {admin}")