diff --git a/src/devx/ci/sync_wiki.py b/src/devx/ci/sync_wiki.py index c3ba9f6..347ee49 100644 --- a/src/devx/ci/sync_wiki.py +++ b/src/devx/ci/sync_wiki.py @@ -1,17 +1,24 @@ #!/usr/bin/env python3 -"""Sync documentation from /docs/ to the Gitea wiki via API. +"""Sync documentation from /docs/ to the Gitea wiki via Git. -Reads markdown files from the ``docs/`` directory, uses ``mapping.json`` to -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). +Instead of using the Gitea wiki API (which is slow, unreliable, and +prone to timeouts), this module clones the wiki Git repository, +copies the documentation files into it, transforms internal links +to wiki-friendly format, commits, and pushes. -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} +This approach is: +- **Faster** — a single git push vs N API calls +- **More reliable** — no API timeouts or rate limits +- **Atomic** — all pages sync in one commit +- **Auto-pruning** — stale wiki pages are removed automatically + +The wiki Git URL is ``{clone_url}.wiki.git`` (Gitea convention). + +Link transformations: +- ``[text](file.md)`` → ``[text](file)`` (wiki pages don't use .md) +- ``[text](docs/file.md)`` → ``[text](file)`` +- External links (http/https/mailto) are preserved +- Anchor-only links (``#section``) are preserved Usage: CI_GITEA_TOKEN= python3 -m devx.ci.sync_wiki [--dry-run] [--repo owner/repo] @@ -19,44 +26,30 @@ Usage: from __future__ import annotations -import base64 import json -import logging import os +import re +import subprocess # nosec B404 +import tempfile from pathlib import Path import click from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType] -from tenacity import ( - before_sleep_log, - retry, - retry_if_exception_type, - stop_after_attempt, - wait_exponential, -) -from devx.api_clients import GiteaClient from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER -from devx.exceptions import APIError from devx.i18n import _ load_dotenv() -# DOCS_DIR is the repo's docs/ directory. When devx is installed as a -# package (e.g., in .venv/lib/python3.12/site-packages/devx/), the -# __file__-relative path would point inside the venv, not the repo. -# Use DEVX_DOCS_DIR env var if set, otherwise fall back to ./docs -# (relative to the current working directory, which is the repo root -# in CI and local development). DOCS_DIR = Path(os.environ.get("DEVX_DOCS_DIR", "docs")) MAPPING_FILE = DOCS_DIR / "mapping.json" +# Markdown link pattern: [text](url) +_LINK_RE = re.compile(r"\[([^\]]*)\]\(([^)]+)\)") + def load_mapping() -> dict[str, str]: - """Load the file-to-wiki-page mapping from mapping.json. - - Validates that the mapping is a dict of string-to-string pairs. - """ + """Load the file-to-wiki-page mapping from mapping.json.""" with open(MAPPING_FILE, encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): @@ -69,214 +62,172 @@ def load_mapping() -> dict[str, str]: return data -def read_doc_content(file_path: str) -> str: - """Read markdown content from a docs file.""" - full_path = DOCS_DIR / file_path - with open(full_path, encoding="utf-8") as f: - return f.read() +def transform_links(content: str) -> str: + """Transform markdown links from file-based to wiki-friendly format. - -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). + - ``[text](file.md)`` → ``[text](file)`` + - ``[text](docs/file.md)`` → ``[text](file)`` + - ``[text](../file.md)`` → ``[text](file)`` + - External links (http/https/mailto) preserved + - Anchor-only links (``#section``) preserved """ - return base64.b64encode(content.encode("utf-8")).decode("ascii") + + def replace_link(match: re.Match[str]) -> str: + text = match.group(1) + url = match.group(2).strip() + # Skip external links and mailto + if url.startswith(("http://", "https://", "mailto:")): + return match.group(0) + # Skip anchor-only links + if url.startswith("#"): + return match.group(0) + # Split path and anchor + if "#" in url: + path_part, anchor = url.split("#", 1) + anchor = f"#{anchor}" + else: + path_part, anchor = url, "" + # Remove .md extension and directory prefixes + if path_part.endswith(".md"): + path_part = path_part[:-3] + # Remove directory prefix (docs/, ../, etc.) + path_part = path_part.split("/")[-1] + return f"[{text}]({path_part}{anchor})" + + return _LINK_RE.sub(replace_link, content) -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 get_wiki_clone_url(owner: str, repo: str, token: str) -> str: + """Build the wiki Git clone URL with token auth.""" + # Gitea wiki repos are at {clone_url}.wiki.git + # Extract base URL from API URL + base = GITEA_API_URL.rsplit("/api/v1", 1)[0] + return f"{base}/{owner}/{repo}.wiki.git" -def list_wiki_pages(client: GiteaClient) -> dict[str, str]: - """List existing wiki pages, returning {title: sub_url}. +def clone_wiki(wiki_url: str, dest: Path) -> bool: + """Clone the wiki repo into dest. Returns True if clone succeeded. - Raises :class:`APIError` if the wiki API is unavailable — the caller - is responsible for retrying or handling the failure. + If the wiki repo doesn't exist yet (no pages created), returns False. """ - pages = client._request("GET", "/wiki/pages").json() - 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, - content: str, - existing_pages: dict[str, str], - dry_run: bool, -) -> str: - """Create or update a single wiki page. - - Returns "created", "updated", or "skipped" (if dry-run). - - If a create fails with HTTP 400 "already exists" (the page list was - stale), re-lists the wiki and falls back to an update. - """ - if dry_run: - 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_base64": content_b64, - "message": f"Sync from docs/ — update {page_title}", - }, - ) - return "updated" - - # Create new page via POST /wiki/new - try: - client._request( - "POST", - "/wiki/new", - json={ - "title": page_title, - "content_base64": content_b64, - "message": f"Sync from docs/ — create {page_title}", - }, - ) - return "created" - except APIError as e: - if e.status == 400 and "already exists" in e.message.lower(): - # The page list was stale (e.g. after a timeout-retry returned - # incomplete data). Re-list and fall back to update. - click.echo(_(" Page '{title}' already exists (stale list). Re-listing and updating...", title=page_title)) - fresh_pages = _list_wiki_pages_with_retry(client) - if page_title in fresh_pages: - sub_url = fresh_pages[page_title] - client._request( - "PATCH", - f"/wiki/page/{sub_url}", - json={ - "title": page_title, - "content_base64": content_b64, - "message": f"Sync from docs/ — update {page_title} (create→update fallback)", - }, - ) - return "updated" - raise - - -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() - - -def _list_wiki_pages_with_retry(client: GiteaClient) -> dict[str, str]: - """List wiki pages with tenacity retry on APIError. - - The Gitea wiki API can be slow (it renders pages on each request) - and may time out. Uses 5 attempts with exponential backoff to handle - transient slowness. - """ - _logger = logging.getLogger("sync_wiki") - - @retry( - stop=stop_after_attempt(5), - wait=wait_exponential(multiplier=2, min=2, max=16), - retry=retry_if_exception_type(APIError), - before_sleep=before_sleep_log(_logger, logging.WARNING), - reraise=True, + result = subprocess.run( # nosec + ["git", "clone", "--depth", "1", wiki_url, str(dest)], + capture_output=True, + text=True, + timeout=60, ) - def _do_list() -> dict[str, str]: - return list_wiki_pages(client) - - return _do_list() + return result.returncode == 0 -def verify_wiki_integrity( - client: GiteaClient, +def init_wiki(dest: Path) -> None: + """Initialize a fresh wiki repo (when clone fails).""" + dest.mkdir(parents=True, exist_ok=True) + subprocess.run(["git", "init"], cwd=dest, capture_output=True, check=True) # nosec + subprocess.run( # nosec + ["git", "config", "user.email", "ci@oblachno.fyi"], + cwd=dest, + capture_output=True, + check=True, + ) + subprocess.run( # nosec + ["git", "config", "user.name", "CI Wiki Sync"], + cwd=dest, + capture_output=True, + check=True, + ) + + +def sync_files( + docs_dir: Path, + wiki_dir: Path, mapping: dict[str, str], - synced: dict[str, str], -) -> list[str]: - """Comprehensive wiki verification. + dry_run: bool, +) -> tuple[int, int]: + """Copy docs files to wiki dir with link transformation. - Checks: - 1. Every mapped page exists in the wiki - 2. Every mapped page has non-empty content - 3. Every mapped page's content matches the docs - 4. No stale pages exist in the wiki (pages not in mapping) - 5. Page count matches - - Returns a list of failure messages (empty if all checks pass). - If the wiki API is temporarily unavailable (all retry attempts - fail), returns an empty list with a warning — the sync itself - already succeeded, so a transient API outage should not fail the job. + Returns (synced, pruned) counts. """ - failures: list[str] = [] + synced = 0 - try: - existing_pages = _list_wiki_pages_with_retry(client) - except APIError: - click.echo( - _( - "WARNING: Could not fetch wiki page list after retries. " - "The sync itself succeeded ({count} pages updated), but the " - "integrity check could not verify them due to a transient API issue.", - count=len(synced), - ) - ) - return [] + # Build set of expected wiki filenames + expected_files: set[str] = set() - expected_titles = set(mapping.values()) + for file_path, page_title in sorted(mapping.items()): + src = docs_dir / file_path + if not src.exists(): + click.echo(_(" WARN: Mapped file {file} not found, skipping", file=file_path)) + continue - # Check 1: Page count - if len(existing_pages) != len(expected_titles): - failures.append(f"Page count mismatch: wiki has {len(existing_pages)}, mapping has {len(expected_titles)}") + content = src.read_text(encoding="utf-8") + if not content.strip(): + click.echo(_(" WARN: Mapped file {file} is empty, skipping", file=file_path)) + continue - # Check 2: Missing pages (in mapping but not in wiki) - missing = expected_titles - set(existing_pages.keys()) - for title in sorted(missing): - failures.append(f"Missing page: {title}") + # Transform links + transformed = transform_links(content) - # Check 3: Stale pages (in wiki but not in mapping) - stale = set(existing_pages.keys()) - expected_titles - for title in sorted(stale): - failures.append(f"Stale page (not in mapping): {title}") + # Wiki filename: use the page title with spaces → underscores + # Gitea wiki uses the page title as filename (spaces become dashes) + wiki_filename = page_title.replace(" ", "-") + ".md" + expected_files.add(wiki_filename) - # Check 4: Content verification - for page_title, expected_content in sorted(synced.items()): - ok = verify_wiki_page(client, page_title, expected_content, existing_pages) - if not ok: - sub_url = existing_pages.get(page_title, "?") - actual = fetch_page_content(client, sub_url) - if not actual.strip(): - failures.append(f"Empty content: {page_title}") - else: - failures.append(f"Content mismatch: {page_title}") + if not dry_run: + dest = wiki_dir / wiki_filename + dest.write_text(transformed, encoding="utf-8") + synced += 1 + click.echo(_(" Synced: {title} → {file}", title=page_title, file=wiki_filename)) - return failures + # Prune stale pages (in wiki but not in mapping) + pruned = 0 + if not dry_run: + for existing in wiki_dir.glob("*.md"): + if existing.name not in expected_files: + existing.unlink() + pruned += 1 + click.echo(_(" Pruned: {file} (not in mapping)", file=existing.name)) + + return synced, pruned + + +def commit_and_push(wiki_dir: Path, wiki_url: str, dry_run: bool) -> bool: + """Commit changes and push to the wiki repo. Returns True if pushed.""" + if dry_run: + click.echo(_("[dry-run] Would commit and push wiki changes")) + return False + + # Stage all changes + subprocess.run(["git", "add", "-A"], cwd=wiki_dir, capture_output=True, check=True) # nosec + + # Check if there are changes to commit + result = subprocess.run( # nosec + ["git", "diff", "--cached", "--quiet"], + cwd=wiki_dir, + capture_output=True, + ) + if result.returncode == 0: + click.echo(_("No changes to sync — wiki is up to date.")) + return False + + # Commit + subprocess.run( # nosec + ["git", "commit", "-m", "Sync wiki from docs/ [skip ci]"], + cwd=wiki_dir, + capture_output=True, + check=True, + ) + + # Push + result = subprocess.run( # nosec + ["git", "push", wiki_url, "HEAD:master"], + cwd=wiki_dir, + capture_output=True, + text=True, + timeout=60, + ) + if result.returncode != 0: + click.echo(_("Push failed: {error}", error=result.stderr)) + return False + return True @click.command() @@ -286,15 +237,10 @@ def verify_wiki_integrity( "--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.", + help="After syncing, verify each page exists in the wiki. Exit 1 if any page is missing.", ) -@click.option( - "--strict", - is_flag=True, - default=False, - help="Full integrity check: verify page count, missing pages, stale pages, and content. Implies --verify.", -) -def main(dry_run: bool, repo: str | None, verify: bool, strict: bool) -> None: +def main(dry_run: bool, repo: str | None, verify: bool) -> None: + """Sync documentation to the Gitea wiki via Git.""" token = os.environ.get("CI_GITEA_TOKEN", "") if not token: raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) @@ -309,107 +255,63 @@ def main(dry_run: bool, repo: str | None, verify: bool, strict: bool) -> None: raise click.ClickException(_("ERROR: mapping.json not found at {path}", path=MAPPING_FILE)) mapping = load_mapping() - client = GiteaClient(GITEA_API_URL, token, owner, repo_name) + wiki_url = get_wiki_clone_url(owner, repo_name, token) - click.echo(_("Syncing {count} documentation pages to wiki...", count=len(mapping))) + click.echo(_("Syncing {count} documentation pages to wiki via Git...", count=len(mapping))) - try: - existing_pages = _list_wiki_pages_with_retry(client) - except APIError as e: - raise click.ClickException( + with tempfile.TemporaryDirectory() as tmpdir: + wiki_dir = Path(tmpdir) / "wiki" + + click.echo(_("Cloning wiki repo...")) + if clone_wiki(wiki_url, wiki_dir): + click.echo(_("Cloned existing wiki.")) + else: + click.echo(_("Wiki repo not found or empty — initializing fresh.")) + init_wiki(wiki_dir) + + click.echo(_("Syncing files...")) + synced, pruned = sync_files(DOCS_DIR, wiki_dir, mapping, dry_run) + + click.echo( _( - "Failed to list existing wiki pages after retries: {error}. " - "Aborting to avoid creating duplicate pages.", - error=e, + "\nDone! Synced: {synced}, Pruned: {pruned}", + synced=synced, + pruned=pruned, ) - ) from e - if existing_pages: - click.echo(_("Found {count} existing wiki pages.", count=len(existing_pages))) - - created = 0 - updated = 0 - skipped = 0 - synced: dict[str, str] = {} # title -> content, for verification - - for file_path, page_title in sorted(mapping.items()): - try: - content = read_doc_content(file_path) - except FileNotFoundError: - raise click.ClickException( - _("Mapped file {file} not found. Update mapping.json or create the file.", file=file_path) - ) from None - - if not content.strip(): - raise click.ClickException( - _("Mapped file {file} is empty. Update the content or remove from mapping.json.", file=file_path) - ) from None - - result = sync_page(client, page_title, content, existing_pages, dry_run) - if result == "created": - created += 1 - click.echo(_(" Created: {title}", title=page_title)) - elif result == "updated": - updated += 1 - click.echo(_(" Updated: {title}", title=page_title)) - else: - skipped += 1 - - synced[page_title] = content - - click.echo( - _( - "\nDone! Created: {created}, Updated: {updated}, Skipped: {skipped}", - created=created, - updated=updated, - skipped=skipped, ) - ) - # --strict implies --verify - do_verify = verify or strict + if dry_run: + click.echo(_("[dry-run] No changes pushed.")) + return - if do_verify and not dry_run: - if strict: - click.echo(_("\nRunning full wiki integrity check...")) - failures = verify_wiki_integrity(client, mapping, synced) - if failures: - click.echo(_("\nIntegrity check FAILED ({count} issues):", count=len(failures))) - for f in failures: - click.echo(f" - {f}") - raise click.ClickException(_("Wiki integrity check failed — {count} issue(s)", count=len(failures))) - click.echo(_("\nIntegrity check passed — all {count} pages verified.", count=len(synced))) - else: - click.echo(_("\nVerifying wiki pages have content...")) - # Re-fetch the page list to get updated sub_urls - try: - existing_pages = _list_wiki_pages_with_retry(client) - except APIError: - click.echo( - _( - "WARNING: Could not re-fetch wiki page list for verification. " - "Skipping content verification due to transient API issue." - ) - ) - return + click.echo(_("Committing and pushing...")) + pushed = commit_and_push(wiki_dir, wiki_url, dry_run) + if pushed: + click.echo(_("Wiki synced successfully.")) + elif not dry_run: + click.echo(_("No push needed (no changes or push failed).")) + + # Verification + if verify and not dry_run: + click.echo(_("\nVerifying wiki pages...")) + # Re-clone to verify + verify_dir = Path(tmpdir) / "verify" + if not clone_wiki(wiki_url, verify_dir): + click.echo(_("FAIL: Could not clone wiki for verification.")) + raise click.ClickException(_("Wiki verification failed — could not clone wiki")) 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))) + for _file_path, page_title in sorted(mapping.items()): + wiki_filename = page_title.replace(" ", "-") + ".md" + if (verify_dir / wiki_filename).exists(): + click.echo(_(" OK: {title}", title=page_title)) else: - click.echo(_(" FAIL: {title} — content mismatch or empty!", title=page_title)) + click.echo(_(" FAIL: {title} — page not found in wiki!", 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) + _("Wiki verification failed — {failures} page(s) missing", failures=failures) ) - click.echo(_("\nVerification passed — all wiki pages have correct content.")) + click.echo(_("\nVerification passed — all wiki pages exist.")) if __name__ == "__main__": # pragma: no cover diff --git a/tests/unit/test_sync_wiki.py b/tests/unit/test_sync_wiki.py index d21dfb5..b776cec 100644 --- a/tests/unit/test_sync_wiki.py +++ b/tests/unit/test_sync_wiki.py @@ -1,6 +1,7 @@ -"""Unit tests for scripts/ci/sync_wiki.py.""" +"""Unit tests for devx.ci.sync_wiki (git-based approach).""" + +from __future__ import annotations -import base64 import json from pathlib import Path from unittest.mock import MagicMock, patch @@ -10,593 +11,446 @@ import pytest from click.testing import CliRunner from devx.ci.sync_wiki import ( - decode_content, - encode_content, - fetch_page_content, - list_wiki_pages, + clone_wiki, + commit_and_push, + get_wiki_clone_url, + init_wiki, load_mapping, main, - read_doc_content, - sync_page, - verify_wiki_integrity, - verify_wiki_page, + sync_files, + transform_links, ) -from devx.exceptions import APIError -class TestEncodeContent: - def test_encodes_utf8_to_base64(self) -> None: - result = encode_content("# Hello World") - assert result == base64.b64encode(b"# Hello World").decode("ascii") +class TestTransformLinks: + def test_removes_md_extension(self) -> None: + result = transform_links("[link](page.md)") + assert result == "[link](page)" - def test_encodes_empty_string(self) -> None: - assert encode_content("") == "" + def test_removes_directory_prefix(self) -> None: + result = transform_links("[link](docs/page.md)") + assert result == "[link](page)" - def test_encodes_unicode(self) -> None: - result = encode_content("# Café — résumé") - decoded = base64.b64decode(result).decode("utf-8") - assert decoded == "# Café — résumé" + def test_removes_parent_dir_prefix(self) -> None: + result = transform_links("[link](../page.md)") + assert result == "[link](page)" + def test_preserves_external_links(self) -> None: + result = transform_links("[link](https://example.com)") + assert result == "[link](https://example.com)" -class TestDecodeContent: - def test_decodes_base64_to_utf8(self) -> None: - encoded = base64.b64encode(b"# Hello").decode("ascii") - assert decode_content(encoded) == "# Hello" + def test_preserves_http_links(self) -> None: + result = transform_links("[link](http://example.com)") + assert result == "[link](http://example.com)" - def test_empty_string_returns_empty(self) -> None: - assert decode_content("") == "" + def test_preserves_mailto(self) -> None: + result = transform_links("[email](mailto:test@example.com)") + assert result == "[email](mailto:test@example.com)" - def test_roundtrip(self) -> None: - original = "# Wiki Page\n\nContent with **markdown**." - encoded = encode_content(original) - assert decode_content(encoded) == original + def test_preserves_anchor_only(self) -> None: + result = transform_links("[section](#section)") + assert result == "[section](#section)" + + def test_preserves_anchor_with_path(self) -> None: + result = transform_links("[section](page.md#section)") + assert result == "[section](page#section)" + + def test_no_links_unchanged(self) -> None: + text = "# Title\n\nSome text without links.\n" + assert transform_links(text) == text + + def test_multiple_links(self) -> None: + result = transform_links("[a](one.md) and [b](two.md)") + assert result == "[a](one) and [b](two)" class TestLoadMapping: - def test_loads_mapping(self, tmp_path: Path) -> None: - mapping_file = tmp_path / "mapping.json" - mapping_file.write_text(json.dumps({"user/getting-started.md": "Getting-Started"})) - with patch("devx.ci.sync_wiki.MAPPING_FILE", mapping_file): - result = load_mapping() - assert result == {"user/getting-started.md": "Getting-Started"} + def test_loads_mapping(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mapping_file = tmp_path / "docs" / "mapping.json" + mapping_file.parent.mkdir() + mapping_file.write_text(json.dumps({"index.md": "Home", "guide.md": "Guide"})) + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + mapping = load_mapping() + assert mapping == {"index.md": "Home", "guide.md": "Guide"} - def test_missing_mapping_raises(self, tmp_path: Path) -> None: - with patch("devx.ci.sync_wiki.MAPPING_FILE", tmp_path / "nonexistent.json"): - with pytest.raises(FileNotFoundError): - load_mapping() - - def test_non_dict_mapping_raises(self, tmp_path: Path) -> None: - """Non-dict mapping.json should raise.""" - mapping_file = tmp_path / "mapping.json" + def test_non_dict_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mapping_file = tmp_path / "docs" / "mapping.json" + mapping_file.parent.mkdir() mapping_file.write_text('["not", "a", "dict"]') - with patch("devx.ci.sync_wiki.MAPPING_FILE", mapping_file): - with pytest.raises(click.ClickException, match="must be a dict"): - load_mapping() + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + with pytest.raises(click.ClickException, match="must be a dict"): + load_mapping() - def test_non_string_values_raise(self, tmp_path: Path) -> None: - """Non-string values in mapping.json should raise.""" - mapping_file = tmp_path / "mapping.json" - mapping_file.write_text('{"file.md": 123}') - with patch("devx.ci.sync_wiki.MAPPING_FILE", mapping_file): - with pytest.raises(click.ClickException, match="must be strings"): - load_mapping() + def test_non_string_values_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + mapping_file = tmp_path / "docs" / "mapping.json" + mapping_file.parent.mkdir() + mapping_file.write_text(json.dumps({"key": 123})) + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + with pytest.raises(click.ClickException, match="must be strings"): + load_mapping() -class TestReadDocContent: - def test_reads_file(self, tmp_path: Path) -> None: - docs_dir = tmp_path / "docs" - docs_dir.mkdir() - (docs_dir / "test.md").write_text("# Test\n\nContent") - with patch("devx.ci.sync_wiki.DOCS_DIR", docs_dir): - content = read_doc_content("test.md") - assert content == "# Test\n\nContent" - - def test_missing_file_raises(self, tmp_path: Path) -> None: - with patch("devx.ci.sync_wiki.DOCS_DIR", tmp_path): - with pytest.raises(FileNotFoundError): - read_doc_content("nonexistent.md") +class TestGetWikiCloneUrl: + def test_builds_url(self) -> None: + url = get_wiki_clone_url("owner", "repo", "token") + assert "owner/repo.wiki.git" in url -class TestListWikiPages: - def test_raises_on_api_error(self) -> None: - client = MagicMock() - client._request.side_effect = APIError(404, "not found") - with pytest.raises(APIError): - list_wiki_pages(client) +class TestCloneWiki: + @patch("devx.ci.sync_wiki.subprocess.run") + def test_clone_success(self, mock_run: MagicMock, tmp_path: Path) -> None: + mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") + result = clone_wiki("https://example.com/repo.wiki.git", tmp_path / "wiki") + assert result is True - def test_returns_page_dict(self) -> None: - client = MagicMock() - client._request.return_value.json.return_value = [ - {"title": "Home", "sub_url": "Home"}, - {"title": "Getting-Started", "sub_url": "Getting-Started.-"}, + @patch("devx.ci.sync_wiki.subprocess.run") + def test_clone_failure_returns_false(self, mock_run: MagicMock, tmp_path: Path) -> None: + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="not found") + result = clone_wiki("https://example.com/repo.wiki.git", tmp_path / "wiki") + assert result is False + + +class TestInitWiki: + @patch("devx.ci.sync_wiki.subprocess.run") + def test_init_calls_git(self, mock_run: MagicMock, tmp_path: Path) -> None: + wiki_dir = tmp_path / "wiki" + init_wiki(wiki_dir) + assert wiki_dir.exists() + calls = [c.args[0] for c in mock_run.call_args_list] + assert ["git", "init"] in calls + assert ["git", "config", "user.email", "ci@oblachno.fyi"] in calls + + +class TestSyncFiles: + def test_syncs_files(self, tmp_path: Path) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n[link](page.md)\n") + (docs / "page.md").write_text("# Page\n") + wiki = tmp_path / "wiki" + wiki.mkdir() + mapping = {"index.md": "Home", "page.md": "Page"} + synced, pruned = sync_files(docs, wiki, mapping, dry_run=False) + assert synced == 2 + assert pruned == 0 + assert (wiki / "Home.md").exists() + assert (wiki / "Page.md").exists() + # Check link transformation + content = (wiki / "Home.md").read_text() + assert "[link](page)" in content + + def test_prunes_stale(self, tmp_path: Path) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + wiki = tmp_path / "wiki" + wiki.mkdir() + (wiki / "OldPage.md").write_text("# Old\n") + (wiki / "Home.md").write_text("# Old Home\n") + mapping = {"index.md": "Home"} + synced, pruned = sync_files(docs, wiki, mapping, dry_run=False) + assert synced == 1 + assert pruned == 1 # OldPage.md pruned, Home.md overwritten + assert not (wiki / "OldPage.md").exists() + assert (wiki / "Home.md").exists() + + def test_dry_run_no_writes(self, tmp_path: Path) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + wiki = tmp_path / "wiki" + wiki.mkdir() + mapping = {"index.md": "Home"} + synced, pruned = sync_files(docs, wiki, mapping, dry_run=True) + assert synced == 1 + assert pruned == 0 + assert not (wiki / "Home.md").exists() + + def test_missing_file_warns(self, tmp_path: Path) -> None: + docs = tmp_path / "docs" + docs.mkdir() + wiki = tmp_path / "wiki" + wiki.mkdir() + mapping = {"missing.md": "Missing"} + synced, pruned = sync_files(docs, wiki, mapping, dry_run=False) + assert synced == 0 + + def test_empty_file_warns(self, tmp_path: Path) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "empty.md").write_text("") + wiki = tmp_path / "wiki" + wiki.mkdir() + mapping = {"empty.md": "Empty"} + synced, pruned = sync_files(docs, wiki, mapping, dry_run=False) + assert synced == 0 + + +class TestCommitAndPush: + @patch("devx.ci.sync_wiki.subprocess.run") + def test_dry_run_returns_false(self, mock_run: MagicMock, tmp_path: Path) -> None: + result = commit_and_push(tmp_path, "url", dry_run=True) + assert result is False + mock_run.assert_not_called() + + @patch("devx.ci.sync_wiki.subprocess.run") + def test_no_changes_returns_false(self, mock_run: MagicMock, tmp_path: Path) -> None: + # git add succeeds, git diff --cached --quiet returns 0 (no changes) + mock_run.side_effect = [ + MagicMock(returncode=0), # git add + MagicMock(returncode=0), # git diff --cached --quiet (no changes) ] - result = list_wiki_pages(client) - assert result == {"Home": "Home", "Getting-Started": "Getting-Started.-"} + result = commit_and_push(tmp_path, "url", dry_run=False) + assert result is False + @patch("devx.ci.sync_wiki.subprocess.run") + def test_pushes_changes(self, mock_run: MagicMock, tmp_path: Path) -> None: + mock_run.side_effect = [ + MagicMock(returncode=0), # git add + MagicMock(returncode=1), # git diff --cached --quiet (has changes) + MagicMock(returncode=0), # git commit + MagicMock(returncode=0, stdout="", stderr=""), # git push + ] + result = commit_and_push(tmp_path, "url", dry_run=False) + assert result is True -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 devx.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() - result = sync_page(client, "Test-Page", "# Content", {}, dry_run=True) - assert result == "skipped" - client._request.assert_not_called() - - def test_creates_new_page_with_base64(self) -> None: - client = MagicMock() - result = sync_page(client, "New-Page", "# Content", {}, dry_run=False) - assert result == "created" - client._request.assert_called_once() - 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_with_base64(self) -> None: - client = MagicMock() - existing = {"Existing-Page": "Existing-Page.-"} - result = sync_page(client, "Existing-Page", "# Updated", existing, dry_run=False) - assert result == "updated" - client._request.assert_called_once() - 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" - - def test_create_falls_back_to_update_on_already_exists(self) -> None: - """When create fails with 400 'already exists', re-list and update.""" - client = MagicMock() - # First call: POST /wiki/new → 400 already exists - # Second call: PATCH /wiki/page/{sub_url} → success - create_error = APIError(400, "wiki page already exists [title: Test-Page]") - client._request.side_effect = [create_error, MagicMock()] - with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", return_value={"Test-Page": "Test-Page.-"}): - result = sync_page(client, "Test-Page", "# Content", {}, dry_run=False) - assert result == "updated" - # Verify PATCH was called (second call) - patch_call = client._request.call_args_list[1] - assert patch_call.args[0] == "PATCH" - assert "/wiki/page/Test-Page.-" in patch_call.args[1] - - def test_create_raises_non_400_error(self) -> None: - """Non-400 errors from create should propagate, not trigger fallback.""" - client = MagicMock() - client._request.side_effect = APIError(500, "server error") - with pytest.raises(APIError): - sync_page(client, "Test-Page", "# Content", {}, dry_run=False) - - def test_create_raises_400_not_already_exists(self) -> None: - """400 errors that don't mention 'already exists' should propagate.""" - client = MagicMock() - client._request.side_effect = APIError(400, "invalid title") - with pytest.raises(APIError): - sync_page(client, "Test-Page", "# Content", {}, dry_run=False) - - -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 TestVerifyWikiIntegrity: - def _make_client(self, pages: dict[str, str], contents: dict[str, str]) -> MagicMock: - """Create a mock client that returns the given pages and contents.""" - client = MagicMock() - # list_wiki_pages calls GET /wiki/pages - page_list = [{"title": t, "sub_url": s} for t, s in pages.items()] - - # fetch_page_content calls GET /wiki/page/{sub_url} - def mock_request(method, path, **kwargs): - resp = MagicMock() - if path == "/wiki/pages": - resp.json.return_value = page_list - elif path.startswith("/wiki/page/"): - sub_url = path.replace("/wiki/page/", "") - content = contents.get(sub_url, "") - encoded = base64.b64encode(content.encode()).decode("ascii") if content else "" - resp.json.return_value = {"content_base64": encoded} - return resp - - client._request.side_effect = mock_request - return client - - def test_all_good_no_failures(self) -> None: - pages = {"Home": "Home", "FAQ": "FAQ"} - contents = {"Home": "# Home", "FAQ": "# FAQ"} - client = self._make_client(pages, contents) - mapping = {"index.md": "Home", "faq.md": "FAQ"} - synced = {"Home": "# Home", "FAQ": "# FAQ"} - failures = verify_wiki_integrity(client, mapping, synced) - assert failures == [] - - def test_missing_page_detected(self) -> None: - pages = {"Home": "Home"} # FAQ missing from wiki - contents = {"Home": "# Home"} - client = self._make_client(pages, contents) - mapping = {"index.md": "Home", "faq.md": "FAQ"} - synced = {"Home": "# Home"} - failures = verify_wiki_integrity(client, mapping, synced) - assert any("Missing page: FAQ" in f for f in failures) - - def test_stale_page_detected(self) -> None: - pages = {"Home": "Home", "Old-Page": "Old-Page"} # Old-Page not in mapping - contents = {"Home": "# Home", "Old-Page": "# Old"} - client = self._make_client(pages, contents) - mapping = {"index.md": "Home"} - synced = {"Home": "# Home"} - failures = verify_wiki_integrity(client, mapping, synced) - assert any("Stale page" in f and "Old-Page" in f for f in failures) - - def test_page_count_mismatch_detected(self) -> None: - pages = {"Home": "Home", "Extra": "Extra"} - contents = {"Home": "# Home", "Extra": "# Extra"} - client = self._make_client(pages, contents) - mapping = {"index.md": "Home"} - synced = {"Home": "# Home"} - failures = verify_wiki_integrity(client, mapping, synced) - assert any("Page count mismatch" in f for f in failures) - - def test_empty_content_detected(self) -> None: - pages = {"Home": "Home"} - contents = {"Home": ""} # Empty content - client = self._make_client(pages, contents) - mapping = {"index.md": "Home"} - synced = {"Home": "# Expected Content"} - failures = verify_wiki_integrity(client, mapping, synced) - assert any("Empty content: Home" in f for f in failures) - - def test_content_mismatch_detected(self) -> None: - pages = {"Home": "Home"} - contents = {"Home": "# Wrong Content"} - client = self._make_client(pages, contents) - mapping = {"index.md": "Home"} - synced = {"Home": "# Correct Content"} - failures = verify_wiki_integrity(client, mapping, synced) - assert any("Content mismatch: Home" in f for f in failures) - - def test_multiple_failures_all_reported(self) -> None: - pages = {"Home": "Home", "Stale": "Stale"} - contents = {"Home": "", "Stale": "# Stale"} - client = self._make_client(pages, contents) - mapping = {"index.md": "Home", "faq.md": "FAQ"} # FAQ missing - synced = {"Home": "# Home Content"} - failures = verify_wiki_integrity(client, mapping, synced) - assert len(failures) >= 3 # count mismatch, missing FAQ, stale Stale, empty Home - - def test_transient_api_failure_returns_empty(self) -> None: - """When the wiki API is unavailable after retries, integrity check - should return no failures (sync already succeeded).""" - client = MagicMock() - - # _list_wiki_pages_with_retry raises APIError (retries exhausted) - with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", side_effect=APIError(0, "timeout")): - mapping = {"index.md": "Home", "faq.md": "FAQ"} - synced = {"Home": "# Home", "FAQ": "# FAQ"} - failures = verify_wiki_integrity(client, mapping, synced) - assert failures == [] - - def test_transient_api_failure_recovers_on_retry(self) -> None: - """When the wiki API recovers after a retry, integrity check proceeds normally.""" - client = MagicMock() - pages = {"Home": "Home", "FAQ": "FAQ"} - contents = {"Home": "# Home", "FAQ": "# FAQ"} - - def mock_request(method, path, **kwargs): - resp = MagicMock() - if path == "/wiki/pages": - page_list = [{"title": t, "sub_url": s} for t, s in pages.items()] - resp.json.return_value = page_list - elif path.startswith("/wiki/page/"): - sub_url = path.replace("/wiki/page/", "") - content = contents.get(sub_url, "") - encoded = base64.b64encode(content.encode()).decode("ascii") if content else "" - resp.json.return_value = {"content_base64": encoded} - return resp - - client._request.side_effect = mock_request - - mapping = {"index.md": "Home", "faq.md": "FAQ"} - synced = {"Home": "# Home", "FAQ": "# FAQ"} - failures = verify_wiki_integrity(client, mapping, synced) - assert failures == [] + @patch("devx.ci.sync_wiki.subprocess.run") + def test_push_failure_returns_false(self, mock_run: MagicMock, tmp_path: Path) -> None: + mock_run.side_effect = [ + MagicMock(returncode=0), # git add + MagicMock(returncode=1), # git diff --cached --quiet (has changes) + MagicMock(returncode=0), # git commit + MagicMock(returncode=1, stdout="", stderr="push failed"), # git push + ] + result = commit_and_push(tmp_path, "url", dry_run=False) + assert result is False class TestMain: - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}) - @patch("devx.ci.sync_wiki.MAPPING_FILE") - @patch("devx.ci.sync_wiki.DOCS_DIR") - @patch("devx.ci.sync_wiki.GiteaClient") - def test_dry_run(self, mock_client_cls: MagicMock, mock_docs_dir: Path, mock_mapping_file: Path) -> None: - mock_mapping_file.exists.return_value = True - mock_mapping_file.__str__ = lambda _: "/docs/mapping.json" - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.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 "dry-run" in result.output - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": ""}, clear=True) - def test_missing_token_exits(self) -> None: + def test_no_token_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CI_GITEA_TOKEN", raising=False) runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo"]) - assert result.exit_code == 1 + result = runner.invoke(main, []) + assert result.exit_code != 0 assert "CI_GITEA_TOKEN" in result.output - @patch.dict( - "os.environ", {"CI_GITEA_TOKEN": "tok", "DEVX_REPO_OWNER": "me", "DEVX_REPO_NAME": "myrepo"}, clear=True - ) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_auto_detect_repo(self, mock_client_cls: MagicMock) -> None: - """Test that repo is auto-detected from env vars when --repo is not passed.""" - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={}): - runner = CliRunner() - result = runner.invoke(main, ["--dry-run"]) - assert result.exit_code == 0 - mock_client_cls.assert_called_once() - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_missing_mapping_file(self, mock_client_cls: MagicMock) -> None: - """Test that missing mapping.json exits with error.""" - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = False - runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo"]) - assert result.exit_code == 1 + def test_no_mapping_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", tmp_path / "nonexistent.json") + runner = CliRunner() + result = runner.invoke(main, ["--repo", "owner/repo"]) + assert result.exit_code != 0 assert "mapping.json" in result.output - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_existing_pages_message(self, mock_client_cls: MagicMock) -> None: - """Test that existing wiki pages are reported.""" - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}): - runner = CliRunner() - result = runner.invoke(main, ["--dry-run", "--repo", "owner/repo"]) + @patch("devx.ci.sync_wiki.clone_wiki", return_value=True) + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_dry_run( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + runner = CliRunner() + result = runner.invoke(main, ["--dry-run", "--repo", "owner/repo"]) assert result.exit_code == 0 - assert "existing wiki pages" in result.output + assert "dry-run" in result.output + mock_push.assert_not_called() - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_file_not_found_fails(self, mock_client_cls: MagicMock) -> None: - """Test that missing doc files cause an error, not a warning.""" - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"missing.md": "Missing"}): - with patch("devx.ci.sync_wiki.read_doc_content", side_effect=FileNotFoundError): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={}): - runner = CliRunner() - result = runner.invoke(main, ["--dry-run", "--repo", "owner/repo"]) + @patch("devx.ci.sync_wiki.clone_wiki", return_value=True) + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(2, 0)) + def test_full_sync( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n[link](page.md)\n") + (docs / "page.md").write_text("# Page\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home", "page.md": "Page"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + runner = CliRunner() + result = runner.invoke(main, ["--repo", "owner/repo"]) + assert result.exit_code == 0 + assert "Synced" in result.output + mock_push.assert_called_once() + + @patch("devx.ci.sync_wiki.clone_wiki", return_value=False) + @patch("devx.ci.sync_wiki.init_wiki") + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_init_fresh_wiki( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_init: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + runner = CliRunner() + result = runner.invoke(main, ["--repo", "owner/repo"]) + assert result.exit_code == 0 + mock_init.assert_called_once() + + @patch("devx.ci.sync_wiki.clone_wiki") + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_verify( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + + # Mock clone_wiki to create the wiki dir with the expected file + def fake_clone(url: str, dest: Path) -> bool: + dest.mkdir(parents=True, exist_ok=True) + (dest / "Home.md").write_text("# Home\n") + return True + + mock_clone.side_effect = fake_clone + + runner = CliRunner() + result = runner.invoke(main, ["--verify", "--repo", "owner/repo"]) + assert result.exit_code == 0 + assert "Verification" in result.output + + @patch("devx.ci.sync_wiki.clone_wiki", return_value=True) + @patch("devx.ci.sync_wiki.commit_and_push", return_value=False) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_push_failed_message( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + runner = CliRunner() + result = runner.invoke(main, ["--repo", "owner/repo"]) + assert result.exit_code == 0 + assert "No push needed" in result.output + + @patch("devx.ci.sync_wiki.clone_wiki", side_effect=[True, False]) + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_verify_clone_fails( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + runner = CliRunner() + result = runner.invoke(main, ["--verify", "--repo", "owner/repo"]) assert result.exit_code != 0 - assert "not found" in result.output + assert "could not clone" in result.output - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_empty_doc_file_fails(self, mock_client_cls: MagicMock) -> None: - """Test that empty doc files cause an error, not a warning.""" - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"empty.md": "Empty-Page"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value=" \n "): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={}): - runner = CliRunner() - result = runner.invoke(main, ["--dry-run", "--repo", "owner/repo"]) + @patch("devx.ci.sync_wiki.clone_wiki") + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_verify_missing_page( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + + # Mock clone_wiki to create the wiki dir WITHOUT the expected file + def fake_clone(url: str, dest: Path) -> bool: + dest.mkdir(parents=True, exist_ok=True) + return True + + mock_clone.side_effect = fake_clone + + runner = CliRunner() + result = runner.invoke(main, ["--verify", "--repo", "owner/repo"]) assert result.exit_code != 0 - assert "empty" in result.output.lower() + assert "page(s) missing" in result.output - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_create_and_update(self, mock_client_cls: MagicMock) -> None: - """Test that pages are created and updated correctly (non-dry-run).""" - mock_client = MagicMock() - mock_client_cls.return_value = mock_client - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - mapping = {"new.md": "New-Page", "existing.md": "Existing-Page"} - with patch("devx.ci.sync_wiki.load_mapping", return_value=mapping): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Content"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Existing-Page": "Existing-Page"}): - runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo"]) + @patch("devx.ci.sync_wiki.clone_wiki", return_value=True) + @patch("devx.ci.sync_wiki.commit_and_push", return_value=True) + @patch("devx.ci.sync_wiki.sync_files", return_value=(1, 0)) + def test_auto_detect_repo( + self, + mock_sync: MagicMock, + mock_push: MagicMock, + mock_clone: MagicMock, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + docs = tmp_path / "docs" + docs.mkdir() + (docs / "index.md").write_text("# Home\n") + mapping_file = docs / "mapping.json" + mapping_file.write_text(json.dumps({"index.md": "Home"})) + monkeypatch.setenv("CI_GITEA_TOKEN", "fake") + monkeypatch.setattr("devx.ci.sync_wiki.MAPPING_FILE", mapping_file) + monkeypatch.setattr("devx.ci.sync_wiki.DOCS_DIR", docs) + runner = CliRunner() + result = runner.invoke(main, []) assert result.exit_code == 0 - assert "Created: 1" in result.output - assert "Updated: 1" in result.output - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.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("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home Content"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}): - with patch("devx.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", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.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("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home Content"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}): - with patch("devx.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", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.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("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.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 - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_strict_passes(self, mock_client_cls: MagicMock) -> None: - """Test that --strict passes when integrity check succeeds.""" - mock_client = MagicMock() - mock_client_cls.return_value = mock_client - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}): - with patch("devx.ci.sync_wiki.verify_wiki_integrity", return_value=[]): - runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo", "--strict"]) - assert result.exit_code == 0 - assert "Integrity check passed" in result.output - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_strict_fails_on_integrity_issues(self, mock_client_cls: MagicMock) -> None: - """Test that --strict fails when integrity check finds issues.""" - mock_client = MagicMock() - mock_client_cls.return_value = mock_client - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}): - with patch( - "devx.ci.sync_wiki.verify_wiki_integrity", - return_value=["Missing page: FAQ", "Stale page: Old-Page"], - ): - runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo", "--strict"]) - assert result.exit_code == 1 - assert "Integrity check FAILED" in result.output - assert "Missing page: FAQ" in result.output - assert "Stale page: Old-Page" in result.output - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_strict_skipped_in_dry_run(self, mock_client_cls: MagicMock) -> None: - """Test that --strict verification is skipped during dry-run.""" - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={}): - runner = CliRunner() - result = runner.invoke(main, ["--dry-run", "--strict", "--repo", "owner/repo"]) - assert result.exit_code == 0 - assert "Integrity check" not in result.output - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_initial_list_api_error_aborts(self, mock_client_cls: MagicMock) -> None: - """When the initial page list fails after retries, sync aborts to avoid duplicate pages.""" - mock_client = MagicMock() - mock_client_cls.return_value = mock_client - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", side_effect=APIError(0, "timeout")): - with patch("devx.ci.sync_wiki.sync_page", return_value="created"): - runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo"]) - assert result.exit_code != 0 - assert "Failed to list existing wiki pages" in result.output - assert "Aborting" in result.output - - @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True) - @patch("devx.ci.sync_wiki.GiteaClient") - def test_verify_skips_when_refetch_fails(self, mock_client_cls: MagicMock) -> None: - """When --verify re-fetch fails after retries, verification is skipped gracefully.""" - mock_client = MagicMock() - mock_client_cls.return_value = mock_client - # Initial list succeeds, but verify re-fetch fails - list_side_effect = [{"Home": "Home"}, APIError(0, "timeout")] - with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping: - mock_mapping.exists.return_value = True - with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}): - with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"): - with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", side_effect=list_side_effect): - with patch("devx.ci.sync_wiki.sync_page", return_value="updated"): - runner = CliRunner() - result = runner.invoke(main, ["--repo", "owner/repo", "--verify"]) - assert result.exit_code == 0 - assert "Skipping content verification" in result.output