feat: the shapes of the four hand-written .wikitool-*.json files live in one record each (config_shapes.py); loaders, doctor and docs/reference/configuration.md read them (#193)
CI / verify (push) Successful in 6m17s
CI / pwsh (push) Successful in 2m0s
Release / release (push) Successful in 34s

The telemetry, remotes, upload and tasks loaders take field names, defaults, coercions and their Expected: text from the record and accept exactly what they did before. doctor WARNs on a telemetry file the policy silently reads differently. configuration.md carries one generated region per file, also checked in an instance. docs reference --apply now refreshes several regions in one page in sequence instead of keeping only the last.

Files changed:
- CHANGES.md
- VERSION
- docs/how-to/set-up-the-mcp-server.md
- docs/reference/README.md
- docs/reference/commands/instance-health.md
- docs/reference/commands/types-instructions-and-docs.md
- docs/reference/configuration.md
- instructions/dev/doc-pull-through.md
- instructions/setup-instance.md
- tools/CONTRACT.md
- tools/README.md
- tools/chemenu/commands/docs_verify.py
- tools/chemenu/commands/doctor.py
- tools/chemenu/commands/git_publish.py
- tools/chemenu/config.py
- tools/chemenu/config_shapes.py
- tools/chemenu/reference_doc.py
- tools/chemenu/tasks/caldav.py
- tools/chemenu/tasks/config.py
- tools/chemenu/tasks/superproductivity.py
- tools/chemenu/telemetry/policy.py
- tools/chemenu/tests/test_config_shapes.py
- tools/chemenu/tests/test_doctor.py
- tools/chemenu/tests/test_reference_doc.py
- tools/chemenu/upload.py

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SnAJ7Z3CpVD3PRbN73QtU2
This commit is contained in:
torbenandClaude Opus 5.5 committed 2026-10-07 08:44:50 +02:00
1 parent 61e18dc4aa
commit 893db6e640
25 files changed
+1646 -311

No files matched your search

+608
View File
@@ -0,0 +1,608 @@
"""`chemenu.config_shapes` - one record per hand-written configuration file,
which the loaders read and `docs reference` renders.
The drop-in half compares every loader against a copy of the loader as it stood
before the records existed (the `_legacy_*` oracles below, kept verbatim in
what they accept): any file it accepted must still be accepted with the same
values, and any file it rejected must still be rejected. Only the wording of a
rejection may differ.
"""
from __future__ import annotations
import ast
import dataclasses
import json
import re
import sys
from pathlib import Path
import pytest
import typer
from typer.testing import CliRunner
from chemenu import blocks, config, config_shapes, reference_doc
from chemenu.cli import app
from chemenu.commands import docs_verify, git_publish
from chemenu.errors import ValidationError
from chemenu.tasks import caldav, superproductivity
from chemenu.tasks import config as tasks_config
from chemenu.telemetry import policy
runner = CliRunner()
PACKAGE = Path(config_shapes.__file__).parent
# --- the module itself -------------------------------------------------------
def test_the_module_imports_only_the_stdlib_and_chemenu_config():
"""A hook handler imports `chemenu.telemetry` on every tool call without
the venv, and the telemetry policy imports this module."""
tree = ast.parse(Path(config_shapes.__file__).read_text(encoding="utf-8"))
imported = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
imported |= {alias.name for alias in node.names}
elif isinstance(node, ast.ImportFrom):
if node.module == "chemenu":
imported |= {f"chemenu.{alias.name}" for alias in node.names}
elif node.module != "__future__":
imported.add(node.module)
foreign = {
name for name in imported
if name != "chemenu.config" and name.split(".")[0] not in sys.stdlib_module_names
}
assert foreign == set()
def test_the_telemetry_policy_reads_the_record():
assert policy.config_shapes is config_shapes
assert policy.DEFAULT_MAX_SESSION_BYTES == config_shapes.TELEMETRY.field("max_session_bytes").default
assert policy.DEFAULT_KEEP_SESSIONS == config_shapes.TELEMETRY.field("keep_sessions").default
def test_one_record_per_file():
assert [shape.key for shape in config_shapes.SHAPES] == [
config.TELEMETRY_FILENAME, config.PUBLISH_REMOTES_FILENAME,
config.UPLOAD_CONFIG_FILENAME, config.TASKS_CONFIG_FILENAME,
]
LOADERS = ("telemetry/policy.py", "commands/git_publish.py", "upload.py", "tasks/config.py",
"tasks/superproductivity.py", "tasks/caldav.py")
def test_no_hand_written_expected_text_is_left_in_a_loader():
pattern = re.compile(r"_EXPECTED_\w*\s*=|\bexpected\s*=\s*\(|Expected: \{\\?\"")
offenders = [
relative for relative in LOADERS
if pattern.search((PACKAGE / relative).read_text(encoding="utf-8"))
]
assert offenders == []
def test_the_known_names_follow_the_records():
assert tasks_config.KNOWN_PROVIDERS == ("superproductivity", "caldav")
assert superproductivity.KNOWN_ACCESS == ("api", "snapshot")
# --- the legacy oracles: the loaders as they were before the records --------
def _legacy_upload(data):
if not isinstance(data, dict):
raise ValidationError("not an object")
try:
identity_header = str(data.get("identity_header") or "X-Forwarded-User")
max_bytes = int(data["max_bytes"])
extensions = tuple(sorted({str(ext).lower() for ext in data["allowed_extensions"]}))
quota = data["quota"]
submissions_per_day = int(quota["submissions_per_day"])
bytes_per_day = int(quota["bytes_per_day"])
except (KeyError, TypeError, ValueError) as exc:
raise ValidationError("misshaped") from exc
if max_bytes <= 0 or submissions_per_day <= 0 or bytes_per_day <= 0:
raise ValidationError("not positive")
if not extensions or [ext for ext in extensions if not ext.startswith(".")]:
raise ValidationError("extensions")
return (identity_header, max_bytes, extensions, submissions_per_day, bytes_per_day)
def _legacy_tasks(data):
if not isinstance(data, dict):
raise ValidationError("not an object")
try:
provider = str(data["provider"])
raw = data["thresholds"]
thresholds = (int(raw["stalled_waiting_days"]), int(raw["unpaged_project_weeks"]),
int(raw["someday_stale_months"]))
except (KeyError, TypeError, ValueError) as exc:
raise ValidationError("misshaped") from exc
if provider not in ("superproductivity", "caldav"):
raise ValidationError("provider")
if any(value <= 0 for value in thresholds):
raise ValidationError("not positive")
if not isinstance(data.get(provider), dict):
raise ValidationError("section")
return (provider, thresholds, data.get(provider))
def _legacy_sp(data):
if not isinstance(data, dict):
raise ValidationError("not an object")
access = data.get("access")
if access not in ("api", "snapshot"):
raise ValidationError("access")
if access == "api":
if set(data) - {"access", "api_base_url", "api_token"}:
raise ValidationError("extra")
token = data.get("api_token")
if not isinstance(token, str) or not token:
raise ValidationError("token")
return ("api", None, str(data.get("api_base_url") or "http://127.0.0.1:3876"), token)
if set(data) - {"access", "backups_dir"}:
raise ValidationError("extra")
if not data.get("backups_dir"):
raise ValidationError("backups_dir")
return ("snapshot", Path(data["backups_dir"]).expanduser(), "http://127.0.0.1:3876", None)
_CALDAV_STRINGS = ("url", "username", "app_password", "inbox_list", "someday_list")
def _legacy_caldav(data):
if not isinstance(data, dict):
raise ValidationError("not an object")
if [k for k in _CALDAV_STRINGS if not isinstance(data.get(k), str) or not data[k]]:
raise ValidationError("missing")
if set(data) - {*_CALDAV_STRINGS, "exclude_lists"}:
raise ValidationError("extra")
exclude = data.get("exclude_lists", [])
if not isinstance(exclude, list) or not all(isinstance(x, str) for x in exclude):
raise ValidationError("exclude_lists")
url = data["url"] if data["url"].endswith("/") else data["url"] + "/"
return (url, data["username"], data["app_password"], data["inbox_list"], data["someday_list"],
tuple(exclude))
def _legacy_remotes(data):
urls = data.get("allowed_push_urls")
if not isinstance(urls, list) or not all(isinstance(u, str) for u in urls):
raise ValidationError("urls")
return urls
def _legacy_telemetry(data):
def positive(value):
try:
parsed = int(value)
except (TypeError, ValueError):
return None
return parsed if parsed > 0 else None
data = data if isinstance(data, dict) else {}
return (
bool(data["enabled"]) if "enabled" in data else None,
positive(data.get("max_session_bytes")),
positive(data.get("keep_sessions")),
)
# --- the loaders today, reduced to the same tuples --------------------------
def _upload(data, tmp_path):
(tmp_path / config.UPLOAD_CONFIG_FILENAME).write_text(json.dumps(data), encoding="utf-8")
from chemenu import upload
cfg = upload.read_config(tmp_path)
return (cfg.identity_header, cfg.max_bytes, cfg.allowed_extensions, cfg.submissions_per_day,
cfg.bytes_per_day)
def _tasks(data, tmp_path):
(tmp_path / config.TASKS_CONFIG_FILENAME).write_text(json.dumps(data), encoding="utf-8")
cfg = tasks_config.read_config(tmp_path)
t = cfg.thresholds
return (cfg.provider, (t.stalled_waiting_days, t.unpaged_project_weeks, t.someday_stale_months),
cfg.provider_config)
def _sp(data, tmp_path):
cfg = superproductivity.SuperProductivityConfig.from_dict(data)
return (cfg.access, cfg.backups_dir, cfg.api_base_url, cfg.api_token)
def _caldav(data, tmp_path):
cfg = caldav.CalDAVConfig.from_dict(data)
return (cfg.url, cfg.username, cfg.app_password, cfg.inbox_list, cfg.someday_list,
cfg.exclude_lists)
def _remotes(data, tmp_path):
(tmp_path / config.PUBLISH_REMOTES_FILENAME).write_text(json.dumps(data), encoding="utf-8")
try:
return git_publish.read_allowed_push_urls()
except typer.Exit as exc:
raise ValidationError("fail()") from exc
def _telemetry(data, tmp_path):
(tmp_path / config.TELEMETRY_FILENAME).write_text(json.dumps(data), encoding="utf-8")
policy.reset_cache()
defaults = policy._root_defaults(tmp_path.resolve())
policy.reset_cache()
return (defaults.file_enabled, defaults.file_max_bytes, defaults.file_keep_sessions)
UPLOAD_OK = {
"schema": 1, "identity_header": "X-Forwarded-User", "max_bytes": 10485760,
"allowed_extensions": [".md", ".txt", ".pdf", ".html", ".csv", ".json", ".png", ".jpg"],
"quota": {"submissions_per_day": 20, "bytes_per_day": 52428800},
}
TASKS_OK = {
"schema": 1, "provider": "superproductivity",
"thresholds": {"stalled_waiting_days": 14, "unpaged_project_weeks": 3, "someday_stale_months": 5},
"superproductivity": {"access": "api", "api_base_url": "http://127.0.0.1:3876", "api_token": "t"},
}
CALDAV_OK = {
"url": "https://host/remote.php/dav/calendars/me", "username": "me", "app_password": "pw",
"inbox_list": "Inbox", "someday_list": "Someday", "exclude_lists": ["Shopping"],
}
def _with(base, **changes):
data = json.loads(json.dumps(base))
for key, value in changes.items():
target = data
*parents, leaf = key.split("__")
for parent in parents:
target = target[parent]
if value is _DROP:
del target[leaf]
else:
target[leaf] = value
return data
_DROP = object()
CASES = [
# upload: the documented file, then every tolerance and every rejection
("upload", UPLOAD_OK),
("upload", _with(UPLOAD_OK, schema=_DROP)),
("upload", _with(UPLOAD_OK, identity_header=_DROP)),
("upload", _with(UPLOAD_OK, identity_header="")),
("upload", _with(UPLOAD_OK, max_bytes="10485760")),
("upload", _with(UPLOAD_OK, quota__bytes_per_day="100")),
("upload", _with(UPLOAD_OK, allowed_extensions=[".MD", ".md", ".Pdf"])),
("upload", _with(UPLOAD_OK, surplus="ignored")),
("upload", _with(UPLOAD_OK, quota__surplus=1)),
("upload", _with(UPLOAD_OK, max_bytes=0)),
("upload", _with(UPLOAD_OK, max_bytes="ten")),
("upload", _with(UPLOAD_OK, max_bytes=_DROP)),
("upload", _with(UPLOAD_OK, allowed_extensions=["md"])),
("upload", _with(UPLOAD_OK, allowed_extensions=[])),
("upload", _with(UPLOAD_OK, allowed_extensions=".md")),
("upload", _with(UPLOAD_OK, quota=[1, 2])),
("upload", _with(UPLOAD_OK, quota__submissions_per_day=-1)),
("upload", [1, 2]),
# tasks
("tasks", TASKS_OK),
("tasks", _with(TASKS_OK, schema=_DROP)),
("tasks", _with(TASKS_OK, thresholds__stalled_waiting_days="14")),
("tasks", _with(TASKS_OK, caldav={"url": "unused"})),
("tasks", _with(TASKS_OK, surplus=True)),
("tasks", _with(TASKS_OK, thresholds__surplus=1)),
("tasks", _with(TASKS_OK, provider="asana")),
("tasks", _with(TASKS_OK, provider=_DROP)),
("tasks", _with(TASKS_OK, thresholds__someday_stale_months=0)),
("tasks", _with(TASKS_OK, thresholds__unpaged_project_weeks=_DROP)),
("tasks", _with(TASKS_OK, thresholds="soon")),
("tasks", _with(TASKS_OK, superproductivity=_DROP)),
("tasks", _with(TASKS_OK, superproductivity="api")),
("tasks", _with(TASKS_OK, provider="caldav")),
("tasks", "not an object"),
# the Super Productivity section
("sp", {"access": "api", "api_base_url": "http://127.0.0.1:3876", "api_token": "t"}),
("sp", {"access": "api", "api_token": "t"}),
("sp", {"access": "api", "api_token": "t", "api_base_url": ""}),
("sp", {"access": "snapshot", "backups_dir": "~/.config/superProductivity/backups"}),
("sp", {"access": "api"}),
("sp", {"access": "api", "api_token": ""}),
("sp", {"access": "api", "api_token": 7}),
("sp", {"access": "api", "api_token": "t", "backups_dir": "/x"}),
("sp", {"access": "snapshot", "backups_dir": "/x", "api_token": "t"}),
("sp", {"access": "snapshot"}),
("sp", {"access": "snapshot", "backups_dir": ""}),
("sp", {"access": "auto", "backups_dir": "/x"}),
("sp", {"backups_dir": "/x"}),
("sp", ["api"]),
# the CalDAV section
("caldav", CALDAV_OK),
("caldav", _with(CALDAV_OK, exclude_lists=_DROP)),
("caldav", _with(CALDAV_OK, url="https://host/dav/")),
("caldav", _with(CALDAV_OK, url=_DROP)),
("caldav", _with(CALDAV_OK, username="")),
("caldav", _with(CALDAV_OK, app_password=5)),
("caldav", _with(CALDAV_OK, exclude_lists="Shopping")),
("caldav", _with(CALDAV_OK, exclude_lists=[1])),
("caldav", _with(CALDAV_OK, backups_dir="/x")),
# remotes
("remotes", {"schema": 1, "allowed_push_urls": ["git@example.org:me/my-wiki.git"]}),
("remotes", {"allowed_push_urls": []}),
("remotes", {"allowed_push_urls": ["a"], "surplus": 1}),
("remotes", {"allowed_push_urls": "not-a-list"}),
("remotes", {"allowed_push_urls": [1]}),
("remotes", {}),
# telemetry: never rejected, only read differently
("telemetry", {"enabled": True, "max_session_bytes": 5242880, "keep_sessions": 250}),
("telemetry", {}),
("telemetry", {"enabled": "no"}),
("telemetry", {"enabled": 0}),
("telemetry", {"enabled": None}),
("telemetry", {"max_session_bytes": "100", "keep_sessions": 3.7}),
("telemetry", {"max_session_bytes": 0, "keep_sessions": -2}),
("telemetry", {"max_session_bytes": "lots", "keep_sessions": [1]}),
("telemetry", {"surplus": 1, "schema": 1}),
("telemetry", [True]),
]
ORACLES = {
"upload": (_legacy_upload, _upload),
"tasks": (_legacy_tasks, _tasks),
"sp": (_legacy_sp, _sp),
"caldav": (_legacy_caldav, _caldav),
"remotes": (_legacy_remotes, _remotes),
"telemetry": (_legacy_telemetry, _telemetry),
}
def _outcome(function, data, *args):
try:
return ("accepted", function(data, *args))
except ValidationError:
return ("rejected", None)
@pytest.mark.parametrize("loader,data", CASES, ids=[f"{n}-{i}" for i, (n, _) in enumerate(CASES)])
def test_every_loader_accepts_and_rejects_exactly_what_it_did_before(loader, data, tmp_path, monkeypatch):
legacy, current = ORACLES[loader]
monkeypatch.setattr(config, "ROOT", tmp_path)
assert _outcome(current, data, tmp_path) == _outcome(legacy, data)
def test_the_cases_cover_both_outcomes_for_every_loader(tmp_path):
for loader, (legacy, _) in ORACLES.items():
if loader == "telemetry":
continue
outcomes = {_outcome(legacy, data)[0] for name, data in CASES if name == loader}
assert outcomes == {"accepted", "rejected"}, loader
def test_a_rejection_keeps_its_cause_and_renders_the_expected_text(tmp_path):
(tmp_path / config.UPLOAD_CONFIG_FILENAME).write_text(
json.dumps(_with(UPLOAD_OK, max_bytes=0)), encoding="utf-8"
)
from chemenu import upload
with pytest.raises(ValidationError) as caught:
upload.read_config(tmp_path)
message = str(caught.value)
assert "max_bytes must be positive" in message
assert f"Expected: {config_shapes.render_expected(config_shapes.UPLOAD)}" in message
# --- the loaders take their expectations from the record --------------------
def _add_required(shape, name="added"):
return dataclasses.replace(
shape, fields=shape.fields + (config_shapes.Field(name, "string", "x", required=True),)
)
def _not_required(shape, name, default):
fields = tuple(
dataclasses.replace(f, required=False, default=default) if f.name == name else f
for f in shape.fields
)
return dataclasses.replace(shape, fields=fields)
def test_upload_follows_a_field_made_required_or_optional(tmp_path, monkeypatch):
from chemenu import upload
(tmp_path / config.UPLOAD_CONFIG_FILENAME).write_text(json.dumps(UPLOAD_OK), encoding="utf-8")
monkeypatch.setattr(config_shapes, "UPLOAD", _add_required(config_shapes.UPLOAD))
with pytest.raises(ValidationError, match="added is missing"):
upload.read_config(tmp_path)
monkeypatch.undo()
(tmp_path / config.UPLOAD_CONFIG_FILENAME).write_text(
json.dumps(_with(UPLOAD_OK, max_bytes=_DROP)), encoding="utf-8"
)
with pytest.raises(ValidationError):
upload.read_config(tmp_path)
monkeypatch.setattr(config_shapes, "UPLOAD", _not_required(config_shapes.UPLOAD, "max_bytes", 99))
assert upload.read_config(tmp_path).max_bytes == 99
def test_tasks_follows_a_field_made_required(tmp_path, monkeypatch):
(tmp_path / config.TASKS_CONFIG_FILENAME).write_text(json.dumps(TASKS_OK), encoding="utf-8")
assert tasks_config.read_config(tmp_path) is not None
monkeypatch.setattr(config_shapes, "TASKS", _add_required(config_shapes.TASKS))
with pytest.raises(ValidationError, match="added is missing"):
tasks_config.read_config(tmp_path)
def test_remotes_follows_a_field_made_required(tmp_path, monkeypatch):
monkeypatch.setattr(config, "ROOT", tmp_path)
(tmp_path / config.PUBLISH_REMOTES_FILENAME).write_text(
json.dumps({"allowed_push_urls": ["u"]}), encoding="utf-8"
)
assert git_publish.read_allowed_push_urls() == ["u"]
monkeypatch.setattr(config_shapes, "REMOTES", _add_required(config_shapes.REMOTES))
with pytest.raises(typer.Exit):
git_publish.read_allowed_push_urls()
def test_caldav_follows_a_field_made_required(monkeypatch):
data = _with(CALDAV_OK, exclude_lists=_DROP)
assert caldav.CalDAVConfig.from_dict(data).exclude_lists == ()
fields = tuple(
dataclasses.replace(f, required=True) if f.name == "exclude_lists" else f
for f in config_shapes.CALDAV.fields
)
monkeypatch.setattr(config_shapes, "CALDAV", dataclasses.replace(config_shapes.CALDAV, fields=fields))
with pytest.raises(ValidationError, match="exclude_lists is missing"):
caldav.CalDAVConfig.from_dict(data)
def test_superproductivity_follows_a_field_added_to_a_variant(monkeypatch):
data = {"access": "api", "api_token": "t", "timeout": "5"}
with pytest.raises(ValidationError, match="does not take"):
superproductivity.SuperProductivityConfig.from_dict(data)
shape = config_shapes.SUPERPRODUCTIVITY
api = dict(shape.variants)["api"] + (config_shapes.Field("timeout", "string", "x"),)
variants = (("api", api),) + tuple(v for v in shape.variants if v[0] != "api")
monkeypatch.setattr(config_shapes, "SUPERPRODUCTIVITY", dataclasses.replace(shape, variants=variants))
assert superproductivity.SuperProductivityConfig.from_dict(data).api_token == "t"
def test_telemetry_follows_a_condition_in_the_record(tmp_path, monkeypatch):
(tmp_path / config.TELEMETRY_FILENAME).write_text(json.dumps({"keep_sessions": 500}), encoding="utf-8")
policy.reset_cache()
assert policy.resolve(tmp_path).keep_sessions == 500
shape = config_shapes.TELEMETRY
fields = tuple(
dataclasses.replace(f, check=lambda v: None if v < 100 else "too many") if f.name == "keep_sessions" else f
for f in shape.fields
)
monkeypatch.setattr(config_shapes, "TELEMETRY", dataclasses.replace(shape, fields=fields))
policy.reset_cache()
assert policy.resolve(tmp_path).keep_sessions == policy.DEFAULT_KEEP_SESSIONS
policy.reset_cache()
# --- deviations: what doctor reports for telemetry -------------------------
@pytest.mark.parametrize("data,field", [
({"enabled": "yes"}, "enabled"),
({"enabled": 1}, "enabled"),
({"max_session_bytes": 0}, "max_session_bytes"),
({"keep_sessions": "250"}, "keep_sessions"),
({"keep_sessions": 2.5}, "keep_sessions"),
({"enabled": True, "verbose": True}, "verbose"),
])
def test_a_telemetry_deviation_names_its_field(data, field):
found = config_shapes.deviations(config_shapes.TELEMETRY, data)
assert found and all(field in entry for entry in found)
def test_a_conforming_telemetry_file_has_no_deviation():
for example in config_shapes.TELEMETRY.examples:
assert config_shapes.deviations(config_shapes.TELEMETRY, example) == []
assert config_shapes.deviations(config_shapes.TELEMETRY, {"schema": 1, "enabled": False}) == []
def test_every_example_passes_its_own_record():
for shape in config_shapes.SHAPES:
for example in shape.examples:
assert config_shapes.check(shape, example).problems == [], shape.key
for section in config_shapes.TASKS.sections:
for example in section.examples:
assert config_shapes.check(section, example).problems == [], section.key
# --- the generated regions --------------------------------------------------
def _page() -> str:
return (config.ROOT / reference_doc.CONFIGURATION).read_text(encoding="utf-8")
def _all_fields(shape, prefix=""):
for spec in shape.fields + tuple(f for _, fs in shape.variants for f in fs):
yield prefix + spec.name
yield from (prefix + spec.name + "." + sub.name for sub in spec.fields)
def test_the_page_carries_one_region_per_file_and_every_field_in_it():
page = _page()
for name, attr in reference_doc.CONFIG_REGIONS:
shape = getattr(config_shapes, attr)
region = blocks.find(page, name)
assert region is not None, name
for path in _all_fields(shape):
assert f"`{path}`" in region, (name, path)
for section in shape.sections:
for path in _all_fields(section):
assert f"`{path}`" in region, (name, path)
@pytest.mark.parametrize("relative", [reference_doc.CONFIGURATION, "docs/how-to/set-up-the-mcp-server.md"])
def test_no_json_example_for_these_files_stands_outside_a_region(relative):
text = blocks.mask_regions((config.ROOT / relative).read_text(encoding="utf-8"))
keys = {f.name for shape in config_shapes.SHAPES for f in shape.fields if f.name != "schema"}
keys |= {f.name for s in config_shapes.TASKS.sections for f in s.fields}
keys |= {f.name for s in config_shapes.TASKS.sections for _, fs in s.variants for f in fs}
for block in re.findall(r"```json\n(.*?)```", text, re.S):
assert not any(f'"{key}"' in block for key in keys), block
def _only_config_regions(monkeypatch):
real = reference_doc.regions
monkeypatch.setattr(
reference_doc, "regions",
lambda *a, **k: [r for r in real(*a, **k) if r.path == reference_doc.CONFIGURATION],
)
def _hand_written(text: str) -> str:
for name, _ in reference_doc.CONFIG_REGIONS:
text = re.sub(
re.escape(blocks.open_marker(name)) + ".*?" + re.escape(blocks.close_marker(name)),
blocks.open_marker(name), text, flags=re.S,
)
return text
def test_apply_fills_every_region_of_the_page_and_a_second_run_changes_nothing(tmp_path, monkeypatch):
"""Four regions in one file: each is refreshed into the text the one
before it left, or only the last would survive."""
target = tmp_path / reference_doc.CONFIGURATION
target.parent.mkdir(parents=True)
real = _page()
emptied = real
for name, _ in reference_doc.CONFIG_REGIONS:
emptied = blocks.replace(emptied, name, blocks.open_marker(name) + "\n" + blocks.close_marker(name) + "\n")
target.write_text(emptied, encoding="utf-8")
monkeypatch.setattr(config, "ROOT", tmp_path)
_only_config_regions(monkeypatch)
result = runner.invoke(app, ["docs", "reference", "--apply"])
assert result.exit_code == 0, result.output
applied = target.read_text(encoding="utf-8")
assert applied == real
assert _hand_written(applied) == _hand_written(emptied)
result = runner.invoke(app, ["docs", "reference", "--apply"])
assert "already current" in result.output
assert target.read_text(encoding="utf-8") == applied
def test_a_changed_record_without_apply_fails_docs_verify(monkeypatch):
assert not [i for i in docs_verify.check_reference_regions() if "configuration.md" in i]
monkeypatch.setattr(config_shapes, "REMOTES", _add_required(config_shapes.REMOTES))
issues = [i for i in docs_verify.check_reference_regions() if reference_doc.CONFIGURATION in i]
assert len(issues) == 1
assert "reference-config-remotes" in issues[0] and "docs reference --apply" in issues[0]
def test_the_configuration_regions_are_checked_in_an_instance_too():
regions = [r for r in reference_doc.regions() if r.path == reference_doc.CONFIGURATION]
assert [r.name for r in regions] == [name for name, _ in reference_doc.CONFIG_REGIONS]
assert not any(r.origin_only for r in regions)
+43
View File
@@ -521,6 +521,49 @@ def test_telemetry_never_fails_either_way(instance):
assert _status(doctor.run_doctor(), "telemetry") == "OK"
def _telemetry_with(text: "str | None"):
"""`check_telemetry` against a `.wikitool-telemetry.json` holding `text`
(absent for None), plus what the policy itself makes of it."""
from chemenu.telemetry import policy
path = config.ROOT / config.TELEMETRY_FILENAME
if text is None:
path.unlink(missing_ok=True)
else:
path.write_text(text, encoding="utf-8")
policy.reset_cache()
try:
return doctor.check_telemetry(), policy.resolve(config.ROOT)
finally:
policy.reset_cache()
@pytest.mark.parametrize("text", [None, '{"enabled": false}', '{"schema": 1, "enabled": true, '
'"max_session_bytes": 1000, "keep_sessions": 9}'])
def test_telemetry_stays_ok_for_an_absent_or_conforming_file(instance, text):
check, _ = _telemetry_with(text)
assert check.status == "OK"
assert check.fix is None
@pytest.mark.parametrize("text,named,policy_reads", [
("{not json", "unreadable", (True, 5 * 1024 * 1024, 250)),
("[true]", "not a JSON object", (True, 5 * 1024 * 1024, 250)),
('{"enabled": "no"}', "enabled", (True, 5 * 1024 * 1024, 250)),
('{"max_session_bytes": 0}', "max_session_bytes", (True, 5 * 1024 * 1024, 250)),
('{"keep_sessions": "12"}', "keep_sessions", (True, 5 * 1024 * 1024, 12)),
('{"enabled": false, "keep_session": 12}', "keep_session", (False, 5 * 1024 * 1024, 250)),
])
def test_telemetry_warns_on_what_the_policy_reads_differently(instance, text, named, policy_reads):
"""The policy stays silent and reads the file exactly as before; `doctor`
is where the deviation becomes visible, by field."""
check, pol = _telemetry_with(text)
assert check.status == "WARN"
assert named in check.detail
assert config.TELEMETRY_FILENAME in check.fix
assert (pol.enabled, pol.max_session_bytes, pol.keep_sessions) == policy_reads
def test_telemetry_reports_the_current_session_count_and_byte_total(instance, monkeypatch):
from chemenu.telemetry.writer import TRACE_FILE, trace_root
@@ -266,6 +266,7 @@ def _page_type(description: str = "A fixture type.") -> reference_doc.PageType:
@pytest.fixture
def tree(tmp_path, monkeypatch):
real_root = config.ROOT
monkeypatch.setattr(config, "ROOT", tmp_path)
records = {"frobnicate": _record("frobnicate"), "gate keep": _record("gate keep", ("keep",))}
monkeypatch.setattr(cli_contract, "all_records", lambda: dict(records))
@@ -283,6 +284,13 @@ def tree(tmp_path, monkeypatch):
path.write_text(
f"{head}\n{blocks.open_marker(name)}\n{blocks.close_marker(name)}\n", encoding="utf-8"
)
# The configuration regions render from the real records, not from the
# fixture's; the current page keeps them out of what these tests count.
# test_config_shapes.py covers that page.
configuration = tmp_path / reference_doc.CONFIGURATION
configuration.write_text(
(real_root / reference_doc.CONFIGURATION).read_text(encoding="utf-8"), encoding="utf-8"
)
return tmp_path, records