torbenandClaude Sonnet 5.5
db09e89223
Files changed: - CHANGES.md - VERSION - docs/how-to/export-to-a-synced-folder.md - docs/reference/commands/exports.md - docs/reference/configuration.md - raw/CONTRACT.md - tools/CONTRACT.md - tools/README.md - tools/chemenu/catalog.py - tools/chemenu/commands/export_cmd.py - tools/chemenu/commands/index_build.py - tools/chemenu/config_shapes.py - tools/chemenu/export_index.py - tools/chemenu/guideline_export.py - tools/chemenu/page_export.py - tools/chemenu/tests/test_config_shapes.py - tools/chemenu/tests/test_export_run.py - tools/chemenu/tests/test_index_build.py Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SnAJ7Z3CpVD3PRbN73QtU2
1221 lines
49 KiB
Python
1221 lines
49 KiB
Python
"""`export run`, the `export` settings section, the `after_publish` step of
|
|
`publish`, the `doctor` line, and the digest check of `raw accept` - against a
|
|
real checkout and target folders in `tmp_path`.
|
|
|
|
The target's path carries a space on purpose: the folder a sync client keeps on
|
|
Windows is `OneDrive - <company>`."""
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
from pathlib import Path
|
|
from urllib.parse import unquote
|
|
|
|
import pytest
|
|
import typer
|
|
import yaml
|
|
from typer.testing import CliRunner
|
|
|
|
from chemenu import catalog, config, config_shapes, guideline_export, page_export
|
|
from chemenu.cli import app
|
|
from chemenu.commands import doctor
|
|
from chemenu.commands.git_publish import publish_command
|
|
from chemenu.frontmatter_io import write_page
|
|
from chemenu.search.service import load_pages_by_path
|
|
from chemenu.type_resolver import resolver
|
|
|
|
runner = CliRunner()
|
|
|
|
ALPHA = """
|
|
# Alpha
|
|
|
|
See [[Beta]], [[Beta|the beta page]] and [[Gamma]] for more, also [[Beta#Scope]].
|
|
Cited[^s-one]. Inline `[[Beta]]` stays.
|
|
|
|
```
|
|
[[Beta]] and [[Gamma]]
|
|
```
|
|
|
|
<!-- wikitool:footnotes -->
|
|
## Footnotes
|
|
|
|
[^s-one]: [[Source - One]]
|
|
<!-- /wikitool:footnotes -->
|
|
"""
|
|
|
|
BETA = """
|
|
# Beta
|
|
|
|
Back to [[Alpha]].
|
|
"""
|
|
|
|
|
|
def _git(root: Path, *args: str) -> str:
|
|
result = subprocess.run(["git", *args], cwd=root, capture_output=True, text=True)
|
|
assert result.returncode == 0, result.stderr
|
|
return result.stdout.strip()
|
|
|
|
|
|
def _commit(root: Path, message: str = "change") -> str:
|
|
_git(root, "add", "-A")
|
|
_git(root, "commit", "-q", "-m", message)
|
|
return _git(root, "rev-parse", "HEAD")
|
|
|
|
|
|
def _page(root: Path, rel: str, body: str, **frontmatter) -> Path:
|
|
path = root / "kb" / rel
|
|
meta = {"type": "types/concept.md", "tags": ["docs"], "summary": f"About {path.stem}.",
|
|
"modified": "2026-10-01", "related": ["Beta"], **frontmatter}
|
|
write_page(path, meta, body)
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def instance(tmp_path, monkeypatch):
|
|
root = tmp_path / "instance"
|
|
root.mkdir()
|
|
_git(root, "init", "-q", "-b", "main")
|
|
_git(root, "config", "user.name", "Test")
|
|
_git(root, "config", "user.email", "test@example.com")
|
|
(root / "incoming").mkdir()
|
|
(root / "raw").mkdir()
|
|
(root / "incoming" / ".gitkeep").write_text("", encoding="utf-8")
|
|
(root / ".gitignore").write_text(
|
|
"/incoming/*\n!/incoming/.gitkeep\n/.wikitool-local.json\n/quarantine/\n/reports/\n",
|
|
encoding="utf-8")
|
|
shutil.copytree(config._PACKAGE_ROOT / "types", root / "types")
|
|
monkeypatch.setattr(config, "TYPES_DIR", root / "types")
|
|
monkeypatch.setattr(resolver, "_repo_root", root)
|
|
_page(root, "concepts/Alpha.md", ALPHA)
|
|
_page(root, "concepts/protocols/Beta.md", BETA)
|
|
_page(root, "concepts/Gamma.md", "\n# Gamma\n\nPrivate.\n", tags=["private"])
|
|
_commit(root, "init")
|
|
(tmp_path / "OneDrive - Example").mkdir()
|
|
monkeypatch.setattr(config, "ROOT", root)
|
|
monkeypatch.setenv("WIKITOOL_SESSION_ID", "test-session")
|
|
return root
|
|
|
|
|
|
def _out(root: Path, name: str = "Pages") -> Path:
|
|
return root.parent / "OneDrive - Example" / name
|
|
|
|
|
|
def _configure(root: Path, **targets) -> None:
|
|
"""One target `pages` by default, with `docs` pages selected."""
|
|
if not targets:
|
|
targets = {"pages": {}}
|
|
section = {}
|
|
for name, extra in targets.items():
|
|
body = {"handler": "markdown", "path": str(_out(root, name)), "include": [["tags=docs"]]}
|
|
body.update(extra)
|
|
section[name] = body
|
|
commands = {body["command"]: ["unset"] for body in section.values() if body.get("command")}
|
|
export = {"targets": section, **({"commands": commands} if commands else {})}
|
|
(root / config.LOCAL_SETTINGS_FILENAME).write_text(
|
|
json.dumps({"schema": 1, "export": export}), encoding="utf-8")
|
|
|
|
|
|
def _set_commands(root: Path, commands: dict) -> None:
|
|
path = root / config.LOCAL_SETTINGS_FILENAME
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
data["export"]["commands"] = commands
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
|
|
def _run(*names: str):
|
|
return runner.invoke(app, ["export", "run", *names])
|
|
|
|
|
|
def _flat(text: str) -> str:
|
|
return " ".join(text.split())
|
|
|
|
|
|
def _snapshot(directory: Path) -> dict:
|
|
return {p.relative_to(directory).as_posix(): p.read_bytes()
|
|
for p in sorted(directory.rglob("*")) if p.is_file()}
|
|
|
|
|
|
def _mtimes(directory: Path) -> dict:
|
|
return {p.relative_to(directory).as_posix(): p.stat().st_mtime_ns
|
|
for p in sorted(directory.rglob("*")) if p.is_file()}
|
|
|
|
|
|
def _manifest(folder: Path) -> dict:
|
|
return json.loads((folder / page_export.MANIFEST_NAME).read_text(encoding="utf-8"))
|
|
|
|
|
|
# --- the section -------------------------------------------------------------
|
|
|
|
|
|
def test_no_section_is_an_error_naming_it_and_doctor_says_ok(instance):
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "no export section" in _flat(result.output)
|
|
assert _check("export").status == "OK"
|
|
|
|
|
|
@pytest.mark.parametrize("target, needle", [
|
|
({"path": "relative/folder"}, "absolute"),
|
|
({"include": []}, "include"),
|
|
({"include": [[]]}, "include"),
|
|
({"include": [["not a predicate"]]}, "predicate"),
|
|
({"handler": "pdf"}, "handler"),
|
|
({"frontmatter": "everything"}, "frontmatter"),
|
|
({"links": "hyper"}, "links"),
|
|
({"after_publish": "yes"}, "after_publish"),
|
|
({"command": "convert"}, "command"),
|
|
({"handler": "command"}, "command"),
|
|
])
|
|
def test_a_broken_section_stops_run_and_doctor_fails(instance, target, needle):
|
|
_configure(instance, pages=target)
|
|
result = _run()
|
|
assert result.exit_code == 1, result.output
|
|
assert needle in _flat(result.output)
|
|
assert _check("export").status == "FAIL"
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
def test_a_target_inside_the_instance_is_refused(instance):
|
|
_configure(instance, pages={"path": str(instance / "site")})
|
|
assert _run().exit_code == 1
|
|
assert _check("export").status == "FAIL"
|
|
|
|
|
|
def test_targets_lying_in_one_another_are_refused(instance):
|
|
_configure(instance, one={"path": str(_out(instance, "one"))},
|
|
two={"path": str(_out(instance, "one") / "inner")})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "lie in one another" in _flat(result.output)
|
|
|
|
|
|
def _local(root: Path) -> dict:
|
|
return json.loads((root / config.LOCAL_SETTINGS_FILENAME).read_text(encoding="utf-8"))
|
|
|
|
|
|
def _save(root: Path, data: dict) -> None:
|
|
(root / config.LOCAL_SETTINGS_FILENAME).write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
|
|
def _backup_section(root: Path) -> Path:
|
|
"""A backup whose target is the shared folder itself; returns that folder."""
|
|
shared = _out(root, "Chemenu")
|
|
data = _local(root)
|
|
data["backup"] = {"target": str(shared), "machine_data": False, "machine_name": "WORK-LAPTOP"}
|
|
_save(root, data)
|
|
return shared
|
|
|
|
|
|
def _channel(root: Path, path: Path) -> None:
|
|
data = _local(root)
|
|
data["intake"] = {"channels": {"inbox": {
|
|
"handler": "folder", "path": str(path), "max_bytes": 1024, "allowed_extensions": [".md"]}}}
|
|
_save(root, data)
|
|
|
|
|
|
def test_export_intake_and_backup_may_share_one_synced_folder(instance):
|
|
shared = _out(instance, "Chemenu")
|
|
_configure(instance, pages={"path": str(shared / "Export")})
|
|
(shared / "Inbox").mkdir(parents=True)
|
|
_backup_section(instance)
|
|
_channel(instance, shared / "Inbox")
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
assert (shared / "Export" / ".wikitool-export.json").is_file()
|
|
for check in (doctor.check_export(), doctor.check_upload_intake(), doctor.check_backup()):
|
|
assert check.status != "FAIL", (check.name, check.detail)
|
|
|
|
|
|
@pytest.mark.parametrize("where", ["git", "git/inner", "WORK-LAPTOP", "WORK-LAPTOP/inner", ""])
|
|
def test_an_export_equal_to_in_or_over_a_folder_the_backup_writes_is_refused(instance, where):
|
|
shared = _out(instance, "Chemenu")
|
|
_configure(instance, pages={"path": str(shared / where) if where else str(shared)})
|
|
_backup_section(instance)
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "lie in one another" in _flat(result.output)
|
|
assert "the backup's" in _flat(result.output)
|
|
assert not (shared / "git").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("where", ["git", "git/inner", "WORK-LAPTOP", "WORK-LAPTOP/inner"])
|
|
def test_doctor_fails_an_intake_folder_in_a_folder_the_backup_writes(instance, where):
|
|
shared = _out(instance, "Chemenu")
|
|
_configure(instance, pages={"path": str(shared / "Export")})
|
|
_backup_section(instance)
|
|
_channel(instance, shared / where)
|
|
check = doctor.check_upload_intake()
|
|
assert check.status == "FAIL" and "the backup's" in check.detail
|
|
|
|
|
|
def test_a_target_in_an_intake_folder_is_refused(instance):
|
|
_configure(instance, pages={"path": str(_out(instance, "pages"))})
|
|
_channel(instance, _out(instance, "pages") / "in")
|
|
assert _run().exit_code == 1
|
|
_channel(instance, _out(instance, "pages").parent)
|
|
assert _run().exit_code == 1
|
|
|
|
|
|
def test_a_target_below_another_machines_folder_is_refused_before_anything_is_written(instance):
|
|
shared = _out(instance, "Chemenu")
|
|
foreign = shared / "OTHER-PC"
|
|
foreign.mkdir(parents=True)
|
|
(foreign / ".chemenu-backup.json").write_text("{}", encoding="utf-8")
|
|
_configure(instance, pages={"path": str(foreign / "deeper" / "Export")})
|
|
_backup_section(instance)
|
|
before = {p: p.read_bytes() for p in shared.rglob("*") if p.is_file()}
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "backup data" in _flat(result.output)
|
|
assert {p: p.read_bytes() for p in shared.rglob("*") if p.is_file()} == before
|
|
assert not (foreign / "deeper").exists()
|
|
|
|
|
|
def test_a_target_in_a_bare_repository_is_refused(instance):
|
|
bare = _out(instance, "repo.git")
|
|
(bare / "objects").mkdir(parents=True)
|
|
(bare / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
|
|
_configure(instance, pages={"path": str(bare / "Export")})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "is a git repository" in _flat(result.output)
|
|
assert not (bare / "Export").exists()
|
|
|
|
|
|
def test_the_operator_only_fields_cannot_be_set_by_the_agent():
|
|
export = config_shapes.EXPORT
|
|
assert export.field("targets").user_only and export.field("commands").user_only
|
|
assert export.field("targets").tier == config_shapes.CHECKOUT
|
|
assert export.field("commands").tier == config_shapes.CHECKOUT
|
|
|
|
|
|
def test_an_unknown_target_name_is_an_error_and_changes_nothing(instance):
|
|
_configure(instance)
|
|
result = _run("pages", "nothing")
|
|
assert result.exit_code == 1
|
|
assert "unknown export target" in _flat(result.output)
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
# --- what a run writes -------------------------------------------------------
|
|
|
|
|
|
def test_a_run_writes_the_selected_pages_with_marker_and_links(instance):
|
|
_configure(instance)
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
folder = _out(instance, "pages")
|
|
assert set(_snapshot(folder)) == {
|
|
"concepts/Alpha.md", "concepts/protocols/Beta.md", page_export.MANIFEST_NAME}
|
|
commit = _git(instance, "rev-parse", "HEAD")
|
|
alpha = (folder / "concepts" / "Alpha.md").read_text(encoding="utf-8")
|
|
lines = alpha.split("\n")
|
|
assert lines[0] == "---"
|
|
assert lines[1] == f'wikitool_export: "kind=pages instance=local commit={commit}"'
|
|
assert "title: Alpha" in alpha and "summary: About Alpha." in alpha
|
|
assert "tags" not in alpha and "related" not in alpha
|
|
assert "[Beta](protocols/Beta.md)" in alpha and "[the beta page](protocols/Beta.md)" in alpha
|
|
assert "also [Beta](protocols/Beta.md)" in alpha
|
|
assert "and Gamma for more" in alpha
|
|
assert "Cited[^s-one]." in alpha and "[^s-one]: " in alpha
|
|
assert "Inline `[[Beta]]` stays." in alpha
|
|
assert "```\n[[Beta]] and [[Gamma]]\n```" in alpha
|
|
assert alpha.endswith("\n") and not alpha.endswith("\n\n") and "\r" not in alpha
|
|
beta = (folder / "concepts" / "protocols" / "Beta.md").read_text(encoding="utf-8")
|
|
assert "[Alpha](../Alpha.md)" in beta
|
|
assert guideline_export.file_is_export(folder / "concepts" / "Alpha.md")
|
|
out = _flat(result.output)
|
|
assert "new (2)" in out and "Alpha" in out and "unchanged: 0" in out
|
|
|
|
|
|
def test_frontmatter_modes_and_links_text(instance):
|
|
_configure(instance, minimal={"frontmatter": "minimal", "links": "text"},
|
|
full={"frontmatter": "full"})
|
|
assert _run().exit_code == 0
|
|
minimal = (_out(instance, "minimal") / "concepts" / "Alpha.md").read_text(encoding="utf-8")
|
|
assert minimal.startswith('---\nwikitool_export: "kind=pages')
|
|
assert minimal.split("---\n")[1].count("\n") == 1
|
|
assert "[Beta](" not in minimal and "See Beta, the beta page and Gamma" in minimal
|
|
full = (_out(instance, "full") / "concepts" / "Alpha.md").read_text(encoding="utf-8")
|
|
assert "tags:" in full and "type: types/concept.md" in full and "related:" in full
|
|
|
|
|
|
def test_selections_are_or_and_predicates_are_and(instance):
|
|
_configure(instance, pages={"include": [["tags=docs", "summary~Alpha"], ["tags=private"]]})
|
|
assert _run().exit_code == 0
|
|
assert set(_snapshot(_out(instance, "pages"))) == {
|
|
"concepts/Alpha.md", "concepts/Gamma.md", page_export.MANIFEST_NAME}
|
|
|
|
|
|
def test_an_unknown_field_is_an_error_at_run(instance):
|
|
_configure(instance, pages={"include": [["no_such_field=x"]]})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "unknown field" in _flat(result.output)
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
def test_a_rerun_on_an_unchanged_corpus_touches_nothing(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
before, stamps = _snapshot(folder), _mtimes(folder)
|
|
result = _run()
|
|
assert result.exit_code == 0
|
|
assert _snapshot(folder) == before and _mtimes(folder) == stamps
|
|
assert "unchanged: 2" in _flat(result.output)
|
|
assert "new (" not in result.output and "changed (" not in result.output
|
|
|
|
|
|
def test_an_instance_commit_that_touches_no_selected_page_changes_nothing(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
stamps = _mtimes(folder)
|
|
(instance / "README.md").write_text("other\n", encoding="utf-8")
|
|
_commit(instance, "unrelated")
|
|
assert _run().exit_code == 0
|
|
assert _mtimes(folder) == stamps
|
|
|
|
|
|
def _marker_commit(path: Path) -> str:
|
|
line = path.read_text(encoding="utf-8").split("\n")[1]
|
|
return line.rsplit("commit=", 1)[1].rstrip('"')
|
|
|
|
|
|
def test_a_commit_to_one_page_rewrites_that_file_alone(instance):
|
|
_configure(instance, pages={"include": [["tags=docs"], ["tags=private"]]})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
before, stamps = _snapshot(folder), _mtimes(folder)
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
commit = _commit(instance, "edit alpha")
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
out = _flat(result.output)
|
|
assert "changed (1): Alpha" in out and "unchanged: 2" in out and "new (" not in out
|
|
after, after_stamps = _snapshot(folder), _mtimes(folder)
|
|
for rel in ("concepts/protocols/Beta.md", "concepts/Gamma.md"):
|
|
assert after[rel] == before[rel] and after_stamps[rel] == stamps[rel]
|
|
assert _marker_commit(folder / "concepts" / "Alpha.md") == commit
|
|
|
|
|
|
def test_each_file_carries_its_own_pages_commit_and_the_manifest_the_newest(instance):
|
|
_page(instance, "concepts/Löwe [1].md", "\n# Löwe\n\nText.\n")
|
|
first = _commit(instance, "lion")
|
|
_page(instance, "concepts/protocols/Beta.md", BETA + "\nMore.\n")
|
|
newest = _commit(instance, "edit beta")
|
|
(instance / "README.md").write_text("other\n", encoding="utf-8")
|
|
_commit(instance, "unrelated")
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
init = _git(instance, "rev-list", "--max-parents=0", "HEAD")
|
|
assert _marker_commit(folder / "concepts" / "Alpha.md") == init
|
|
assert _marker_commit(folder / "concepts" / "Löwe [1].md") == first
|
|
assert _marker_commit(folder / "concepts" / "protocols" / "Beta.md") == newest
|
|
assert _manifest(folder)["commit"] == newest
|
|
|
|
|
|
def test_a_page_entering_the_selection_rewrites_only_the_files_whose_text_changes(instance):
|
|
_configure(instance, pages={"include": [["tags=docs"], ["summary~Gamma"]]})
|
|
_page(instance, "concepts/Gamma.md", "\n# Gamma\n\nPrivate.\n", tags=["private"], summary="Hidden.")
|
|
_commit(instance, "hide gamma")
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
before, stamps = _snapshot(folder), _mtimes(folder)
|
|
assert "and Gamma for more" in before["concepts/Alpha.md"].decode("utf-8")
|
|
_page(instance, "concepts/Gamma.md", "\n# Gamma\n\nPrivate.\n", tags=["private"])
|
|
_commit(instance, "show gamma")
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
out = _flat(result.output)
|
|
assert "new (1): Gamma" in out and "changed (1): Alpha" in out and "unchanged: 1" in out
|
|
after, after_stamps = _snapshot(folder), _mtimes(folder)
|
|
assert "[Gamma](Gamma.md)" in after["concepts/Alpha.md"].decode("utf-8")
|
|
beta = "concepts/protocols/Beta.md"
|
|
assert after[beta] == before[beta] and after_stamps[beta] == stamps[beta]
|
|
|
|
|
|
def test_a_renamed_page_carries_the_commit_of_the_rename(instance):
|
|
(instance / "README.md").write_text("other\n", encoding="utf-8")
|
|
_commit(instance, "unrelated")
|
|
result = runner.invoke(app, ["rename", "--from", "Beta", "--to", "Epsilon"])
|
|
assert result.exit_code == 0, result.output
|
|
renamed = _commit(instance, "rename")
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
assert _marker_commit(folder / "concepts" / "protocols" / "Epsilon.md") == renamed
|
|
|
|
|
|
def test_the_git_calls_do_not_grow_with_the_pages(instance, monkeypatch):
|
|
_configure(instance)
|
|
calls = []
|
|
real = guideline_export._git
|
|
|
|
def counted(args, root, stdin=None):
|
|
calls.append(args)
|
|
return real(args, root, stdin)
|
|
|
|
monkeypatch.setattr(guideline_export, "_git", counted)
|
|
assert _run().exit_code == 0
|
|
few = len(calls)
|
|
for index in range(30):
|
|
_page(instance, f"concepts/Many{index}.md", f"\n# Many{index}\n\nText.\n")
|
|
_commit(instance, "many")
|
|
calls.clear()
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
assert "new (30)" in _flat(result.output)
|
|
assert len(calls) == few
|
|
assert sum(1 for args in calls if "log" in args) == 1
|
|
|
|
|
|
def test_a_changed_page_a_new_page_and_a_removed_page(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_page(instance, "concepts/Delta.md", "\n# Delta\n\nNew.\n")
|
|
_page(instance, "concepts/protocols/Beta.md", BETA, tags=["private"])
|
|
commit = _commit(instance, "edit")
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
out = _flat(result.output)
|
|
assert "new (1): Delta" in out and "changed (1):" in out and "removed (1):" in out
|
|
assert (folder / "concepts" / "Delta.md").is_file()
|
|
assert not (folder / "concepts" / "protocols" / "Beta.md").exists()
|
|
assert not (folder / "concepts" / "protocols").exists()
|
|
assert "More." in (folder / "concepts" / "Alpha.md").read_text(encoding="utf-8")
|
|
assert _manifest(folder)["commit"] == commit
|
|
assert [e["path"] for e in _manifest(folder)["files"]] == ["concepts/Alpha.md", "concepts/Delta.md"]
|
|
|
|
|
|
def test_the_selection_growing_needs_no_gate(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
for index in range(30):
|
|
_page(instance, f"concepts/Many{index}.md", f"\n# Many{index}\n\nText.\n")
|
|
_commit(instance, "many")
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
assert "new (30)" in _flat(result.output)
|
|
|
|
|
|
def test_foreign_files_stay_and_only_created_directories_are_pruned(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
(folder / "notes.txt").write_bytes(b"mine\r\n")
|
|
(folder / "concepts" / "mine.txt").write_bytes(b"also mine")
|
|
(folder / "concepts" / "protocols" / "mine.txt").write_bytes(b"in a created dir")
|
|
_page(instance, "concepts/protocols/Beta.md", BETA, tags=["private"])
|
|
_commit(instance, "drop beta")
|
|
assert _run().exit_code == 0
|
|
assert (folder / "notes.txt").read_bytes() == b"mine\r\n"
|
|
assert (folder / "concepts" / "mine.txt").read_bytes() == b"also mine"
|
|
assert (folder / "concepts" / "protocols" / "mine.txt").read_bytes() == b"in a created dir"
|
|
assert (folder / "concepts" / "Alpha.md").is_file()
|
|
|
|
|
|
def test_a_directory_that_existed_before_the_export_is_kept(instance):
|
|
_page(instance, "concepts/protocols/Beta.md", BETA, tags=["private"])
|
|
_commit(instance, "no beta yet")
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
(folder / "concepts" / "protocols").mkdir()
|
|
_page(instance, "concepts/protocols/Beta.md", BETA)
|
|
_commit(instance, "beta")
|
|
assert _run().exit_code == 0
|
|
assert "concepts/protocols" not in _manifest(folder)["dirs"]
|
|
_page(instance, "concepts/protocols/Beta.md", BETA, tags=["private"])
|
|
_commit(instance, "drop beta")
|
|
assert _run().exit_code == 0
|
|
assert (folder / "concepts" / "protocols").is_dir()
|
|
|
|
|
|
def test_a_non_empty_folder_without_a_manifest_is_refused(instance):
|
|
folder = _out(instance, "pages")
|
|
folder.mkdir()
|
|
(folder / "mine.md").write_text("mine\n", encoding="utf-8")
|
|
_configure(instance)
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "not empty" in _flat(result.output) and "mine.md" not in _snapshot(folder).keys() - {"mine.md"}
|
|
assert _snapshot(folder) == {"mine.md": b"mine\n"}
|
|
|
|
|
|
def test_an_empty_existing_folder_is_taken(instance):
|
|
_out(instance, "pages").mkdir()
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
|
|
|
|
def test_a_file_changed_by_hand_aborts_the_target_before_any_write(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
edited = folder / "concepts" / "protocols" / "Beta.md"
|
|
edited.write_text("my own words\n", encoding="utf-8")
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_commit(instance, "edit")
|
|
before = _snapshot(folder)
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "concepts/protocols/Beta.md" in _flat(result.output)
|
|
assert _snapshot(folder) == before
|
|
|
|
|
|
def test_a_missing_manifest_file_is_rewritten(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
expected = _snapshot(folder)
|
|
(folder / "concepts" / "Alpha.md").unlink()
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
assert _snapshot(folder) == expected
|
|
|
|
|
|
def test_a_crash_between_the_files_and_the_manifest_is_recovered(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
old_manifest = (folder / page_export.MANIFEST_NAME).read_bytes()
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_commit(instance, "edit")
|
|
assert _run().exit_code == 0
|
|
new_state = _snapshot(folder)
|
|
(folder / page_export.MANIFEST_NAME).write_bytes(old_manifest)
|
|
assert _run().exit_code == 0
|
|
assert _snapshot(folder) == new_state
|
|
|
|
|
|
def test_zero_selected_pages_is_an_error_and_deletes_nothing(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
before = _snapshot(folder)
|
|
_configure(instance, pages={"include": [["tags=nothing-has-this"]]})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "no page matches" in _flat(result.output)
|
|
assert _snapshot(folder) == before
|
|
|
|
|
|
def test_a_dirty_kb_is_refused(instance):
|
|
_configure(instance)
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nunsaved\n")
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "uncommitted" in _flat(result.output)
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
def test_a_page_with_unreadable_frontmatter_stops_the_run(instance):
|
|
_configure(instance)
|
|
(instance / "kb" / "concepts" / "Broken.md").write_text("---\ntype: [unclosed\n---\nBody\n", encoding="utf-8")
|
|
_commit(instance, "broken")
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "Broken.md" in result.output
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
def test_targets_are_independent_and_any_failure_is_exit_1(instance):
|
|
bad = _out(instance, "bad")
|
|
bad.mkdir()
|
|
(bad / "mine.md").write_text("mine\n", encoding="utf-8")
|
|
_configure(instance, bad={}, good={})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert (_out(instance, "good") / "concepts" / "Alpha.md").is_file()
|
|
assert _snapshot(bad) == {"mine.md": b"mine\n"}
|
|
assert "bad" in result.output and "failed" in result.output
|
|
|
|
|
|
def test_named_targets_run_alone(instance):
|
|
_configure(instance, one={}, two={})
|
|
assert _run("two").exit_code == 0
|
|
assert _out(instance, "two").is_dir() and not _out(instance, "one").exists()
|
|
|
|
|
|
def test_a_case_only_collision_is_refused(instance):
|
|
_page(instance, "concepts/alpha.md", "\n# alpha\n\nLower.\n")
|
|
_commit(instance, "case")
|
|
_configure(instance)
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "differ only in case" in _flat(result.output)
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
# --- the command handler -----------------------------------------------------
|
|
|
|
|
|
def _converter(root: Path, body: str) -> list[str]:
|
|
script = root.parent / "convert.py"
|
|
script.write_text(textwrap.dedent(body), encoding="utf-8")
|
|
return [sys.executable, str(script)]
|
|
|
|
|
|
def _command_target(root: Path, program: list[str]) -> None:
|
|
_configure(root, pages={"handler": "command", "command": "convert"})
|
|
_set_commands(root, {"convert": program})
|
|
|
|
|
|
COPY_AS_TXT = """
|
|
import pathlib, sys
|
|
src, out = map(pathlib.Path, sys.argv[1:3])
|
|
for f in src.rglob("*.md"):
|
|
dest = out / f.relative_to(src).with_suffix(".txt")
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(f.read_text(encoding="utf-8").upper(), encoding="utf-8")
|
|
"""
|
|
|
|
|
|
def test_the_command_handler_exports_what_the_program_leaves(instance):
|
|
_command_target(instance, _converter(instance, COPY_AS_TXT))
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
folder = _out(instance, "pages")
|
|
assert set(_snapshot(folder)) == {
|
|
"concepts/Alpha.txt", "concepts/protocols/Beta.txt", page_export.MANIFEST_NAME}
|
|
assert "WIKITOOL_EXPORT" in (folder / "concepts" / "Alpha.txt").read_text(encoding="utf-8")
|
|
stamps = _mtimes(folder)
|
|
assert _run().exit_code == 0 and _mtimes(folder) == stamps
|
|
|
|
|
|
def test_the_command_gets_two_directories_no_stdin_and_the_staging_is_removed(instance, tmp_path):
|
|
record = tmp_path / "record.json"
|
|
program = _converter(instance, f"""
|
|
import json, os, pathlib, sys
|
|
src, out = pathlib.Path(sys.argv[1]), pathlib.Path(sys.argv[2])
|
|
stdin = sys.stdin.read()
|
|
pathlib.Path({str(record)!r}).write_text(json.dumps({{
|
|
"args": sys.argv[1:], "stdin": stdin, "cwd": os.getcwd(),
|
|
"src_files": sorted(p.name for p in src.rglob("*.md")), "out_files": os.listdir(out)}}))
|
|
(out / "all.txt").write_text("done")
|
|
""")
|
|
_command_target(instance, program)
|
|
assert _run().exit_code == 0
|
|
seen = json.loads(record.read_text(encoding="utf-8"))
|
|
assert seen["stdin"] == "" and seen["out_files"] == [] and seen["src_files"] == ["Alpha.md", "Beta.md"]
|
|
assert Path(seen["args"][0]).name == "markdown" and Path(seen["args"][1]).name == "output"
|
|
assert not Path(seen["args"][0]).parent.exists()
|
|
|
|
|
|
def test_a_failing_command_leaves_the_target_as_it_was(instance):
|
|
_command_target(instance, _converter(instance, COPY_AS_TXT))
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
before = _snapshot(folder)
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_commit(instance, "edit")
|
|
_set_commands(instance, {"convert": _converter(instance, """
|
|
import sys
|
|
sys.stderr.write("x" * 5000 + "boom at the end")
|
|
sys.exit(3)
|
|
""")})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
text = _flat(result.output)
|
|
assert "exit 3" in text and "boom at the end" in text and "x" * 3000 not in text
|
|
assert _snapshot(folder) == before
|
|
|
|
|
|
def test_a_missing_program_and_a_timeout_are_target_errors(instance, monkeypatch):
|
|
_command_target(instance, ["/nonexistent/converter"])
|
|
result = _run()
|
|
assert result.exit_code == 1 and "could not be started" in _flat(result.output)
|
|
_set_commands(instance, {"convert": _converter(instance, "import time; time.sleep(30)")})
|
|
monkeypatch.setattr(page_export, "COMMAND_TIMEOUT_SECONDS", 1)
|
|
result = _run()
|
|
assert result.exit_code == 1 and "did not finish" in _flat(result.output)
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="symlinks")
|
|
def test_symlinks_in_the_output_are_skipped_and_reported(instance):
|
|
_command_target(instance, _converter(instance, """
|
|
import os, pathlib, sys
|
|
out = pathlib.Path(sys.argv[2])
|
|
(out / "real.txt").write_text("real")
|
|
os.symlink("/etc/hostname", out / "link.txt")
|
|
"""))
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
assert set(_snapshot(_out(instance, "pages"))) == {"real.txt", page_export.MANIFEST_NAME}
|
|
assert "link.txt" in result.output
|
|
|
|
|
|
def test_a_converter_that_leaves_nothing_is_an_error(instance):
|
|
_command_target(instance, _converter(instance, "pass"))
|
|
assert _run().exit_code == 1
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
# --- publish -----------------------------------------------------------------
|
|
|
|
|
|
def _publish(**overrides):
|
|
kwargs = dict(message="change", push=False, confirm=None, confirm_rebase=None, yes=False,
|
|
threshold=10, remote="origin", branch="main", path=None)
|
|
kwargs.update(overrides)
|
|
publish_command(**kwargs)
|
|
|
|
|
|
def test_publish_exports_after_the_commit_with_no_push_too(instance):
|
|
_configure(instance, pages={"after_publish": True}, manual={})
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_publish()
|
|
folder = _out(instance, "pages")
|
|
head = _git(instance, "rev-parse", "HEAD")
|
|
assert _manifest(folder)["commit"] == head
|
|
assert "More." in (folder / "concepts" / "Alpha.md").read_text(encoding="utf-8")
|
|
assert not _out(instance, "manual").exists()
|
|
|
|
|
|
def test_a_failed_export_after_publish_is_a_warning_and_the_commit_stands(instance, capsys):
|
|
folder = _out(instance, "pages")
|
|
folder.mkdir()
|
|
(folder / "mine.md").write_text("mine\n", encoding="utf-8")
|
|
_configure(instance, pages={"after_publish": True})
|
|
head = _git(instance, "rev-parse", "HEAD")
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_publish()
|
|
out = capsys.readouterr().out
|
|
assert "WARN" in out and "export run" in out
|
|
assert _git(instance, "rev-parse", "HEAD") != head
|
|
|
|
|
|
def test_publish_without_after_publish_exports_nothing(instance):
|
|
_configure(instance)
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_publish()
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
def test_a_malformed_section_stops_publish_only_when_a_target_asks_for_it(instance):
|
|
_configure(instance, pages={"after_publish": True, "include": []})
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
head = _git(instance, "rev-parse", "HEAD")
|
|
with pytest.raises(typer.Exit) as exc:
|
|
_publish()
|
|
assert exc.value.exit_code == 1 and _git(instance, "rev-parse", "HEAD") == head
|
|
_configure(instance, pages={"include": []})
|
|
_publish()
|
|
assert _git(instance, "rev-parse", "HEAD") != head
|
|
|
|
|
|
# --- doctor ------------------------------------------------------------------
|
|
|
|
|
|
def _check(name: str):
|
|
assert name == "export"
|
|
return doctor.check_export()
|
|
|
|
|
|
def test_doctor_lists_the_targets(instance):
|
|
_configure(instance, pages={"after_publish": True})
|
|
check = _check("export")
|
|
assert check.status == "OK"
|
|
assert "pages" in check.detail and "after publish" in check.detail
|
|
|
|
|
|
# --- raw accept --------------------------------------------------------------
|
|
|
|
|
|
def _accept(path: Path):
|
|
return runner.invoke(app, ["raw", "accept", str(path), "--fidelity", "verbatim",
|
|
"--authority", "normative"])
|
|
|
|
|
|
def test_raw_accept_refuses_a_file_carrying_the_page_export_marker(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
exported = (_out(instance, "pages") / "concepts" / "Alpha.md").read_bytes()
|
|
(instance / "incoming" / "Alpha.md").write_bytes(exported)
|
|
result = _accept(instance / "incoming" / "Alpha.md")
|
|
assert result.exit_code == 1
|
|
assert "export" in _flat(result.output) and "nothing was moved" in _flat(result.output)
|
|
assert (instance / "incoming" / "Alpha.md").is_file()
|
|
assert not any((instance / "raw").rglob("*.md"))
|
|
|
|
|
|
def test_raw_accept_refuses_an_exported_file_whose_marker_a_converter_dropped(instance):
|
|
_command_target(instance, _converter(instance, COPY_AS_TXT))
|
|
assert _run().exit_code == 0
|
|
exported = (_out(instance, "pages") / "concepts" / "Alpha.txt").read_bytes()
|
|
folder = instance / "incoming" / "pack"
|
|
folder.mkdir()
|
|
(folder / "note.txt").write_text("ordinary\n", encoding="utf-8")
|
|
(folder / "Alpha.txt").write_bytes(exported)
|
|
(instance / "incoming" / "Alpha.txt").write_bytes(exported)
|
|
for target in (folder, instance / "incoming" / "Alpha.txt"):
|
|
before = _snapshot(instance / "incoming")
|
|
result = _accept(target)
|
|
assert result.exit_code == 1, result.output
|
|
assert "configured export target" in _flat(result.output)
|
|
assert "Alpha.txt" in result.output
|
|
assert _snapshot(instance / "incoming") == before
|
|
assert not any(p.is_file() for p in (instance / "raw").rglob("*"))
|
|
|
|
|
|
def test_raw_accept_warns_and_goes_on_when_a_manifest_is_missing(instance):
|
|
_configure(instance)
|
|
(instance / "incoming" / "note.md").write_text("# Note\n\nOrdinary.\n", encoding="utf-8")
|
|
result = _accept(instance / "incoming" / "note.md")
|
|
assert "WARN" in result.output and "no manifest" in _flat(result.output)
|
|
assert "configured export target" not in result.output
|
|
|
|
|
|
def test_raw_accept_warns_and_goes_on_when_the_section_is_broken(instance):
|
|
_configure(instance, pages={"include": []})
|
|
(instance / "incoming" / "note.md").write_text("# Note\n\nOrdinary.\n", encoding="utf-8")
|
|
result = _accept(instance / "incoming" / "note.md")
|
|
assert "WARN" in result.output and "export section is broken" in _flat(result.output)
|
|
assert "configured export target" not in result.output
|
|
|
|
|
|
def test_known_digests_lists_the_manifest_files(instance):
|
|
_configure(instance)
|
|
assert _run().exit_code == 0
|
|
digests, warnings = page_export.known_digests()
|
|
alpha = (_out(instance, "pages") / "concepts" / "Alpha.md").read_bytes()
|
|
assert hashlib.sha256(alpha).hexdigest() in digests and warnings == []
|
|
|
|
|
|
# --- the marker --------------------------------------------------------------
|
|
|
|
|
|
FRONT = b'---\nwikitool_export: "kind=pages instance=local commit=abc"\ntitle: A\n---\n\nBody\n'
|
|
|
|
|
|
@pytest.mark.parametrize("data, expected", [
|
|
(b"<!-- wikitool:export kind=guidelines -->\nbody", True),
|
|
(FRONT, True),
|
|
(b"\xef\xbb\xbf" + FRONT, True),
|
|
(FRONT.replace(b"\n", b"\r\n"), True),
|
|
(b"---\ntitle: A\n---\nwikitool_export: x\n", False),
|
|
(b"---\ntitle: A\nwikitool_export: x\n", False),
|
|
(b"# Title\n---\nwikitool_export: x\n---\n", False),
|
|
(b"---\ntitle: A\n---\n\nBody\n", False),
|
|
(b"", False),
|
|
])
|
|
def test_is_export_recognises_both_forms(data, expected):
|
|
assert guideline_export.is_export(data) is expected
|
|
|
|
|
|
def test_the_key_beyond_the_scan_window_is_not_seen():
|
|
padding = b"x: " + b"y" * 100 + b"\n"
|
|
data = b"---\n" + padding * (guideline_export.SCAN_BYTES // len(padding) + 2) + b"wikitool_export: z\n---\n"
|
|
assert guideline_export.is_export(data) is False
|
|
|
|
|
|
def test_file_is_export_reads_files(tmp_path):
|
|
(tmp_path / "a.md").write_bytes(FRONT)
|
|
(tmp_path / "b.md").write_bytes(b"plain\n")
|
|
assert guideline_export.file_is_export(tmp_path / "a.md")
|
|
assert not guideline_export.file_is_export(tmp_path / "b.md")
|
|
|
|
|
|
def test_the_marker_key_is_defined_once():
|
|
package = Path(guideline_export.__file__).parent
|
|
definitions = [
|
|
path for path in package.rglob("*.py")
|
|
if "tests" not in path.parts and "wikitool_export:" in path.read_text("utf-8")
|
|
and 'b"wikitool_export:"' in path.read_text("utf-8")
|
|
]
|
|
assert definitions == [Path(guideline_export.__file__)]
|
|
|
|
|
|
# --- include "all" and the index files -----------------------------------------
|
|
|
|
META_FILES = {
|
|
"index.md": "# Wiki Index\n", "log.md": "# Log\n", "provenance.md": "# Provenance\n",
|
|
"CONTRACT.md": "# Contract\n", "CONVENTIONS.md": "# Conventions\n",
|
|
"concepts/INDEX.md": "# concepts/ - Index\n", "concepts/COLLECTION.md": "# Concepts\n",
|
|
}
|
|
INDEX_NAMES = ("index.md", "INDEX.md")
|
|
|
|
|
|
def _all(root: Path, **extra) -> dict:
|
|
return {"include": "all", **extra}
|
|
|
|
|
|
def _add_meta_and_stray(root: Path) -> None:
|
|
for rel, text in META_FILES.items():
|
|
(root / "kb" / rel).write_text(text, encoding="utf-8")
|
|
_page(root, "Stray.md", "\n# Stray\n\nDirectly under kb.\n")
|
|
_commit(root, "meta and stray")
|
|
|
|
|
|
def _links(folder: Path, rel: str) -> list[str]:
|
|
text = (folder / rel).read_text(encoding="utf-8")
|
|
return re.findall(r"\]\(([^)]*)\)", text)
|
|
|
|
|
|
def _resolve(rel: str, link: str) -> tuple[str, str]:
|
|
path, _, fragment = link.partition("#")
|
|
return posixpath.normpath(posixpath.join(posixpath.dirname(rel), unquote(path))), unquote(fragment)
|
|
|
|
|
|
def _headings(text: str) -> set[str]:
|
|
return {catalog.anchor(m.group(1)) for m in re.finditer(r"^#+ (.+)$", text, re.MULTILINE)}
|
|
|
|
|
|
def _index_files(folder: Path) -> list[str]:
|
|
return sorted(rel for rel in _snapshot(folder) if posixpath.basename(rel) in INDEX_NAMES
|
|
and guideline_export.file_is_export(folder / rel)
|
|
and "kind=index" in (folder / rel).read_text(encoding="utf-8").split("\n")[1])
|
|
|
|
|
|
def test_all_writes_every_page_search_knows_and_no_generated_or_meta_file(instance):
|
|
_add_meta_and_stray(instance)
|
|
_configure(instance, whole=_all(instance))
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
pages = load_pages_by_path(instance / "kb", instance)
|
|
expected = {page.path.relative_to(instance / "kb").as_posix() for page in pages.values()}
|
|
assert "Stray.md" in expected and "concepts/Gamma.md" in expected
|
|
written = set(_snapshot(_out(instance, "whole")))
|
|
assert written == expected | {page_export.MANIFEST_NAME}
|
|
for meta in META_FILES:
|
|
assert meta not in written
|
|
|
|
|
|
@pytest.mark.parametrize("include", [None, [], [[]], "All", "*", "everything", "", 1, True])
|
|
def test_every_other_include_than_all_or_selections_is_an_error_and_writes_nothing(instance, include):
|
|
_configure(instance, pages={})
|
|
path = instance / config.LOCAL_SETTINGS_FILENAME
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if include is None:
|
|
del data["export"]["targets"]["pages"]["include"]
|
|
else:
|
|
data["export"]["targets"]["pages"]["include"] = include
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
result = _run()
|
|
assert result.exit_code == 1, result.output
|
|
message = _flat(result.output)
|
|
assert '"all"' in message and "array of selections" in message
|
|
assert _check("export").status == "FAIL"
|
|
assert not _out(instance, "pages").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["yes", 1, "true", None])
|
|
def test_index_must_be_a_boolean(instance, value):
|
|
_configure(instance, pages={"index": value})
|
|
result = _run()
|
|
assert result.exit_code == 1 and "`index` must be true or false" in _flat(result.output)
|
|
assert _check("export").status == "FAIL"
|
|
|
|
|
|
def test_without_index_or_with_false_no_index_file_is_written(instance):
|
|
_add_meta_and_stray(instance)
|
|
_configure(instance, absent=_all(instance), off=_all(instance, index=False))
|
|
assert _run().exit_code == 0
|
|
for name in ("absent", "off"):
|
|
assert not [rel for rel in _snapshot(_out(instance, name))
|
|
if posixpath.basename(rel) in INDEX_NAMES]
|
|
|
|
|
|
def _many_pages(root: Path, count: int) -> None:
|
|
for number in range(count):
|
|
_page(root, f"concepts/protocols/P{number:02d}.md", f"\n# P{number:02d}\n\nText.\n")
|
|
_commit(root, "many")
|
|
|
|
|
|
def test_every_link_of_every_index_file_lands_and_every_page_is_reachable(instance):
|
|
_add_meta_and_stray(instance)
|
|
_many_pages(instance, catalog.SHARD_THRESHOLD)
|
|
_page(instance, "concepts/areas/Solo.md", "\n# Solo\n\nSmall area.\n")
|
|
_commit(instance, "small area")
|
|
_configure(instance, whole=_all(instance, index=True))
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "whole")
|
|
files = _snapshot(folder)
|
|
indexes = _index_files(folder)
|
|
assert "index.md" in indexes and "concepts/INDEX.md" in indexes
|
|
assert "concepts/protocols/INDEX.md" in indexes # over the threshold: a file of its own
|
|
assert "concepts/areas/INDEX.md" not in indexes # under it: an anchor
|
|
reached, queue = {"index.md"}, ["index.md"]
|
|
while queue:
|
|
rel = queue.pop()
|
|
for link in _links(folder, rel):
|
|
target, fragment = _resolve(rel, link)
|
|
assert target in files, f"{rel} links to {link}, which is not in the target"
|
|
if fragment:
|
|
assert fragment in _headings(files[target].decode("utf-8")), f"{rel}: no heading for #{fragment}"
|
|
if target in indexes and target not in reached:
|
|
reached.add(target)
|
|
queue.append(target)
|
|
pages = set(files) - set(indexes) - {page_export.MANIFEST_NAME}
|
|
linked = {_resolve(rel, link)[0] for rel in reached for link in _links(folder, rel)}
|
|
assert pages <= linked
|
|
root_text = files["index.md"].decode("utf-8")
|
|
assert "concepts/INDEX.md#" in root_text and "concepts/protocols/INDEX.md" in root_text
|
|
assert "## Pages outside any collection" in root_text and "(Stray.md)" in root_text
|
|
|
|
|
|
def test_the_index_names_only_what_is_exported(instance):
|
|
_page(instance, "other/Delta.md", "\n# Delta\n\nNot shared.\n", tags=["private"])
|
|
_page(instance, "concepts/areas/Hidden.md", "\n# Hidden\n\nNot shared.\n", tags=["private"])
|
|
_commit(instance, "private pages")
|
|
_configure(instance, pages={"index": True})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
assert not (folder / "other").exists()
|
|
for rel in _index_files(folder):
|
|
text = (folder / rel).read_text(encoding="utf-8")
|
|
assert not any(word in text for word in ("Gamma", "Delta", "Hidden", "other/", "areas")), rel
|
|
assert "Alpha" in (folder / "concepts" / "INDEX.md").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_an_index_file_is_marked_and_cannot_come_back_in(instance):
|
|
_configure(instance, pages={"index": True, "frontmatter": "full"})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
commit = _git(instance, "rev-parse", "HEAD")
|
|
for rel in ("index.md", "concepts/INDEX.md"):
|
|
text = (folder / rel).read_text(encoding="utf-8")
|
|
assert text.startswith("---\n")
|
|
frontmatter = text.split("---\n")[1]
|
|
assert yaml.safe_load(frontmatter) == {
|
|
"wikitool_export": f"kind=index instance=local commit={commit}"}
|
|
assert guideline_export.file_is_export(folder / rel)
|
|
(instance / "incoming" / "index.md").write_bytes((folder / "index.md").read_bytes())
|
|
before = (instance / "incoming" / "index.md").read_bytes()
|
|
result = _accept(instance / "incoming" / "index.md")
|
|
assert result.exit_code == 1 and "export" in _flat(result.output)
|
|
assert (instance / "incoming" / "index.md").read_bytes() == before
|
|
assert not any((instance / "raw").rglob("*.md"))
|
|
|
|
|
|
def test_a_second_run_touches_nothing_and_no_index_holds_a_date_of_its_own(instance):
|
|
_configure(instance, pages={"index": True})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
stamps, before = _mtimes(folder), _snapshot(folder)
|
|
assert _run().exit_code == 0
|
|
assert _mtimes(folder) == stamps and _snapshot(folder) == before
|
|
for rel in _index_files(folder):
|
|
text = (folder / rel).read_text(encoding="utf-8")
|
|
assert datetime.date.today().isoformat() not in text.replace("2026-10-01", "")
|
|
assert "Generated" not in text and "wikitool" not in text.split("---\n", 2)[2]
|
|
|
|
|
|
def test_a_page_without_dates_or_summary_has_empty_cells_and_never_the_file_time(instance):
|
|
path = instance / "kb" / "concepts" / "Bare.md"
|
|
write_page(path, {"type": "types/concept.md", "tags": ["docs"]}, "\n# Bare\n\nNothing.\n")
|
|
_commit(instance, "bare")
|
|
os.utime(path, (86400 * 365, 86400 * 365))
|
|
_configure(instance, pages={"index": True})
|
|
assert _run().exit_code == 0
|
|
text = (_out(instance, "pages") / "concepts" / "INDEX.md").read_text(encoding="utf-8")
|
|
row = next(line for line in text.splitlines() if "[Bare]" in line)
|
|
assert row.endswith("| | |") and "1971" not in text
|
|
|
|
|
|
def test_a_commit_to_one_collection_leaves_the_index_of_another_byte_equal(instance):
|
|
_page(instance, "other/Delta.md", "\n# Delta\n\nSecond collection.\n")
|
|
_commit(instance, "second collection")
|
|
_configure(instance, whole=_all(instance, index=True))
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "whole")
|
|
before = _snapshot(folder)
|
|
_page(instance, "concepts/Alpha.md", ALPHA + "\nMore.\n")
|
|
_commit(instance, "touch concepts only")
|
|
assert _run().exit_code == 0
|
|
after = _snapshot(folder)
|
|
assert after["other/INDEX.md"] == before["other/INDEX.md"]
|
|
assert after["other/Delta.md"] == before["other/Delta.md"]
|
|
assert after["concepts/INDEX.md"] != before["concepts/INDEX.md"]
|
|
assert after["index.md"] != before["index.md"]
|
|
|
|
|
|
def test_switching_index_off_removes_the_index_files_and_nothing_else(instance):
|
|
_configure(instance, pages={"index": True})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
indexes = _index_files(folder)
|
|
assert indexes
|
|
kept = {rel: data for rel, data in _snapshot(folder).items()
|
|
if rel not in indexes and rel != page_export.MANIFEST_NAME}
|
|
_configure(instance, pages={"index": False})
|
|
result = _run()
|
|
assert result.exit_code == 0, result.output
|
|
after = _snapshot(folder)
|
|
assert not [rel for rel in after if posixpath.basename(rel) in INDEX_NAMES]
|
|
assert {rel: data for rel, data in after.items() if rel != page_export.MANIFEST_NAME} == kept
|
|
assert f"removed ({len(indexes)})" in _flat(result.output)
|
|
|
|
|
|
def test_a_page_index_md_beside_the_generated_one_stops_the_target_before_any_write(instance):
|
|
_configure(instance, pages={})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
before = _snapshot(folder)
|
|
_page(instance, "concepts/index.md", "\n# Index\n\nA page called index.\n")
|
|
_commit(instance, "index page")
|
|
_configure(instance, pages={"index": True})
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
message = _flat(result.output)
|
|
assert "concepts/index.md" in message and "concepts/INDEX.md" in message
|
|
assert _snapshot(folder) == before
|
|
|
|
|
|
def test_a_stray_page_that_differs_from_index_md_only_in_case_stops_the_target(instance):
|
|
_page(instance, "Index.md", "\n# Index\n\nStray.\n")
|
|
_commit(instance, "stray")
|
|
_configure(instance, whole=_all(instance, index=True))
|
|
result = _run()
|
|
assert result.exit_code == 1
|
|
assert "Index.md" in _flat(result.output) and "index.md" in _flat(result.output)
|
|
assert not _out(instance, "whole").exists()
|
|
|
|
|
|
def test_with_links_text_pages_hold_no_link_to_an_exported_page_but_the_index_links(instance):
|
|
_configure(instance, pages={"index": True, "links": "text"})
|
|
assert _run().exit_code == 0
|
|
folder = _out(instance, "pages")
|
|
indexes = set(_index_files(folder))
|
|
for rel in _snapshot(folder):
|
|
if rel in indexes or rel == page_export.MANIFEST_NAME:
|
|
continue
|
|
assert not re.search(r"\]\([^)]*\.md[^)]*\)", (folder / rel).read_text(encoding="utf-8")), rel
|
|
assert "](Alpha.md)" in (folder / "concepts" / "INDEX.md").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_the_command_handler_gets_the_index_files_with_the_pages(instance):
|
|
record = instance.parent / "files.json"
|
|
program = _converter(instance, f"""
|
|
import json, pathlib, sys
|
|
src, out = pathlib.Path(sys.argv[1]), pathlib.Path(sys.argv[2])
|
|
pathlib.Path({str(record)!r}).write_text(json.dumps(sorted(
|
|
p.relative_to(src).as_posix() for p in src.rglob("*") if p.is_file())))
|
|
(out / "done.txt").write_text("done")
|
|
""")
|
|
_configure(instance, pages={"handler": "command", "command": "convert", "index": True})
|
|
_set_commands(instance, {"convert": program})
|
|
assert _run().exit_code == 0
|
|
seen = json.loads(record.read_text(encoding="utf-8"))
|
|
assert "index.md" in seen and "concepts/INDEX.md" in seen and "concepts/Alpha.md" in seen
|
|
|
|
|
|
def test_the_run_names_the_index_files_by_path(instance):
|
|
_configure(instance, pages={"index": True})
|
|
out = _flat(_run().output)
|
|
assert "index.md" in out and "concepts/INDEX.md" in out
|