GRM-39: fix: use content_base64 for Gitea wiki API, add --verify flag (#33)
This commit is contained in:
emil
@@ -22,7 +22,7 @@ jobs:
|
||||
PYTHONPATH: src
|
||||
run: |
|
||||
. .venv/bin/activate
|
||||
python3 scripts/ci/sync_wiki.py --repo "${{ github.repository }}"
|
||||
python3 scripts/ci/sync_wiki.py --repo "${{ github.repository }}" --verify
|
||||
- name: Tag wiki on release
|
||||
if: startsWith(github.ref, 'refs/tags/v')
|
||||
env:
|
||||
|
||||
+93
-7
@@ -6,18 +6,20 @@ map file paths to wiki page titles, and creates/updates wiki pages via the
|
||||
Gitea API. Pages that exist in the wiki but not in the mapping are left
|
||||
untouched (not deleted).
|
||||
|
||||
Gitea 1.26 wiki API endpoints:
|
||||
- Create: POST /repos/{owner}/{repo}/wiki/new {title, content, message}
|
||||
- Update: PATCH /repos/{owner}/{repo}/wiki/page/{sub_url} {title, content, message}
|
||||
Gitea 1.26 wiki API endpoints (all use content_base64, NOT content):
|
||||
- Create: POST /repos/{owner}/{repo}/wiki/new {title, content_base64, message}
|
||||
- Update: PATCH /repos/{owner}/{repo}/wiki/page/{sub_url} {title, content_base64, message}
|
||||
- List: GET /repos/{owner}/{repo}/wiki/pages → [{title, sub_url, ...}]
|
||||
- Fetch: GET /repos/{owner}/{repo}/wiki/page/{sub_url} → {title, content_base64, ...}
|
||||
- Delete: DELETE /repos/{owner}/{repo}/wiki/page/{sub_url}
|
||||
|
||||
Usage:
|
||||
REPO_TOKEN=<token> python3 scripts/sync_wiki.py [--dry-run] [--repo owner/repo]
|
||||
REPO_TOKEN=<token> python3 scripts/ci/sync_wiki.py [--dry-run] [--repo owner/repo]
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
@@ -49,6 +51,23 @@ def read_doc_content(file_path: str) -> str:
|
||||
return f.read()
|
||||
|
||||
|
||||
def encode_content(content: str) -> str:
|
||||
"""Encode content as base64 for the Gitea wiki API.
|
||||
|
||||
The Gitea wiki API requires content_base64, not plain content.
|
||||
Sending plain content silently fails (pages are created/updated
|
||||
but with empty content).
|
||||
"""
|
||||
return base64.b64encode(content.encode("utf-8")).decode("ascii")
|
||||
|
||||
|
||||
def decode_content(content_b64: str) -> str:
|
||||
"""Decode base64 content from the Gitea wiki API."""
|
||||
if not content_b64:
|
||||
return ""
|
||||
return base64.b64decode(content_b64).decode("utf-8")
|
||||
|
||||
|
||||
def list_wiki_pages(client: GiteaClient) -> dict[str, str]:
|
||||
"""List existing wiki pages, returning {title: sub_url}."""
|
||||
try:
|
||||
@@ -58,6 +77,15 @@ def list_wiki_pages(client: GiteaClient) -> dict[str, str]:
|
||||
return {page.get("title", ""): page.get("sub_url", page.get("title", "")) for page in pages}
|
||||
|
||||
|
||||
def fetch_page_content(client: GiteaClient, sub_url: str) -> str:
|
||||
"""Fetch a wiki page's content by sub_url, decoded from base64."""
|
||||
try:
|
||||
page = client._request("GET", f"/wiki/page/{sub_url}").json()
|
||||
return decode_content(page.get("content_base64", ""))
|
||||
except APIError:
|
||||
return ""
|
||||
|
||||
|
||||
def sync_page(
|
||||
client: GiteaClient,
|
||||
page_title: str,
|
||||
@@ -73,13 +101,19 @@ def sync_page(
|
||||
click.echo(_("[dry-run] Would sync page: {title} ({chars} chars)", title=page_title, chars=len(content)))
|
||||
return "skipped"
|
||||
|
||||
content_b64 = encode_content(content)
|
||||
|
||||
if page_title in existing_pages:
|
||||
# Update existing page via PATCH
|
||||
sub_url = existing_pages[page_title]
|
||||
client._request(
|
||||
"PATCH",
|
||||
f"/wiki/page/{sub_url}",
|
||||
json={"title": page_title, "content": content, "message": f"Sync from docs/ — update {page_title}"},
|
||||
json={
|
||||
"title": page_title,
|
||||
"content_base64": content_b64,
|
||||
"message": f"Sync from docs/ — update {page_title}",
|
||||
},
|
||||
)
|
||||
return "updated"
|
||||
|
||||
@@ -87,15 +121,37 @@ def sync_page(
|
||||
client._request(
|
||||
"POST",
|
||||
"/wiki/new",
|
||||
json={"title": page_title, "content": content, "message": f"Sync from docs/ — create {page_title}"},
|
||||
json={
|
||||
"title": page_title,
|
||||
"content_base64": content_b64,
|
||||
"message": f"Sync from docs/ — create {page_title}",
|
||||
},
|
||||
)
|
||||
return "created"
|
||||
|
||||
|
||||
def verify_wiki_page(client: GiteaClient, page_title: str, expected_content: str, existing_pages: dict[str, str]) -> bool:
|
||||
"""Verify that a wiki page has non-empty content matching the docs.
|
||||
|
||||
Returns True if the page content matches, False otherwise.
|
||||
"""
|
||||
if page_title not in existing_pages:
|
||||
return False
|
||||
sub_url = existing_pages[page_title]
|
||||
actual = fetch_page_content(client, sub_url)
|
||||
return actual.strip() == expected_content.strip()
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option("--dry-run", is_flag=True, default=False, help="Show what would happen without making changes.")
|
||||
@click.option("--repo", default=None, help="Repository in owner/name format (auto-detected if omitted).")
|
||||
def main(dry_run: bool, repo: str | None) -> None:
|
||||
@click.option(
|
||||
"--verify",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="After syncing, verify each page has non-empty content. Exit 1 if any page is empty or mismatched.",
|
||||
)
|
||||
def main(dry_run: bool, repo: str | None, verify: bool) -> None:
|
||||
token = os.environ.get("REPO_TOKEN", "")
|
||||
if not token:
|
||||
raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
|
||||
@@ -121,6 +177,7 @@ def main(dry_run: bool, repo: str | None) -> None:
|
||||
created = 0
|
||||
updated = 0
|
||||
skipped = 0
|
||||
synced: dict[str, str] = {} # title -> content, for verification
|
||||
|
||||
for file_path, page_title in sorted(mapping.items()):
|
||||
try:
|
||||
@@ -130,6 +187,11 @@ def main(dry_run: bool, repo: str | None) -> None:
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
if not content.strip():
|
||||
click.echo(_("WARNING: File {file} is empty — skipping.", file=file_path))
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
result = sync_page(client, page_title, content, existing_pages, dry_run)
|
||||
if result == "created":
|
||||
created += 1
|
||||
@@ -140,6 +202,8 @@ def main(dry_run: bool, repo: str | None) -> None:
|
||||
else:
|
||||
skipped += 1
|
||||
|
||||
synced[page_title] = content
|
||||
|
||||
click.echo(
|
||||
_(
|
||||
"\nDone! Created: {created}, Updated: {updated}, Skipped: {skipped}",
|
||||
@@ -149,6 +213,28 @@ def main(dry_run: bool, repo: str | None) -> None:
|
||||
)
|
||||
)
|
||||
|
||||
if verify and not dry_run:
|
||||
click.echo(_("\nVerifying wiki pages have content..."))
|
||||
# Re-fetch the page list to get updated sub_urls
|
||||
existing_pages = list_wiki_pages(client)
|
||||
failures = 0
|
||||
for page_title, expected_content in sorted(synced.items()):
|
||||
ok = verify_wiki_page(client, page_title, expected_content, existing_pages)
|
||||
if ok:
|
||||
click.echo(_(" OK: {title} ({chars} chars)", title=page_title, chars=len(expected_content)))
|
||||
else:
|
||||
click.echo(_(" FAIL: {title} — content mismatch or empty!", title=page_title))
|
||||
failures += 1
|
||||
if failures > 0:
|
||||
click.echo(
|
||||
_(
|
||||
"\nVerification FAILED: {failures} page(s) have empty or mismatched content!",
|
||||
failures=failures,
|
||||
)
|
||||
)
|
||||
raise click.ClickException(_("Wiki verification failed — {failures} page(s) empty or mismatched", failures=failures))
|
||||
click.echo(_("\nVerification passed — all wiki pages have correct content."))
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Unit tests for scripts/ci/sync_wiki.py."""
|
||||
|
||||
import base64
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
@@ -8,14 +9,46 @@ import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from scripts.ci.sync_wiki import (
|
||||
decode_content,
|
||||
encode_content,
|
||||
fetch_page_content,
|
||||
list_wiki_pages,
|
||||
load_mapping,
|
||||
main,
|
||||
read_doc_content,
|
||||
sync_page,
|
||||
verify_wiki_page,
|
||||
)
|
||||
|
||||
|
||||
class TestEncodeContent:
|
||||
def test_encodes_utf8_to_base64(self) -> None:
|
||||
result = encode_content("# Hello World")
|
||||
assert result == base64.b64encode(b"# Hello World").decode("ascii")
|
||||
|
||||
def test_encodes_empty_string(self) -> None:
|
||||
assert encode_content("") == ""
|
||||
|
||||
def test_encodes_unicode(self) -> None:
|
||||
result = encode_content("# Café — résumé")
|
||||
decoded = base64.b64decode(result).decode("utf-8")
|
||||
assert decoded == "# Café — résumé"
|
||||
|
||||
|
||||
class TestDecodeContent:
|
||||
def test_decodes_base64_to_utf8(self) -> None:
|
||||
encoded = base64.b64encode(b"# Hello").decode("ascii")
|
||||
assert decode_content(encoded) == "# Hello"
|
||||
|
||||
def test_empty_string_returns_empty(self) -> None:
|
||||
assert decode_content("") == ""
|
||||
|
||||
def test_roundtrip(self) -> None:
|
||||
original = "# Wiki Page\n\nContent with **markdown**."
|
||||
encoded = encode_content(original)
|
||||
assert decode_content(encoded) == original
|
||||
|
||||
|
||||
class TestLoadMapping:
|
||||
def test_loads_mapping(self, tmp_path: Path) -> None:
|
||||
mapping_file = tmp_path / "mapping.json"
|
||||
@@ -64,6 +97,27 @@ class TestListWikiPages:
|
||||
assert result == {"Home": "Home", "Getting-Started": "Getting-Started.-"}
|
||||
|
||||
|
||||
class TestFetchPageContent:
|
||||
def test_fetches_and_decodes_content(self) -> None:
|
||||
client = MagicMock()
|
||||
encoded = base64.b64encode(b"# Hello Wiki").decode("ascii")
|
||||
client._request.return_value.json.return_value = {"content_base64": encoded}
|
||||
result = fetch_page_content(client, "Home")
|
||||
assert result == "# Hello Wiki"
|
||||
|
||||
def test_returns_empty_on_api_error(self) -> None:
|
||||
from gitea_runner_manager.exceptions import APIError
|
||||
|
||||
client = MagicMock()
|
||||
client._request.side_effect = APIError(404, "not found")
|
||||
assert fetch_page_content(client, "Missing") == ""
|
||||
|
||||
def test_returns_empty_for_empty_content(self) -> None:
|
||||
client = MagicMock()
|
||||
client._request.return_value.json.return_value = {"content_base64": ""}
|
||||
assert fetch_page_content(client, "Home") == ""
|
||||
|
||||
|
||||
class TestSyncPage:
|
||||
def test_dry_run_skips(self) -> None:
|
||||
client = MagicMock()
|
||||
@@ -71,7 +125,7 @@ class TestSyncPage:
|
||||
assert result == "skipped"
|
||||
client._request.assert_not_called()
|
||||
|
||||
def test_creates_new_page(self) -> None:
|
||||
def test_creates_new_page_with_base64(self) -> None:
|
||||
client = MagicMock()
|
||||
result = sync_page(client, "New-Page", "# Content", {}, dry_run=False)
|
||||
assert result == "created"
|
||||
@@ -79,8 +133,13 @@ class TestSyncPage:
|
||||
call_args = client._request.call_args
|
||||
assert call_args.args[0] == "POST"
|
||||
assert call_args.args[1] == "/wiki/new"
|
||||
# Verify content_base64 is used, not content
|
||||
payload = call_args.kwargs["json"]
|
||||
assert "content_base64" in payload
|
||||
assert "content" not in payload
|
||||
assert base64.b64decode(payload["content_base64"]).decode("utf-8") == "# Content"
|
||||
|
||||
def test_updates_existing_page(self) -> None:
|
||||
def test_updates_existing_page_with_base64(self) -> None:
|
||||
client = MagicMock()
|
||||
existing = {"Existing-Page": "Existing-Page.-"}
|
||||
result = sync_page(client, "Existing-Page", "# Updated", existing, dry_run=False)
|
||||
@@ -89,6 +148,37 @@ class TestSyncPage:
|
||||
call_args = client._request.call_args
|
||||
assert call_args.args[0] == "PATCH"
|
||||
assert "/wiki/page/Existing-Page.-" in call_args.args[1]
|
||||
# Verify content_base64 is used
|
||||
payload = call_args.kwargs["json"]
|
||||
assert "content_base64" in payload
|
||||
assert "content" not in payload
|
||||
assert base64.b64decode(payload["content_base64"]).decode("utf-8") == "# Updated"
|
||||
|
||||
|
||||
class TestVerifyWikiPage:
|
||||
def test_verifies_matching_content(self) -> None:
|
||||
client = MagicMock()
|
||||
encoded = base64.b64encode(b"# Hello Wiki").decode("ascii")
|
||||
client._request.return_value.json.return_value = {"content_base64": encoded}
|
||||
existing = {"Home": "Home"}
|
||||
assert verify_wiki_page(client, "Home", "# Hello Wiki", existing) is True
|
||||
|
||||
def test_fails_on_mismatch(self) -> None:
|
||||
client = MagicMock()
|
||||
encoded = base64.b64encode(b"# Old Content").decode("ascii")
|
||||
client._request.return_value.json.return_value = {"content_base64": encoded}
|
||||
existing = {"Home": "Home"}
|
||||
assert verify_wiki_page(client, "Home", "# New Content", existing) is False
|
||||
|
||||
def test_fails_on_empty_wiki_content(self) -> None:
|
||||
client = MagicMock()
|
||||
client._request.return_value.json.return_value = {"content_base64": ""}
|
||||
existing = {"Home": "Home"}
|
||||
assert verify_wiki_page(client, "Home", "# Expected", existing) is False
|
||||
|
||||
def test_fails_when_page_not_in_existing(self) -> None:
|
||||
client = MagicMock()
|
||||
assert verify_wiki_page(client, "Missing", "# Content", {}) is False
|
||||
|
||||
|
||||
class TestMain:
|
||||
@@ -168,6 +258,21 @@ class TestMain:
|
||||
assert "not found" in result.output
|
||||
assert "Skipped: 1" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
|
||||
@patch("scripts.ci.sync_wiki.GiteaClient")
|
||||
def test_empty_doc_file_skipped(self, mock_client_cls: MagicMock) -> None:
|
||||
"""Test that empty doc files are skipped with a warning."""
|
||||
with patch("scripts.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
|
||||
mock_mapping.exists.return_value = True
|
||||
with patch("scripts.ci.sync_wiki.load_mapping", return_value={"empty.md": "Empty-Page"}):
|
||||
with patch("scripts.ci.sync_wiki.read_doc_content", return_value=" \n "):
|
||||
with patch("scripts.ci.sync_wiki.list_wiki_pages", return_value={}):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--dry-run", "--repo", "owner/repo"])
|
||||
assert result.exit_code == 0
|
||||
assert "empty" in result.output.lower()
|
||||
assert "Skipped: 1" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
|
||||
@patch("scripts.ci.sync_wiki.GiteaClient")
|
||||
def test_create_and_update(self, mock_client_cls: MagicMock) -> None:
|
||||
@@ -185,3 +290,54 @@ class TestMain:
|
||||
assert result.exit_code == 0
|
||||
assert "Created: 1" in result.output
|
||||
assert "Updated: 1" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
|
||||
@patch("scripts.ci.sync_wiki.GiteaClient")
|
||||
def test_verify_passes(self, mock_client_cls: MagicMock) -> None:
|
||||
"""Test that --verify passes when content matches."""
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
encoded = base64.b64encode(b"# Home Content").decode("ascii")
|
||||
# list_wiki_pages returns {"Home": "Home"}, fetch returns encoded content
|
||||
mock_client._request.return_value.json.return_value = {"content_base64": encoded}
|
||||
with patch("scripts.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
|
||||
mock_mapping.exists.return_value = True
|
||||
with patch("scripts.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}):
|
||||
with patch("scripts.ci.sync_wiki.read_doc_content", return_value="# Home Content"):
|
||||
with patch("scripts.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}):
|
||||
with patch("scripts.ci.sync_wiki.verify_wiki_page", return_value=True):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--repo", "owner/repo", "--verify"])
|
||||
assert result.exit_code == 0
|
||||
assert "Verification passed" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
|
||||
@patch("scripts.ci.sync_wiki.GiteaClient")
|
||||
def test_verify_fails_on_empty_content(self, mock_client_cls: MagicMock) -> None:
|
||||
"""Test that --verify fails when wiki pages have empty content."""
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
with patch("scripts.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
|
||||
mock_mapping.exists.return_value = True
|
||||
with patch("scripts.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}):
|
||||
with patch("scripts.ci.sync_wiki.read_doc_content", return_value="# Home Content"):
|
||||
with patch("scripts.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}):
|
||||
with patch("scripts.ci.sync_wiki.verify_wiki_page", return_value=False):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--repo", "owner/repo", "--verify"])
|
||||
assert result.exit_code == 1
|
||||
assert "FAIL" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
|
||||
@patch("scripts.ci.sync_wiki.GiteaClient")
|
||||
def test_verify_skipped_in_dry_run(self, mock_client_cls: MagicMock) -> None:
|
||||
"""Test that --verify is skipped during dry-run."""
|
||||
with patch("scripts.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
|
||||
mock_mapping.exists.return_value = True
|
||||
with patch("scripts.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}):
|
||||
with patch("scripts.ci.sync_wiki.read_doc_content", return_value="# Home"):
|
||||
with patch("scripts.ci.sync_wiki.list_wiki_pages", return_value={}):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--dry-run", "--verify", "--repo", "owner/repo"])
|
||||
assert result.exit_code == 0
|
||||
assert "Verification" not in result.output
|
||||
|
||||
Reference in New Issue
Block a user