Make the wiki publisher a scoped workflow with an opt-in gate

Embed the publish script in .gitea/scoped_workflows/publish-wiki.yml and
remove the repository workflow and tools/publish_wiki.py. A repository
publishes only when wiki/.publish exists; the word prune in it is the only
thing that lets the workflow delete pages. The organisation's
TOKEN_FOR_GITEA_WIKI is used with no fallback. This repository opts in with
prune. Tests run the embedded script against a fake Gitea.

Task: MIL-006#2
Task: MIL-006#3

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
-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)
+from workflow_source import PUBLISH_WORKFLOW, REPO_ROOT, load_publish_module
 
 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):
-        with pytest.raises(publish_wiki.WorkflowError, match="missing"):
-            publish_wiki.read_pages(tmp_path / "nope")
-
-    def test_rejects_empty_directory_so_the_wiki_is_never_wiped(self, tmp_path):
+    def test_rejects_a_directory_without_pages_so_the_wiki_is_never_wiped(
+        self, publish_wiki, tmp_path
+    ):
+        (tmp_path / ".publish").write_text("prune", encoding="utf-8")
         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):
-    monkeypatch.setattr(publish_wiki, "request_json", gitea)
+def run_main(publish_wiki, gitea):
+    publish_wiki.request_json = gitea
     return publish_wiki.main()
 
 
-class TestMain:
-    def test_creates_updates_and_deletes_to_match_the_source(
-        self, monkeypatch, environment, capsys
+class TestOptIn:
+    def test_a_repository_without_the_marker_is_never_written_to(
+        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(
-        self, monkeypatch, environment
+    def test_without_prune_pages_outside_the_source_are_kept(
+        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):
-        gitea = FakeGitea(wiki={"Home": b64("home v2"), "Credentials": b64("credentials")})
+    def test_does_nothing_when_the_wiki_is_already_current(self, publish_wiki, environment):
+        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):
-        seen = []
+    def test_sends_the_token_and_never_prints_it(self, publish_wiki, environment, capsys):
+        gitea = FakeGitea(publish_wiki)
 
-        class Recording(FakeGitea):
-            def __call__(self, url, method="GET", headers=None, body=None):
-                seen.append(headers["Authorization"])
-                return super().__call__(url, method, headers, body)
+        assert run_main(publish_wiki, gitea) == 0
 
-        assert run_main(monkeypatch, Recording()) == 0
-
-        assert set(seen) == {"token wiki-token"}
+        assert {call[3]["Authorization"] for call in gitea.calls} == {"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):
-        gitea = FakeGitea(has_wiki=False)
+    def test_fails_when_the_wiki_is_not_enabled_and_writes_nothing(
+        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(
-        self, monkeypatch, 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
+    def test_an_empty_source_never_deletes_pages_even_with_prune(
+        self, publish_wiki, 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):
-        self.raise_http_error(monkeypatch, code, reason)
+    def test_categorizes_http_errors(self, publish_wiki, monkeypatch, code, reason, category):
+        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 job["permissions"] == {"contents": "read"}
+        assert document["jobs"]["publish-wiki"]["permissions"] == {"contents": "read"}
 
-    def test_checks_the_pages_before_publishing_them(self, document):
-        runs = [step.get("run", "") for step in document["jobs"]["publish-wiki"]["steps"]]
-        assert runs.index("python3 tools/build_wiki.py --check") < runs.index(
-            "python3 tools/publish_wiki.py"
-        )
-
-    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 }}"
-        )
+    def test_secret_is_passed_only_through_the_environment_and_has_no_fallback(self, document):
+        text = PUBLISH_WORKFLOW.read_text(encoding="utf-8")
+        assert "secrets.TOKEN_FOR_GITEA }}" not in text
+        publish_step = document["jobs"]["publish-wiki"]["steps"][-1]
+        assert publish_step["env"]["WIKI_CREDENTIALS"] == "${{ secrets.TOKEN_FOR_GITEA_WIKI }}"
         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"