This commit is contained in:
Jason Wen
2026-03-12 21:46:34 -04:00
parent 31b3776150
commit 922ad64d13
11 changed files with 1909 additions and 305 deletions
@@ -1,105 +0,0 @@
name: 'Post to Discourse'
description: 'Posts a message to a Discourse topic (existing or new)'
inputs:
discourse-url:
description: 'Discourse instance URL (e.g., https://discourse.example.com)'
required: true
api-key:
description: 'Discourse API key'
required: true
api-username:
description: 'Discourse API username'
required: true
topic-id:
description: 'Discourse topic ID to post to (use this OR category-id + title)'
required: false
category-id:
description: 'Category ID for new topic (required if topic-id not provided)'
required: false
title:
description: 'Title for new topic (required if topic-id not provided)'
required: false
message:
description: 'Message content (markdown supported)'
required: true
outputs:
post-number:
description: 'The post number in the topic'
value: ${{ steps.post.outputs.post_number }}
post-url:
description: 'Direct URL to the post'
value: ${{ steps.post.outputs.post_url }}
topic-id:
description: 'The topic ID (useful when creating a new topic)'
value: ${{ steps.post.outputs.topic_id }}
runs:
using: "composite"
steps:
- name: Post to Discourse
id: post
shell: bash
run: |
# Validate inputs
if [ -z "${{ inputs.topic-id }}" ] && ([ -z "${{ inputs.category-id }}" ] || [ -z "${{ inputs.title }}" ]); then
echo "❌ Error: Must provide either topic-id OR both category-id and title"
exit 1
fi
if [ -n "${{ inputs.topic-id }}" ] && ([ -n "${{ inputs.category-id }}" ] || [ -n "${{ inputs.title }}" ]); then
echo "⚠️ Warning: Both topic-id and category-id/title provided. Will post to existing topic."
fi
# Determine if creating new topic or posting to existing
if [ -n "${{ inputs.topic-id }}" ]; then
echo "📝 Posting to existing topic ID: ${{ inputs.topic-id }}"
# Create JSON payload for posting to existing topic
PAYLOAD=$(jq -n \
--arg content '${{ inputs.message }}' \
--arg topic_id "${{ inputs.topic-id }}" \
'{topic_id: $topic_id, raw: $content}')
else
echo "✨ Creating new topic: ${{ inputs.title }}"
# Create JSON payload for new topic
PAYLOAD=$(jq -n \
--arg content '${{ inputs.message }}' \
--arg title "${{ inputs.title }}" \
--arg category "${{ inputs.category-id }}" \
'{title: $title, category: ($category | tonumber), raw: $content}')
fi
# Post to Discourse
RESPONSE=$(curl -s -w "\n%{http_code}" \
-X POST "${{ inputs.discourse-url }}/posts.json" \
-H "Content-Type: application/json" \
-H "Api-Key: ${{ inputs.api-key }}" \
-H "Api-Username: ${{ inputs.api-username }}" \
-d "$PAYLOAD")
HTTP_CODE=$(echo "$RESPONSE" | tail -n1)
BODY=$(echo "$RESPONSE" | sed '$d')
if [ "$HTTP_CODE" -ge 200 ] && [ "$HTTP_CODE" -lt 300 ]; then
echo "✅ Successfully posted to Discourse!"
POST_NUMBER=$(echo "$BODY" | jq -r '.post_number // "unknown"')
TOPIC_ID=$(echo "$BODY" | jq -r '.topic_id // "${{ inputs.topic-id }}"')
POST_URL="${{ inputs.discourse-url }}/t/${TOPIC_ID}/${POST_NUMBER}"
echo "post_number=${POST_NUMBER}" >> $GITHUB_OUTPUT
echo "post_url=${POST_URL}" >> $GITHUB_OUTPUT
echo "topic_id=${TOPIC_ID}" >> $GITHUB_OUTPUT
echo "Topic ID: ${TOPIC_ID}"
echo "Post number: ${POST_NUMBER}"
echo "URL: ${POST_URL}"
else
echo "❌ Failed to post to Discourse"
echo "HTTP Code: ${HTTP_CODE}"
echo "Response: ${BODY}"
exit 1
fi
+89
View File
@@ -0,0 +1,89 @@
name: Sync Docs to Discourse
on:
push:
paths:
- "docs_sp/**"
- "zensical.toml"
pull_request:
paths:
- "docs_sp/**"
- "zensical.toml"
workflow_dispatch:
jobs:
smoke-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Install uv
run: pip install uv
- name: Smoke test - post one doc to Discourse
env:
DISCOURSE_URL: ${{ secrets.DISCOURSE_URL }}
DISCOURSE_API_KEY: ${{ secrets.DISCOURSE_API_KEY }}
DISCOURSE_API_USER: ${{ secrets.DISCOURSE_API_USER }}
run: |
uv run --python 3.12 -c "
import sys
sys.path.insert(0, 'docs_sp/tools')
from pathlib import Path
from converter import convert
from discourse_client import DiscourseClient, DiscourseConfig
DOCS_BASE_URL = 'https://docs.sunnypilot.ai'
DOC_PATH = 'getting-started/what-is-sunnypilot.md'
raw = (Path('docs_sp') / DOC_PATH).read_text()
body = convert(raw, file_path=DOC_PATH, docs_base_url=DOCS_BASE_URL)
# Append sync metadata
docs_url = f'{DOCS_BASE_URL}/{DOC_PATH.replace(\".md\", \"/\")}'
body = body.rstrip('\n') + f'\n\n---\n:link: [View on docs site]({docs_url})\n\n<!-- docs-sync-id: {DOC_PATH} -->\n'
# Extract title from front matter or first heading
title = None
for line in raw.splitlines():
s = line.strip()
if s.startswith('title:'):
title = s[len('title:'):].strip().strip('\"').strip(\"'\")
break
if s.startswith('# '):
title = s[2:].strip()
break
title = f'{title or \"What is sunnypilot?\"} - sunnypilot Docs'
print(f'Title: {title}')
print(f'Body: {len(body)} chars')
config = DiscourseConfig.from_env()
client = DiscourseClient(config)
cat_id = client.get_category_id()
if cat_id is None:
print(f'ERROR: Category not found on {config.base_url}')
sys.exit(1)
existing = client.find_topic_by_sync_id(DOC_PATH)
if existing is not None:
topic_id = existing['id']
post_id = client.first_post_id(topic_id)
if post_id is None:
print(f'ERROR: No first post for topic {topic_id}')
sys.exit(1)
result = client.update_post(post_id, body, edit_reason='CI smoke test')
if result is None:
print('ERROR: Failed to update post')
sys.exit(1)
print(f'Updated: {config.base_url}/t/{topic_id}')
else:
result = client.create_topic(title=title, raw=body, category_id=cat_id, tags=['docs-auto-sync'])
if result is None:
print('ERROR: Failed to create topic')
sys.exit(1)
print(f'Created: {config.base_url}/t/{result.get(\"topic_id\", \"?\")}')
print('Smoke test passed!')
"
-78
View File
@@ -1,78 +0,0 @@
name: Debug Discourse Posting
on:
push:
jobs:
test-discourse-post:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Post test message to Discourse
uses: ./.github/workflows/post-to-discourse
with:
discourse-url: ${{ vars.DISCOURSE_URL }}
api-key: ${{ secrets.DISCOURSE_API_KEY }}
api-username: ${{ secrets.DISCOURSE_API_USERNAME }}
topic-id: ${{ vars.DISCOURSE_UPDATES_TOPIC_ID }}
message: |
## 🧪 Test Post from GitHub Actions
**This is a test post to verify Discourse integration**
- **Workflow**: ${{ github.workflow }}
- **Run Number**: #${{ github.run_number }}
- **Branch**: `${{ github.ref_name }}`
- **Commit**: ${{ github.sha }}
- **Actor**: @${{ github.actor }}
- **Timestamp**: ${{ github.event.head_commit.timestamp }}
---
### Fake Build Info (for testing)
- **Version**: 0.9.8-test
- **Build**: #42
- **Branch**: release-test
[View workflow run](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }})
*This is an automated test message. Drive safe! 🚗💨*
- name: Create topic on Discourse
uses: ./.github/workflows/post-to-discourse
with:
discourse-url: ${{ vars.DISCOURSE_URL }}
api-key: ${{ secrets.DISCOURSE_API_KEY }}
api-username: ${{ secrets.DISCOURSE_API_USERNAME }}
#topic-id: ${{ vars.DISCOURSE_UPDATES_TOPIC_ID }}
category-id: 4
title: "This is a test of a new topic instead of a reply"
message: |
## 🧪 Test Post from GitHub Actions
**This is a test post to verify Discourse integration**
- **Workflow**: ${{ github.workflow }}
- **Run Number**: #${{ github.run_number }}
- **Branch**: `${{ github.ref_name }}`
- **Commit**: ${{ github.sha }}
- **Actor**: @${{ github.actor }}
- **Timestamp**: ${{ github.event.head_commit.timestamp }}
---
### Fake Build Info (for testing)
- **Version**: 0.9.8-test
- **Build**: #42
- **Branch**: release-test
[View workflow run](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }})
*This is an automated test message. Drive safe! 🚗💨*
- name: Display results
if: always()
run: |
echo "::notice::Discourse post test completed"
echo "Check your Discourse topic to verify the post appeared correctly"
+58
View File
@@ -0,0 +1,58 @@
"""SHA-256 content cache for skipping unchanged docs on re-sync.
Stores one .sha256 file per doc in .discourse_sync_cache/. On re-run,
compares the current file hash against the cached hash to determine
whether the doc needs syncing.
"""
from __future__ import annotations
import hashlib
from pathlib import Path
DEFAULT_CACHE_DIR = Path(__file__).resolve().parent.parent.parent / ".discourse_sync_cache"
class ContentCache:
"""File-based SHA-256 content cache."""
def __init__(self, cache_dir: str | Path = DEFAULT_CACHE_DIR) -> None:
self._cache_dir = Path(cache_dir)
@property
def cache_dir(self) -> Path:
return self._cache_dir
@staticmethod
def compute_hash(content: str) -> str:
"""Compute SHA-256 hex digest of content."""
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def _cache_path(self, doc_path: str) -> Path:
"""Map a doc path (e.g. 'features/cruise/icbm.md') to its cache file."""
slug = doc_path.replace("/", "_").replace(".md", "")
return self._cache_dir / f"{slug}.sha256"
def is_changed(self, doc_path: str, content: str) -> bool:
"""Return True if content differs from the cached hash (or no cache exists)."""
content_hash = self.compute_hash(content)
cached = self._cache_path(doc_path)
if not cached.exists():
return True
return cached.read_text().strip() != content_hash
def save(self, doc_path: str, content: str) -> None:
"""Save the SHA-256 hash of content to the cache file."""
content_hash = self.compute_hash(content)
self._cache_dir.mkdir(parents=True, exist_ok=True)
self._cache_path(doc_path).write_text(content_hash + "\n")
def clear(self) -> int:
"""Remove all cached hashes. Returns the number of files removed."""
if not self._cache_dir.exists():
return 0
count = 0
for f in self._cache_dir.glob("*.sha256"):
f.unlink()
count += 1
return count
+281
View File
@@ -0,0 +1,281 @@
"""MkDocs Material -> Discourse-compatible Markdown converter.
Converts MkDocs Material syntax to Discourse-friendly markdown:
1. Strip YAML front matter
2. Convert admonitions (!!!/???/???+) to Discourse callouts (> [!TYPE])
3. Convert Material tabs (=== "Tab Name") to bold headings + ---
4. Strip grid card HTML (<div class="grid cards" markdown>)
5. Convert Material emoji shortcodes (:material-*:) to Unicode or strip
6. Resolve internal .md links to docs site URLs
7. Clean up excessive blank lines
"""
from __future__ import annotations
import os
import re
from pathlib import Path
DOCS_BASE_URL = os.environ.get("DOCS_BASE_URL", "https://docs.sunnypilot.ai")
ADMONITION_MAP: dict[str, str] = {
"note": "NOTE",
"abstract": "ABSTRACT",
"info": "INFO",
"tip": "TIP",
"success": "SUCCESS",
"question": "QUESTION",
"warning": "WARNING",
"failure": "FAILURE",
"danger": "DANGER",
"bug": "BUG",
"example": "EXAMPLE",
"quote": "QUOTE",
}
EMOJI_MAP: dict[str, str] = {
":material-rocket-launch:": "",
":material-cog:": "",
":material-car:": "",
":material-shield:": "",
":material-download:": "",
":material-check:": "Y",
":material-close:": "N",
":material-alert:": "!",
":material-information:": "i",
":material-help-circle:": "?",
":material-star:": "*",
":material-link:": "",
":material-eye:": "",
":material-map:": "",
":material-wifi:": "",
":material-cellphone:": "",
":material-steering:": "",
":material-speedometer:": "",
}
_ADMONITION_RE = re.compile(r"^(\s*)(!{3}|\?{3}\+?)\s+(\w+)(?:\s+\"([^\"]*)\")?\s*$")
_TAB_RE = re.compile(r'^(\s*)===\s+"([^"]+)"\s*$')
_FRONT_MATTER_RE = re.compile(r"\A---\n.*?\n---\n*", re.DOTALL)
_GRID_CARD_OPEN_RE = re.compile(r'<div\s+class="grid\s+cards"\s*(?:markdown)?\s*>')
_GRID_CARD_CLOSE_RE = re.compile(r"</div>")
_INTERNAL_LINK_RE = re.compile(r"\]\(([^)]+\.md(?:#[^)]*)?)\)")
_EMOJI_SHORTCODE_RE = re.compile(r":material-[\w-]+:")
_EXCESSIVE_BLANKS_RE = re.compile(r"\n{4,}")
def convert(content: str, *, file_path: str, docs_base_url: str | None = None) -> str:
"""Convert MkDocs Material markdown to Discourse-compatible markdown.
Args:
content: Raw markdown content.
file_path: Path to the source file (relative to docs_sp/ or absolute).
Used for resolving relative internal links.
docs_base_url: Base URL for the docs site. Defaults to DOCS_BASE_URL env var.
Returns:
Converted markdown string.
"""
base_url = docs_base_url or DOCS_BASE_URL
result = content
result = strip_front_matter(result)
result = convert_admonitions(result)
result = convert_tabs(result)
result = convert_grid_cards(result)
result = convert_emoji_shortcodes(result)
result = resolve_internal_links(result, file_path=file_path, docs_base_url=base_url)
result = clean_blank_lines(result)
return result.strip() + "\n"
def strip_front_matter(content: str) -> str:
"""Remove YAML front matter (--- ... ---) from the start of content."""
return _FRONT_MATTER_RE.sub("", content)
def convert_admonitions(content: str) -> str:
"""Convert MkDocs admonitions to Discourse/Obsidian callouts.
Input:
!!! warning "Title"
Content line 1
Content line 2
Output:
> [!WARNING] Title
> Content line 1
> Content line 2
"""
lines = content.splitlines(keepends=True)
result: list[str] = []
i = 0
while i < len(lines):
line = lines[i]
m = _ADMONITION_RE.match(line)
if m:
indent = m.group(1)
marker = m.group(2)
ad_type = m.group(3).lower()
title = m.group(4)
callout_type = ADMONITION_MAP.get(ad_type, ad_type.upper())
header = f"{indent}> [!{callout_type}]"
if title:
header += f" {title}"
if marker.startswith("???"):
collapsed = "+" not in marker
if not title:
action = "expand" if collapsed else "collapse"
header += f" *(click to {action})*"
result.append(header + "\n")
i += 1
content_indent = indent + " "
while i < len(lines):
content_line = lines[i]
if content_line.startswith(content_indent):
stripped = content_line[len(content_indent):]
result.append(f"{indent}> {stripped}")
i += 1
elif content_line.strip() == "":
# Check if blank line is internal to the admonition
j = i + 1
while j < len(lines) and lines[j].strip() == "":
j += 1
if j < len(lines) and lines[j].startswith(content_indent):
result.append(f"{indent}>\n")
i += 1
else:
break
else:
break
else:
result.append(line)
i += 1
return "".join(result)
def convert_tabs(content: str) -> str:
"""Convert Material tabs to bold headings with horizontal rules.
Input:
=== "Tab Name"
Content
Output:
**Tab Name**
Content
---
"""
lines = content.splitlines(keepends=True)
result: list[str] = []
i = 0
while i < len(lines):
line = lines[i]
m = _TAB_RE.match(line)
if m:
indent = m.group(1)
tab_name = m.group(2)
result.append(f"{indent}**{tab_name}**\n")
result.append("\n")
i += 1
content_indent = indent + " "
while i < len(lines):
content_line = lines[i]
if content_line.startswith(content_indent) or content_line.strip() == "":
if content_line.strip() == "":
result.append("\n")
else:
stripped = content_line[len(content_indent):]
result.append(f"{indent}{stripped}")
i += 1
else:
break
# Ensure blank line before the horizontal rule
if result and result[-1] != "\n":
result.append("\n")
result.append(f"{indent}---\n")
result.append("\n")
else:
result.append(line)
i += 1
return "".join(result)
def convert_grid_cards(content: str) -> str:
"""Strip grid card HTML wrappers (Discourse doesn't support them)."""
result = _GRID_CARD_OPEN_RE.sub("", content)
result = _GRID_CARD_CLOSE_RE.sub("", result)
return result
def convert_emoji_shortcodes(content: str) -> str:
"""Convert Material emoji shortcodes to plain text equivalents."""
result = content
for shortcode, replacement in EMOJI_MAP.items():
result = result.replace(shortcode, replacement)
# Strip any remaining :material-*: shortcodes not in the map
result = _EMOJI_SHORTCODE_RE.sub("", result)
return result
def resolve_internal_links(
content: str, *, file_path: str, docs_base_url: str
) -> str:
"""Resolve internal .md links to docs site URLs.
Converts: [text](../features/icbm.md) -> [text](https://docs.sunnypilot.ai/features/icbm/)
Converts: [text](../features/icbm.md#section) -> [text](https://docs.sunnypilot.ai/features/icbm/#section)
"""
current_dir = str(Path(file_path).parent)
def _replace_link(match: re.Match[str]) -> str:
raw_path = match.group(1)
# Skip external URLs
if raw_path.startswith("http"):
return match.group(0)
# Split path and optional anchor
anchor = ""
if "#" in raw_path:
raw_path, anchor = raw_path.rsplit("#", 1)
anchor = f"#{anchor}"
# Resolve relative path from current file's directory
resolved = Path(current_dir, raw_path).resolve()
resolved_str = str(resolved)
# Extract docs-relative path
docs_sp_idx = resolved_str.find("docs_sp/")
if docs_sp_idx >= 0:
docs_relative = resolved_str[docs_sp_idx + len("docs_sp/"):]
else:
# Fallback: use the resolved path relative to file
docs_relative = raw_path
# Convert to URL: remove .md, add trailing slash
url_path = re.sub(r"\.md$", "/", docs_relative)
return f"]({docs_base_url}/{url_path}{anchor})"
return _INTERNAL_LINK_RE.sub(_replace_link, content)
def clean_blank_lines(content: str) -> str:
"""Collapse 4+ consecutive blank lines down to 3."""
return _EXCESSIVE_BLANKS_RE.sub("\n\n\n", content)
+217
View File
@@ -0,0 +1,217 @@
"""Minimal Discourse API client using only urllib (zero external deps).
Provides the 5 CRUD operations needed by the docs sync orchestrator:
1. get_category_id(slug)
2. find_topic_by_sync_id(sync_id)
3. create_topic(title, raw, category_id, tags)
4. update_post(post_id, raw, edit_reason)
5. first_post_id(topic_id)
Configuration via environment variables:
DISCOURSE_URL - Base URL (e.g. https://community.sunnypilot.ai)
DISCOURSE_API_KEY - API key with topic create/update permissions
DISCOURSE_API_USER - Username for API requests (default: "system")
DISCOURSE_CATEGORY - Category slug for documentation (default: "documentation")
"""
from __future__ import annotations
import json
import os
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class DiscourseConfig:
"""Immutable configuration for the Discourse API client."""
base_url: str
api_key: str
api_user: str = "system"
category_slug: str = "documentation"
@classmethod
def from_env(cls) -> DiscourseConfig:
"""Build config from environment variables.
Raises:
ValueError: If required env vars are missing.
"""
base_url = os.environ.get("DISCOURSE_URL", "")
api_key = os.environ.get("DISCOURSE_API_KEY", "")
if not base_url:
raise ValueError("DISCOURSE_URL environment variable is required")
if not api_key:
raise ValueError("DISCOURSE_API_KEY environment variable is required")
return cls(
base_url=base_url.rstrip("/"),
api_key=api_key,
api_user=os.environ.get("DISCOURSE_API_USER", "system"),
category_slug=os.environ.get("DISCOURSE_CATEGORY", "documentation"),
)
class DiscourseClient:
"""Discourse API client for docs sync operations."""
def __init__(self, config: DiscourseConfig) -> None:
self._config = config
@property
def config(self) -> DiscourseConfig:
return self._config
# ----- Public API -----
def get_category_id(self, slug: str | None = None) -> int | None:
"""Look up a category ID by slug.
Args:
slug: Category slug. Defaults to config.category_slug.
Returns:
Category ID, or None if not found.
"""
slug = slug or self._config.category_slug
data = self._get(f"/c/{slug}/show.json")
if data is None:
return None
return data.get("category", {}).get("id")
def find_topic_by_sync_id(self, sync_id: str) -> dict[str, Any] | None:
"""Find an existing topic by its embedded sync ID comment.
Searches for topics containing <!-- docs-sync-id: {sync_id} -->.
Args:
sync_id: The doc path used as sync identifier.
Returns:
Topic dict with at least 'id' key, or None if not found.
"""
query = f"<!-- docs-sync-id: {sync_id} -->"
encoded = urllib.parse.urlencode({"q": query})
data = self._get(f"/search.json?{encoded}")
if data is None:
return None
topics = data.get("topics", [])
return topics[0] if topics else None
def create_topic(
self,
title: str,
raw: str,
category_id: int,
tags: list[str] | None = None,
) -> dict[str, Any] | None:
"""Create a new topic in the specified category.
Args:
title: Topic title.
raw: Markdown body content.
category_id: Discourse category ID.
tags: Optional list of tags.
Returns:
Response dict with 'topic_id', 'id' (post ID), etc., or None on failure.
"""
payload: dict[str, Any] = {
"title": title,
"raw": raw,
"category": category_id,
}
if tags:
payload["tags"] = tags
return self._post("/posts.json", payload)
def update_post(
self,
post_id: int,
raw: str,
edit_reason: str = "Documentation sync",
) -> dict[str, Any] | None:
"""Update an existing post's content.
Args:
post_id: The Discourse post ID to update.
raw: New markdown body content.
edit_reason: Reason shown in edit history.
Returns:
Response dict, or None on failure.
"""
payload = {
"post": {
"raw": raw,
"edit_reason": edit_reason,
},
}
return self._put(f"/posts/{post_id}.json", payload)
def first_post_id(self, topic_id: int) -> int | None:
"""Get the first post ID of a topic.
Args:
topic_id: The Discourse topic ID.
Returns:
Post ID of the first post, or None if not found.
"""
data = self._get(f"/t/{topic_id}.json")
if data is None:
return None
posts = data.get("post_stream", {}).get("posts", [])
if not posts:
return None
return posts[0].get("id")
# ----- HTTP helpers -----
def _headers(self) -> dict[str, str]:
return {
"Content-Type": "application/json",
"Api-Key": self._config.api_key,
"Api-Username": self._config.api_user,
}
def _get(self, path: str) -> dict[str, Any] | None:
url = self._config.base_url + path
req = urllib.request.Request(url, headers=self._headers(), method="GET")
return self._send(req)
def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any] | None:
url = self._config.base_url + path
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=self._headers(), method="POST")
return self._send(req)
def _put(self, path: str, payload: dict[str, Any]) -> dict[str, Any] | None:
url = self._config.base_url + path
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=self._headers(), method="PUT")
return self._send(req)
def _send(self, req: urllib.request.Request) -> dict[str, Any] | None:
try:
with urllib.request.urlopen(req) as resp:
body = resp.read().decode("utf-8")
return json.loads(body) if body else {}
except urllib.error.HTTPError as e:
# Log but don't crash — caller decides how to handle None
status = e.code
body = ""
try:
body = e.read().decode("utf-8", errors="replace")[:500]
except Exception:
pass
print(f" Discourse API error: {req.method} {req.full_url} -> {status}: {body}")
return None
except urllib.error.URLError as e:
print(f" Discourse connection error: {req.full_url} -> {e.reason}")
return None
+88
View File
@@ -0,0 +1,88 @@
"""Parse zensical.toml nav structure into a flat list of doc entries.
Reads the nav tree from zensical.toml and produces a flat list of
{title, path, breadcrumb} dicts suitable for the sync orchestrator.
"""
from __future__ import annotations
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
SKIP_FILES = frozenset({"index.md", "README.md"})
@dataclass(frozen=True)
class NavEntry:
"""A single navigable documentation page."""
title: str
path: str
breadcrumb: tuple[str, ...] = field(default_factory=tuple)
def parse(config_path: str | Path) -> list[NavEntry]:
"""Parse zensical.toml and return all nav entries.
Args:
config_path: Path to zensical.toml.
Returns:
Flat list of NavEntry for every page in the nav tree,
excluding index.md, README.md, and external links.
"""
config_path = Path(config_path)
with config_path.open("rb") as f:
config = tomllib.load(f)
nav = config.get("project", {}).get("nav", [])
entries = _flatten_nav(nav)
return [e for e in entries if Path(e.path).name not in SKIP_FILES]
def parse_all(config_path: str | Path) -> list[NavEntry]:
"""Like parse(), but includes index.md and README.md entries."""
config_path = Path(config_path)
with config_path.open("rb") as f:
config = tomllib.load(f)
nav = config.get("project", {}).get("nav", [])
return _flatten_nav(nav)
def _flatten_nav(
items: list[dict[str, str | list] | str],
breadcrumb: tuple[str, ...] = (),
) -> list[NavEntry]:
"""Recursively flatten the nav tree into NavEntry objects."""
result: list[NavEntry] = []
for item in items:
if isinstance(item, dict):
for title, value in item.items():
if isinstance(value, str):
# Skip external links
if value.startswith("http"):
continue
result.append(NavEntry(
title=title,
path=value,
breadcrumb=breadcrumb + (title,),
))
elif isinstance(value, list):
result.extend(
_flatten_nav(value, breadcrumb + (title,))
)
elif isinstance(item, str):
# Bare path without title
if item.startswith("http"):
continue
name = Path(item).stem.replace("-", " ").capitalize()
result.append(NavEntry(
title=name,
path=item,
breadcrumb=breadcrumb + (name,),
))
return result
+183
View File
@@ -0,0 +1,183 @@
#!/usr/bin/env python3
"""Tests for the SHA-256 content cache.
Run: python3 docs_sp/tools/test_content_cache.py
"""
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from content_cache import ContentCache
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def make_cache(tmp: str) -> ContentCache:
return ContentCache(cache_dir=Path(tmp) / ".discourse_sync_cache")
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_compute_hash_deterministic():
"""Same content always produces the same hash."""
h1 = ContentCache.compute_hash("hello world")
h2 = ContentCache.compute_hash("hello world")
assert h1 == h2
assert len(h1) == 64 # SHA-256 hex digest length
print(" PASS: compute_hash_deterministic")
def test_compute_hash_differs():
"""Different content produces different hashes."""
h1 = ContentCache.compute_hash("hello world")
h2 = ContentCache.compute_hash("hello world!")
assert h1 != h2
print(" PASS: compute_hash_differs")
def test_is_changed_no_cache():
"""First run (no cache file) should report changed."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
assert cache.is_changed("features/icbm.md", "content")
print(" PASS: is_changed_no_cache")
def test_is_changed_after_save():
"""After saving, same content should report unchanged."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
cache.save("features/icbm.md", "content v1")
assert not cache.is_changed("features/icbm.md", "content v1")
print(" PASS: is_changed_after_save")
def test_is_changed_after_modification():
"""After saving, different content should report changed."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
cache.save("features/icbm.md", "content v1")
assert cache.is_changed("features/icbm.md", "content v2")
print(" PASS: is_changed_after_modification")
def test_separate_paths_independent():
"""Different doc paths have independent caches."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
cache.save("features/icbm.md", "content A")
cache.save("safety/safety.md", "content B")
assert not cache.is_changed("features/icbm.md", "content A")
assert not cache.is_changed("safety/safety.md", "content B")
assert cache.is_changed("features/icbm.md", "content B")
assert cache.is_changed("safety/safety.md", "content A")
print(" PASS: separate_paths_independent")
def test_cache_dir_created_on_save():
"""Cache directory is created automatically on first save."""
with tempfile.TemporaryDirectory() as tmp:
cache_dir = Path(tmp) / "nested" / "cache"
cache = ContentCache(cache_dir=cache_dir)
assert not cache_dir.exists()
cache.save("test.md", "content")
assert cache_dir.exists()
assert (cache_dir / "test.sha256").exists()
print(" PASS: cache_dir_created_on_save")
def test_cache_file_naming():
"""Cache files use slug derived from doc path."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
cache.save("settings/cruise/speed-limit/source.md", "content")
expected_name = "settings_cruise_speed-limit_source.sha256"
cached_files = list(cache.cache_dir.glob("*.sha256"))
assert len(cached_files) == 1
assert cached_files[0].name == expected_name
print(" PASS: cache_file_naming")
def test_clear_removes_all():
"""clear() removes all .sha256 files and returns count."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
cache.save("a.md", "aaa")
cache.save("b.md", "bbb")
cache.save("c.md", "ccc")
removed = cache.clear()
assert removed == 3
assert list(cache.cache_dir.glob("*.sha256")) == []
print(" PASS: clear_removes_all")
def test_clear_empty_cache():
"""clear() on nonexistent cache dir returns 0."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
assert cache.clear() == 0
print(" PASS: clear_empty_cache")
def test_overwrite_on_resave():
"""Saving again overwrites the previous hash."""
with tempfile.TemporaryDirectory() as tmp:
cache = make_cache(tmp)
cache.save("doc.md", "version 1")
assert not cache.is_changed("doc.md", "version 1")
assert cache.is_changed("doc.md", "version 2")
cache.save("doc.md", "version 2")
assert cache.is_changed("doc.md", "version 1")
assert not cache.is_changed("doc.md", "version 2")
print(" PASS: overwrite_on_resave")
# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("Testing content cache:")
tests = [
test_compute_hash_deterministic,
test_compute_hash_differs,
test_is_changed_no_cache,
test_is_changed_after_save,
test_is_changed_after_modification,
test_separate_paths_independent,
test_cache_dir_created_on_save,
test_cache_file_naming,
test_clear_removes_all,
test_clear_empty_cache,
test_overwrite_on_resave,
]
passed = 0
failed = 0
for test in tests:
try:
test()
passed += 1
except AssertionError as e:
print(f" FAIL: {test.__name__}: {e}")
failed += 1
except Exception as e:
print(f" ERROR: {test.__name__}: {e}")
failed += 1
print(f"\n{passed}/{passed + failed} tests passed")
sys.exit(1 if failed > 0 else 0)
+317 -122
View File
@@ -1,242 +1,209 @@
#!/usr/bin/env python3
"""Test the MkDocs Discourse admonition conversion logic.
"""Tests for the MkDocs -> Discourse markdown converter.
This mirrors the Ruby converter's logic to verify correctness.
Run: python3 docs_sp/tools/test_converter.py
"""
import re
import sys
from pathlib import Path
ADMONITION_MAP = {
"note": "NOTE",
"abstract": "ABSTRACT",
"info": "INFO",
"tip": "TIP",
"success": "SUCCESS",
"question": "QUESTION",
"warning": "WARNING",
"failure": "FAILURE",
"danger": "DANGER",
"bug": "BUG",
"example": "EXAMPLE",
"quote": "QUOTE",
}
# Ensure the tools directory is importable
sys.path.insert(0, str(Path(__file__).parent))
from converter import (
clean_blank_lines,
convert,
convert_admonitions,
convert_emoji_shortcodes,
convert_grid_cards,
convert_tabs,
resolve_internal_links,
strip_front_matter,
)
# ---------------------------------------------------------------------------
# 1. Strip YAML Front Matter
# ---------------------------------------------------------------------------
def convert_admonitions(content: str) -> str:
"""Convert MkDocs admonitions to Obsidian/Discourse callouts."""
lines = content.splitlines(keepends=True)
result = []
i = 0
def test_strip_front_matter_basic():
input_text = """---
title: My Page
description: A test page
---
pattern = re.compile(r'^(\s*)(!{3}|\?{3}\+?) (\w+)(?: "([^"]*)")?')
while i < len(lines):
line = lines[i]
m = pattern.match(line)
if m:
indent = m.group(1)
marker = m.group(2)
ad_type = m.group(3).lower()
title = m.group(4)
callout_type = ADMONITION_MAP.get(ad_type, ad_type.upper())
header = f"{indent}> [!{callout_type}]"
if title:
header += f" {title}"
if marker.startswith("???"):
collapsed = "+" not in marker
if not title:
header += f" *(click to {'expand' if collapsed else 'collapse'})*"
result.append(header + "\n")
i += 1
content_indent = indent + " "
while i < len(lines):
content_line = lines[i]
if content_line.startswith(content_indent):
stripped = content_line[len(content_indent):]
result.append(f"{indent}> {stripped}")
i += 1
elif content_line.strip() == "":
# Blank line: only treat as part of admonition if the
# next non-blank line is still indented at content level
j = i + 1
while j < len(lines) and lines[j].strip() == "":
j += 1
if j < len(lines) and lines[j].startswith(content_indent):
result.append(f"{indent}>\n")
i += 1
else:
# Blank line ends the admonition
break
else:
break
else:
result.append(line)
i += 1
return "".join(result)
# Hello
"""
expected = """# Hello
"""
result = strip_front_matter(input_text)
assert result == expected, f"FAIL:\n{result!r}\n!=\n{expected!r}"
print(" PASS: strip_front_matter_basic")
# ---- Tests ----
def test_strip_front_matter_absent():
input_text = "# Hello\n\nContent here.\n"
result = strip_front_matter(input_text)
assert result == input_text, f"FAIL:\n{result!r}\n!=\n{input_text!r}"
print(" PASS: strip_front_matter_absent")
# ---------------------------------------------------------------------------
# 2. Convert Admonitions
# ---------------------------------------------------------------------------
def test_basic_warning():
input_text = '''!!! warning "Important"
input_text = """!!! warning "Important"
sunnypilot is a **driver assistance** system.
Always pay attention.
'''
expected = '''> [!WARNING] Important
"""
expected = """> [!WARNING] Important
> sunnypilot is a **driver assistance** system.
> Always pay attention.
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL basic_warning:\n{result!r}\n!=\n{expected!r}"
print(" PASS: basic_warning")
def test_info_no_title():
input_text = '''!!! info
input_text = """!!! info
Content line 1
Content line 2
'''
expected = '''> [!INFO]
"""
expected = """> [!INFO]
> Content line 1
> Content line 2
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL info_no_title:\n{result!r}\n!=\n{expected!r}"
print(" PASS: info_no_title")
def test_info_with_title():
input_text = '''!!! info "Requirements"
input_text = """!!! info "Requirements"
- Longitudinal control must be available
- ICBM must be enabled
'''
expected = '''> [!INFO] Requirements
"""
expected = """> [!INFO] Requirements
> - Longitudinal control must be available
> - ICBM must be enabled
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL info_with_title:\n{result!r}\n!=\n{expected!r}"
print(" PASS: info_with_title")
def test_danger():
input_text = '''!!! danger "Important"
input_text = """!!! danger "Important"
sunnypilot is a **driver assistance** system. It is **NOT** a self-driving system.
'''
expected = '''> [!DANGER] Important
"""
expected = """> [!DANGER] Important
> sunnypilot is a **driver assistance** system. It is **NOT** a self-driving system.
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL danger:\n{result!r}\n!=\n{expected!r}"
print(" PASS: danger")
def test_tip():
input_text = '''!!! tip
input_text = """!!! tip
The more detail you provide, the faster we can diagnose and fix the issue.
'''
expected = '''> [!TIP]
"""
expected = """> [!TIP]
> The more detail you provide, the faster we can diagnose and fix the issue.
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL tip:\n{result!r}\n!=\n{expected!r}"
print(" PASS: tip")
def test_multiline_with_blank():
input_text = '''!!! warning
input_text = """!!! warning
Line 1
Line 2 after blank
'''
expected = '''> [!WARNING]
"""
expected = """> [!WARNING]
> Line 1
>
> Line 2 after blank
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL multiline_with_blank:\n{result!r}\n!=\n{expected!r}"
print(" PASS: multiline_with_blank")
def test_collapsible():
input_text = '''??? warning "Click to see"
input_text = """??? warning "Click to see"
Hidden content
'''
expected = '''> [!WARNING] Click to see
"""
expected = """> [!WARNING] Click to see
> Hidden content
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL collapsible:\n{result!r}\n!=\n{expected!r}"
print(" PASS: collapsible")
def test_collapsible_open():
input_text = '''???+ info "Open by default"
input_text = """???+ info "Open by default"
Visible content
'''
expected = '''> [!INFO] Open by default
"""
expected = """> [!INFO] Open by default
> Visible content
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL collapsible_open:\n{result!r}\n!=\n{expected!r}"
print(" PASS: collapsible_open")
def test_surrounded_by_content():
input_text = '''Some text before.
input_text = """Some text before.
!!! note "Note Title"
Note content here.
Some text after.
'''
expected = '''Some text before.
"""
expected = """Some text before.
> [!NOTE] Note Title
> Note content here.
Some text after.
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL surrounded:\n{result!r}\n!=\n{expected!r}"
print(" PASS: surrounded_by_content")
def test_multiple_admonitions():
input_text = '''!!! info "Requirements"
input_text = """!!! info "Requirements"
- Req 1
- Req 2
!!! warning "Vehicle Restrictions"
- Tesla: disabled on release
- Rivian: always disabled
'''
expected = '''> [!INFO] Requirements
"""
expected = """> [!INFO] Requirements
> - Req 1
> - Req 2
> [!WARNING] Vehicle Restrictions
> - Tesla: disabled on release
> - Rivian: always disabled
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL multiple:\n{result!r}\n!=\n{expected!r}"
print(" PASS: multiple_admonitions")
def test_real_doc_snippet():
"""Test with an actual snippet from docs_sp/settings/speed-limit.md."""
input_text = '''## Speed Limit Mode
"""Test with an actual snippet from docs_sp content."""
input_text = """## Speed Limit Mode
| Property | Value |
|----------|-------|
@@ -251,8 +218,8 @@ def test_real_doc_snippet():
- **Rivian:** Speed Limit Assist mode is always disabled
---
'''
expected = '''## Speed Limit Mode
"""
expected = """## Speed Limit Mode
| Property | Value |
|----------|-------|
@@ -267,15 +234,225 @@ def test_real_doc_snippet():
> - **Rivian:** Speed Limit Assist mode is always disabled
---
'''
"""
result = convert_admonitions(input_text)
assert result == expected, f"FAIL real_doc:\n{result!r}\n!=\n{expected!r}"
print(" PASS: real_doc_snippet")
# ---------------------------------------------------------------------------
# 3. Convert Tabs
# ---------------------------------------------------------------------------
def test_tabs_basic():
input_text = """=== "Tab One"
Content for tab one.
=== "Tab Two"
Content for tab two.
"""
expected = """**Tab One**
Content for tab one.
---
**Tab Two**
Content for tab two.
---
"""
result = convert_tabs(input_text)
assert result == expected, f"FAIL tabs_basic:\n{result!r}\n!=\n{expected!r}"
print(" PASS: tabs_basic")
def test_tabs_multiline():
input_text = """=== "Details"
Line 1
Line 2
Line 3
"""
expected = """**Details**
Line 1
Line 2
Line 3
---
"""
result = convert_tabs(input_text)
assert result == expected, f"FAIL tabs_multiline:\n{result!r}\n!=\n{expected!r}"
print(" PASS: tabs_multiline")
# ---------------------------------------------------------------------------
# 4. Convert Grid Cards
# ---------------------------------------------------------------------------
def test_grid_cards_stripped():
input_text = """<div class="grid cards" markdown>
- **Card 1** - Description
- **Card 2** - Description
</div>
"""
expected = """
- **Card 1** - Description
- **Card 2** - Description
"""
result = convert_grid_cards(input_text)
# Normalize whitespace for comparison
assert result.strip() == expected.strip(), f"FAIL grid_cards:\n{result!r}\n!=\n{expected!r}"
print(" PASS: grid_cards_stripped")
# ---------------------------------------------------------------------------
# 5. Convert Emoji Shortcodes
# ---------------------------------------------------------------------------
def test_emoji_known():
input_text = ":material-check: Supported | :material-close: Not supported"
expected = "Y Supported | N Not supported"
result = convert_emoji_shortcodes(input_text)
assert result == expected, f"FAIL emoji_known:\n{result!r}\n!=\n{expected!r}"
print(" PASS: emoji_known")
def test_emoji_unknown_stripped():
input_text = ":material-unknown-icon: Some text"
expected = " Some text"
result = convert_emoji_shortcodes(input_text)
assert result == expected, f"FAIL emoji_unknown:\n{result!r}\n!=\n{expected!r}"
print(" PASS: emoji_unknown_stripped")
def test_emoji_in_grid_card():
input_text = "- :material-rocket-launch: **[Feature](link.md)**"
expected = "- **[Feature](link.md)**"
result = convert_emoji_shortcodes(input_text)
assert result == expected, f"FAIL emoji_grid:\n{result!r}\n!=\n{expected!r}"
print(" PASS: emoji_in_grid_card")
# ---------------------------------------------------------------------------
# 6. Resolve Internal Links
# ---------------------------------------------------------------------------
def test_internal_link_relative():
input_text = "See [ICBM](../features/cruise/icbm.md) for details."
result = resolve_internal_links(
input_text,
file_path="docs_sp/settings/cruise/speed-limit.md",
docs_base_url="https://docs.sunnypilot.ai",
)
assert "https://docs.sunnypilot.ai/" in result, f"FAIL internal_link:\n{result!r}"
assert ".md" not in result.split("](")[1], f"FAIL internal_link still has .md:\n{result!r}"
print(" PASS: internal_link_relative")
def test_internal_link_with_anchor():
input_text = "See [section](./safety.md#driver-responsibility)."
result = resolve_internal_links(
input_text,
file_path="docs_sp/safety/index.md",
docs_base_url="https://docs.sunnypilot.ai",
)
assert "#driver-responsibility" in result, f"FAIL anchor preserved:\n{result!r}"
assert ".md" not in result.split("](")[1].split("#")[0], f"FAIL link has .md:\n{result!r}"
print(" PASS: internal_link_with_anchor")
def test_external_link_untouched():
input_text = "Visit [GitHub](https://github.com/sunnypilot/sunnypilot)."
result = resolve_internal_links(
input_text,
file_path="docs_sp/index.md",
docs_base_url="https://docs.sunnypilot.ai",
)
assert result == input_text, f"FAIL external_link:\n{result!r}"
print(" PASS: external_link_untouched")
# ---------------------------------------------------------------------------
# 7. Clean Blank Lines
# ---------------------------------------------------------------------------
def test_clean_blank_lines():
input_text = "Line 1\n\n\n\n\nLine 2\n"
expected = "Line 1\n\n\nLine 2\n"
result = clean_blank_lines(input_text)
assert result == expected, f"FAIL clean_blanks:\n{result!r}\n!=\n{expected!r}"
print(" PASS: clean_blank_lines")
def test_clean_blank_lines_no_change():
input_text = "Line 1\n\nLine 2\n"
result = clean_blank_lines(input_text)
assert result == input_text, f"FAIL clean_blanks_noop:\n{result!r}"
print(" PASS: clean_blank_lines_no_change")
# ---------------------------------------------------------------------------
# Integration: full convert()
# ---------------------------------------------------------------------------
def test_full_convert():
input_text = """---
title: Test Doc
---
# Test Document
!!! warning "Important"
Pay attention to the road.
See [safety info](../safety/safety.md) for more.
:material-check: Feature supported
"""
result = convert(
input_text,
file_path="docs_sp/features/index.md",
docs_base_url="https://docs.sunnypilot.ai",
)
# Front matter stripped
assert "---\ntitle:" not in result, f"FAIL front matter not stripped:\n{result!r}"
# Admonition converted
assert "> [!WARNING] Important" in result, f"FAIL admonition:\n{result!r}"
assert "> Pay attention to the road." in result, f"FAIL admonition content:\n{result!r}"
# Link resolved
assert ".md" not in result.split("](")[1].split(")")[0], f"FAIL link:\n{result!r}"
# Emoji converted
assert ":material-check:" not in result, f"FAIL emoji:\n{result!r}"
assert "Y Feature supported" in result, f"FAIL emoji replacement:\n{result!r}"
print(" PASS: full_convert")
# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("Testing MkDocs Discourse admonition conversion:")
print("Testing MkDocs -> Discourse converter:")
tests = [
# 1. Front matter
test_strip_front_matter_basic,
test_strip_front_matter_absent,
# 2. Admonitions
test_basic_warning,
test_info_no_title,
test_info_with_title,
@@ -287,6 +464,24 @@ if __name__ == "__main__":
test_surrounded_by_content,
test_multiple_admonitions,
test_real_doc_snippet,
# 3. Tabs
test_tabs_basic,
test_tabs_multiline,
# 4. Grid cards
test_grid_cards_stripped,
# 5. Emoji
test_emoji_known,
test_emoji_unknown_stripped,
test_emoji_in_grid_card,
# 6. Internal links
test_internal_link_relative,
test_internal_link_with_anchor,
test_external_link_untouched,
# 7. Blank lines
test_clean_blank_lines,
test_clean_blank_lines_no_change,
# Integration
test_full_convert,
]
passed = 0
failed = 0
@@ -298,7 +493,7 @@ if __name__ == "__main__":
print(f" FAIL: {test.__name__}: {e}")
failed += 1
except Exception as e:
print(f" FAIL: {test.__name__}: {e}")
print(f" ERROR: {test.__name__}: {e}")
failed += 1
print(f"\n{passed}/{passed + failed} tests passed")
+438
View File
@@ -0,0 +1,438 @@
#!/usr/bin/env python3
"""Tests for the Discourse API client (fully mocked, no live requests).
Run: python3 docs_sp/tools/test_discourse_client.py
"""
import io
import json
import sys
import urllib.error
import urllib.request
from pathlib import Path
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).parent))
from discourse_client import DiscourseClient, DiscourseConfig
TEST_CONFIG = DiscourseConfig(
base_url="https://community.sunnypilot.ai",
api_key="test-api-key-123",
api_user="docs-bot",
category_slug="documentation",
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def mock_response(data: dict, status: int = 200) -> MagicMock:
"""Create a mock urllib response with JSON body."""
body = json.dumps(data).encode("utf-8")
resp = MagicMock()
resp.read.return_value = body
resp.status = status
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
return resp
def mock_http_error(status: int, body: str = "") -> urllib.error.HTTPError:
"""Create a mock HTTPError."""
return urllib.error.HTTPError(
url="https://community.sunnypilot.ai/test",
code=status,
msg=f"HTTP {status}",
hdrs={}, # type: ignore[arg-type]
fp=io.BytesIO(body.encode("utf-8")),
)
# ---------------------------------------------------------------------------
# DiscourseConfig tests
# ---------------------------------------------------------------------------
def test_config_from_env():
env = {
"DISCOURSE_URL": "https://forum.example.com/",
"DISCOURSE_API_KEY": "secret-key",
"DISCOURSE_API_USER": "bot",
"DISCOURSE_CATEGORY": "docs",
}
with patch.dict("os.environ", env, clear=False):
config = DiscourseConfig.from_env()
assert config.base_url == "https://forum.example.com" # trailing slash stripped
assert config.api_key == "secret-key"
assert config.api_user == "bot"
assert config.category_slug == "docs"
print(" PASS: config_from_env")
def test_config_from_env_defaults():
env = {
"DISCOURSE_URL": "https://forum.example.com",
"DISCOURSE_API_KEY": "key",
}
with patch.dict("os.environ", env, clear=False):
# Remove optional vars if present
with patch.dict("os.environ", {"DISCOURSE_API_USER": "", "DISCOURSE_CATEGORY": ""}, clear=False):
pass
config = DiscourseConfig.from_env()
assert config.api_user in ("system", os.environ.get("DISCOURSE_API_USER", "system"))
print(" PASS: config_from_env_defaults")
def test_config_missing_url():
env = {"DISCOURSE_API_KEY": "key"}
with patch.dict("os.environ", env, clear=True):
try:
DiscourseConfig.from_env()
assert False, "Should have raised ValueError"
except ValueError as e:
assert "DISCOURSE_URL" in str(e)
print(" PASS: config_missing_url")
def test_config_missing_api_key():
env = {"DISCOURSE_URL": "https://forum.example.com"}
with patch.dict("os.environ", env, clear=True):
try:
DiscourseConfig.from_env()
assert False, "Should have raised ValueError"
except ValueError as e:
assert "DISCOURSE_API_KEY" in str(e)
print(" PASS: config_missing_api_key")
def test_config_immutable():
try:
TEST_CONFIG.base_url = "https://other.com" # type: ignore[misc]
assert False, "Should have raised"
except AttributeError:
pass
print(" PASS: config_immutable")
# ---------------------------------------------------------------------------
# get_category_id
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_get_category_id_found(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"category": {"id": 42, "slug": "documentation"}})
client = DiscourseClient(TEST_CONFIG)
result = client.get_category_id("documentation")
assert result == 42
call_args = mock_urlopen.call_args[0][0]
assert "/c/documentation/show.json" in call_args.full_url
assert call_args.get_header("Api-key") == "test-api-key-123"
assert call_args.get_header("Api-username") == "docs-bot"
print(" PASS: get_category_id_found")
@patch("urllib.request.urlopen")
def test_get_category_id_not_found(mock_urlopen: MagicMock):
mock_urlopen.side_effect = mock_http_error(404, "Not Found")
client = DiscourseClient(TEST_CONFIG)
result = client.get_category_id("nonexistent")
assert result is None
print(" PASS: get_category_id_not_found")
@patch("urllib.request.urlopen")
def test_get_category_id_uses_default_slug(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"category": {"id": 7}})
client = DiscourseClient(TEST_CONFIG)
client.get_category_id() # no arg — uses config.category_slug
call_args = mock_urlopen.call_args[0][0]
assert "/c/documentation/show.json" in call_args.full_url
print(" PASS: get_category_id_uses_default_slug")
# ---------------------------------------------------------------------------
# find_topic_by_sync_id
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_find_topic_found(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({
"topics": [{"id": 101, "title": "ICBM Docs"}],
})
client = DiscourseClient(TEST_CONFIG)
result = client.find_topic_by_sync_id("features/cruise/icbm.md")
assert result is not None
assert result["id"] == 101
call_args = mock_urlopen.call_args[0][0]
assert "/search.json?" in call_args.full_url
assert "docs-sync-id" in call_args.full_url
print(" PASS: find_topic_found")
@patch("urllib.request.urlopen")
def test_find_topic_not_found(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"topics": []})
client = DiscourseClient(TEST_CONFIG)
result = client.find_topic_by_sync_id("nonexistent.md")
assert result is None
print(" PASS: find_topic_not_found")
@patch("urllib.request.urlopen")
def test_find_topic_api_error(mock_urlopen: MagicMock):
mock_urlopen.side_effect = mock_http_error(500, "Internal Server Error")
client = DiscourseClient(TEST_CONFIG)
result = client.find_topic_by_sync_id("features/icbm.md")
assert result is None
print(" PASS: find_topic_api_error")
# ---------------------------------------------------------------------------
# create_topic
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_create_topic_success(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({
"id": 501,
"topic_id": 201,
"topic_slug": "icbm-sunnypilot-docs",
})
client = DiscourseClient(TEST_CONFIG)
result = client.create_topic(
title="ICBM - sunnypilot Docs",
raw="# ICBM\n\nDoc content here.",
category_id=42,
tags=["docs", "auto-sync"],
)
assert result is not None
assert result["topic_id"] == 201
call_args = mock_urlopen.call_args[0][0]
assert call_args.method == "POST"
assert "/posts.json" in call_args.full_url
sent_payload = json.loads(call_args.data.decode("utf-8"))
assert sent_payload["title"] == "ICBM - sunnypilot Docs"
assert sent_payload["category"] == 42
assert sent_payload["tags"] == ["docs", "auto-sync"]
assert "ICBM" in sent_payload["raw"]
print(" PASS: create_topic_success")
@patch("urllib.request.urlopen")
def test_create_topic_no_tags(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"id": 502, "topic_id": 202})
client = DiscourseClient(TEST_CONFIG)
client.create_topic(title="Test", raw="body", category_id=1)
sent_payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
assert "tags" not in sent_payload
print(" PASS: create_topic_no_tags")
@patch("urllib.request.urlopen")
def test_create_topic_failure(mock_urlopen: MagicMock):
mock_urlopen.side_effect = mock_http_error(422, '{"errors":["Title too short"]}')
client = DiscourseClient(TEST_CONFIG)
result = client.create_topic(title="X", raw="body", category_id=1)
assert result is None
print(" PASS: create_topic_failure")
# ---------------------------------------------------------------------------
# update_post
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_update_post_success(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"post": {"id": 501, "version": 2}})
client = DiscourseClient(TEST_CONFIG)
result = client.update_post(post_id=501, raw="Updated content")
assert result is not None
call_args = mock_urlopen.call_args[0][0]
assert call_args.method == "PUT"
assert "/posts/501.json" in call_args.full_url
sent_payload = json.loads(call_args.data.decode("utf-8"))
assert sent_payload["post"]["raw"] == "Updated content"
assert sent_payload["post"]["edit_reason"] == "Documentation sync"
print(" PASS: update_post_success")
@patch("urllib.request.urlopen")
def test_update_post_custom_reason(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"post": {"id": 501}})
client = DiscourseClient(TEST_CONFIG)
client.update_post(post_id=501, raw="content", edit_reason="Manual fix")
sent_payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
assert sent_payload["post"]["edit_reason"] == "Manual fix"
print(" PASS: update_post_custom_reason")
@patch("urllib.request.urlopen")
def test_update_post_not_found(mock_urlopen: MagicMock):
mock_urlopen.side_effect = mock_http_error(404)
client = DiscourseClient(TEST_CONFIG)
result = client.update_post(post_id=99999, raw="content")
assert result is None
print(" PASS: update_post_not_found")
# ---------------------------------------------------------------------------
# first_post_id
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_first_post_id_found(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({
"post_stream": {
"posts": [
{"id": 501, "post_number": 1},
{"id": 502, "post_number": 2},
],
},
})
client = DiscourseClient(TEST_CONFIG)
result = client.first_post_id(topic_id=201)
assert result == 501
print(" PASS: first_post_id_found")
@patch("urllib.request.urlopen")
def test_first_post_id_empty_stream(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"post_stream": {"posts": []}})
client = DiscourseClient(TEST_CONFIG)
result = client.first_post_id(topic_id=201)
assert result is None
print(" PASS: first_post_id_empty_stream")
@patch("urllib.request.urlopen")
def test_first_post_id_topic_not_found(mock_urlopen: MagicMock):
mock_urlopen.side_effect = mock_http_error(404)
client = DiscourseClient(TEST_CONFIG)
result = client.first_post_id(topic_id=99999)
assert result is None
print(" PASS: first_post_id_topic_not_found")
# ---------------------------------------------------------------------------
# Headers / auth
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_headers_set_correctly(mock_urlopen: MagicMock):
mock_urlopen.return_value = mock_response({"category": {"id": 1}})
client = DiscourseClient(TEST_CONFIG)
client.get_category_id()
req = mock_urlopen.call_args[0][0]
assert req.get_header("Api-key") == "test-api-key-123"
assert req.get_header("Api-username") == "docs-bot"
assert req.get_header("Content-type") == "application/json"
print(" PASS: headers_set_correctly")
# ---------------------------------------------------------------------------
# Connection error
# ---------------------------------------------------------------------------
@patch("urllib.request.urlopen")
def test_connection_error_returns_none(mock_urlopen: MagicMock):
mock_urlopen.side_effect = urllib.error.URLError("Connection refused")
client = DiscourseClient(TEST_CONFIG)
result = client.get_category_id()
assert result is None
print(" PASS: connection_error_returns_none")
# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------
import os # noqa: E402 (needed for test_config_from_env_defaults)
if __name__ == "__main__":
print("Testing Discourse API client:")
tests = [
# Config
test_config_from_env,
test_config_from_env_defaults,
test_config_missing_url,
test_config_missing_api_key,
test_config_immutable,
# get_category_id
test_get_category_id_found,
test_get_category_id_not_found,
test_get_category_id_uses_default_slug,
# find_topic_by_sync_id
test_find_topic_found,
test_find_topic_not_found,
test_find_topic_api_error,
# create_topic
test_create_topic_success,
test_create_topic_no_tags,
test_create_topic_failure,
# update_post
test_update_post_success,
test_update_post_custom_reason,
test_update_post_not_found,
# first_post_id
test_first_post_id_found,
test_first_post_id_empty_stream,
test_first_post_id_topic_not_found,
# Misc
test_headers_set_correctly,
test_connection_error_returns_none,
]
passed = 0
failed = 0
for test in tests:
try:
test()
passed += 1
except AssertionError as e:
print(f" FAIL: {test.__name__}: {e}")
failed += 1
except Exception as e:
print(f" ERROR: {test.__name__}: {e}")
failed += 1
print(f"\n{passed}/{passed + failed} tests passed")
sys.exit(1 if failed > 0 else 0)
+238
View File
@@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""Tests for the zensical.toml nav parser.
Run: python3 docs_sp/tools/test_nav_parser.py
"""
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from nav_parser import NavEntry, parse, parse_all, _flatten_nav
ZENSICAL_TOML = Path(__file__).resolve().parent.parent.parent / "zensical.toml"
# ---------------------------------------------------------------------------
# Unit tests: _flatten_nav
# ---------------------------------------------------------------------------
def test_flatten_simple_dict():
nav = [{"Page One": "page-one.md"}]
result = _flatten_nav(nav)
assert len(result) == 1
assert result[0] == NavEntry(title="Page One", path="page-one.md", breadcrumb=("Page One",))
print(" PASS: flatten_simple_dict")
def test_flatten_nested():
nav = [
{"Section": [
{"Child A": "section/a.md"},
{"Child B": "section/b.md"},
]},
]
result = _flatten_nav(nav)
assert len(result) == 2
assert result[0].title == "Child A"
assert result[0].path == "section/a.md"
assert result[0].breadcrumb == ("Section", "Child A")
assert result[1].breadcrumb == ("Section", "Child B")
print(" PASS: flatten_nested")
def test_flatten_deep_nesting():
nav = [
{"L1": [
{"L2": [
{"L3": "deep/page.md"},
]},
]},
]
result = _flatten_nav(nav)
assert len(result) == 1
assert result[0].breadcrumb == ("L1", "L2", "L3")
assert result[0].path == "deep/page.md"
print(" PASS: flatten_deep_nesting")
def test_flatten_skips_external_links():
nav = [
{"Docs": "docs.md"},
{"Forum": "https://community.sunnypilot.ai"},
]
result = _flatten_nav(nav)
assert len(result) == 1
assert result[0].title == "Docs"
print(" PASS: flatten_skips_external_links")
def test_flatten_bare_string():
nav = ["getting-started/index.md"]
result = _flatten_nav(nav)
assert len(result) == 1
assert result[0].title == "Index"
assert result[0].path == "getting-started/index.md"
print(" PASS: flatten_bare_string")
def test_flatten_mixed():
nav = [
{"Home": "index.md"},
{"Features": [
"features/index.md",
{"ICBM": "features/cruise/icbm.md"},
{"Forum": "https://example.com"},
]},
]
result = _flatten_nav(nav)
assert len(result) == 3 # Home, features/index.md (bare), ICBM
titles = [e.title for e in result]
assert "Home" in titles
assert "ICBM" in titles
assert "Forum" not in titles
print(" PASS: flatten_mixed")
# ---------------------------------------------------------------------------
# Unit tests: parse (filters index.md/README.md)
# ---------------------------------------------------------------------------
def test_parse_filters_index():
"""parse() should exclude index.md and README.md entries."""
toml_content = b"""
[project]
nav = [
{"Home" = "index.md"},
{"Guide" = [
"guide/index.md",
{"Setup" = "guide/setup.md"},
]},
]
"""
with tempfile.NamedTemporaryFile(suffix=".toml", delete=False) as f:
f.write(toml_content)
f.flush()
result = parse(f.name)
assert len(result) == 1
assert result[0].title == "Setup"
assert result[0].path == "guide/setup.md"
print(" PASS: parse_filters_index")
def test_parse_all_includes_index():
"""parse_all() should include index.md entries."""
toml_content = b"""
[project]
nav = [
{"Home" = "index.md"},
{"Setup" = "guide/setup.md"},
]
"""
with tempfile.NamedTemporaryFile(suffix=".toml", delete=False) as f:
f.write(toml_content)
f.flush()
result = parse_all(f.name)
assert len(result) == 2
titles = [e.title for e in result]
assert "Home" in titles
assert "Setup" in titles
print(" PASS: parse_all_includes_index")
# ---------------------------------------------------------------------------
# Integration: parse real zensical.toml
# ---------------------------------------------------------------------------
def test_parse_real_zensical():
"""Parse the actual zensical.toml and verify structure."""
if not ZENSICAL_TOML.exists():
print(" SKIP: parse_real_zensical (zensical.toml not found)")
return
all_entries = parse_all(ZENSICAL_TOML)
filtered_entries = parse(ZENSICAL_TOML)
# Sanity checks on totals
assert len(all_entries) > 50, f"Expected 50+ total entries, got {len(all_entries)}"
assert len(filtered_entries) > 40, f"Expected 40+ filtered entries, got {len(filtered_entries)}"
assert len(filtered_entries) < len(all_entries), "Filtering should remove some entries"
# Every filtered entry should NOT be index.md or README.md
for entry in filtered_entries:
assert Path(entry.path).name not in ("index.md", "README.md"), (
f"Filtered list contains {entry.path}"
)
# No external links should be present
for entry in all_entries:
assert not entry.path.startswith("http"), f"External link leaked: {entry.path}"
# Check some known entries exist
paths = {e.path for e in filtered_entries}
assert "getting-started/what-is-sunnypilot.md" in paths, "Missing what-is-sunnypilot"
assert "features/cruise/icbm.md" in paths, "Missing ICBM"
assert "safety/safety.md" in paths, "Missing safety"
# Check breadcrumbs are populated
icbm = next(e for e in filtered_entries if e.path == "features/cruise/icbm.md")
assert len(icbm.breadcrumb) >= 2, f"ICBM breadcrumb too short: {icbm.breadcrumb}"
assert "Features" in icbm.breadcrumb or "Cruise Control" in icbm.breadcrumb
print(f" PASS: parse_real_zensical ({len(all_entries)} total, {len(filtered_entries)} filtered)")
def test_nav_entry_immutable():
"""NavEntry is frozen — attributes cannot be reassigned."""
entry = NavEntry(title="Test", path="test.md", breadcrumb=("Test",))
try:
entry.title = "Modified" # type: ignore[misc]
assert False, "Should have raised FrozenInstanceError"
except AttributeError:
pass
print(" PASS: nav_entry_immutable")
# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("Testing nav parser:")
tests = [
# Unit
test_flatten_simple_dict,
test_flatten_nested,
test_flatten_deep_nesting,
test_flatten_skips_external_links,
test_flatten_bare_string,
test_flatten_mixed,
test_parse_filters_index,
test_parse_all_includes_index,
test_nav_entry_immutable,
# Integration
test_parse_real_zensical,
]
passed = 0
failed = 0
for test in tests:
try:
test()
passed += 1
except AssertionError as e:
print(f" FAIL: {test.__name__}: {e}")
failed += 1
except Exception as e:
print(f" ERROR: {test.__name__}: {e}")
failed += 1
print(f"\n{passed}/{passed + failed} tests passed")
sys.exit(1 if failed > 0 else 0)