From ac20d1c2a1c6764f73e05f4430b645b8d0c9eabb Mon Sep 17 00:00:00 2001 From: Jens Tirsvad Nielsen Date: Sat, 10 Oct 2026 20:30:32 +0800 Subject: [PATCH] Move the sync script out of the scoped workflow into a file tools/sync_github_metadata.py holds the 221 lines that were embedded in sync-github-metadata.yml, unchanged in behaviour and messages. The scoped workflow now checks out TirSystem/github-action at main and runs the file (33 lines instead of 230). The tests import the file instead of executing the YAML. Verified live from TirSystem-BashScript/RepoFoundry with that organisation's secrets. Task: MIL-004#6 --- .../scoped_workflows/sync-github-metadata.yml | 221 +----------------- tests/test_sync_github_metadata.py | 2 +- tests/test_workflows.py | 16 +- tests/workflow_source.py | 44 +++- tools/sync_github_metadata.py | 221 ++++++++++++++++++ 5 files changed, 285 insertions(+), 219 deletions(-) create mode 100644 tools/sync_github_metadata.py diff --git a/.gitea/scoped_workflows/sync-github-metadata.yml b/.gitea/scoped_workflows/sync-github-metadata.yml index 2971a99..6b38215 100644 --- a/.gitea/scoped_workflows/sync-github-metadata.yml +++ b/.gitea/scoped_workflows/sync-github-metadata.yml @@ -1,5 +1,9 @@ name: Sync GitHub mirror metadata +# Scoped workflow: it runs in every repository of an organization. The script +# is a file of TirSystem/github-action (tools/sync_github_metadata.py), fetched +# with a checkout so that it is readable, lintable and tested as a file +# (MIL-004 task 6). The repository is public, so no token is needed. on: push: branches: [ main ] @@ -13,218 +17,17 @@ jobs: contents: read runs-on: ubuntu-latest steps: + - name: Fetch the shared script + uses: actions/checkout@v4 + with: + repository: TirSystem/github-action + ref: main + path: shared-scripts + - name: Sync description and topics env: GITEA_API_URL: ${{ gitea.api_url }} GITEA_CREDENTIALS: ${{ secrets.TOKEN_FOR_GITEA }} SOURCE_REPOSITORY: ${{ gitea.repository }} PAT_FOR_GITHUB: ${{ secrets.PAT_FOR_GITHUB }} - run: | - python3 - <<'PY' - import json - import os - import sys - import urllib.error - import urllib.parse - import urllib.request - - - class WorkflowError(RuntimeError): - """A failure with a category. - - configuration: a secret or repository setting is missing or malformed. - credentials: a service rejected the token (HTTP 401). - permissions: the token is valid but not allowed (HTTP 403). - api: any other API or network failure. - """ - - def __init__(self, category, message): - super().__init__(f"[{category}] {message}") - self.category = category - - - def service_name(url): - hostname = urllib.parse.urlsplit(url).hostname - return "GitHub" if hostname == "api.github.com" else "Gitea" - - - def request_json(url, method="GET", headers=None, body=None): - request = urllib.request.Request( - url, - data=json.dumps(body).encode("utf-8") if body is not None else None, - headers=headers or {}, - method=method, - ) - service = service_name(url) - try: - with urllib.request.urlopen(request, timeout=30) as response: - content = response.read() - return json.loads(content) if content else None - except urllib.error.HTTPError as error: - detail = ( - f"{service} API request failed with HTTP " - f"{error.code} ({error.reason})" - ) - if error.code == 401: - raise WorkflowError( - "credentials", f"{detail}. The token was rejected." - ) from None - if error.code == 403: - raise WorkflowError( - "permissions", - f"{detail}. The token lacks a required permission.", - ) from None - if error.code == 404: - raise WorkflowError( - "configuration", - f"{detail}. The repository was not found or the " - "token cannot see it.", - ) from None - raise WorkflowError("api", detail) from None - except (urllib.error.URLError, TimeoutError) as error: - raise WorkflowError( - "api", f"{service} API is unreachable ({error})." - ) from None - - - def parse_github_token(raw_token): - token = (raw_token or "").strip() - if not token: - raise WorkflowError( - "configuration", "PAT_FOR_GITHUB is missing or empty." - ) - return token - - - def parse_gitea_token(raw_token): - token = (raw_token or "").strip() - if not token: - raise WorkflowError( - "configuration", "TOKEN_FOR_GITEA is missing or empty." - ) - return token - - - def split_repository(full_name): - owner, separator, repo = (full_name or "").partition("/") - if not separator or not owner or not repo: - raise WorkflowError( - "configuration", - "Could not determine the Gitea source repository.", - ) - return owner, repo - - - def find_github_targets(mirrors): - targets = [] - for mirror in mirrors: - remote_address = mirror.get("remote_address", "") - if remote_address.startswith("git@github.com:"): - mirror_path = remote_address.split(":", 1)[1] - else: - parsed_remote = urllib.parse.urlsplit(remote_address) - if parsed_remote.hostname != "github.com": - continue - mirror_path = parsed_remote.path.lstrip("/") - - mirror_path = mirror_path.removesuffix(".git").strip("/") - path_parts = mirror_path.split("/") - if len(path_parts) != 2 or not all(path_parts): - raise WorkflowError( - "configuration", - "Could not determine the GitHub owner and repository " - "from a configured push mirror.", - ) - targets.append(tuple(path_parts)) - - if not targets: - return None - if len(targets) != 1: - raise WorkflowError( - "configuration", - "Expected at most one GitHub push mirror for this repository; " - f"found {len(targets)}.", - ) - return targets[0] - - - def main(): - gitea_token = parse_gitea_token(os.environ.get("GITEA_CREDENTIALS")) - source_owner, source_repo = split_repository( - os.environ.get("SOURCE_REPOSITORY") - ) - - gitea_api_url = os.environ["GITEA_API_URL"].rstrip("/") - source_url = f"{gitea_api_url}/repos/{source_owner}/{source_repo}" - gitea_headers = { - "Authorization": f"token {gitea_token}", - "Accept": "application/json", - } - source = request_json(source_url, headers=gitea_headers) - mirrors = request_json( - f"{source_url}/push_mirrors", - headers=gitea_headers, - ) - if not isinstance(mirrors, list): - print( - "WARNING: Gitea returned an invalid push mirror list; " - "GitHub metadata was not synced." - ) - return 0 - - target = find_github_targets(mirrors) - if target is None: - print( - "No GitHub push mirror configured for " - f"{source_owner}/{source_repo}; nothing to sync." - ) - return 0 - - github_token = parse_github_token(os.environ.get("PAT_FOR_GITHUB")) - github_owner, github_repo = target - github_api_url = ( - "https://api.github.com/repos/" - f"{urllib.parse.quote(github_owner, safe='')}/" - f"{urllib.parse.quote(github_repo, safe='')}" - ) - github_headers = { - "Authorization": f"Bearer {github_token}", - "Accept": "application/vnd.github+json", - "X-GitHub-Api-Version": "2022-11-28", - "Content-Type": "application/json", - } - - github_repository = request_json(github_api_url, headers=github_headers) - permissions = (github_repository or {}).get("permissions") - if isinstance(permissions, dict) and not permissions.get("admin"): - raise WorkflowError( - "permissions", - f"The GitHub token cannot administer {github_owner}/{github_repo}; " - "editing the description and topics needs administration access.", - ) - print("Preflight passed: Gitea and GitHub accepted the credentials.") - - request_json( - github_api_url, - method="PATCH", - headers=github_headers, - body={"description": source.get("description") or ""}, - ) - request_json( - f"{github_api_url}/topics", - method="PUT", - headers=github_headers, - body={"names": source.get("topics") or []}, - ) - - print(f"Synced description and topics to {github_owner}/{github_repo}.") - return 0 - - - if __name__ == "__main__": - try: - sys.exit(main()) - except RuntimeError as error: - print(f"ERROR: {error}", file=sys.stderr) - sys.exit(1) - PY + run: python3 shared-scripts/tools/sync_github_metadata.py diff --git a/tests/test_sync_github_metadata.py b/tests/test_sync_github_metadata.py index b1716d3..6a7e213 100644 --- a/tests/test_sync_github_metadata.py +++ b/tests/test_sync_github_metadata.py @@ -1,4 +1,4 @@ -"""Offline tests for the Python embedded in sync-github-metadata.yml.""" +"""Offline tests for tools/sync_github_metadata.py.""" import io import json diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 7a7f858..35423a3 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -5,7 +5,7 @@ import ast import pytest import yaml -from workflow_source import REPO_ROOT, SYNC_WORKFLOW, embedded_python, workflow_files +from workflow_source import REPO_ROOT, SYNC_SCRIPT, SYNC_WORKFLOW, embedded_python, workflow_files WORKFLOWS = workflow_files() @@ -33,8 +33,18 @@ class TestEachWorkflow: assert "permissions" in job, f"job {name} has no explicit permissions" -def test_sync_workflow_embeds_exactly_one_script(): - assert len(embedded_python(SYNC_WORKFLOW)) == 1 +def test_sync_workflow_embeds_no_script_and_runs_the_file(): + assert embedded_python(SYNC_WORKFLOW) == [] + document = yaml.safe_load(SYNC_WORKFLOW.read_text(encoding="utf-8")) + steps = document["jobs"]["sync-metadata"]["steps"] + assert steps[0]["with"]["repository"] == "TirSystem/github-action" + assert steps[-1]["run"] == "python3 shared-scripts/tools/sync_github_metadata.py" + + +def test_sync_script_compiles_and_is_a_file_of_this_repository(): + source = SYNC_SCRIPT.read_text(encoding="utf-8") + ast.parse(source, filename=str(SYNC_SCRIPT)) + assert "def main()" in source and 'if __name__ == "__main__":' in source def test_each_workflow_name_has_one_source(): diff --git a/tests/workflow_source.py b/tests/workflow_source.py index 3a1b771..7278077 100644 --- a/tests/workflow_source.py +++ b/tests/workflow_source.py @@ -1,5 +1,6 @@ """Helpers that read workflow files and the Python embedded in them.""" +import importlib.util import re import textwrap from pathlib import Path @@ -11,6 +12,7 @@ WORKFLOW_DIRS = ( REPO_ROOT / ".gitea" / "workflows", ) SYNC_WORKFLOW = WORKFLOW_DIRS[0] / "sync-github-metadata.yml" +SYNC_SCRIPT = REPO_ROOT / "tools" / "sync_github_metadata.py" PUBLISH_WORKFLOW = WORKFLOW_DIRS[0] / "publish-wiki.yml" _HEREDOC = re.compile( @@ -37,12 +39,42 @@ def embedded_python(path: Path) -> list[str]: ] -def load_sync_module(name: str = "embedded_sync") -> dict[str, Any]: - """Execute the embedded sync script without running `main`.""" - (source,) = embedded_python(SYNC_WORKFLOW) - namespace: dict[str, Any] = {"__name__": name} - exec(compile(source, str(SYNC_WORKFLOW), "exec"), namespace) - return namespace +class ModuleView: + """A script module, readable and writable by item and by attribute. + + Tests write `sync["request_json"] = fake` or `module.request_json = fake`; + both set the attribute of the real module, so the functions of the script + see the replacement. + """ + + def __init__(self, module: Any) -> None: + object.__setattr__(self, "_module", module) + + def __getitem__(self, name: str) -> Any: + return getattr(self._module, name) + + def __setitem__(self, name: str, value: Any) -> None: + setattr(self._module, name, value) + + def __getattr__(self, name: str) -> Any: + return getattr(self._module, name) + + def __setattr__(self, name: str, value: Any) -> None: + setattr(self._module, name, value) + + +def load_script(path: Path, name: str) -> ModuleView: + """Import a script file without running its `main` (its name is not `__main__`).""" + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return ModuleView(module) + + +def load_sync_module(name: str = "sync_github_metadata") -> ModuleView: + """Import tools/sync_github_metadata.py.""" + return load_script(SYNC_SCRIPT, name) class ScriptNamespace(dict): diff --git a/tools/sync_github_metadata.py b/tools/sync_github_metadata.py new file mode 100644 index 0000000..db96168 --- /dev/null +++ b/tools/sync_github_metadata.py @@ -0,0 +1,221 @@ +"""Copy the description and topics of a Gitea repository to its GitHub push mirror. + +Run by the scoped workflow sync-github-metadata.yml, which checks out this +repository and runs this file (MIL-004 task 6). It reads its settings from the +environment: + + GITEA_API_URL Gitea API base, for example https://host/api/v1 + GITEA_CREDENTIALS the Gitea token, as clear text (TOKEN_FOR_GITEA) + SOURCE_REPOSITORY owner/repo of the repository whose metadata is synced + PAT_FOR_GITHUB the GitHub token, as clear text + +Every failure is printed as ``ERROR: [category] message`` with the categories +configuration, credentials, permissions and api; no secret value is printed. +""" + +import json +import os +import sys +import urllib.error +import urllib.parse +import urllib.request + + +class WorkflowError(RuntimeError): + """A failure with a category. + + configuration: a secret or repository setting is missing or malformed. + credentials: a service rejected the token (HTTP 401). + permissions: the token is valid but not allowed (HTTP 403). + api: any other API or network failure. + """ + + def __init__(self, category, message): + super().__init__(f"[{category}] {message}") + self.category = category + + +def service_name(url): + hostname = urllib.parse.urlsplit(url).hostname + return "GitHub" if hostname == "api.github.com" else "Gitea" + + +def request_json(url, method="GET", headers=None, body=None): + request = urllib.request.Request( + url, + data=json.dumps(body).encode("utf-8") if body is not None else None, + headers=headers or {}, + method=method, + ) + service = service_name(url) + try: + with urllib.request.urlopen(request, timeout=30) as response: + content = response.read() + return json.loads(content) if content else None + except urllib.error.HTTPError as error: + detail = ( + f"{service} API request failed with HTTP " + f"{error.code} ({error.reason})" + ) + if error.code == 401: + raise WorkflowError( + "credentials", f"{detail}. The token was rejected." + ) from None + if error.code == 403: + raise WorkflowError( + "permissions", + f"{detail}. The token lacks a required permission.", + ) from None + if error.code == 404: + raise WorkflowError( + "configuration", + f"{detail}. The repository was not found or the " + "token cannot see it.", + ) from None + raise WorkflowError("api", detail) from None + except (urllib.error.URLError, TimeoutError) as error: + raise WorkflowError( + "api", f"{service} API is unreachable ({error})." + ) from None + + +def parse_github_token(raw_token): + token = (raw_token or "").strip() + if not token: + raise WorkflowError( + "configuration", "PAT_FOR_GITHUB is missing or empty." + ) + return token + + +def parse_gitea_token(raw_token): + token = (raw_token or "").strip() + if not token: + raise WorkflowError( + "configuration", "TOKEN_FOR_GITEA is missing or empty." + ) + return token + + +def split_repository(full_name): + owner, separator, repo = (full_name or "").partition("/") + if not separator or not owner or not repo: + raise WorkflowError( + "configuration", + "Could not determine the Gitea source repository.", + ) + return owner, repo + + +def find_github_targets(mirrors): + targets = [] + for mirror in mirrors: + remote_address = mirror.get("remote_address", "") + if remote_address.startswith("git@github.com:"): + mirror_path = remote_address.split(":", 1)[1] + else: + parsed_remote = urllib.parse.urlsplit(remote_address) + if parsed_remote.hostname != "github.com": + continue + mirror_path = parsed_remote.path.lstrip("/") + + mirror_path = mirror_path.removesuffix(".git").strip("/") + path_parts = mirror_path.split("/") + if len(path_parts) != 2 or not all(path_parts): + raise WorkflowError( + "configuration", + "Could not determine the GitHub owner and repository " + "from a configured push mirror.", + ) + targets.append(tuple(path_parts)) + + if not targets: + return None + if len(targets) != 1: + raise WorkflowError( + "configuration", + "Expected at most one GitHub push mirror for this repository; " + f"found {len(targets)}.", + ) + return targets[0] + + +def main(): + gitea_token = parse_gitea_token(os.environ.get("GITEA_CREDENTIALS")) + source_owner, source_repo = split_repository( + os.environ.get("SOURCE_REPOSITORY") + ) + + gitea_api_url = os.environ["GITEA_API_URL"].rstrip("/") + source_url = f"{gitea_api_url}/repos/{source_owner}/{source_repo}" + gitea_headers = { + "Authorization": f"token {gitea_token}", + "Accept": "application/json", + } + source = request_json(source_url, headers=gitea_headers) + mirrors = request_json( + f"{source_url}/push_mirrors", + headers=gitea_headers, + ) + if not isinstance(mirrors, list): + print( + "WARNING: Gitea returned an invalid push mirror list; " + "GitHub metadata was not synced." + ) + return 0 + + target = find_github_targets(mirrors) + if target is None: + print( + "No GitHub push mirror configured for " + f"{source_owner}/{source_repo}; nothing to sync." + ) + return 0 + + github_token = parse_github_token(os.environ.get("PAT_FOR_GITHUB")) + github_owner, github_repo = target + github_api_url = ( + "https://api.github.com/repos/" + f"{urllib.parse.quote(github_owner, safe='')}/" + f"{urllib.parse.quote(github_repo, safe='')}" + ) + github_headers = { + "Authorization": f"Bearer {github_token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "Content-Type": "application/json", + } + + github_repository = request_json(github_api_url, headers=github_headers) + permissions = (github_repository or {}).get("permissions") + if isinstance(permissions, dict) and not permissions.get("admin"): + raise WorkflowError( + "permissions", + f"The GitHub token cannot administer {github_owner}/{github_repo}; " + "editing the description and topics needs administration access.", + ) + print("Preflight passed: Gitea and GitHub accepted the credentials.") + + request_json( + github_api_url, + method="PATCH", + headers=github_headers, + body={"description": source.get("description") or ""}, + ) + request_json( + f"{github_api_url}/topics", + method="PUT", + headers=github_headers, + body={"names": source.get("topics") or []}, + ) + + print(f"Synced description and topics to {github_owner}/{github_repo}.") + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except RuntimeError as error: + print(f"ERROR: {error}", file=sys.stderr) + sys.exit(1) -- 2.54.0