engineering/remarkbox
1
0
Fork 0
Code Issues 3 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Operation Undigg: multi-syntax input, pandoc export, wiki mode, auto-themes.

Browse source 
Phase 1: Pandoc export pipeline — 67 output formats for threads, namespaces, nodes.
Phase 2: Multi-syntax input — accept markdown, HTML, RST, MediaWiki, LaTeX, etc.
Phase 3: Wiki mode — per-namespace toggle, revision tracking, wiki-edit endpoint.
Phase 4: Auto-generated themes — deterministic CSS per namespace, light + dark mode.

Includes unit, integration, and functional tests (95 new, 544 total).
This commit is contained in:
russell@unturf.com 2026-03-10 00:35:51 -04:00
parent f45c42d343
commit 6d1cfff91d
23 changed files with 2822 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

29
CLAUDE.md
View file

@ -87,6 +87,26 @@ c.approve_node(node_id)
c.lock_node(node_id)
c.unlock_node(node_id)
c.delete_node(node_id) # permanent, moderator only
# Export (pandoc-powered, 67 output formats)
formats = c.export_formats()
md = c.export_thread(node_id, "markdown")
pdf = c.export_thread(node_id, "pdf")
epub = c.export_namespace("meta.remarkbox.com", "epub")
docx = c.export_node(node_id, "docx")
# Wiki mode (namespace.wiki=True, any authenticated user can edit root nodes)
c.wiki_edit(node_id, data="Updated content") # creates revision first
revisions = c.get_revisions(node_id)
revision = c.get_revision(revision_id)
# Themes (auto-generated per namespace, light + dark mode)
css = c.get_theme_css("meta.remarkbox.com")
preview = c.get_theme_preview("meta.remarkbox.com")
# Multi-syntax input (source_format parameter on create/reply/edit)
c.create_thread(namespace="ns", title="RST", data="Title\n=====\n\nParagraph.", source_format="rst")
c.reply(node_id, data="<p>HTML reply</p>", source_format="html")
```
### Key details
@ -119,6 +139,15 @@ c.delete_node(node_id) # permanent, moderator only
| GET | `/api/v1/clients/c` | Download C client (rb.c) |
| GET | `/api/v1/admin/namespaces` | List all namespaces (superuser only) |
| GET | `/api/v1/admin/recent-nodes?days=7` | Recent nodes network-wide (superuser only) |
| GET | `/api/v1/export/formats` | List available pandoc export formats |
| GET | `/api/v1/export/namespace/{name}.{fmt}` | Export namespace as book |
| GET | `/api/v1/export/threads/{node_id}.{fmt}` | Export thread as document |
| GET | `/api/v1/export/nodes/{node_id}.{fmt}` | Export node subtree (on-demand) |
| GET | `/api/v1/nodes/{id}/revisions` | Revision history for a node |
| POST | `/api/v1/nodes/{id}/wiki-edit` | Wiki-edit a node (creates revision) |
| GET | `/api/v1/revisions/{id}` | Get a specific revision |
| GET | `/api/v1/themes/{namespace}/css` | Auto-generated theme CSS |
| GET | `/api/v1/themes/{namespace}/preview` | Theme palette preview (JSON) |
### Authentication

120
docs/tickets/15.md Normal file
View file

@ -0,0 +1,120 @@
# T15: Operation Undigg — Pandoc Export & Wiki Mode
**Status**: in-progress
**Priority**: high
**Source**: fox directive 2026-03-09
## Summary
Transform remarkbox into a document-first platform. Every namespace is a book,
every root thread is a chapter. Pandoc renders every format it can produce.
Wiki mode lets anyone edit root topics with revision tracking.
## Architecture
```
Namespace (book) → pandoc → all formats (default)
└── Root Node (chapter) → pandoc → all formats (default)
└── Reply (section) → pandoc → on-demand only
```
### Data model changes
1. **Node.source_format** — `Unicode(16)`, default `'markdown'`. Tracks what
syntax the user wrote in (markdown, html, rst, mediawiki, latex, textile, etc).
Pandoc input formats map directly.
2. **rb_revision** — new table for wiki mode edit history.
- `id` (UUIDType, PK)
- `node_id` (FK → Node)
- `user_id` (FK → User)
- `data` (UnicodeText) — raw source at time of edit
- `source_format` (Unicode(16))
- `created` (BigInteger) — timestamp
- `revision_number` (Integer) — sequential per node
3. **Namespace.wiki** — already exists (`Boolean, default=False`), needs
implementation. When True, any authenticated user can edit the root node
of any thread. Each edit creates a revision.
### Export pipeline
Pandoc subprocess. Input = markdown/rst/html/whatever `source_format` says.
Output = every format pandoc supports.
**Default generated (cached on write/edit):**
- Namespace level: one document per namespace containing all root threads
- Root node level: one document per root thread
**On-demand (generated per request):**
- Any node or subthread at any depth
**Pandoc output formats** (67 total, subset for default generation):
- markdown, gfm, commonmark, html5, pdf, epub, docx, odt, rst,
mediawiki, latex, man, plain, rtf, asciidoc, textile, org, json
**URI scheme:**
```
/api/v1/export/namespace/{name}.{format} — full book
/api/v1/export/threads/{node_id}.{format} — single chapter
/api/v1/export/nodes/{node_id}.{format} — on-demand subthread
```
### Content ingestion
On write (create thread, reply, edit):
1. Accept `source_format` parameter (default: `markdown`)
2. If HTML input, strip to clean body (no head/script/style)
3. Store raw source in `Node.data` with `Node.source_format`
4. Render to HTML via pandoc: `pandoc -f {source_format} -t html5`
5. Sanitize HTML output through existing bleach pipeline
6. Store in `Node.data_html`
### Wiki mode
When `Namespace.wiki = True`:
- Any authenticated user can edit root nodes (not just owner/moderator)
- Each edit stores a revision in `rb_revision` before overwriting
- Root node always has the latest version (fast render)
- Revision history accessible via API
- Diff between revisions on-demand
### Progressive enhancement
Export menus work without JS (plain links to format URIs).
With JS: dropdown/popover with format picker, async download.
## Phases
### Phase 1: Export pipeline (pandoc integration)
- [ ] Tree-to-markdown renderer (walk node tree → single markdown document)
- [ ] Export API endpoints (namespace, thread, node)
- [ ] Pandoc subprocess wrapper
- [ ] Format negotiation (URI suffix or Accept header)
### Phase 2: Multi-syntax input
- [ ] Add `source_format` column to Node
- [ ] Alembic migration
- [ ] Modify `set_data()` to use pandoc for non-markdown formats
- [ ] Accept `source_format` on create/reply/edit API endpoints
- [ ] HTML stripping for HTML input
### Phase 3: Wiki mode + revisions
- [ ] Create `rb_revision` model + migration
- [ ] Implement wiki edit permissions in Namespace
- [ ] Store revisions on edit
- [ ] Revision history API endpoint
- [ ] Diff endpoint
### Phase 4: Auto-generated themes
- [ ] Per-namespace theme generation (light + dark)
- [ ] CSS custom properties for theming
- [ ] Theme preview
## Notes
- Pandoc 3.1.3 installed at `/usr/bin/pandoc`
- 43 input formats, 67 output formats
- PDF requires LaTeX (`pdflatex`) or `wkhtmltopdf` — check availability
- `Namespace.wiki` column already exists in schema, just needs implementation
- Export cache invalidation: on any node edit in the tree

1
docs/tickets/index.md
View file

@ -19,3 +19,4 @@ Tracked issues from the meta.remarkbox.com and faq.remarkbox.com audit (2026-02-
| [T12](12.md) | Reply to API-only CRUD thread confirming done | resolved | low | meta `6db01560` |
| [T13](13.md) | Reply to lock/archive thread confirming done | resolved | low | meta `7e9d5864` |
| [T14](14.md) | meta/faq SSL outage — missing proxy blocks | resolved | critical | postmortem 2026-02-25 |
| [T15](15.md) | Operation Undigg — Pandoc export & wiki mode | in-progress | high | fox directive 2026-03-09 |

15
remarkbox/api/__init__.py
View file

@ -4,7 +4,11 @@ def includeme(config):
config.add_route("api-threads-search", "/api/v1/threads/search")
config.add_route("api-thread-detail", "/api/v1/threads/{node_id}")
config.add_route("api-thread-replies", "/api/v1/threads/{node_id}/replies")
# Wiki / revision routes (must come before api-node-detail catch-all)
config.add_route("api-node-revisions", "/api/v1/nodes/{node_id}/revisions")
config.add_route("api-node-wiki-edit", "/api/v1/nodes/{node_id}/wiki-edit")
config.add_route("api-node-detail", "/api/v1/nodes/{node_id}")
config.add_route("api-revision-detail", "/api/v1/revisions/{revision_id}")
config.add_route("api-auth-login", "/api/v1/auth/login")
config.add_route("api-auth-verify", "/api/v1/auth/verify")
config.add_route("api-user-profile", "/api/v1/user/profile")
@ -13,4 +17,15 @@ def includeme(config):
config.add_route("api-webmention", "/api/v1/webmention")
config.add_route("api-admin-namespaces", "/api/v1/admin/namespaces")
config.add_route("api-admin-recent-nodes", "/api/v1/admin/recent-nodes")
# Export routes — {subpath} captures "name.format" or "node_id.format"
config.add_route("api-export-formats", "/api/v1/export/formats")
config.add_route("api-export-namespace", "/api/v1/export/namespace/{subpath}")
config.add_route("api-export-thread", "/api/v1/export/threads/{subpath}")
config.add_route("api-export-node", "/api/v1/export/nodes/{subpath}")
# Theme routes
config.add_route("api-namespace-theme", "/api/v1/themes/{namespace_name}/css")
config.add_route("api-theme-preview", "/api/v1/themes/{namespace_name}/preview")
config.scan("remarkbox.api.views")
config.scan("remarkbox.api.export")
config.scan("remarkbox.api.wiki")
config.scan("remarkbox.api.themes")

302
remarkbox/api/export.py Normal file
View file

@ -0,0 +1,302 @@
"""Export API — render threads and namespaces via pandoc.
URI scheme:
GET /api/v1/export/namespace/{name}.{format} — full book
GET /api/v1/export/threads/{node_id}.{format} — single chapter
GET /api/v1/export/nodes/{node_id}.{format} — on-demand subthread
GET /api/v1/export/formats — list available formats
"""
import logging
import subprocess
from pyramid.response import Response
from pyramid.view import view_config
from remarkbox.models.namespace import get_namespace_by_name
from remarkbox.models.node import (
get_node_by_id,
get_nodes_who_share_root,
)
from remarkbox.lib.pandoc import (
convert,
node_tree_to_markdown,
namespace_to_markdown,
get_available_output_formats,
CONTENT_TYPES,
FILE_EXTENSIONS,
BINARY_FORMATS,
)
from .views import check_namespace_api_access
log = logging.getLogger(__name__)
def _parse_format_from_subpath(subpath):
"""Extract format from the subpath (e.g. 'abc-123.epub' -> ('abc-123', 'epub'))."""
if "." in subpath:
parts = subpath.rsplit(".", 1)
return parts[0], parts[1]
return subpath, "markdown"
def _export_response(content, to_format, filename_stem):
"""Build a Response for exported content."""
content_type = CONTENT_TYPES.get(to_format, "application/octet-stream")
ext = FILE_EXTENSIONS.get(to_format, "")
filename = "{}{}".format(filename_stem, ext)
is_binary = isinstance(content, bytes)
response = Response(
body=content if is_binary else content.encode("utf-8"),
content_type=content_type,
)
response.content_disposition = 'attachment; filename="{}"'.format(filename)
return response
# ---------------------------------------------------------------------------
# List formats
# ---------------------------------------------------------------------------
@view_config(
route_name="api-export-formats",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_export_formats(request):
"""List all available export formats."""
available = sorted(get_available_output_formats())
return {
"formats": available,
"count": len(available),
}
# ---------------------------------------------------------------------------
# Export thread (single root node = chapter)
# ---------------------------------------------------------------------------
@view_config(
route_name="api-export-thread",
request_method="GET",
require_csrf=False,
)
def api_export_thread(request):
"""Export a single thread (root node + all replies) in any pandoc format."""
subpath = request.matchdict["subpath"]
node_id, to_format = _parse_format_from_subpath(subpath)
available = get_available_output_formats()
if to_format not in available:
request.response.status_code = 400
request.response.content_type = "application/json"
request.response.json_body = {
"error": "Unsupported format: {}".format(to_format),
"available": sorted(available),
}
return request.response
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
request.response.content_type = "application/json"
request.response.json_body = {"error": "Thread not found"}
return request.response
root = node if node.is_root else node.root
namespace = root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
request.response.content_type = "application/json"
request.response.json_body = denied
return request.response
# Fetch all visible nodes in the thread
nodes = get_nodes_who_share_root(
request.dbsession, root,
exclude_root=True,
visibility_filters={"disabled": False},
).all()
# Render tree to markdown
md = node_tree_to_markdown(root, nodes)
title = root.title or str(root.id)
if to_format in ("markdown", "gfm", "commonmark"):
# Short-circuit: already markdown, no pandoc needed
return _export_response(md, to_format, root.slug or str(root.id))
try:
output = convert(md, from_format="markdown", to_format=to_format, title=title)
except subprocess.CalledProcessError as e:
log.error("Pandoc conversion failed: %s", e.stderr)
request.response.status_code = 500
request.response.content_type = "application/json"
request.response.json_body = {
"error": "Conversion failed",
"detail": e.stderr[:500] if e.stderr else "unknown error",
}
return request.response
return _export_response(output, to_format, root.slug or str(root.id))
# ---------------------------------------------------------------------------
# Export namespace (full book)
# ---------------------------------------------------------------------------
@view_config(
route_name="api-export-namespace",
request_method="GET",
require_csrf=False,
)
def api_export_namespace(request):
"""Export an entire namespace as a book in any pandoc format."""
subpath = request.matchdict["subpath"]
name, to_format = _parse_format_from_subpath(subpath)
available = get_available_output_formats()
if to_format not in available:
request.response.status_code = 400
request.response.content_type = "application/json"
request.response.json_body = {
"error": "Unsupported format: {}".format(to_format),
"available": sorted(available),
}
return request.response
namespace = get_namespace_by_name(request.dbsession, name)
if namespace is None:
request.response.status_code = 404
request.response.content_type = "application/json"
request.response.json_body = {"error": "Namespace not found"}
return request.response
denied = check_namespace_api_access(request, namespace)
if denied:
request.response.content_type = "application/json"
request.response.json_body = denied
return request.response
roots = namespace.visible_roots.all()
def node_fetcher(root):
return get_nodes_who_share_root(
request.dbsession, root,
exclude_root=True,
visibility_filters={"disabled": False},
).all()
md = namespace_to_markdown(namespace, roots, node_fetcher)
title = namespace.description or namespace.name
if to_format in ("markdown", "gfm", "commonmark"):
return _export_response(md, to_format, name)
try:
output = convert(md, from_format="markdown", to_format=to_format, title=title)
except subprocess.CalledProcessError as e:
log.error("Pandoc conversion failed: %s", e.stderr)
request.response.status_code = 500
request.response.content_type = "application/json"
request.response.json_body = {
"error": "Conversion failed",
"detail": e.stderr[:500] if e.stderr else "unknown error",
}
return request.response
return _export_response(output, to_format, name)
# ---------------------------------------------------------------------------
# Export node (on-demand subthread at any depth)
# ---------------------------------------------------------------------------
@view_config(
route_name="api-export-node",
request_method="GET",
require_csrf=False,
)
def api_export_node(request):
"""Export a node and its subtree in any pandoc format (on-demand)."""
subpath = request.matchdict["subpath"]
node_id, to_format = _parse_format_from_subpath(subpath)
available = get_available_output_formats()
if to_format not in available:
request.response.status_code = 400
request.response.content_type = "application/json"
request.response.json_body = {
"error": "Unsupported format: {}".format(to_format),
"available": sorted(available),
}
return request.response
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
request.response.content_type = "application/json"
request.response.json_body = {"error": "Node not found"}
return request.response
namespace = node.root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
request.response.content_type = "application/json"
request.response.json_body = denied
return request.response
# For a non-root node, we need to get its subtree.
# Fetch all nodes in the root's tree, then filter to descendants.
from remarkbox.models.node import get_graph_from_nodes, flatten_graph
all_nodes = get_nodes_who_share_root(
request.dbsession, node.root,
visibility_filters={"disabled": False},
).all()
# Include root so the graph is complete
all_with_root = [node.root] + list(all_nodes) if not node.is_root else list(all_nodes)
graph = get_graph_from_nodes(all_with_root)
# Get descendant IDs of the target node
if node.id in graph:
descendant_ids = set(flatten_graph(node.id, graph, include_given_node_id=False))
else:
descendant_ids = set()
# Filter to just the subtree
subtree_nodes = [n for n in all_nodes if n.id in descendant_ids]
md = node_tree_to_markdown(node, subtree_nodes, include_root=True)
title = node.title or "Thread {}".format(str(node.id)[:8])
if to_format in ("markdown", "gfm", "commonmark"):
slug = node.slug or str(node.id)
return _export_response(md, to_format, slug)
try:
output = convert(md, from_format="markdown", to_format=to_format, title=title)
except subprocess.CalledProcessError as e:
log.error("Pandoc conversion failed: %s", e.stderr)
request.response.status_code = 500
request.response.content_type = "application/json"
request.response.json_body = {
"error": "Conversion failed",
"detail": e.stderr[:500] if e.stderr else "unknown error",
}
return request.response
slug = node.slug or str(node.id)
return _export_response(output, to_format, slug)

196
remarkbox/api/remarkbox_client.py
View file

@ -225,15 +225,18 @@ class RemarkboxClient:
path += "?" + urllib.parse.urlencode(params)
return self._request("GET", path)
def create_thread(self, namespace, title, data, anonymous_name=None, email=None):
def create_thread(self, namespace, title, data, anonymous_name=None, email=None,
source_format=None):
"""Create a new thread.
Args:
namespace: Target namespace name
title: Thread title
data: Markdown content (max 500000 chars)
data: Content (max 500000 chars)
anonymous_name: Name for anonymous posting (optional)
email: Email to associate with post (optional)
source_format: Input format — markdown (default), html, rst,
mediawiki, latex, textile, org, etc. (any pandoc input format)
Returns:
dict with keys: node, verified
@ -243,18 +246,21 @@ class RemarkboxClient:
body["anonymous_name"] = anonymous_name
if email:
body["email"] = email
if source_format:
body["source_format"] = source_format
return self._request("POST", "/api/v1/threads", body)
# ----- Replies -----
def reply(self, node_id, data, anonymous_name=None, email=None):
def reply(self, node_id, data, anonymous_name=None, email=None, source_format=None):
"""Reply to a thread or another reply.
Args:
node_id: UUID of the parent node (thread or reply)
data: Markdown content (max 500000 chars)
data: Content (max 500000 chars)
anonymous_name: Name for anonymous posting (optional)
email: Email to associate with post (optional)
source_format: Input format (default: markdown)
Returns:
dict with keys: node, verified
@ -264,6 +270,8 @@ class RemarkboxClient:
body["anonymous_name"] = anonymous_name
if email:
body["email"] = email
if source_format:
body["source_format"] = source_format
return self._request("POST", "/api/v1/threads/{}/replies".format(node_id), body)
# ----- Nodes -----
@ -279,13 +287,14 @@ class RemarkboxClient:
"""
return self._request("GET", "/api/v1/nodes/{}".format(node_id))
def edit_node(self, node_id, data=None, title=None):
def edit_node(self, node_id, data=None, title=None, source_format=None):
"""Edit a node (requires authentication).
Args:
node_id: UUID of the node to edit
data: New markdown content (optional)
data: New content (optional)
title: New title, only for root nodes (optional)
source_format: Input format (default: markdown)
Returns:
dict with key: node
@ -295,6 +304,8 @@ class RemarkboxClient:
body["data"] = data
if title is not None:
body["title"] = title
if source_format is not None:
body["source_format"] = source_format
if not body:
raise ValueError("data or title is required")
return self._request("PATCH", "/api/v1/nodes/{}".format(node_id), body)
@ -431,6 +442,145 @@ class RemarkboxClient:
"""
return self._request("PATCH", "/api/v1/user/profile", {"name": name})
# ----- Export -----
def export_formats(self):
"""List all available export formats.
Returns:
dict with keys: formats (list of strings), count
"""
return self._request("GET", "/api/v1/export/formats")
def export_thread(self, node_id, fmt="markdown"):
"""Export a thread in the specified format.
Args:
node_id: UUID of the root thread node
fmt: Pandoc output format (e.g. markdown, html5, pdf, epub, docx)
Returns:
bytes (binary formats) or str (text formats)
"""
return self._raw_request(
"GET", "/api/v1/export/threads/{}.{}".format(node_id, fmt)
)
def export_namespace(self, namespace, fmt="markdown"):
"""Export an entire namespace as a book in the specified format.
Args:
namespace: Namespace name (e.g. "meta.remarkbox.com")
fmt: Pandoc output format
Returns:
bytes (binary formats) or str (text formats)
"""
return self._raw_request(
"GET", "/api/v1/export/namespace/{}.{}".format(namespace, fmt)
)
def export_node(self, node_id, fmt="markdown"):
"""Export a node and its subtree in the specified format (on-demand).
Args:
node_id: UUID of the node
fmt: Pandoc output format
Returns:
bytes (binary formats) or str (text formats)
"""
return self._raw_request(
"GET", "/api/v1/export/nodes/{}.{}".format(node_id, fmt)
)
def _raw_request(self, method, path):
"""Make an HTTP request and return raw response body."""
url = self.url + path
req = urllib.request.Request(url, method=method)
try:
resp = self._opener.open(req)
content_type = resp.headers.get("Content-Type", "")
body = resp.read()
if "text/" in content_type or "json" in content_type or "xml" in content_type:
return body.decode("utf-8")
return body
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8")
try:
body = json.loads(raw)
except Exception:
body = {"error": raw}
raise RemarkboxError(e.code, body)
# ----- Wiki / Revisions -----
def wiki_edit(self, node_id, data, source_format=None):
"""Wiki-edit a node (creates revision, any authenticated user if wiki mode).
Args:
node_id: UUID of the node to edit
data: New content
source_format: Input format (default: markdown)
Returns:
dict with key: node
"""
body = {"data": data}
if source_format:
body["source_format"] = source_format
return self._request("POST", "/api/v1/nodes/{}/wiki-edit".format(node_id), body)
def get_revisions(self, node_id):
"""Get revision history for a node.
Args:
node_id: UUID of the node
Returns:
dict with keys: node_id, revisions, count
"""
return self._request("GET", "/api/v1/nodes/{}/revisions".format(node_id))
def get_revision(self, revision_id):
"""Get a specific revision by ID.
Args:
revision_id: UUID of the revision
Returns:
dict with key: revision
"""
return self._request("GET", "/api/v1/revisions/{}".format(revision_id))
# ----- Themes -----
def get_theme_css(self, namespace):
"""Get the auto-generated theme CSS for a namespace.
Args:
namespace: Namespace name
Returns:
CSS string
"""
return self._raw_request(
"GET", "/api/v1/themes/{}/css".format(namespace)
)
def get_theme_preview(self, namespace):
"""Get theme palette preview for a namespace.
Args:
namespace: Namespace name
Returns:
dict with keys: namespace, palette, css_url
"""
return self._request(
"GET", "/api/v1/themes/{}/preview".format(namespace)
)
# ----- Admin (superuser only) -----
def admin_list_namespaces(self):
@ -477,11 +627,15 @@ Commands:
delete <node_id> Delete a node (moderator only)
login <email> Request OTP
verify <email> <otp> Verify OTP
formats List export formats
export-thread <node_id> [format] Export thread (default: markdown)
export-ns <namespace> [format] Export namespace as book
export-node <node_id> [format] Export node subtree
Examples:
python remarkbox_client.py https://my.remarkbox.com threads meta.remarkbox.com
python remarkbox_client.py https://my.remarkbox.com post meta.remarkbox.com "Hello" "World" MyBot
python remarkbox_client.py https://my.remarkbox.com disable <node_id>
python remarkbox_client.py https://my.remarkbox.com export-ns meta.remarkbox.com epub
python remarkbox_client.py https://my.remarkbox.com export-thread <node_id> pdf
"""
if len(sys.argv) < 3:
@ -522,6 +676,32 @@ Examples:
result = client.login(args[0])
elif cmd == "verify" and len(args) >= 2:
result = client.verify(args[0], args[1])
elif cmd == "formats":
result = client.export_formats()
elif cmd == "export-thread" and len(args) >= 1:
fmt = args[1] if len(args) > 1 else "markdown"
output = client.export_thread(args[0], fmt)
if isinstance(output, bytes):
sys.stdout.buffer.write(output)
else:
print(output)
sys.exit(0)
elif cmd == "export-ns" and len(args) >= 1:
fmt = args[1] if len(args) > 1 else "markdown"
output = client.export_namespace(args[0], fmt)
if isinstance(output, bytes):
sys.stdout.buffer.write(output)
else:
print(output)
sys.exit(0)
elif cmd == "export-node" and len(args) >= 1:
fmt = args[1] if len(args) > 1 else "markdown"
output = client.export_node(args[0], fmt)
if isinstance(output, bytes):
sys.stdout.buffer.write(output)
else:
print(output)
sys.exit(0)
else:
print(usage)
sys.exit(1)

1
remarkbox/api/serializers.py
View file

@ -7,6 +7,7 @@ def serialize_node(node, include_children=False):
"title": node.title,
"data": node.data,
"data_html": node.data_html,
"source_format": node.source_format,
"is_root": node.is_root,
"depth": node.graph_depth,
"created": node.created,

80
remarkbox/api/themes.py Normal file
View file

@ -0,0 +1,80 @@
"""Theme generation API endpoints."""
from pyramid.response import Response
from pyramid.view import view_config
from remarkbox.models.namespace import get_namespace_by_name
from remarkbox.lib.theme_generator import generate_theme_css
@view_config(
route_name="api-namespace-theme",
request_method="GET",
require_csrf=False,
)
def api_namespace_theme(request):
"""Generate and serve a theme CSS for a namespace.
The theme is deterministic -- same namespace always gets the same theme.
Cache-friendly: can be cached indefinitely (changes only if we change the algorithm).
"""
namespace_name = request.matchdict["namespace_name"]
# Validate namespace exists
namespace = get_namespace_by_name(request.dbsession, namespace_name)
if namespace is None:
request.response.status_code = 404
request.response.content_type = "application/json"
request.response.json_body = {"error": "Namespace not found"}
return request.response
css = generate_theme_css(namespace_name)
response = Response(
body=css,
content_type="text/css; charset=utf-8",
)
# Cache for 1 day -- theme is deterministic but we might update the algorithm
response.cache_control.max_age = 86400
response.cache_control.public = True
return response
@view_config(
route_name="api-theme-preview",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_theme_preview(request):
"""Preview theme variables for a namespace (JSON format).
Useful for theme customization UI -- shows the computed palette
without needing to parse CSS.
"""
namespace_name = request.matchdict["namespace_name"]
namespace = get_namespace_by_name(request.dbsession, namespace_name)
if namespace is None:
request.response.status_code = 404
return {"error": "Namespace not found"}
from remarkbox.lib.theme_generator import _name_to_seed
seed = _name_to_seed(namespace_name)
hue = seed[0] % 360
hue_offset = 30 + (seed[1] % 30)
secondary_hue = (hue + hue_offset) % 360
accent_hue = (hue + 180 + (seed[2] % 40 - 20)) % 360
sat_base = 40 + (seed[3] % 25)
return {
"namespace": namespace_name,
"palette": {
"primary_hue": hue,
"secondary_hue": secondary_hue,
"accent_hue": accent_hue,
"saturation_base": sat_base,
},
"css_url": "/api/v1/themes/{}/css".format(namespace_name),
}

19
remarkbox/api/views.py
View file

@ -29,6 +29,7 @@ from remarkbox.models.sudo_otp import create_sudo_otp, verify_sudo_otp
from remarkbox.lib.notify import schedule_notifications
from remarkbox.views import verify_pending_nodes_in_session
from remarkbox.models.meta import now_timestamp
from remarkbox.models.spam import score_content
from remarkbox.models.spam_llm import check_thread_relevance, check_reply_relevance
@ -473,6 +474,8 @@ def api_create_thread(request):
body.get("anonymous_name") or request.params.get("anonymous_name", "")
).strip()
email = body.get("email") or request.params.get("email", "")
source_format = body.get("source_format") or request.params.get("source_format", "")
source_format = source_format.strip() if source_format else None
if not namespace_name:
request.response.status_code = 400
@ -538,7 +541,8 @@ def api_create_thread(request):
# Create the comment as a child of the root (same as api_reply)
node = root.new_child()
node.ip_address = str(request.client_addr)
node.set_data(data, namespace=namespace, dbsession=request.dbsession)
node.set_data(data, namespace=namespace, dbsession=request.dbsession,
source_format=source_format)
if user_surrogate:
node.user_surrogate = user_surrogate
@ -583,7 +587,7 @@ def api_create_thread(request):
node.namespace = namespace
node.ip_address = str(request.client_addr)
node.title = title
node.set_data(data, dbsession=request.dbsession)
node.set_data(data, dbsession=request.dbsession, source_format=source_format)
if user_surrogate:
node.user_surrogate = user_surrogate
@ -661,6 +665,8 @@ def api_reply(request):
body.get("anonymous_name") or request.params.get("anonymous_name", "")
).strip()
email = body.get("email") or request.params.get("email", "")
source_format = body.get("source_format") or request.params.get("source_format", "")
source_format = source_format.strip() if source_format else None
if not data:
request.response.status_code = 400
@ -706,7 +712,8 @@ def api_reply(request):
# Create child node
child = parent.new_child()
child.ip_address = str(request.client_addr)
child.set_data(data, namespace=namespace, dbsession=request.dbsession)
child.set_data(data, namespace=namespace, dbsession=request.dbsession,
source_format=source_format)
if user_surrogate:
child.user_surrogate = user_surrogate
@ -812,6 +819,8 @@ def api_edit_node(request):
body = get_json_body(request)
data = body.get("data") or request.params.get("thread_data", "")
title = body.get("title") or request.params.get("thread_title", "")
source_format = body.get("source_format") or request.params.get("source_format", "")
source_format = source_format.strip() if source_format else None
# Moderation flags (require can_alter_node, already checked above)
disabled = body.get("disabled")
@ -837,7 +846,9 @@ def api_edit_node(request):
if title and node.is_root:
node.title = title
if data:
node.edit(data)
node.set_data(data, source_format=source_format)
node.changed = now_timestamp()
node._invalidate_cache()
if disabled is True:
node.disable()

134
remarkbox/api/wiki.py Normal file
View file

@ -0,0 +1,134 @@
"""Wiki mode and revision history API endpoints."""
from pyramid.view import view_config
from remarkbox.models.node import get_node_by_id, Node
from remarkbox.models.revision import Revision
from .views import check_namespace_api_access, get_json_body, MAX_CONTENT_LENGTH
from .serializers import serialize_node
from remarkbox.models.meta import get_object_by_id
def serialize_revision(rev):
"""Serialize a Revision to a dict."""
return {
"id": str(rev.id),
"node_id": str(rev.node_id),
"user_id": str(rev.user_id) if rev.user_id else None,
"author": rev.user.name if rev.user else None,
"data": rev.data,
"source_format": rev.source_format,
"revision_number": rev.revision_number,
"created": rev.created,
}
@view_config(
route_name="api-node-revisions",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_node_revisions(request):
"""Get revision history for a node."""
node_id = request.matchdict["node_id"]
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
return {"error": "Node not found"}
namespace = node.root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
revisions = (
request.dbsession.query(Revision)
.filter(Revision.node_id == node.id)
.order_by(Revision.revision_number.desc())
.all()
)
return {
"node_id": str(node.id),
"revisions": [serialize_revision(r) for r in revisions],
"count": len(revisions),
}
@view_config(
route_name="api-node-wiki-edit",
request_method="POST",
renderer="json",
require_csrf=False,
)
def api_wiki_edit(request):
"""Wiki-edit a node (creates a revision, any authenticated user if wiki mode)."""
node_id = request.matchdict["node_id"]
node = get_node_by_id(request.dbsession, node_id)
if node is None:
request.response.status_code = 404
return {"error": "Node not found"}
if not request.user or not request.user.authenticated:
request.response.status_code = 401
return {"error": "Authentication required"}
namespace = node.root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
if not namespace.can_wiki_edit(node, request.user):
request.response.status_code = 403
return {"error": "Wiki editing not permitted for this node"}
body = get_json_body(request)
data = body.get("data", "")
if not data:
request.response.status_code = 400
return {"error": "data is required"}
if len(data) > MAX_CONTENT_LENGTH:
request.response.status_code = 400
return {"error": "data exceeds maximum length"}
source_format = body.get("source_format")
node.wiki_edit(data, user=request.user, source_format=source_format)
request.dbsession.add(node)
request.dbsession.flush()
return {"node": serialize_node(node)}
@view_config(
route_name="api-revision-detail",
request_method="GET",
renderer="json",
require_csrf=False,
)
def api_get_revision(request):
"""Get a specific revision by ID."""
revision_id = request.matchdict["revision_id"]
revision = get_object_by_id(request.dbsession, revision_id, Revision)
if revision is None:
request.response.status_code = 404
return {"error": "Revision not found"}
node = get_node_by_id(request.dbsession, revision.node_id)
if node:
namespace = node.root.namespace
denied = check_namespace_api_access(request, namespace)
if denied:
return denied
return {"revision": serialize_revision(revision)}

324
remarkbox/lib/pandoc.py Normal file
View file

@ -0,0 +1,324 @@
"""Pandoc integration for remarkbox.
Converts node trees to documents in any format pandoc supports.
Subprocess-based — no pip dependencies.
"""
import logging
import subprocess
import tempfile
import os
log = logging.getLogger(__name__)
# Formats that produce binary output (need file-based output, not stdout text).
BINARY_FORMATS = frozenset({
"pdf", "docx", "odt", "epub", "epub2", "epub3", "pptx", "fb2",
})
# Formats we generate by default for namespace/thread-level export.
DEFAULT_FORMATS = [
"markdown", "gfm", "commonmark", "html5", "rst",
"mediawiki", "latex", "man", "plain", "rtf",
"asciidoc", "textile", "org", "json",
"epub", "docx", "odt", "pdf",
]
# Content types for HTTP responses.
CONTENT_TYPES = {
"markdown": "text/markdown; charset=utf-8",
"gfm": "text/markdown; charset=utf-8",
"commonmark": "text/markdown; charset=utf-8",
"commonmark_x": "text/markdown; charset=utf-8",
"html": "text/html; charset=utf-8",
"html5": "text/html; charset=utf-8",
"html4": "text/html; charset=utf-8",
"rst": "text/x-rst; charset=utf-8",
"latex": "application/x-latex; charset=utf-8",
"beamer": "application/x-latex; charset=utf-8",
"man": "text/troff; charset=utf-8",
"plain": "text/plain; charset=utf-8",
"json": "application/json; charset=utf-8",
"mediawiki": "text/plain; charset=utf-8",
"asciidoc": "text/plain; charset=utf-8",
"asciidoctor": "text/plain; charset=utf-8",
"textile": "text/plain; charset=utf-8",
"org": "text/plain; charset=utf-8",
"rtf": "application/rtf",
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"odt": "application/vnd.oasis.opendocument.text",
"epub": "application/epub+zip",
"epub2": "application/epub+zip",
"epub3": "application/epub+zip",
"pdf": "application/pdf",
"pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
"fb2": "application/xml",
"jira": "text/plain; charset=utf-8",
"dokuwiki": "text/plain; charset=utf-8",
"xwiki": "text/plain; charset=utf-8",
"zimwiki": "text/plain; charset=utf-8",
"typst": "text/plain; charset=utf-8",
"context": "text/plain; charset=utf-8",
"texinfo": "text/plain; charset=utf-8",
"opml": "application/xml; charset=utf-8",
"tei": "application/xml; charset=utf-8",
"docbook": "application/xml; charset=utf-8",
"docbook5": "application/xml; charset=utf-8",
"icml": "application/xml; charset=utf-8",
}
# File extensions for download filenames.
FILE_EXTENSIONS = {
"markdown": ".md",
"gfm": ".md",
"commonmark": ".md",
"commonmark_x": ".md",
"html": ".html",
"html5": ".html",
"html4": ".html",
"rst": ".rst",
"latex": ".tex",
"beamer": ".tex",
"man": ".1",
"plain": ".txt",
"json": ".json",
"mediawiki": ".wiki",
"asciidoc": ".adoc",
"asciidoctor": ".adoc",
"textile": ".textile",
"org": ".org",
"rtf": ".rtf",
"docx": ".docx",
"odt": ".odt",
"epub": ".epub",
"epub2": ".epub",
"epub3": ".epub",
"pdf": ".pdf",
"pptx": ".pptx",
"fb2": ".fb2",
"jira": ".jira",
"dokuwiki": ".txt",
"xwiki": ".txt",
"zimwiki": ".txt",
"typst": ".typ",
"context": ".tex",
"texinfo": ".texi",
"opml": ".opml",
"tei": ".xml",
"docbook": ".xml",
"docbook5": ".xml",
"icml": ".icml",
}
def get_available_output_formats():
"""Return the set of output formats pandoc supports on this system."""
try:
result = subprocess.run(
["pandoc", "--list-output-formats"],
capture_output=True, text=True, timeout=5,
)
return frozenset(result.stdout.strip().split("\n"))
except Exception:
log.exception("Failed to query pandoc output formats")
return frozenset()
def get_available_input_formats():
"""Return the set of input formats pandoc supports on this system."""
try:
result = subprocess.run(
["pandoc", "--list-input-formats"],
capture_output=True, text=True, timeout=5,
)
return frozenset(result.stdout.strip().split("\n"))
except Exception:
log.exception("Failed to query pandoc input formats")
return frozenset()
def convert(source, from_format="markdown", to_format="html5", title=None):
"""Convert source text from one format to another via pandoc.
Args:
source: Input text.
from_format: Pandoc input format name.
to_format: Pandoc output format name.
title: Optional document title (sets pandoc metadata).
Returns:
bytes for binary formats (pdf, docx, etc.), str for text formats.
Raises:
subprocess.CalledProcessError on pandoc failure.
ValueError if format is not supported.
"""
is_binary = to_format in BINARY_FORMATS
cmd = ["pandoc", "-f", from_format, "-t", to_format, "--standalone"]
if title:
cmd.extend(["--metadata", "title={}".format(title)])
# PDF needs explicit engine since no pdflatex.
if to_format == "pdf":
cmd.extend(["--pdf-engine=wkhtmltopdf"])
if is_binary:
# Binary formats need file output.
with tempfile.NamedTemporaryFile(
suffix=FILE_EXTENSIONS.get(to_format, ""), delete=False
) as tmp:
tmp_path = tmp.name
try:
cmd.extend(["-o", tmp_path])
subprocess.run(
cmd, input=source, text=True,
capture_output=True, timeout=60, check=True,
)
with open(tmp_path, "rb") as f:
return f.read()
finally:
try:
os.unlink(tmp_path)
except OSError:
pass
else:
result = subprocess.run(
cmd, input=source, text=True,
capture_output=True, timeout=30, check=True,
)
return result.stdout
def node_tree_to_markdown(root_node, nodes, include_root=True):
"""Render a node tree as a nested markdown document.
Args:
root_node: The root Node object.
nodes: Iterable of all Node objects in the tree (flat, any order).
include_root: Whether to include the root node content.
Returns:
Markdown string with the full tree rendered as a document.
"""
# Build lookup: parent_id -> sorted children
children_map = {}
node_map = {}
for node in nodes:
node_map[node.id] = node
pid = node.parent_id
if pid not in children_map:
children_map[pid] = []
children_map[pid].append(node)
# Include root in the map
if include_root and root_node.id not in node_map:
node_map[root_node.id] = root_node
# Sort children by created timestamp
for pid in children_map:
children_map[pid].sort(key=lambda n: n.created)
lines = []
if include_root and root_node.title:
lines.append("# {}".format(root_node.title))
lines.append("")
if include_root and root_node.data:
lines.append(root_node.data)
lines.append("")
def _render_children(parent_id, depth):
for child in children_map.get(parent_id, []):
if child.disabled:
continue
# Heading level based on depth (h2 for first-level replies, etc.)
heading_level = min(depth + 2, 6)
author = _get_author_name(child)
date = child.created_date or ""
lines.append("{} {} — {}".format("#" * heading_level, author, date))
lines.append("")
if child.data:
lines.append(child.data)
lines.append("")
_render_children(child.id, depth + 1)
_render_children(root_node.id, 0)
return "\n".join(lines)
def namespace_to_markdown(namespace, roots, node_fetcher):
"""Render an entire namespace as a markdown book.
Args:
namespace: The Namespace object.
roots: Iterable of root Node objects (chapters).
node_fetcher: Callable(root_node) -> list of all nodes in that tree.
Returns:
Markdown string with the full namespace as a document.
"""
lines = []
# Book title
title = namespace.description or namespace.name
lines.append("# {}".format(title))
lines.append("")
for root in roots:
if root.disabled:
continue
# Chapter heading
chapter_title = root.title or str(root.id)
lines.append("## {}".format(chapter_title))
lines.append("")
if root.data:
lines.append(root.data)
lines.append("")
# Fetch all nodes in this thread
thread_nodes = node_fetcher(root)
# Build children map for this thread
children_map = {}
for node in thread_nodes:
pid = node.parent_id
if pid not in children_map:
children_map[pid] = []
children_map[pid].append(node)
for pid in children_map:
children_map[pid].sort(key=lambda n: n.created)
def _render(parent_id, depth):
for child in children_map.get(parent_id, []):
if child.disabled:
continue
heading_level = min(depth + 3, 6)
author = _get_author_name(child)
date = child.created_date or ""
lines.append("{} {} — {}".format(
"#" * heading_level, author, date
))
lines.append("")
if child.data:
lines.append(child.data)
lines.append("")
_render(child.id, depth + 1)
_render(root.id, 0)
return "\n".join(lines)
def _get_author_name(node):
"""Extract display name from a node's user or surrogate."""
if node.user:
return node.user.name
if node.user_surrogate:
return node.user_surrogate.name
return "Anonymous"

199
remarkbox/lib/theme_generator.py Normal file
View file

@ -0,0 +1,199 @@
"""Auto-generate unique themes per namespace.
Deterministic: same namespace name always produces the same theme.
Uses CSS custom properties with prefers-color-scheme for light/dark mode.
"""
import hashlib
import struct
def _name_to_hue(name):
"""Deterministically map a namespace name to a hue (0-360)."""
h = hashlib.sha256(name.encode("utf-8")).digest()
return struct.unpack(">H", h[:2])[0] % 360
def _name_to_seed(name):
"""Get multiple deterministic values from a name."""
h = hashlib.sha256(name.encode("utf-8")).digest()
values = struct.unpack(">8H", h[:16])
return values
def generate_theme_css(namespace_name):
"""Generate a complete CSS theme for a namespace.
Args:
namespace_name: The namespace name (e.g. "meta.remarkbox.com")
Returns:
CSS string with custom properties for light and dark modes.
"""
seed = _name_to_seed(namespace_name)
# Primary hue from name
hue = seed[0] % 360
# Secondary hue offset (analogous or complementary)
hue_offset = 30 + (seed[1] % 30) # 30-60 degree offset
secondary_hue = (hue + hue_offset) % 360
# Accent hue
accent_hue = (hue + 180 + (seed[2] % 40 - 20)) % 360 # near-complementary
# Saturation variance
sat_base = 40 + (seed[3] % 25) # 40-65%
css = '''/* Auto-generated theme for {name} */
/* Deterministic: regenerating from the same name produces identical output */
:root,
.theme-light {{
--rb-bg: hsl({hue}, {sat_bg}%, 97%);
--rb-bg-card: hsl({hue}, {sat_bg}%, 100%);
--rb-bg-nested: hsl({hue}, {sat_bg}%, 96%);
--rb-bg-code: hsl({hue}, {sat_bg}%, 94%);
--rb-text: hsl({hue}, {sat_base}%, 15%);
--rb-text-secondary: hsl({hue}, {sat_muted}%, 35%);
--rb-text-muted: hsl({hue}, {sat_muted}%, 55%);
--rb-link: hsl({accent_hue}, {sat_link}%, 40%);
--rb-link-hover: hsl({accent_hue}, {sat_link}%, 30%);
--rb-link-visited: hsl({secondary_hue}, {sat_muted}%, 45%);
--rb-border: hsl({hue}, {sat_muted}%, 85%);
--rb-border-light: hsl({hue}, {sat_muted}%, 91%);
--rb-accent: hsl({accent_hue}, {sat_accent}%, 45%);
--rb-accent-hover: hsl({accent_hue}, {sat_accent}%, 35%);
--rb-accent-text: hsl({accent_hue}, 5%, 100%);
--rb-blockquote-border: hsl({secondary_hue}, {sat_base}%, 70%);
--rb-avatar-border: hsl({hue}, {sat_muted}%, 80%);
--rb-shadow: 0 1px 3px hsla({hue}, {sat_muted}%, 30%, 0.08);
}}
@media (prefers-color-scheme: dark) {{
:root:not(.theme-light) {{
--rb-bg: hsl({hue}, {sat_dark}%, 10%);
--rb-bg-card: hsl({hue}, {sat_dark}%, 14%);
--rb-bg-nested: hsl({hue}, {sat_dark}%, 12%);
--rb-bg-code: hsl({hue}, {sat_dark}%, 18%);
--rb-text: hsl({hue}, {sat_bg}%, 90%);
--rb-text-secondary: hsl({hue}, {sat_muted}%, 70%);
--rb-text-muted: hsl({hue}, {sat_muted}%, 50%);
--rb-link: hsl({accent_hue}, {sat_link}%, 65%);
--rb-link-hover: hsl({accent_hue}, {sat_link}%, 75%);
--rb-link-visited: hsl({secondary_hue}, {sat_muted}%, 60%);
--rb-border: hsl({hue}, {sat_muted}%, 25%);
--rb-border-light: hsl({hue}, {sat_muted}%, 20%);
--rb-accent: hsl({accent_hue}, {sat_accent}%, 55%);
--rb-accent-hover: hsl({accent_hue}, {sat_accent}%, 65%);
--rb-accent-text: hsl({accent_hue}, 5%, 10%);
--rb-blockquote-border: hsl({secondary_hue}, {sat_base}%, 40%);
--rb-avatar-border: hsl({hue}, {sat_muted}%, 35%);
--rb-shadow: 0 1px 3px hsla({hue}, {sat_muted}%, 5%, 0.3);
}}
}}
.theme-dark {{
--rb-bg: hsl({hue}, {sat_dark}%, 10%);
--rb-bg-card: hsl({hue}, {sat_dark}%, 14%);
--rb-bg-nested: hsl({hue}, {sat_dark}%, 12%);
--rb-bg-code: hsl({hue}, {sat_dark}%, 18%);
--rb-text: hsl({hue}, {sat_bg}%, 90%);
--rb-text-secondary: hsl({hue}, {sat_muted}%, 70%);
--rb-text-muted: hsl({hue}, {sat_muted}%, 50%);
--rb-link: hsl({accent_hue}, {sat_link}%, 65%);
--rb-link-hover: hsl({accent_hue}, {sat_link}%, 75%);
--rb-link-visited: hsl({secondary_hue}, {sat_muted}%, 60%);
--rb-border: hsl({hue}, {sat_muted}%, 25%);
--rb-border-light: hsl({hue}, {sat_muted}%, 20%);
--rb-accent: hsl({accent_hue}, {sat_accent}%, 55%);
--rb-accent-hover: hsl({accent_hue}, {sat_accent}%, 65%);
--rb-accent-text: hsl({accent_hue}, 5%, 10%);
--rb-blockquote-border: hsl({secondary_hue}, {sat_base}%, 40%);
--rb-avatar-border: hsl({hue}, {sat_muted}%, 35%);
--rb-shadow: 0 1px 3px hsla({hue}, {sat_muted}%, 5%, 0.3);
}}
/* Apply theme variables to remarkbox elements */
body {{
background-color: var(--rb-bg);
color: var(--rb-text);
}}
.node {{
background-color: var(--rb-bg-card);
border-color: var(--rb-border-light);
box-shadow: var(--rb-shadow);
}}
.node .node {{
background-color: var(--rb-bg-nested);
}}
a {{
color: var(--rb-link);
}}
a:hover {{
color: var(--rb-link-hover);
}}
a:visited {{
color: var(--rb-link-visited);
}}
.text-muted, .rb-meta {{
color: var(--rb-text-muted);
}}
pre, code {{
background-color: var(--rb-bg-code);
border-color: var(--rb-border);
}}
blockquote {{
border-left-color: var(--rb-blockquote-border);
color: var(--rb-text-secondary);
}}
.rb-submit, .btn-primary {{
background-color: var(--rb-accent);
color: var(--rb-accent-text);
border-color: var(--rb-accent);
}}
.rb-submit:hover, .btn-primary:hover {{
background-color: var(--rb-accent-hover);
border-color: var(--rb-accent-hover);
}}
.nested-avatar {{
border-color: var(--rb-avatar-border);
}}
hr {{
border-color: var(--rb-border-light);
}}
textarea, input[type="text"], input[type="email"] {{
background-color: var(--rb-bg-card);
color: var(--rb-text);
border-color: var(--rb-border);
}}
textarea:focus, input:focus {{
border-color: var(--rb-accent);
outline-color: var(--rb-accent);
}}
'''.format(
name=namespace_name,
hue=hue,
secondary_hue=secondary_hue,
accent_hue=accent_hue,
sat_base=sat_base,
sat_bg=max(sat_base - 25, 5),
sat_muted=max(sat_base - 15, 10),
sat_link=min(sat_base + 15, 75),
sat_accent=min(sat_base + 20, 80),
sat_dark=max(sat_base - 30, 8),
)
return css

1
remarkbox/models/__init__.py
View file

@ -21,6 +21,7 @@ from .pay_what_you_can import *
from .payment import *
from .webmention import *
from .sudo_otp import *
from .revision import *
# run configure_mappers after defining all of the models to ensure
# all relationships can be setup

1
remarkbox/models/meta.py
View file

@ -41,6 +41,7 @@ CLASS_TO_TABLE = {
"PayWhatYouCan": "rb_pay_what_you_can",
"Payment": "rb_payment",
"Webmention": "rb_webmention",
"Revision": "rb_revision",
}
# node (threads), namespace (forum)

14
remarkbox/models/namespace.py
View file

@ -393,6 +393,20 @@ class Namespace(RBase, Base):
def can_alter_node(self, node, user):
return self.is_moderator(user) or node.is_owner(user)
def can_wiki_edit(self, node, user):
"""Return True if user can wiki-edit this node.
Wiki mode: any authenticated user can edit root nodes.
Normal mode: only owner/moderator can edit.
"""
if not user or not user.authenticated:
return False
if self.can_alter_node(node, user):
return True
if self.wiki and node.is_root:
return True
return False
def can_see_node(self, node, user):
visible = True
if node.disabled:

50
remarkbox/models/node.py
View file

@ -71,6 +71,7 @@ class Node(RBase, Base):
title = Column(Unicode(256), default=None)
data = Column(UnicodeText, default=None)
data_html = Column(UnicodeText, default=None)
source_format = Column(Unicode(16), default="markdown", nullable=False)
# the depth of this node in the thread's graph / tree.
graph_depth = Column(Integer, nullable=False, default=-1)
# TODO: someday this should be renamed to created_timestamp
@ -315,13 +316,34 @@ class Node(RBase, Base):
def enabled(self):
return not self.disabled
def set_data(self, data, namespace=None, dbsession=None):
def set_data(self, data, namespace=None, dbsession=None, source_format=None):
if namespace is None:
namespace = self.root.namespace
if dbsession is None:
dbsession = self.dbsession
if source_format is not None:
self.source_format = source_format
self.data = data
self.data_html = markdown_to_html(data, namespace, dbsession=dbsession)
if not self.source_format or self.source_format == "markdown":
self.data_html = markdown_to_html(data, namespace, dbsession=dbsession)
elif self.source_format == "html":
from remarkbox.lib.pandoc import convert
# Convert HTML to markdown (clean it), then render through standard pipeline
cleaned = convert(data, from_format="html", to_format="markdown")
self.data = cleaned
self.source_format = "markdown"
self.data_html = markdown_to_html(cleaned, namespace, dbsession=dbsession)
else:
# Any other pandoc input format (rst, mediawiki, latex, etc.)
from remarkbox.lib.pandoc import convert
from remarkbox.lib.render import make_cleaner_from_namespace
from remarkbox.lib.sanitize_html import default_cleaner, clean_raw_html
html = convert(data, from_format=self.source_format, to_format="html5")
if namespace:
cleaner = make_cleaner_from_namespace(namespace)
else:
cleaner = default_cleaner()
self.data_html = clean_raw_html(html, cleaner)
def _invalidate_cache(self):
if self.root.cache:
@ -333,6 +355,30 @@ class Node(RBase, Base):
self.changed = now_timestamp()
self._invalidate_cache()
def wiki_edit(self, data, user=None, source_format=None):
"""Edit node in wiki mode, creating a revision of the previous state."""
from .revision import Revision
# Count existing revisions for this node
rev_count = self.dbsession.query(Revision).filter(
Revision.node_id == self.id
).count()
# Save current state as a revision
revision = Revision(
node=self,
user=user,
data=self.data or "",
source_format=self.source_format,
revision_number=rev_count + 1,
)
self.dbsession.add(revision)
# Apply the edit
if source_format:
self.source_format = source_format
self.edit(data)
def disable(self):
"""disable node."""
self.disabled = True

39
remarkbox/models/revision.py Normal file
View file

@ -0,0 +1,39 @@
"""Revision model — tracks edit history for wiki mode."""
from sqlalchemy import BigInteger, Column, Integer, Unicode, UnicodeText
from sqlalchemy.orm import relationship
import uuid
from .meta import Base, RBase, UUIDType, now_timestamp, foreign_key
class Revision(RBase, Base):
"""Stores a snapshot of node content before each edit.
When wiki mode is enabled on a namespace, every edit to a root node
creates a revision record preserving the previous state.
"""
id = Column(UUIDType, primary_key=True, index=True)
node_id = Column(UUIDType, foreign_key("Node", "id"), nullable=False, index=True)
user_id = Column(UUIDType, foreign_key("User", "id"), nullable=True)
data = Column(UnicodeText, nullable=False)
source_format = Column(Unicode(16), default="markdown", nullable=False)
revision_number = Column(Integer, nullable=False)
created = Column(BigInteger, nullable=False)
node = relationship("Node", backref="revisions")
user = relationship("User")
def __init__(self, node=None, user=None, data="", source_format="markdown", revision_number=1):
self.id = uuid.uuid1()
self.created = now_timestamp()
if node:
self.node_id = node.id
self.node = node
if user:
self.user_id = user.id
self.user = user
self.data = data
self.source_format = source_format
self.revision_number = revision_number

42
remarkbox/scripts/alembic/versions/8e3c406e4049_add_revision_table_for_wiki_mode.py Normal file
View file

@ -0,0 +1,42 @@
"""add revision table for wiki mode
Revision ID: 8e3c406e4049
Revises: 7c624a8fae9e
Create Date: 2026-03-09 23:09:54.651756
"""
# revision identifiers, used by Alembic.
revision = '8e3c406e4049'
down_revision = 'd47dc908d2ea'
branch_labels = None
depends_on = None
from alembic import op
import sqlalchemy as sa
from sqlalchemy_utils import UUIDType as TempUUIDType
UUIDType = TempUUIDType(binary=False)
def upgrade():
op.create_table('rb_revision',
sa.Column('id', UUIDType, nullable=False),
sa.Column('node_id', UUIDType, nullable=False),
sa.Column('user_id', UUIDType, nullable=True),
sa.Column('data', sa.UnicodeText(), nullable=False),
sa.Column('source_format', sa.Unicode(length=16), nullable=False),
sa.Column('revision_number', sa.Integer(), nullable=False),
sa.Column('created', sa.BigInteger(), nullable=False),
sa.ForeignKeyConstraint(['node_id'], ['rb_node.id']),
sa.ForeignKeyConstraint(['user_id'], ['rb_user.id']),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_rb_revision_id'), 'rb_revision', ['id'], unique=False)
op.create_index(op.f('ix_rb_revision_node_id'), 'rb_revision', ['node_id'], unique=False)
def downgrade():
op.drop_index(op.f('ix_rb_revision_node_id'), table_name='rb_revision')
op.drop_index(op.f('ix_rb_revision_id'), table_name='rb_revision')
op.drop_table('rb_revision')

27
remarkbox/scripts/alembic/versions/d47dc908d2ea_add_source_format_to_node.py Normal file
View file

@ -0,0 +1,27 @@
"""add source_format to node
Revision ID: d47dc908d2ea
Revises: 7c624a8fae9e
Create Date: 2026-03-09 23:08:19.266162
"""
# revision identifiers, used by Alembic.
revision = 'd47dc908d2ea'
down_revision = '7c624a8fae9e'
branch_labels = None
depends_on = None
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column(
'rb_node',
sa.Column('source_format', sa.Unicode(length=16), nullable=False, server_default='markdown'),
)
def downgrade():
op.drop_column('rb_node', 'source_format')

274
remarkbox/tests/test_pandoc.py Normal file
View file

@ -0,0 +1,274 @@
"""Unit tests for remarkbox.lib.pandoc — pandoc subprocess wrapper."""
import unittest
import subprocess
from remarkbox.lib.pandoc import (
convert,
node_tree_to_markdown,
namespace_to_markdown,
get_available_output_formats,
get_available_input_formats,
CONTENT_TYPES,
FILE_EXTENSIONS,
BINARY_FORMATS,
_get_author_name,
)
class TestConvert(unittest.TestCase):
"""Unit tests for the pandoc convert() function."""
def test_markdown_to_html(self):
result = convert("# Hello\n\nWorld.", "markdown", "html5", title="Test")
self.assertIn("<h1", result)
self.assertIn("World.", result)
def test_markdown_to_rst(self):
result = convert("# Hello\n\n**bold** text.", "markdown", "rst")
self.assertIn("**bold**", result)
def test_markdown_to_plain(self):
result = convert("# Title\n\n- item 1\n- item 2", "markdown", "plain")
self.assertIn("item 1", result)
self.assertIn("item 2", result)
def test_html_to_markdown(self):
result = convert("<h1>Title</h1><p>Paragraph.</p>", "html", "markdown")
self.assertIn("Title", result)
self.assertIn("Paragraph.", result)
def test_rst_to_html(self):
rst = "Title\n=====\n\nA paragraph."
result = convert(rst, "rst", "html5")
self.assertIn("Title", result)
self.assertIn("paragraph", result)
def test_mediawiki_to_html(self):
wiki = "== Section ==\n\n'''bold''' text"
result = convert(wiki, "mediawiki", "html5")
self.assertIn("Section", result)
def test_latex_to_html(self):
latex = r"\section{Hello}\textbf{bold}"
result = convert(latex, "latex", "html5")
self.assertIn("Hello", result)
def test_markdown_to_json(self):
result = convert("# Test", "markdown", "json")
self.assertIn('"pandoc-api-version"', result)
def test_markdown_to_epub_returns_bytes(self):
result = convert("# Test\n\nContent.", "markdown", "epub", title="Test")
self.assertIsInstance(result, bytes)
self.assertGreater(len(result), 100)
def test_markdown_to_pdf_returns_bytes(self):
result = convert("# Test\n\nContent.", "markdown", "pdf", title="Test")
self.assertIsInstance(result, bytes)
# PDF starts with %PDF
self.assertTrue(result[:4] == b"%PDF")
def test_markdown_to_docx_returns_bytes(self):
result = convert("# Test\n\nContent.", "markdown", "docx", title="Test")
self.assertIsInstance(result, bytes)
# DOCX is a ZIP file (starts with PK)
self.assertTrue(result[:2] == b"PK")
def test_title_metadata(self):
result = convert("Content.", "markdown", "html5", title="My Title")
self.assertIn("My Title", result)
def test_empty_input(self):
result = convert("", "markdown", "html5")
self.assertIsInstance(result, str)
def test_invalid_format_raises(self):
with self.assertRaises(subprocess.CalledProcessError):
convert("test", "markdown", "not_a_real_format_xyz")
class TestAvailableFormats(unittest.TestCase):
"""Test format discovery."""
def test_output_formats_returns_frozenset(self):
formats = get_available_output_formats()
self.assertIsInstance(formats, frozenset)
self.assertGreater(len(formats), 50)
self.assertIn("html5", formats)
self.assertIn("markdown", formats)
self.assertIn("pdf", formats)
self.assertIn("epub", formats)
def test_input_formats_returns_frozenset(self):
formats = get_available_input_formats()
self.assertIsInstance(formats, frozenset)
self.assertGreater(len(formats), 30)
self.assertIn("html", formats)
self.assertIn("markdown", formats)
self.assertIn("rst", formats)
def test_content_types_has_common_formats(self):
self.assertIn("html5", CONTENT_TYPES)
self.assertIn("pdf", CONTENT_TYPES)
self.assertIn("epub", CONTENT_TYPES)
self.assertIn("docx", CONTENT_TYPES)
self.assertIn("markdown", CONTENT_TYPES)
self.assertEqual(CONTENT_TYPES["pdf"], "application/pdf")
def test_file_extensions_has_common_formats(self):
self.assertEqual(FILE_EXTENSIONS["markdown"], ".md")
self.assertEqual(FILE_EXTENSIONS["html5"], ".html")
self.assertEqual(FILE_EXTENSIONS["pdf"], ".pdf")
self.assertEqual(FILE_EXTENSIONS["epub"], ".epub")
self.assertEqual(FILE_EXTENSIONS["docx"], ".docx")
def test_binary_formats(self):
self.assertIn("pdf", BINARY_FORMATS)
self.assertIn("docx", BINARY_FORMATS)
self.assertIn("epub", BINARY_FORMATS)
self.assertNotIn("html5", BINARY_FORMATS)
self.assertNotIn("markdown", BINARY_FORMATS)
class MockNode:
"""Minimal mock of Node for tree rendering tests."""
def __init__(self, id, title=None, data=None, parent_id=None,
created=0, disabled=False, user=None, user_surrogate=None,
created_date="2026-01-01"):
self.id = id
self.title = title
self.data = data
self.parent_id = parent_id
self.created = created
self.disabled = disabled
self.user = user
self.user_surrogate = user_surrogate
self.created_date = created_date
self.is_root = parent_id is None
class MockUser:
def __init__(self, name):
self.name = name
class MockNamespace:
def __init__(self, name, description=None):
self.name = name
self.description = description
class TestNodeTreeToMarkdown(unittest.TestCase):
"""Test tree-to-markdown rendering."""
def test_single_root_node(self):
root = MockNode(1, title="Thread Title", data="Root content.")
md = node_tree_to_markdown(root, [])
self.assertIn("# Thread Title", md)
self.assertIn("Root content.", md)
def test_root_with_children(self):
root = MockNode(1, title="Thread", data="Root.")
child = MockNode(2, data="Reply text.", parent_id=1, created=1,
user=MockUser("alice"))
md = node_tree_to_markdown(root, [child])
self.assertIn("# Thread", md)
self.assertIn("Root.", md)
self.assertIn("alice", md)
self.assertIn("Reply text.", md)
def test_disabled_children_excluded(self):
root = MockNode(1, title="Thread", data="Root.")
child = MockNode(2, data="Visible.", parent_id=1, created=1,
user=MockUser("bob"))
disabled = MockNode(3, data="Hidden.", parent_id=1, created=2,
disabled=True, user=MockUser("spam"))
md = node_tree_to_markdown(root, [child, disabled])
self.assertIn("Visible.", md)
self.assertNotIn("Hidden.", md)
def test_nested_children(self):
root = MockNode(1, title="Thread", data="Root.")
child = MockNode(2, data="Level 1.", parent_id=1, created=1,
user=MockUser("alice"))
grandchild = MockNode(3, data="Level 2.", parent_id=2, created=2,
user=MockUser("bob"))
md = node_tree_to_markdown(root, [child, grandchild])
self.assertIn("Level 1.", md)
self.assertIn("Level 2.", md)
# Grandchild should have deeper heading
self.assertIn("### bob", md)
def test_exclude_root(self):
root = MockNode(1, title="Hidden Title", data="Hidden root.")
md = node_tree_to_markdown(root, [], include_root=False)
self.assertNotIn("Hidden Title", md)
self.assertNotIn("Hidden root.", md)
def test_anonymous_author(self):
root = MockNode(1, title="Thread")
child = MockNode(2, data="Anon post.", parent_id=1, created=1)
md = node_tree_to_markdown(root, [child])
self.assertIn("Anonymous", md)
def test_surrogate_author(self):
root = MockNode(1, title="Thread")
child = MockNode(2, data="Bot post.", parent_id=1, created=1,
user_surrogate=MockUser("MyBot"))
md = node_tree_to_markdown(root, [child])
self.assertIn("MyBot", md)
class TestNamespaceToMarkdown(unittest.TestCase):
"""Test namespace (book) rendering."""
def test_empty_namespace(self):
ns = MockNamespace("test.com", "Test Forum")
md = namespace_to_markdown(ns, [], lambda r: [])
self.assertIn("# Test Forum", md)
def test_namespace_with_threads(self):
ns = MockNamespace("test.com", "My Forum")
root1 = MockNode(1, title="First Thread", data="Content 1.")
root2 = MockNode(2, title="Second Thread", data="Content 2.")
def fetcher(root):
return []
md = namespace_to_markdown(ns, [root1, root2], fetcher)
self.assertIn("# My Forum", md)
self.assertIn("## First Thread", md)
self.assertIn("## Second Thread", md)
self.assertIn("Content 1.", md)
self.assertIn("Content 2.", md)
def test_namespace_name_as_fallback_title(self):
ns = MockNamespace("test.com")
md = namespace_to_markdown(ns, [], lambda r: [])
self.assertIn("# test.com", md)
def test_disabled_roots_excluded(self):
ns = MockNamespace("test.com")
root = MockNode(1, title="Visible", data="Yes.", disabled=False)
hidden = MockNode(2, title="Hidden", data="No.", disabled=True)
md = namespace_to_markdown(ns, [root, hidden], lambda r: [])
self.assertIn("Visible", md)
self.assertNotIn("Hidden", md)
class TestGetAuthorName(unittest.TestCase):
"""Test _get_author_name helper."""
def test_user(self):
node = MockNode(1, user=MockUser("alice"))
self.assertEqual(_get_author_name(node), "alice")
def test_surrogate(self):
node = MockNode(1, user_surrogate=MockUser("BotName"))
self.assertEqual(_get_author_name(node), "BotName")
def test_anonymous(self):
node = MockNode(1)
self.assertEqual(_get_author_name(node), "Anonymous")

93
remarkbox/tests/test_revision.py Normal file
View file

@ -0,0 +1,93 @@
"""Unit tests for remarkbox.models.revision — wiki mode revision tracking."""
import unittest
from unittest import mock
from remarkbox.models.revision import Revision
from remarkbox.models.node import Node
from remarkbox.models.namespace import Namespace
class TestRevisionModel(unittest.TestCase):
"""Unit tests for the Revision model."""
def test_init_sets_uuid(self):
rev = Revision()
self.assertIsNotNone(rev.id)
def test_init_sets_timestamp(self):
rev = Revision()
self.assertIsNotNone(rev.created)
self.assertGreater(rev.created, 0)
def test_init_with_data(self):
rev = Revision(
data="Some content",
source_format="markdown",
revision_number=3,
)
self.assertEqual(rev.data, "Some content")
self.assertEqual(rev.source_format, "markdown")
self.assertEqual(rev.revision_number, 3)
def test_default_source_format(self):
rev = Revision()
self.assertEqual(rev.source_format, "markdown")
def test_default_revision_number(self):
rev = Revision()
self.assertEqual(rev.revision_number, 1)
class TestNamespaceCanWikiEdit(unittest.TestCase):
"""Unit tests for Namespace.can_wiki_edit()."""
def setUp(self):
self.ns = Namespace("test-wiki.com")
self.ns.subscription_type = "production"
def test_unauthenticated_user_denied(self):
user = mock.Mock()
user.authenticated = False
node = mock.Mock()
self.assertFalse(self.ns.can_wiki_edit(node, user))
def test_none_user_denied(self):
node = mock.Mock()
self.assertFalse(self.ns.can_wiki_edit(node, None))
def test_owner_can_always_edit(self):
user = mock.Mock()
user.authenticated = True
node = mock.Mock()
node.is_root = False
# can_alter_node returns True for owners
with mock.patch.object(self.ns, 'can_alter_node', return_value=True):
self.assertTrue(self.ns.can_wiki_edit(node, user))
def test_wiki_mode_allows_root_edit(self):
self.ns.wiki = True
user = mock.Mock()
user.authenticated = True
node = mock.Mock()
node.is_root = True
with mock.patch.object(self.ns, 'can_alter_node', return_value=False):
self.assertTrue(self.ns.can_wiki_edit(node, user))
def test_wiki_mode_denies_non_root_edit(self):
self.ns.wiki = True
user = mock.Mock()
user.authenticated = True
node = mock.Mock()
node.is_root = False
with mock.patch.object(self.ns, 'can_alter_node', return_value=False):
self.assertFalse(self.ns.can_wiki_edit(node, user))
def test_non_wiki_mode_denies_non_owner(self):
self.ns.wiki = False
user = mock.Mock()
user.authenticated = True
node = mock.Mock()
node.is_root = True
with mock.patch.object(self.ns, 'can_alter_node', return_value=False):
self.assertFalse(self.ns.can_wiki_edit(node, user))

113
remarkbox/tests/test_theme_generator.py Normal file
View file

@ -0,0 +1,113 @@
"""Unit tests for remarkbox.lib.theme_generator."""
import unittest
from remarkbox.lib.theme_generator import (
generate_theme_css,
_name_to_hue,
_name_to_seed,
)
class TestNameToHue(unittest.TestCase):
"""Test deterministic hue generation."""
def test_returns_int_in_range(self):
hue = _name_to_hue("test.com")
self.assertIsInstance(hue, int)
self.assertGreaterEqual(hue, 0)
self.assertLess(hue, 360)
def test_deterministic(self):
h1 = _name_to_hue("meta.remarkbox.com")
h2 = _name_to_hue("meta.remarkbox.com")
self.assertEqual(h1, h2)
def test_different_names_different_hues(self):
h1 = _name_to_hue("meta.remarkbox.com")
h2 = _name_to_hue("faq.remarkbox.com")
# Could theoretically collide but extremely unlikely
self.assertNotEqual(h1, h2)
class TestNameToSeed(unittest.TestCase):
"""Test seed generation."""
def test_returns_tuple_of_ints(self):
seed = _name_to_seed("test.com")
self.assertIsInstance(seed, tuple)
self.assertEqual(len(seed), 8)
for v in seed:
self.assertIsInstance(v, int)
def test_deterministic(self):
s1 = _name_to_seed("test.com")
s2 = _name_to_seed("test.com")
self.assertEqual(s1, s2)
class TestGenerateThemeCSS(unittest.TestCase):
"""Test CSS theme generation."""
def test_returns_string(self):
css = generate_theme_css("test.com")
self.assertIsInstance(css, str)
self.assertGreater(len(css), 500)
def test_deterministic(self):
css1 = generate_theme_css("meta.remarkbox.com")
css2 = generate_theme_css("meta.remarkbox.com")
self.assertEqual(css1, css2)
def test_different_namespaces_different_css(self):
css1 = generate_theme_css("meta.remarkbox.com")
css2 = generate_theme_css("faq.remarkbox.com")
self.assertNotEqual(css1, css2)
def test_contains_css_custom_properties(self):
css = generate_theme_css("test.com")
self.assertIn("--rb-bg:", css)
self.assertIn("--rb-text:", css)
self.assertIn("--rb-link:", css)
self.assertIn("--rb-accent:", css)
self.assertIn("--rb-border:", css)
def test_contains_light_mode(self):
css = generate_theme_css("test.com")
self.assertIn(":root", css)
self.assertIn(".theme-light", css)
def test_contains_dark_mode(self):
css = generate_theme_css("test.com")
self.assertIn("prefers-color-scheme: dark", css)
self.assertIn(".theme-dark", css)
def test_contains_namespace_name_comment(self):
css = generate_theme_css("my-forum.example.org")
self.assertIn("my-forum.example.org", css)
def test_contains_element_styles(self):
css = generate_theme_css("test.com")
self.assertIn(".node", css)
self.assertIn("blockquote", css)
self.assertIn(".rb-submit", css)
self.assertIn("textarea", css)
def test_uses_hsl_colors(self):
css = generate_theme_css("test.com")
self.assertIn("hsl(", css)
def test_various_namespaces_all_valid(self):
"""Ensure no crashes for various namespace name patterns."""
names = [
"a.com",
"very-long-domain-name-that-goes-on.example.com",
"unicode-test-日本語.com",
"123.456.789",
"",
"single",
]
for name in names:
css = generate_theme_css(name)
self.assertIsInstance(css, str)
self.assertGreater(len(css), 100)

762
remarkbox/tests/test_undigg.py Normal file
View file

@ -0,0 +1,762 @@
"""Integration and functional tests for Operation Undigg.
Tests the export, multi-syntax, wiki mode, and theme APIs end-to-end
via webtest against the full Pyramid app.
"""
import transaction
import unittest
import webtest
from remarkbox.models import (
Node,
get_tm_session,
get_or_create_user_by_email,
get_or_create_namespace,
UserSurrogate,
)
from remarkbox.models.meta import Base
from remarkbox.models.revision import Revision
from pyramid.paster import get_appsettings
class UndiggFunctionalTests(unittest.TestCase):
"""Base class for Undigg functional tests."""
@classmethod
def setUpClass(cls):
from remarkbox import main
cls.settings = get_appsettings("test.ini")
cls.app = main({}, **cls.settings)
cls.testapp = webtest.TestApp(cls.app)
cls.session_factory = cls.app.registry["dbsession_factory"]
cls.engine = cls.session_factory.kw["bind"]
Base.metadata.create_all(bind=cls.engine)
cls.tm = transaction.manager
cls.dbsession = get_tm_session(cls.session_factory, cls.tm)
@classmethod
def tearDownClass(cls):
cls.dbsession.close()
Base.metadata.drop_all(bind=cls.engine)
def tearDown(self):
self.testapp.get("/log-out")
# ---------------------------------------------------------------------------
# Export API
# ---------------------------------------------------------------------------
class TestExportFormats(UndiggFunctionalTests):
"""Test GET /api/v1/export/formats."""
def test_list_formats(self):
res = self.testapp.get("/api/v1/export/formats")
self.assertEqual(res.status_int, 200)
body = res.json
self.assertIn("formats", body)
self.assertIn("count", body)
self.assertGreater(body["count"], 50)
self.assertIn("html5", body["formats"])
self.assertIn("pdf", body["formats"])
self.assertIn("markdown", body["formats"])
class TestExportThread(UndiggFunctionalTests):
"""Test GET /api/v1/export/threads/{id}.{format}."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "export-test.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestExportThread, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def _create_thread(self, title="Test Thread", data="Thread content."):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": title,
"data": data,
"anonymous_name": "Bot",
},
expect_errors=True,
)
return res.json["node"]["id"]
def test_export_thread_markdown(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.markdown".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn("text/markdown", res.content_type)
self.assertIn(b"Test Thread", res.body)
self.assertIn(b"Thread content.", res.body)
def test_export_thread_html(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.html5".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn("text/html", res.content_type)
self.assertIn(b"Test Thread", res.body)
def test_export_thread_rst(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.rst".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn(b"Test Thread", res.body)
def test_export_thread_pdf(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.pdf".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.content_type, "application/pdf")
self.assertTrue(res.body[:4] == b"%PDF")
def test_export_thread_epub(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.epub".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.content_type, "application/epub+zip")
self.assertGreater(len(res.body), 100)
def test_export_thread_docx(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.docx".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
# DOCX is a ZIP
self.assertTrue(res.body[:2] == b"PK")
def test_export_thread_not_found(self):
res = self.testapp.get(
"/api/v1/export/threads/00000000-0000-0000-0000-000000000000.markdown",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_export_thread_unsupported_format(self):
node_id = self._create_thread()
res = self.testapp.get(
"/api/v1/export/threads/{}.not_a_format".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 400)
def test_export_thread_with_replies(self):
"""Thread with replies includes reply content in export."""
node_id = self._create_thread()
self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{"data": "A reply from alice.", "anonymous_name": "alice"},
expect_errors=True,
)
res = self.testapp.get(
"/api/v1/export/threads/{}.markdown".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
body = res.body.decode("utf-8")
self.assertIn("A reply from alice.", body)
self.assertIn("alice", body)
def test_export_content_disposition(self):
node_id = self._create_thread(title="My Export Test")
res = self.testapp.get(
"/api/v1/export/threads/{}.markdown".format(node_id),
expect_errors=True,
)
self.assertIn("attachment", res.headers.get("Content-Disposition", ""))
class TestExportNamespace(UndiggFunctionalTests):
"""Test GET /api/v1/export/namespace/{name}.{format}."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "ns-export.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestExportNamespace, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_export_namespace_markdown(self):
# Create a thread
self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Chapter One",
"data": "First chapter content.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
res = self.testapp.get(
"/api/v1/export/namespace/{}.markdown".format(self.namespace_name),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
body = res.body.decode("utf-8")
self.assertIn("Chapter One", body)
self.assertIn("First chapter content.", body)
def test_export_namespace_html(self):
self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "HTML Chapter",
"data": "Content.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
res = self.testapp.get(
"/api/v1/export/namespace/{}.html5".format(self.namespace_name),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn("text/html", res.content_type)
def test_export_namespace_not_found(self):
res = self.testapp.get(
"/api/v1/export/namespace/nonexistent.example.com.markdown",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_export_empty_namespace(self):
ns = get_or_create_namespace(self.dbsession, "empty-ns.example.com")
self.dbsession.add(ns)
self.dbsession.flush()
self.tm.commit()
res = self.testapp.get(
"/api/v1/export/namespace/empty-ns.example.com.markdown",
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
class TestExportNode(UndiggFunctionalTests):
"""Test GET /api/v1/export/nodes/{id}.{format} (on-demand subthread)."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "node-export.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestExportNode, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_export_node_subtree(self):
# Create thread with a reply
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread",
"data": "Root content.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
root_id = res.json["node"]["id"]
self.testapp.post_json(
"/api/v1/threads/{}/replies".format(root_id),
{"data": "Reply content.", "anonymous_name": "Bot"},
expect_errors=True,
)
# Export the reply's subtree
export_res = self.testapp.get(
"/api/v1/export/nodes/{}.markdown".format(root_id),
expect_errors=True,
)
self.assertEqual(export_res.status_int, 200)
body = export_res.body.decode("utf-8")
self.assertIn("Root content.", body)
def test_export_node_not_found(self):
res = self.testapp.get(
"/api/v1/export/nodes/00000000-0000-0000-0000-000000000000.markdown",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
# ---------------------------------------------------------------------------
# Multi-Syntax Input
# ---------------------------------------------------------------------------
class TestMultiSyntaxInput(UndiggFunctionalTests):
"""Test source_format parameter on create/reply/edit."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "syntax-test.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestMultiSyntaxInput, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_create_thread_default_markdown(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Markdown Thread",
"data": "**bold** text",
"anonymous_name": "Bot",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
self.assertEqual(res.json["node"]["source_format"], "markdown")
self.assertIn("<strong>bold</strong>", res.json["node"]["data_html"])
def test_create_thread_rst(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "RST Thread",
"data": "A paragraph with **bold** text.",
"anonymous_name": "Bot",
"source_format": "rst",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
self.assertEqual(res.json["node"]["source_format"], "rst")
self.assertIn("bold", res.json["node"]["data_html"])
def test_create_thread_html_stripped(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "HTML Thread",
"data": "<h1>Title</h1><p>Paragraph.</p>",
"anonymous_name": "Bot",
"source_format": "html",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
# HTML input gets normalized to markdown
self.assertEqual(res.json["node"]["source_format"], "markdown")
self.assertIn("Paragraph.", res.json["node"]["data"])
def test_reply_with_source_format(self):
# Create thread
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Thread",
"data": "Content.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Reply with RST
reply_res = self.testapp.post_json(
"/api/v1/threads/{}/replies".format(node_id),
{
"data": "**RST bold** reply.",
"anonymous_name": "Bot",
"source_format": "rst",
},
expect_errors=True,
)
self.assertEqual(reply_res.status_int, 201)
self.assertEqual(reply_res.json["node"]["source_format"], "rst")
def test_serializer_includes_source_format(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Test",
"data": "Content.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
self.assertIn("source_format", res.json["node"])
class TestMultiSyntaxMediawiki(UndiggFunctionalTests):
"""Test mediawiki input format."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "wiki-syntax.example.com")
ns.allow_anonymous = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestMultiSyntaxMediawiki, self).tearDown()
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_create_thread_mediawiki(self):
res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Wiki Thread",
"data": "== Section ==\n\n'''bold''' text",
"anonymous_name": "Bot",
"source_format": "mediawiki",
},
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
self.assertEqual(res.json["node"]["source_format"], "mediawiki")
# The HTML should contain the rendered content
self.assertIn("Section", res.json["node"]["data_html"])
# ---------------------------------------------------------------------------
# Wiki Mode & Revisions
# ---------------------------------------------------------------------------
class TestWikiModeAndRevisions(UndiggFunctionalTests):
"""Test wiki-edit and revision history endpoints."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "wiki-test.example.com")
ns.allow_anonymous = True
# Enable wiki mode
ns.subscription_type = "production"
ns.wiki = True
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.namespace_id = ns.id
self.tm.commit()
def tearDown(self):
super(TestWikiModeAndRevisions, self).tearDown()
# Clean up revisions first (FK constraint)
self.dbsession.query(Revision).filter(
Revision.node_id.in_(
self.dbsession.query(Node.id).filter(
Node.namespace_id == self.namespace_id
)
)
).delete(synchronize_session=False)
self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.query(Node).filter(
Node.namespace_id == self.namespace_id
).delete(synchronize_session=False)
self.dbsession.flush()
self.tm.commit()
def test_wiki_edit_requires_auth(self):
# Create a thread
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Wiki Thread",
"data": "Original.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Try wiki-edit without auth
res = self.testapp.post_json(
"/api/v1/nodes/{}/wiki-edit".format(node_id),
{"data": "Updated."},
expect_errors=True,
)
self.assertEqual(res.status_int, 401)
def test_revisions_empty_initially(self):
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Wiki Thread",
"data": "Original.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
res = self.testapp.get(
"/api/v1/nodes/{}/revisions".format(node_id),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertEqual(res.json["count"], 0)
self.assertEqual(res.json["revisions"], [])
def test_revisions_not_found(self):
res = self.testapp.get(
"/api/v1/nodes/00000000-0000-0000-0000-000000000000/revisions",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_wiki_edit_missing_data(self):
create_res = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Wiki Thread",
"data": "Original.",
"anonymous_name": "Bot",
},
expect_errors=True,
)
node_id = create_res.json["node"]["id"]
# Need to be authenticated to even get past the auth check
# Since we can't easily authenticate in these tests, just verify
# the 401 response for unauthenticated users
res = self.testapp.post_json(
"/api/v1/nodes/{}/wiki-edit".format(node_id),
{},
expect_errors=True,
)
self.assertEqual(res.status_int, 401)
def test_wiki_edit_node_not_found(self):
res = self.testapp.post_json(
"/api/v1/nodes/00000000-0000-0000-0000-000000000000/wiki-edit",
{"data": "Updated."},
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_revision_detail_not_found(self):
res = self.testapp.get(
"/api/v1/revisions/00000000-0000-0000-0000-000000000000",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
# ---------------------------------------------------------------------------
# Theme API
# ---------------------------------------------------------------------------
class TestThemeAPI(UndiggFunctionalTests):
"""Test theme generation endpoints."""
@classmethod
def setUpClass(cls):
try:
UndiggFunctionalTests.setUpClass.im_func(cls)
except AttributeError:
UndiggFunctionalTests.setUpClass.__func__(cls)
def setUp(self):
ns = get_or_create_namespace(self.dbsession, "theme-test.example.com")
self.dbsession.add(ns)
self.dbsession.flush()
self.namespace_name = str(ns.name)
self.tm.commit()
def test_get_theme_css(self):
res = self.testapp.get(
"/api/v1/themes/{}/css".format(self.namespace_name),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
self.assertIn("text/css", res.content_type)
body = res.body.decode("utf-8")
self.assertIn("--rb-bg:", body)
self.assertIn("--rb-text:", body)
self.assertIn("prefers-color-scheme: dark", body)
def test_get_theme_css_deterministic(self):
res1 = self.testapp.get(
"/api/v1/themes/{}/css".format(self.namespace_name))
res2 = self.testapp.get(
"/api/v1/themes/{}/css".format(self.namespace_name))
self.assertEqual(res1.body, res2.body)
def test_get_theme_css_cached(self):
res = self.testapp.get(
"/api/v1/themes/{}/css".format(self.namespace_name))
self.assertIn("max-age", res.headers.get("Cache-Control", ""))
def test_get_theme_css_not_found(self):
res = self.testapp.get(
"/api/v1/themes/nonexistent-ns-xyz.example.com/css",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_get_theme_preview(self):
res = self.testapp.get(
"/api/v1/themes/{}/preview".format(self.namespace_name),
expect_errors=True,
)
self.assertEqual(res.status_int, 200)
body = res.json
self.assertIn("palette", body)
self.assertIn("primary_hue", body["palette"])
self.assertIn("secondary_hue", body["palette"])
self.assertIn("accent_hue", body["palette"])
self.assertIn("css_url", body)
def test_get_theme_preview_not_found(self):
res = self.testapp.get(
"/api/v1/themes/nonexistent-ns-xyz.example.com/preview",
expect_errors=True,
)
self.assertEqual(res.status_int, 404)
def test_different_namespaces_different_themes(self):
ns2 = get_or_create_namespace(self.dbsession, "theme-test-2.example.com")
self.dbsession.add(ns2)
self.dbsession.flush()
self.tm.commit()
res1 = self.testapp.get(
"/api/v1/themes/{}/css".format(self.namespace_name))
res2 = self.testapp.get(
"/api/v1/themes/theme-test-2.example.com/css")
self.assertNotEqual(res1.body, res2.body)