Merge pull request 'Make the wiki publisher a scoped workflow with an opt-in gate' (#62) from mil-006-scoped-publisher into main

Reviewed-on: #62

.gitea/scoped_workflows/publish-wiki.yml

@@ -0,0 +1,288 @@
+name: Publish wiki
+
+# Shared through scoped delivery: it runs in every repository of an
+# organization, but writes only where the repository has opted in with
+# wiki/.publish (ADR-0004). TOKEN_FOR_GITEA_WIKI is read from the organization
+# secrets; there is no fallback to the read-only sync token.
+on:
+  push:
+    branches: [ main ]
+    paths:
+      - "wiki/**"
+  workflow_dispatch:
+
+concurrency:
+  group: publish-wiki
+  cancel-in-progress: false
+
+jobs:
+  publish-wiki:
+    permissions:
+      contents: read
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out the repository
+        uses: actions/checkout@v4
+
+      - name: Publish the pages to the wiki
+        env:
+          GITEA_API_URL: ${{ gitea.api_url }}
+          SOURCE_REPOSITORY: ${{ gitea.repository }}
+          WIKI_CREDENTIALS: ${{ secrets.TOKEN_FOR_GITEA_WIKI }}
+        run: |
+          python3 - <<'PY'
+          import base64
+          import json
+          import os
+          import re
+          import sys
+          import urllib.error
+          import urllib.parse
+          import urllib.request
+          from pathlib import Path
+
+          PAGE_LIMIT = 50
+          REQUEST_TIMEOUT_SECONDS = 30
+          COMMIT_MESSAGE = "Publish wiki from the repository"
+          MARKER_NAME = ".publish"
+          PRUNE_WORD = "prune"
+
+
+          class WorkflowError(RuntimeError):
+              """A failure with a category.
+
+              configuration: a secret, a setting or the wiki source is missing or malformed.
+              credentials:   Gitea 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
+                  self.detail = message
+
+
+          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,
+              )
+              try:
+                  with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+                      content = response.read()
+                      return json.loads(content) if content else None
+              except urllib.error.HTTPError as error:
+                  detail = f"Gitea API request failed with HTTP {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 or page 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"Gitea API is unreachable ({error}).") from None
+
+
+          def parse_marker(text):
+              """Return True when the marker asks for pruning, False when it is empty."""
+              word = (text or "").strip()
+              if not word:
+                  return False
+              if word == PRUNE_WORD:
+                  return True
+              raise WorkflowError(
+                  "configuration",
+                  f"wiki/{MARKER_NAME} may only be empty or contain the word '{PRUNE_WORD}'.",
+              )
+
+
+          def parse_gitea_token(raw_token):
+              token = (raw_token or "").strip()
+              if not token:
+                  raise WorkflowError(
+                      "configuration",
+                      "TOKEN_FOR_GITEA_WIKI is missing or empty. Store it as a secret of "
+                      "the organization of this repository.",
+                  )
+              return token
+
+
+          def describe_token(token):
+              """Describe the shape of a token without revealing any of its characters."""
+              facts = [f"{len(token)} characters"]
+              if re.fullmatch(r"[0-9a-f]{40}", token):
+                  facts.append("looks like a Gitea token (40 hexadecimal characters)")
+                  return ", ".join(facts)
+              facts.append("a Gitea token has 40 hexadecimal characters")
+              if set(token) <= {"*", "x", "X", "•"}:
+                  facts.append("only mask characters: the masked text was stored")
+              if "=" in token:
+                  facts.append("contains '=': the secret name may be part of the value")
+              if re.search(r"[\"'`]", token):
+                  facts.append("contains a quote character")
+              if re.search(r"\s", token):
+                  facts.append("contains whitespace")
+              return ", ".join(facts)
+
+
+          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 repository.")
+              return owner, repo
+
+
+          def read_pages(wiki_dir):
+              """Return {page title: content} for every Markdown file in wiki_dir."""
+              pages = {path.stem: path.read_bytes() for path in sorted(wiki_dir.glob("*.md"))}
+              if not pages:
+                  # Publishing an empty directory would delete the whole wiki.
+                  raise WorkflowError(
+                      "configuration", f"The wiki source directory {wiki_dir} has no Markdown pages."
+                  )
+              return pages
+
+
+          def list_wiki_titles(wiki_url, headers, send):
+              """Return the titles of the pages that exist in the wiki.
+
+              Gitea answers 404 for a wiki that has never been written to; that is an
+              empty wiki, not a failure.
+              """
+              titles = set()
+              page = 1
+              while True:
+                  url = f"{wiki_url}/pages?page={page}&limit={PAGE_LIMIT}"
+                  try:
+                      batch = send(url, headers=headers)
+                  except WorkflowError as error:
+                      if error.category == "configuration" and page == 1:
+                          return titles
+                      raise
+                  if not isinstance(batch, list):
+                      raise WorkflowError("api", "Gitea returned an invalid wiki page list.")
+                  titles.update(str(item.get("title", "")) for item in batch)
+                  if len(batch) < PAGE_LIMIT:
+                      return titles
+                  page += 1
+
+
+          def read_wiki_page(wiki_url, title, headers, send):
+              quoted = urllib.parse.quote(title, safe="")
+              page = send(f"{wiki_url}/page/{quoted}", headers=headers)
+              return base64.b64decode((page or {}).get("content_base64", ""))
+
+
+          def publish(pages, wiki_url, headers, send, prune):
+              """Create and update the pages; delete the others only when prune is set.
+
+              Returns the titles created, updated and deleted, and the titles that were
+              left in the wiki because pruning is off.
+              """
+              existing = list_wiki_titles(wiki_url, headers, send)
+              current = {
+                  title: read_wiki_page(wiki_url, title, headers, send)
+                  for title in sorted(existing & pages.keys())
+              }
+
+              result = {"created": [], "updated": [], "deleted": [], "kept": []}
+              for title, content in pages.items():
+                  body = {
+                      "title": title,
+                      "content_base64": base64.b64encode(content).decode("ascii"),
+                      "message": COMMIT_MESSAGE,
+                  }
+                  if title not in existing:
+                      send(f"{wiki_url}/new", method="POST", headers=headers, body=body)
+                      result["created"].append(title)
+                  elif current[title] != content:
+                      quoted = urllib.parse.quote(title, safe="")
+                      send(f"{wiki_url}/page/{quoted}", method="PATCH", headers=headers, body=body)
+                      result["updated"].append(title)
+
+              for title in sorted(existing - pages.keys()):
+                  if not prune:
+                      result["kept"].append(title)
+                      continue
+                  quoted = urllib.parse.quote(title, safe="")
+                  send(f"{wiki_url}/page/{quoted}", method="DELETE", headers=headers)
+                  result["deleted"].append(title)
+              return result
+
+
+          def main():
+              wiki_dir = Path(os.environ.get("WIKI_DIR", "wiki"))
+              owner, repo = split_repository(os.environ.get("SOURCE_REPOSITORY"))
+
+              marker = wiki_dir / MARKER_NAME
+              if not marker.is_file():
+                  print(
+                      f"No {wiki_dir}/{MARKER_NAME} marker in {owner}/{repo}; "
+                      "wiki publishing is not enabled for this repository."
+                  )
+                  return 0
+              prune = parse_marker(marker.read_text(encoding="utf-8"))
+
+              pages = read_pages(wiki_dir)
+              token = parse_gitea_token(os.environ.get("WIKI_CREDENTIALS"))
+
+              api_url = os.environ["GITEA_API_URL"].rstrip("/")
+              repository_url = (
+                  f"{api_url}/repos/{urllib.parse.quote(owner, safe='')}/"
+                  f"{urllib.parse.quote(repo, safe='')}"
+              )
+              headers = {
+                  "Authorization": f"token {token}",
+                  "Accept": "application/json",
+                  "Content-Type": "application/json",
+              }
+
+              try:
+                  repository = request_json(repository_url, headers=headers)
+                  if isinstance(repository, dict) and repository.get("has_wiki") is False:
+                      raise WorkflowError(
+                          "configuration",
+                          f"The wiki is not enabled on {owner}/{repo}. Enable it in the "
+                          "repository settings.",
+                      )
+                  print("Preflight passed: Gitea accepted the token and the wiki is enabled.")
+
+                  result = publish(pages, f"{repository_url}/wiki", headers, request_json, prune)
+              except WorkflowError as error:
+                  if error.category == "credentials":
+                      raise WorkflowError(
+                          "credentials", f"{error.detail} Token shape: {describe_token(token)}."
+                      ) from None
+                  raise
+
+              print(
+                  f"Published {len(pages)} page(s) to the wiki of {owner}/{repo}: "
+                  f"{len(result['created'])} created, {len(result['updated'])} updated, "
+                  f"{len(result['deleted'])} deleted."
+              )
+              if result["kept"]:
+                  print(
+                      f"{len(result['kept'])} page(s) are in the wiki but not in "
+                      f"{wiki_dir}/ and were kept; put the word '{PRUNE_WORD}' in "
+                      f"{wiki_dir}/{MARKER_NAME} to delete them."
+                  )
+              return 0
+
+
+          if __name__ == "__main__":
+              try:
+                  sys.exit(main())
+              except WorkflowError as error:
+                  print(f"ERROR: {error}", file=sys.stderr)
+                  sys.exit(1)
+          PY

.gitea/workflows/publish-wiki.yml

@@ -1,32 +0,0 @@
-name: Publish wiki
-
-on:
-  push:
-    branches: [ main ]
-    paths:
-      - "wiki/**"
-  workflow_dispatch:
-
-concurrency:
-  group: publish-wiki
-  cancel-in-progress: false
-
-jobs:
-  publish-wiki:
-    permissions:
-      contents: read
-    runs-on: ubuntu-latest
-    steps:
-      - name: Check out the repository
-        uses: actions/checkout@v4
-
-      - name: Check that the English pages are current
-        run: python3 tools/build_wiki.py --check
-
-      - name: Publish the pages to the wiki
-        env:
-          GITEA_API_URL: ${{ gitea.api_url }}
-          SOURCE_REPOSITORY: ${{ gitea.repository }}
-          # TOKEN_FOR_GITEA_WIKI wins when it exists; otherwise TOKEN_FOR_GITEA is reused.
-          WIKI_CREDENTIALS: ${{ secrets.TOKEN_FOR_GITEA_WIKI || secrets.TOKEN_FOR_GITEA }}
-        run: python3 tools/publish_wiki.py

tests/test_publish_wiki.py

@@ -1,56 +1,68 @@
-"""Offline tests for tools/publish_wiki.py and its workflow."""
+"""Offline tests for the scoped publish-wiki workflow and its embedded script."""
 
 import base64
-import importlib.util
 import io
-import json
 import urllib.error
-from pathlib import Path
 
 import pytest
 import yaml
 
-REPO_ROOT = Path(__file__).resolve().parent.parent
+from workflow_source import PUBLISH_WORKFLOW, REPO_ROOT, load_publish_module
-WORKFLOW = REPO_ROOT / ".gitea" / "workflows" / "publish-wiki.yml"
-_SPEC = importlib.util.spec_from_file_location(
-    "publish_wiki", REPO_ROOT / "tools" / "publish_wiki.py"
-)
-assert _SPEC is not None and _SPEC.loader is not None
-publish_wiki = importlib.util.module_from_spec(_SPEC)
-_SPEC.loader.exec_module(publish_wiki)
 
 API = "https://git.example.test/api/v1"
 REPO_URL = f"{API}/repos/TirSystem/repo"
 WIKI_URL = f"{REPO_URL}/wiki"
 
 
+@pytest.fixture()
+def publish_wiki():
+    return load_publish_module()
+
+
 def b64(text: str) -> str:
     return base64.b64encode(text.encode()).decode()
 
 
+class TestParseMarker:
+    @pytest.mark.parametrize(
+        ("text", "expected"),
+        [("", False), ("  \n", False), (None, False), ("prune", True), (" prune\n", True)],
+    )
+    def test_reads_the_empty_marker_and_the_prune_word(self, publish_wiki, text, expected):
+        assert publish_wiki.parse_marker(text) is expected
+
+    @pytest.mark.parametrize("text", ["Prune", "prune: true", "delete", "prune\nmore"])
+    def test_rejects_anything_else(self, publish_wiki, text):
+        with pytest.raises(publish_wiki.WorkflowError) as caught:
+            publish_wiki.parse_marker(text)
+        assert caught.value.category == "configuration"
+        assert "prune" in str(caught.value)
+
+
 class TestParseGiteaToken:
-    def test_returns_the_stripped_token(self):
+    def test_returns_the_stripped_token(self, publish_wiki):
         assert publish_wiki.parse_gitea_token(" abc \n") == "abc"
 
     @pytest.mark.parametrize("raw", [None, "", "  "])
-    def test_rejects_missing_or_empty_token(self, raw):
+    def test_rejects_missing_or_empty_token(self, publish_wiki, raw):
         with pytest.raises(publish_wiki.WorkflowError) as caught:
             publish_wiki.parse_gitea_token(raw)
         assert caught.value.category == "configuration"
-        assert "No token for the wiki" in str(caught.value)
+        assert "TOKEN_FOR_GITEA_WIKI is missing or empty" in str(caught.value)
 
 
 class TestReadPages:
-    def test_reads_markdown_files_by_stem(self, tmp_path):
+    def test_reads_markdown_files_by_stem_and_ignores_the_marker(self, publish_wiki, tmp_path):
         (tmp_path / "Home.md").write_bytes(b"home")
+        (tmp_path / "Home.da.md").write_bytes(b"hjem")
+        (tmp_path / ".publish").write_bytes(b"prune")
         (tmp_path / "Notes.txt").write_bytes(b"ignored")
-        assert publish_wiki.read_pages(tmp_path) == {"Home": b"home"}
+        assert publish_wiki.read_pages(tmp_path) == {"Home": b"home", "Home.da": b"hjem"}
 
-    def test_rejects_missing_directory(self, tmp_path):
+    def test_rejects_a_directory_without_pages_so_the_wiki_is_never_wiped(
-        with pytest.raises(publish_wiki.WorkflowError, match="missing"):
+        self, publish_wiki, tmp_path
-            publish_wiki.read_pages(tmp_path / "nope")
+    ):
-
+        (tmp_path / ".publish").write_text("prune", encoding="utf-8")
-    def test_rejects_empty_directory_so_the_wiki_is_never_wiped(self, tmp_path):
         with pytest.raises(publish_wiki.WorkflowError, match="no Markdown pages"):
             publish_wiki.read_pages(tmp_path)
 
@@ -58,19 +70,20 @@ class TestReadPages:
 class FakeGitea:
     """Answers like the Gitea repository and wiki endpoints."""
 
-    def __init__(self, wiki=None, has_wiki=True, wiki_initialised=True):
+    def __init__(self, publish_wiki, wiki=None, has_wiki=True, wiki_initialised=True):
+        self.error = publish_wiki.WorkflowError
         self.wiki = dict(wiki or {})
         self.has_wiki = has_wiki
         self.wiki_initialised = wiki_initialised
         self.calls = []
 
     def __call__(self, url, method="GET", headers=None, body=None):
-        self.calls.append((method, url, body))
+        self.calls.append((method, url, body, headers))
         if url == REPO_URL:
             return {"has_wiki": self.has_wiki}
         if url.startswith(f"{WIKI_URL}/pages"):
             if not self.wiki_initialised:
-                raise publish_wiki.WorkflowError("configuration", "404")
+                raise self.error("configuration", "404")
             return [{"title": title} for title in self.wiki]
         if method == "POST" and url == f"{WIKI_URL}/new":
             self.wiki[body["title"]] = body["content_base64"]
@@ -86,6 +99,10 @@ class FakeGitea:
             return None
         raise AssertionError(f"unexpected request {method} {url}")
 
+    @property
+    def writes(self):
+        return [call for call in self.calls if call[0] != "GET"]
+
 
 @pytest.fixture()
 def environment(monkeypatch, tmp_path):
@@ -93,6 +110,7 @@ def environment(monkeypatch, tmp_path):
     wiki_dir.mkdir()
     (wiki_dir / "Home.md").write_text("home v2", encoding="utf-8")
     (wiki_dir / "Credentials.md").write_text("credentials", encoding="utf-8")
+    (wiki_dir / ".publish").write_text("prune\n", encoding="utf-8")
     monkeypatch.setenv("WIKI_DIR", str(wiki_dir))
     monkeypatch.setenv("GITEA_API_URL", f"{API}/")
     monkeypatch.setenv("SOURCE_REPOSITORY", "TirSystem/repo")
@@ -100,20 +118,73 @@ def environment(monkeypatch, tmp_path):
     return wiki_dir
 
 
-def run_main(monkeypatch, gitea):
+def run_main(publish_wiki, gitea):
-    monkeypatch.setattr(publish_wiki, "request_json", gitea)
+    publish_wiki.request_json = gitea
     return publish_wiki.main()
 
 
-class TestMain:
+class TestOptIn:
-    def test_creates_updates_and_deletes_to_match_the_source(
+    def test_a_repository_without_the_marker_is_never_written_to(
-        self, monkeypatch, environment, capsys
+        self, publish_wiki, environment, capsys
+    ):
+        (environment / ".publish").unlink()
+        gitea = FakeGitea(publish_wiki, wiki={"Old": b64("hand written")})
+
+        assert run_main(publish_wiki, gitea) == 0
+
+        assert gitea.calls == []
+        assert "wiki publishing is not enabled" in capsys.readouterr().out
+        assert gitea.wiki == {"Old": b64("hand written")}
+
+    def test_a_repository_without_the_marker_does_not_need_the_secret(
+        self, publish_wiki, environment, monkeypatch
+    ):
+        (environment / ".publish").unlink()
+        monkeypatch.setenv("WIKI_CREDENTIALS", "")
+        assert run_main(publish_wiki, FakeGitea(publish_wiki)) == 0
+
+    def test_a_repository_without_a_wiki_directory_is_skipped(
+        self, publish_wiki, monkeypatch, tmp_path
+    ):
+        monkeypatch.setenv("WIKI_DIR", str(tmp_path / "missing"))
+        monkeypatch.setenv("SOURCE_REPOSITORY", "TirSystem/repo")
+        gitea = FakeGitea(publish_wiki)
+        assert run_main(publish_wiki, gitea) == 0
+        assert gitea.calls == []
+
+    def test_an_opted_in_repository_without_the_secret_fails_before_any_request(
+        self, publish_wiki, environment, monkeypatch
+    ):
+        monkeypatch.setenv("WIKI_CREDENTIALS", "")
+        gitea = FakeGitea(publish_wiki)
+
+        with pytest.raises(publish_wiki.WorkflowError, match="TOKEN_FOR_GITEA_WIKI"):
+            run_main(publish_wiki, gitea)
+
+        assert gitea.calls == []
+
+    def test_an_unknown_marker_content_is_refused_before_any_request(
+        self, publish_wiki, environment
+    ):
+        (environment / ".publish").write_text("mirror everything", encoding="utf-8")
+        gitea = FakeGitea(publish_wiki)
+
+        with pytest.raises(publish_wiki.WorkflowError, match="prune"):
+            run_main(publish_wiki, gitea)
+
+        assert gitea.calls == []
+
+
+class TestPublish:
+    def test_prune_creates_updates_and_deletes_to_match_the_source(
+        self, publish_wiki, environment, capsys
     ):
         gitea = FakeGitea(
-            wiki={"Home": b64("home v1"), "Credentials": b64("credentials"), "Old": b64("x")}
+            publish_wiki,
+            wiki={"Home": b64("home v1"), "Credentials": b64("credentials"), "Old": b64("x")},
         )
 
-        assert run_main(monkeypatch, gitea) == 0
+        assert run_main(publish_wiki, gitea) == 0
 
         assert gitea.wiki == {"Home": b64("home v2"), "Credentials": b64("credentials")}
         methods = [(call[0], call[1].rsplit("/", 1)[-1]) for call in gitea.calls]
@@ -122,73 +193,80 @@ class TestMain:
         assert ("PATCH", "Credentials") not in methods
         assert "0 created, 1 updated, 1 deleted" in capsys.readouterr().out
 
-    def test_creates_every_page_when_the_wiki_was_never_written(
+    def test_without_prune_pages_outside_the_source_are_kept(
-        self, monkeypatch, environment
+        self, publish_wiki, environment, capsys
     ):
-        gitea = FakeGitea(wiki_initialised=False)
+        (environment / ".publish").write_text("", encoding="utf-8")
+        gitea = FakeGitea(
+            publish_wiki, wiki={"Home": b64("home v1"), "Old": b64("hand written")}
+        )
 
-        assert run_main(monkeypatch, gitea) == 0
+        assert run_main(publish_wiki, gitea) == 0
+
+        assert gitea.wiki["Old"] == b64("hand written")
+        assert gitea.wiki["Home"] == b64("home v2")
+        assert "Credentials" in gitea.wiki
+        assert not [call for call in gitea.calls if call[0] == "DELETE"]
+        out = capsys.readouterr().out
+        assert "1 created, 1 updated, 0 deleted" in out
+        assert "1 page(s) are in the wiki but not in" in out
+        assert "prune" in out
+
+    def test_creates_every_page_when_the_wiki_was_never_written(
+        self, publish_wiki, environment
+    ):
+        gitea = FakeGitea(publish_wiki, wiki_initialised=False)
+
+        assert run_main(publish_wiki, gitea) == 0
 
         assert set(gitea.wiki) == {"Home", "Credentials"}
         assert [call[0] for call in gitea.calls].count("POST") == 2
 
-    def test_does_nothing_when_the_wiki_is_already_current(self, monkeypatch, environment):
+    def test_does_nothing_when_the_wiki_is_already_current(self, publish_wiki, environment):
-        gitea = FakeGitea(wiki={"Home": b64("home v2"), "Credentials": b64("credentials")})
+        gitea = FakeGitea(
+            publish_wiki, wiki={"Home": b64("home v2"), "Credentials": b64("credentials")}
+        )
 
-        assert run_main(monkeypatch, gitea) == 0
+        assert run_main(publish_wiki, gitea) == 0
 
-        assert all(call[0] == "GET" for call in gitea.calls)
+        assert gitea.writes == []
 
-    def test_sends_the_token_and_never_prints_it(self, monkeypatch, environment, capsys):
+    def test_sends_the_token_and_never_prints_it(self, publish_wiki, environment, capsys):
-        seen = []
+        gitea = FakeGitea(publish_wiki)
 
-        class Recording(FakeGitea):
+        assert run_main(publish_wiki, gitea) == 0
-            def __call__(self, url, method="GET", headers=None, body=None):
-                seen.append(headers["Authorization"])
-                return super().__call__(url, method, headers, body)
 
-        assert run_main(monkeypatch, Recording()) == 0
+        assert {call[3]["Authorization"] for call in gitea.calls} == {"token wiki-token"}
-
-        assert set(seen) == {"token wiki-token"}
         output = capsys.readouterr()
         assert "wiki-token" not in output.out + output.err
 
-    def test_fails_when_the_wiki_is_not_enabled(self, monkeypatch, environment):
+    def test_fails_when_the_wiki_is_not_enabled_and_writes_nothing(
-        gitea = FakeGitea(has_wiki=False)
+        self, publish_wiki, environment
+    ):
+        gitea = FakeGitea(publish_wiki, has_wiki=False)
 
         with pytest.raises(publish_wiki.WorkflowError) as caught:
-            run_main(monkeypatch, gitea)
+            run_main(publish_wiki, gitea)
 
         assert caught.value.category == "configuration"
-        assert all(call[0] == "GET" for call in gitea.calls)
+        assert gitea.writes == []
 
-    def test_fails_before_any_request_when_the_secret_is_missing(
+    def test_an_empty_source_never_deletes_pages_even_with_prune(
-        self, monkeypatch, environment
+        self, publish_wiki, environment
-    ):
-        monkeypatch.setenv("WIKI_CREDENTIALS", "")
-        gitea = FakeGitea()
-
-        with pytest.raises(publish_wiki.WorkflowError, match="No token for the wiki"):
-            run_main(monkeypatch, gitea)
-
-        assert gitea.calls == []
-
-    def test_fails_before_any_request_when_the_source_is_empty(
-        self, monkeypatch, environment
     ):
         for path in environment.glob("*.md"):
             path.unlink()
-        gitea = FakeGitea(wiki={"Home": b64("keep me")})
+        gitea = FakeGitea(publish_wiki, wiki={"Home": b64("keep me")})
 
         with pytest.raises(publish_wiki.WorkflowError, match="no Markdown pages"):
-            run_main(monkeypatch, gitea)
+            run_main(publish_wiki, gitea)
 
         assert gitea.calls == []
         assert gitea.wiki == {"Home": b64("keep me")}
 
 
 class TestDescribeToken:
-    def test_recognises_a_gitea_token_shape(self):
+    def test_recognises_a_gitea_token_shape(self, publish_wiki):
         description = publish_wiki.describe_token("a1b2c3d4e5" * 4)
         assert "40 characters" in description
         assert "looks like a Gitea token" in description
@@ -202,20 +280,24 @@ class TestDescribeToken:
             ("abc def", "whitespace"),
         ],
     )
-    def test_names_the_likely_mistake_without_echoing_the_token(self, token, expected):
+    def test_names_the_likely_mistake_without_echoing_the_token(
+        self, publish_wiki, token, expected
+    ):
         description = publish_wiki.describe_token(token)
         assert expected in description
         assert token not in description
 
 
 class TestRejectedToken:
-    def test_rejection_adds_the_token_shape_but_not_the_token(self, monkeypatch, environment):
+    def test_rejection_adds_the_token_shape_but_not_the_token(
+        self, publish_wiki, environment, monkeypatch
+    ):
         monkeypatch.setenv("WIKI_CREDENTIALS", "*******")
 
         def reject(url, method="GET", headers=None, body=None):
             raise publish_wiki.WorkflowError("credentials", "Gitea rejected the token.")
 
-        monkeypatch.setattr(publish_wiki, "request_json", reject)
+        publish_wiki.request_json = reject
 
         with pytest.raises(publish_wiki.WorkflowError) as caught:
             publish_wiki.main()
@@ -226,11 +308,11 @@ class TestRejectedToken:
         assert "only mask characters" in message
         assert "*******" not in message.split("Token shape:")[0]
 
-    def test_other_categories_are_not_changed(self, monkeypatch, environment):
+    def test_other_categories_are_not_changed(self, publish_wiki, environment):
         def forbid(url, method="GET", headers=None, body=None):
             raise publish_wiki.WorkflowError("permissions", "Forbidden.")
 
-        monkeypatch.setattr(publish_wiki, "request_json", forbid)
+        publish_wiki.request_json = forbid
 
         with pytest.raises(publish_wiki.WorkflowError) as caught:
             publish_wiki.main()
@@ -240,7 +322,7 @@ class TestRejectedToken:
 
 class TestRequestJson:
     @staticmethod
-    def raise_http_error(monkeypatch, code, reason):
+    def raise_http_error(publish_wiki, monkeypatch, code, reason):
         def fake_urlopen(request, timeout):
             raise urllib.error.HTTPError(
                 request.full_url, code, reason, {}, io.BytesIO(b"secret detail")
@@ -257,8 +339,8 @@ class TestRequestJson:
             (500, "Server Error", "api"),
         ],
     )
-    def test_categorizes_http_errors(self, monkeypatch, code, reason, category):
+    def test_categorizes_http_errors(self, publish_wiki, monkeypatch, code, reason, category):
-        self.raise_http_error(monkeypatch, code, reason)
+        self.raise_http_error(publish_wiki, monkeypatch, code, reason)
 
         with pytest.raises(publish_wiki.WorkflowError) as caught:
             publish_wiki.request_json(REPO_URL)
@@ -267,7 +349,7 @@ class TestRequestJson:
         assert f"HTTP {code}" in str(caught.value)
         assert "secret detail" not in str(caught.value)
 
-    def test_reports_unreachable_service_as_api_failure(self, monkeypatch):
+    def test_reports_unreachable_service_as_api_failure(self, publish_wiki, monkeypatch):
         def fake_urlopen(request, timeout):
             raise urllib.error.URLError("no route")
 
@@ -282,7 +364,7 @@ class TestRequestJson:
 class TestWorkflow:
     @pytest.fixture()
     def document(self):
-        return yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
+        return yaml.safe_load(PUBLISH_WORKFLOW.read_text(encoding="utf-8"))
 
     def test_runs_only_on_main_when_wiki_changes_or_by_hand(self, document):
         triggers = document[True]  # PyYAML reads the key `on` as True
@@ -291,21 +373,23 @@ class TestWorkflow:
         assert "workflow_dispatch" in triggers
 
     def test_job_has_read_only_permissions(self, document):
-        job = document["jobs"]["publish-wiki"]
+        assert document["jobs"]["publish-wiki"]["permissions"] == {"contents": "read"}
-        assert job["permissions"] == {"contents": "read"}
 
-    def test_checks_the_pages_before_publishing_them(self, document):
+    def test_secret_is_passed_only_through_the_environment_and_has_no_fallback(self, document):
-        runs = [step.get("run", "") for step in document["jobs"]["publish-wiki"]["steps"]]
+        text = PUBLISH_WORKFLOW.read_text(encoding="utf-8")
-        assert runs.index("python3 tools/build_wiki.py --check") < runs.index(
+        assert "secrets.TOKEN_FOR_GITEA }}" not in text
-            "python3 tools/publish_wiki.py"
+        publish_step = document["jobs"]["publish-wiki"]["steps"][-1]
-        )
+        assert publish_step["env"]["WIKI_CREDENTIALS"] == "${{ secrets.TOKEN_FOR_GITEA_WIKI }}"
-
-    def test_secret_is_passed_only_through_the_environment(self, document):
-        text = WORKFLOW.read_text(encoding="utf-8")
-        assert text.count("secrets.TOKEN_FOR_GITEA_WIKI") == 1
-        steps = document["jobs"]["publish-wiki"]["steps"]
-        publish_step = steps[-1]
-        assert publish_step["env"]["WIKI_CREDENTIALS"] == (
-            "${{ secrets.TOKEN_FOR_GITEA_WIKI || secrets.TOKEN_FOR_GITEA }}"
-        )
         assert "secrets" not in publish_step["run"]
+
+    def test_the_workflow_has_one_source_and_needs_nothing_from_tools(self):
+        assert not (REPO_ROOT / ".gitea" / "workflows" / "publish-wiki.yml").exists()
+        assert not (REPO_ROOT / "tools" / "publish_wiki.py").exists()
+        script = PUBLISH_WORKFLOW.read_text(encoding="utf-8").split("python3 - <<'PY'", 1)[1]
+        assert "tools/" not in script
+
+
+def test_this_repository_opts_in_and_prunes():
+    """The wiki of this repository must equal wiki/ (ADR-0001, ADR-0004)."""
+    marker = REPO_ROOT / "wiki" / ".publish"
+    assert marker.read_text(encoding="utf-8").strip() == "prune"

tests/workflow_source.py

@@ -11,6 +11,7 @@ WORKFLOW_DIRS = (
     REPO_ROOT / ".gitea" / "workflows",
 )
 SYNC_WORKFLOW = WORKFLOW_DIRS[0] / "sync-github-metadata.yml"
+PUBLISH_WORKFLOW = WORKFLOW_DIRS[0] / "publish-wiki.yml"
 
 _HEREDOC = re.compile(
     r"^[ \t]*python3 - <<'PY'\r?\n(?P<code>.*?)^[ \t]*PY[ \t]*\r?$",
@@ -42,3 +43,28 @@ def load_sync_module(name: str = "embedded_sync") -> dict[str, Any]:
     namespace: dict[str, Any] = {"__name__": name}
     exec(compile(source, str(SYNC_WORKFLOW), "exec"), namespace)
     return namespace
+
+
+class ScriptNamespace(dict):
+    """The globals of an embedded script, readable and writable as attributes.
+
+    Functions of the script look names up in this dict, so a test that sets
+    `namespace.request_json = fake` changes what the script calls.
+    """
+
+    def __getattr__(self, name: str) -> Any:
+        try:
+            return self[name]
+        except KeyError:
+            raise AttributeError(name) from None
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        self[name] = value
+
+
+def load_publish_module(name: str = "embedded_publish") -> ScriptNamespace:
+    """Execute the embedded publish script without running `main`."""
+    (source,) = embedded_python(PUBLISH_WORKFLOW)
+    namespace = ScriptNamespace({"__name__": name})
+    exec(compile(source, str(PUBLISH_WORKFLOW), "exec"), namespace)
+    return namespace

tools/publish_wiki.py

@@ -1,256 +0,0 @@
-"""Publish the pages in ``wiki/`` to the Gitea wiki of this repository.
-
-The pages are written through the Gitea wiki REST API. A page that is missing
-in ``wiki/`` is removed from the wiki, so the wiki always equals the reviewed
-source; edits made in the web UI are overwritten (MIL-005).
-
-Environment:
-    GITEA_API_URL        Gitea API base, for example https://host/api/v1
-    SOURCE_REPOSITORY    owner/repo of the repository whose wiki is written
-    WIKI_CREDENTIALS     the Gitea token, as clear text
-    WIKI_DIR             directory with the pages (default: wiki)
-
-Exit codes: 0 published or nothing to do, 1 failed (message names a category).
-"""
-
-import base64
-import json
-import os
-import re
-import sys
-import urllib.error
-import urllib.parse
-import urllib.request
-from collections.abc import Callable
-from pathlib import Path
-from typing import Any
-
-PAGE_LIMIT = 50
-REQUEST_TIMEOUT_SECONDS = 30
-COMMIT_MESSAGE = "Publish wiki from the repository"
-
-JsonRequest = Callable[..., Any]
-
-
-class WorkflowError(RuntimeError):
-    """A failure with a category.
-
-    configuration: a secret, a setting or the wiki source is missing or malformed.
-    credentials:   Gitea 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: str, message: str) -> None:
-        super().__init__(f"[{category}] {message}")
-        self.category = category
-        self.detail = message
-
-
-def request_json(
-    url: str,
-    method: str = "GET",
-    headers: dict[str, str] | None = None,
-    body: dict[str, Any] | None = None,
-) -> Any:
-    """Send one JSON request to Gitea and return the decoded response."""
-    request = urllib.request.Request(
-        url,
-        data=json.dumps(body).encode("utf-8") if body is not None else None,
-        headers=headers or {},
-        method=method,
-    )
-    try:
-        with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
-            content = response.read()
-            return json.loads(content) if content else None
-    except urllib.error.HTTPError as error:
-        detail = f"Gitea API request failed with HTTP {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 or page 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"Gitea API is unreachable ({error}).") from None
-
-
-def parse_gitea_token(raw_token: str | None) -> str:
-    """Return the token, or fail when the secret is missing or empty."""
-    token = (raw_token or "").strip()
-    if not token:
-        raise WorkflowError(
-            "configuration",
-            "No token for the wiki: set TOKEN_FOR_GITEA_WIKI, or TOKEN_FOR_GITEA.",
-        )
-    return token
-
-
-def describe_token(token: str) -> str:
-    """Describe the shape of a token without revealing any of its characters.
-
-    Used when Gitea rejects the token, so a wrongly stored secret (the masked
-    text, ``NAME=value``, quotes) can be recognised from the log.
-    """
-    facts = [f"{len(token)} characters"]
-    if re.fullmatch(r"[0-9a-f]{40}", token):
-        facts.append("looks like a Gitea token (40 hexadecimal characters)")
-        return ", ".join(facts)
-    facts.append("a Gitea token has 40 hexadecimal characters")
-    if set(token) <= {"*", "x", "X", "•"}:
-        facts.append("only mask characters: the masked text was stored")
-    if "=" in token:
-        facts.append("contains '=': the secret name may be part of the value")
-    if re.search(r"[\"'`]", token):
-        facts.append("contains a quote character")
-    if re.search(r"\s", token):
-        facts.append("contains whitespace")
-    return ", ".join(facts)
-
-
-def split_repository(full_name: str | None) -> tuple[str, str]:
-    """Split ``owner/repo`` into its two parts."""
-    owner, separator, repo = (full_name or "").partition("/")
-    if not separator or not owner or not repo:
-        raise WorkflowError("configuration", "Could not determine the Gitea repository.")
-    return owner, repo
-
-
-def read_pages(wiki_dir: Path) -> dict[str, bytes]:
-    """Return ``{page title: content}`` for every Markdown file in ``wiki_dir``."""
-    if not wiki_dir.is_dir():
-        raise WorkflowError("configuration", f"The wiki source directory {wiki_dir} is missing.")
-    pages = {path.stem: path.read_bytes() for path in sorted(wiki_dir.glob("*.md"))}
-    if not pages:
-        # Publishing an empty directory would delete the whole wiki.
-        raise WorkflowError(
-            "configuration", f"The wiki source directory {wiki_dir} has no Markdown pages."
-        )
-    return pages
-
-
-def list_wiki_titles(wiki_url: str, headers: dict[str, str], send: JsonRequest) -> set[str]:
-    """Return the titles of the pages that exist in the wiki.
-
-    Gitea answers 404 for a wiki that has never been written to; that is an
-    empty wiki, not a failure.
-    """
-    titles: set[str] = set()
-    page = 1
-    while True:
-        url = f"{wiki_url}/pages?page={page}&limit={PAGE_LIMIT}"
-        try:
-            batch = send(url, headers=headers)
-        except WorkflowError as error:
-            if error.category == "configuration" and page == 1:
-                return titles
-            raise
-        if not isinstance(batch, list):
-            raise WorkflowError("api", "Gitea returned an invalid wiki page list.")
-        titles.update(str(item.get("title", "")) for item in batch)
-        if len(batch) < PAGE_LIMIT:
-            return titles
-        page += 1
-
-
-def read_wiki_page(
-    wiki_url: str, title: str, headers: dict[str, str], send: JsonRequest
-) -> bytes:
-    """Return the current content of one wiki page."""
-    quoted = urllib.parse.quote(title, safe="")
-    page = send(f"{wiki_url}/page/{quoted}", headers=headers)
-    encoded = (page or {}).get("content_base64", "")
-    return base64.b64decode(encoded)
-
-
-def publish(
-    pages: dict[str, bytes],
-    wiki_url: str,
-    headers: dict[str, str],
-    send: JsonRequest,
-) -> dict[str, list[str]]:
-    """Make the wiki equal ``pages``; return the titles created, updated, deleted."""
-    existing = list_wiki_titles(wiki_url, headers, send)
-    current = {
-        title: read_wiki_page(wiki_url, title, headers, send)
-        for title in sorted(existing & pages.keys())
-    }
-
-    result: dict[str, list[str]] = {"created": [], "updated": [], "deleted": []}
-    for title, content in pages.items():
-        body = {
-            "title": title,
-            "content_base64": base64.b64encode(content).decode("ascii"),
-            "message": COMMIT_MESSAGE,
-        }
-        if title not in existing:
-            send(f"{wiki_url}/new", method="POST", headers=headers, body=body)
-            result["created"].append(title)
-        elif current[title] != content:
-            quoted = urllib.parse.quote(title, safe="")
-            send(f"{wiki_url}/page/{quoted}", method="PATCH", headers=headers, body=body)
-            result["updated"].append(title)
-
-    for title in sorted(existing - pages.keys()):
-        quoted = urllib.parse.quote(title, safe="")
-        send(f"{wiki_url}/page/{quoted}", method="DELETE", headers=headers)
-        result["deleted"].append(title)
-    return result
-
-
-def main() -> int:
-    """Run the publish; return the process exit code."""
-    wiki_dir = Path(os.environ.get("WIKI_DIR", "wiki"))
-    pages = read_pages(wiki_dir)
-    token = parse_gitea_token(os.environ.get("WIKI_CREDENTIALS"))
-    owner, repo = split_repository(os.environ.get("SOURCE_REPOSITORY"))
-
-    api_url = os.environ["GITEA_API_URL"].rstrip("/")
-    repository_url = (
-        f"{api_url}/repos/{urllib.parse.quote(owner, safe='')}/"
-        f"{urllib.parse.quote(repo, safe='')}"
-    )
-    headers = {
-        "Authorization": f"token {token}",
-        "Accept": "application/json",
-        "Content-Type": "application/json",
-    }
-
-    try:
-        repository = request_json(repository_url, headers=headers)
-        if isinstance(repository, dict) and repository.get("has_wiki") is False:
-            raise WorkflowError(
-                "configuration",
-                f"The wiki is not enabled on {owner}/{repo}. Enable it in the repository settings.",
-            )
-        print("Preflight passed: Gitea accepted the token and the wiki is enabled.")
-
-        result = publish(pages, f"{repository_url}/wiki", headers, request_json)
-    except WorkflowError as error:
-        if error.category == "credentials":
-            raise WorkflowError(
-                "credentials", f"{error.detail} Token shape: {describe_token(token)}."
-            ) from None
-        raise
-    print(
-        f"Published {len(pages)} page(s) to the wiki of {owner}/{repo}: "
-        f"{len(result['created'])} created, {len(result['updated'])} updated, "
-        f"{len(result['deleted'])} deleted."
-    )
-    return 0
-
-
-if __name__ == "__main__":
-    try:
-        sys.exit(main())
-    except WorkflowError as error:
-        print(f"ERROR: {error}", file=sys.stderr)
-        sys.exit(1)

wiki/.publish

@@ -0,0 +1 @@
+prune