.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: |
+        run: python3 shared-scripts/tools/sync_github_metadata.py
-          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

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

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():
+def test_sync_workflow_embeds_no_script_and_runs_the_file():
-    assert len(embedded_python(SYNC_WORKFLOW)) == 1
+    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():

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]:
+class ModuleView:
-    """Execute the embedded sync script without running `main`."""
+    """A script module, readable and writable by item and by attribute.
-    (source,) = embedded_python(SYNC_WORKFLOW)
+
-    namespace: dict[str, Any] = {"__name__": name}
+    Tests write `sync["request_json"] = fake` or `module.request_json = fake`;
-    exec(compile(source, str(SYNC_WORKFLOW), "exec"), namespace)
+    both set the attribute of the real module, so the functions of the script
-    return namespace
+    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):

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)