raw accept: Datums-Shard statt Typverzeichnis, fidelity/authority am Drop-Punkt (Teil 1/3, #67)
Files changed: - .gitignore - CHANGES.md - VERSION - instructions/bootstrap.md - instructions/wiki-ingest/SKILL.md - kb/CONTRACT.md - raw/CONTRACT.md - tools/CONTRACT.md - tools/chemenu/commands/dist_cmd.py - tools/chemenu/commands/docs_verify.py - tools/chemenu/commands/new_page.py - tools/chemenu/commands/raw_cmd.py - tools/chemenu/commands/touch.py - tools/chemenu/lint_core.py - tools/chemenu/tests/test_dist_cmd.py - tools/chemenu/tests/test_docs_verify.py - tools/chemenu/tests/test_lint.py - tools/chemenu/tests/test_new_page.py - tools/chemenu/tests/test_provenance.py - tools/chemenu/tests/test_raw_cmd.py - tools/chemenu/tests/test_touch.py - tools/chemenu/tests/test_type_resolver.py - tools/chemenu/type_resolver.py - types/source.md - types/source.schema.yaml
This commit is contained in:
@@ -133,14 +133,6 @@ def _is_coverage_output(filename: str) -> bool:
|
||||
# reconstructs it afterwards, unlike the marker-block content below.
|
||||
INSTRUCTIONS_EXCLUDE_DIRS = {"dev"}
|
||||
|
||||
# Fixed by raw/CONTRACT.md's routing table, unlike kb/'s areas (which are
|
||||
# organic - see kb/CONTRACT.md - so `export` does not manufacture them).
|
||||
# `docs verify` (check_raw_subdirs) holds the table to this tuple in both
|
||||
# directions, and `incoming/` (raw/CONTRACT.md "Getting a file in",
|
||||
# raw_cmd.py) mirrors it as the set of type subdirectories a human may drop a
|
||||
# file into - so this is the one place all three read the list from.
|
||||
RAW_SUBDIRS = ("articles", "documents", "notes", "assets")
|
||||
|
||||
# Stage contracts that are not collections and carry no pages: copied as a
|
||||
# single file each, nothing else from their directory. `kb/` is excluded here
|
||||
# - it is a content stage too, but it has collections underneath it, so its
|
||||
@@ -416,14 +408,14 @@ def build_plan(origin: Optional[Origin] = None) -> dict[str, PlannedFile]:
|
||||
if source.is_file():
|
||||
plan[relative] = _read_planned_file(source, relative)
|
||||
|
||||
for sub in RAW_SUBDIRS:
|
||||
plan[f"raw/{sub}/.gitkeep"] = PlannedFile("")
|
||||
# `incoming/` mirrors raw/'s type subdirectories (raw/CONTRACT.md
|
||||
# "Getting a file in") - seeded the same way, though `.gitignore`
|
||||
# (also exported, see ROOT_FILES) excludes the whole directory again
|
||||
# once the instance is a git repo, which is why bootstrap.md re-creates
|
||||
# it for a plain clone that never had this export step at all.
|
||||
plan[f"incoming/{sub}/.gitkeep"] = PlannedFile("")
|
||||
# `raw/` and `incoming/` are both flat now (Gitea #67 removes type
|
||||
# subdirectories from the addressing scheme entirely - a file's location
|
||||
# under `raw/` is a date shard computed by `raw accept`, never a hand-picked
|
||||
# type). `.gitignore` (also exported, see ROOT_FILES) excludes `incoming/`
|
||||
# again once the instance is a git repo, which is why bootstrap.md
|
||||
# re-creates it for a plain clone that never had this export step at all.
|
||||
plan["raw/.gitkeep"] = PlannedFile("")
|
||||
plan["incoming/.gitkeep"] = PlannedFile("")
|
||||
|
||||
plan["kb/log.md"] = PlannedFile((DIST_TEMPLATES_DIR / "log.md").read_text(encoding="utf-8"))
|
||||
plan["CHANGES.md"] = PlannedFile((DIST_TEMPLATES_DIR / "CHANGES.md").read_text(encoding="utf-8"))
|
||||
|
||||
@@ -38,7 +38,6 @@ from typing import Optional
|
||||
import typer
|
||||
|
||||
from chemenu import config, conventions, kb_collections, version as version_mod
|
||||
from chemenu.commands import dist_cmd
|
||||
from chemenu.commands._util import fail, rel_path, success
|
||||
|
||||
app = typer.Typer(help="Verify documentation that mirrors the code or repo layout.")
|
||||
@@ -113,8 +112,9 @@ REQUIRED_IGNORE_CANARIES = (
|
||||
# raw/ itself, a file here must never be committed - promotion via
|
||||
# `wikitool raw accept` is what makes it immutable, not the drop - so this
|
||||
# is the one canary in this tuple asserting the *opposite* of raw/'s own
|
||||
# backstop a few lines above.
|
||||
"incoming/documents/probe.pdf",
|
||||
# backstop a few lines above. Flat since Gitea #67 - incoming/ no longer
|
||||
# has type subdirectories, so the probe sits directly in it.
|
||||
"incoming/probe.pdf",
|
||||
)
|
||||
REQUIRED_TRACKED_PATHS = (
|
||||
"reports/CONTRACT.md",
|
||||
@@ -165,11 +165,6 @@ LEGACY_TYPE_RE = re.compile(r"^type:\s*(entity|concept|source|comparison)\s*$",
|
||||
# First backticked cell of a markdown table row, e.g. "| `xref add --a ...` | ... |"
|
||||
TABLE_CELL_RE = re.compile(r"^\|\s*`([^`]+)`", re.MULTILINE)
|
||||
|
||||
# A raw/CONTRACT.md routing-table cell naming a bare type subdirectory, e.g.
|
||||
# "| `articles/` | ... |" - deliberately narrower than TABLE_CELL_RE, which
|
||||
# would also match a command example elsewhere on the page.
|
||||
RAW_DIR_CELL_RE = re.compile(r"^\|\s*`([a-zA-Z0-9_-]+)/`\s*\|", re.MULTILINE)
|
||||
|
||||
|
||||
def registered_commands() -> set[str]:
|
||||
"""Every command path the CLI exposes, e.g. {'new', 'xref add', ...}.
|
||||
@@ -329,34 +324,6 @@ def check_stack_required_types() -> list[str]:
|
||||
return issues
|
||||
|
||||
|
||||
def documented_raw_subdirs(text: str) -> list[str]:
|
||||
return [match.group(1) for match in RAW_DIR_CELL_RE.finditer(text)]
|
||||
|
||||
|
||||
def check_raw_subdirs() -> list[str]:
|
||||
"""`raw/CONTRACT.md`'s routing table and `dist_cmd.RAW_SUBDIRS` must name
|
||||
the same set of type subdirectories (Gitea #58) - the table is meant to
|
||||
read as behaviour derived from the tuple, not as a second place the list
|
||||
could drift (AGENTS.md invariant 8). Skipped if the contract itself is
|
||||
missing; `check_collection_contracts` already reports that.
|
||||
"""
|
||||
contract_path = config.ROOT / "raw" / "CONTRACT.md"
|
||||
if not contract_path.exists():
|
||||
return []
|
||||
documented = set(documented_raw_subdirs(contract_path.read_text(encoding="utf-8")))
|
||||
declared = set(dist_cmd.RAW_SUBDIRS)
|
||||
issues = [
|
||||
f"raw/CONTRACT.md's routing table is missing `{missing}/` - dist_cmd.RAW_SUBDIRS names it"
|
||||
for missing in sorted(declared - documented)
|
||||
]
|
||||
issues += [
|
||||
f"raw/CONTRACT.md's routing table lists `{extra}/`, but dist_cmd.RAW_SUBDIRS does not - "
|
||||
"the two must name the same set"
|
||||
for extra in sorted(documented - declared)
|
||||
]
|
||||
return issues
|
||||
|
||||
|
||||
def check_legacy_type_blocks() -> list[str]:
|
||||
issues = []
|
||||
guarded = [
|
||||
@@ -625,7 +592,6 @@ def verify():
|
||||
check_cli_readme()
|
||||
+ check_readmes_have_no_command_table()
|
||||
+ check_collection_contracts()
|
||||
+ check_raw_subdirs()
|
||||
+ check_legacy_type_blocks()
|
||||
+ check_ignored_content()
|
||||
+ check_version_changelog()
|
||||
|
||||
@@ -298,6 +298,33 @@ def new_page_command(
|
||||
|
||||
frontmatter = _build_frontmatter(type_path, schema, today, explicit)
|
||||
|
||||
# Capture fields (Gitea #67, e.g. `fidelity`/`authority` on a source page)
|
||||
# are deliberately absent from `required:` - putting them there would make
|
||||
# every existing instance's source pages stop validating, a
|
||||
# boundary-crossing change (version-parts.md). The requirement is instead
|
||||
# enforced here, in the tool, exactly like `source_type`'s no-default
|
||||
# refusal (#66) reads from the schema alone: without a `default:` and
|
||||
# omitted from `explicit`, a capture field simply never lands in
|
||||
# `frontmatter`, so its absence has to be caught before it is silently
|
||||
# written as a page with no capture record at all.
|
||||
try:
|
||||
capture_fields = resolver.get_capture_fields(type_path)
|
||||
except ValueError as exc:
|
||||
fail(str(exc))
|
||||
missing_capture = [f for f in capture_fields if not frontmatter.get(f)]
|
||||
if missing_capture:
|
||||
fail(
|
||||
f"Type {type_path} requires capture field(s) {', '.join(missing_capture)} - pass them "
|
||||
"explicitly, e.g. --set fidelity=verbatim --set authority=reporting. "
|
||||
"new does not guess them (see raw/CONTRACT.md)."
|
||||
)
|
||||
guessed_unknown = [f for f in capture_fields if frontmatter.get(f) == "unknown"]
|
||||
if guessed_unknown:
|
||||
fail(
|
||||
f"Capture field(s) {', '.join(guessed_unknown)} cannot be set to 'unknown' here - that "
|
||||
"value is backfill-only, written only by `wikitool touch` on a page predating this rule."
|
||||
)
|
||||
|
||||
target_dir = _target_dir(type_path, frontmatter)
|
||||
_type_spec, template = _load_type_or_fail(type_path, target_dir)
|
||||
_validate_or_fail(frontmatter, type_path, target_dir)
|
||||
|
||||
@@ -1,34 +1,61 @@
|
||||
"""`wikitool raw accept` - promote one or more files from `incoming/` into
|
||||
`raw/`, with the destination computed rather than chosen by hand (Gitea #58).
|
||||
`raw/`, with the destination computed rather than chosen by hand (Gitea #58),
|
||||
sharded by the accept date rather than by a hand-picked type (Gitea #67).
|
||||
|
||||
A human classifies a file only by which type subdirectory of `incoming/` they
|
||||
drop it into - `incoming/articles/`, `incoming/documents/`, `incoming/notes/`,
|
||||
`incoming/assets/`, mirroring raw/CONTRACT.md's routing table. Everything past
|
||||
that is this command's job:
|
||||
A human no longer classifies a file at all: `incoming/` is flat, and a
|
||||
subdirectory dropped under it (an old `incoming/documents/` habit, a script
|
||||
that still writes one) is accepted and ignored rather than inspected -
|
||||
promoting `raw/` from a routing decision to an address computed purely from
|
||||
*when* the file was accepted:
|
||||
|
||||
- **Single file, no bundle.** One file promoted alone lands as
|
||||
`raw/<type>/<name>` - no directory of its own.
|
||||
`raw/<YYYY>/<MM>/<name>` - no directory of its own.
|
||||
- **Bundle from the second file on.** Several files of one source promoted in
|
||||
the same call land under `raw/<type>/<stem>/`, named after the first file's
|
||||
stem.
|
||||
the same call land under `raw/<YYYY>/<MM>/<stem>/`, named after the first
|
||||
file's stem.
|
||||
- **Growing an existing single file into a bundle.** `--page` extends an
|
||||
existing source page's `raw_files:`. If that raises the page from one file
|
||||
to more than one, the file it already had is folded into the new bundle
|
||||
alongside the ones just promoted, in the same call - at no point does
|
||||
`raw_files:` point at a path that does not exist.
|
||||
to more than one, the file it already had is folded into a bundle at its
|
||||
own parent directory - `raw/<its-existing-location>/<stem>/` - not at
|
||||
today's shard, so a bundle never mixes an old capture date with today's
|
||||
(Gitea #67 decision, "Datums-Shard" § "Bündelort").
|
||||
|
||||
Existing files under `raw/` are never moved by this change (Gitea #67
|
||||
"Altbestand bleibt stehen"): `raw/articles/`, `raw/documents/`, `raw/notes/`
|
||||
and `raw/assets/` keep whatever they already held, and stay valid promotion
|
||||
targets for `--replaces`.
|
||||
|
||||
**Every promotion now also carries `--fidelity` and `--authority`** (Gitea
|
||||
#67): how faithfully the material was captured, and what it is entitled to
|
||||
claim about its subject. Neither has a default and neither may be `unknown`
|
||||
here - that value is backfill-only, written only by `wikitool touch` on a
|
||||
page predating this rule. Given `--page`, both are written straight onto the
|
||||
target page (once - fill-once, like `touch`, see `touch._capture_field_or_fail`);
|
||||
without `--page` there is no page yet to write them onto (`wiki-ingest`
|
||||
creates it afterwards), so this command instead prints the exact
|
||||
`wikitool new source --set fidelity=... --set authority=...` follow-up line,
|
||||
and `new source` itself refuses to scaffold a source page without both.
|
||||
`--replaces` is the one path where both become *optional*: passing them there
|
||||
is the sanctioned way to correct an already-set capture value on a later
|
||||
edition, the same way `touch`'s own refusal points back to `--replaces`.
|
||||
|
||||
Multi-owner raw files (`provenance.duplicate_raw_file_owners`) are refused
|
||||
rather than silently moved: relocating a file another page also claims would
|
||||
break that page's `raw_files:` without it ever being consulted.
|
||||
|
||||
**Stem uniqueness at `raw/<type>/` level** (Gitea #64) closes the gap this
|
||||
leaves open: without it, a second, unrelated source whose primary file happens
|
||||
not to collide on the exact filename slips silently into an existing bundle,
|
||||
because the per-file `dst.exists()` check above never looks at the bundle
|
||||
directory itself. The set of names occupied at `raw/<type>/` level - file
|
||||
stems and bundle directory names alike - must stay unique; `_occupied_stems()`
|
||||
and the check built on it enforce that, while still allowing a call to grow a
|
||||
bundle it already owns (via `--page`, or by continuing an existing bundle).
|
||||
**Stem uniqueness across the whole of `raw/`** (Gitea #64, widened by #67):
|
||||
without it, a second, unrelated source whose primary file happens not to
|
||||
collide on the exact filename slips silently into an existing bundle, because
|
||||
the per-file `dst.exists()` check above never looks at the bundle directory
|
||||
itself. The set of names occupied anywhere under `raw/` - file stems and
|
||||
bundle directory names alike, at whichever level directly holds files - must
|
||||
stay unique; `_occupied_stems()` and the check built on it enforce that,
|
||||
while still allowing a call to grow a bundle it already owns (via `--page`,
|
||||
or by continuing an existing bundle). It used to be scoped to one type
|
||||
directory; #67 removes type directories from the addressing scheme entirely,
|
||||
so uniqueness now has to span old-style flat directories and the new date
|
||||
shard together, or `--replaces` on a bygone stem would be ambiguous across
|
||||
shards.
|
||||
|
||||
**`--replaces <raw-path>`** is the only sanctioned way past that rule: whether
|
||||
a new file is a later edition of an existing source or a second, separate one
|
||||
@@ -37,6 +64,8 @@ and names both routes rather than choosing one (Gitea #64 decision 2).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
@@ -44,13 +73,20 @@ import typer
|
||||
|
||||
from chemenu import config
|
||||
from chemenu.commands._util import fail, rel_path, success
|
||||
from chemenu.commands.dist_cmd import RAW_SUBDIRS
|
||||
from chemenu.frontmatter_io import write_page
|
||||
from chemenu.kb_scan import load_kb_pages
|
||||
from chemenu.provenance import citing_pages, source_pages_by_raw_file, source_raw_files
|
||||
from chemenu.type_resolver import resolver
|
||||
|
||||
app = typer.Typer(help="Promote raw material out of incoming/ into raw/.")
|
||||
|
||||
# The type this command always promotes into eventually - hardcoded rather
|
||||
# than derived from --page, because there is no page yet in the common case
|
||||
# (see module docstring): the follow-up hint and the capture-field/enum
|
||||
# validation below both need *a* type to read from, and `source` is the only
|
||||
# one `raw accept` has ever scaffolded towards.
|
||||
_SOURCE_TYPE_PATH = "types/source.md"
|
||||
|
||||
|
||||
def _incoming_dir() -> Path:
|
||||
return config.ROOT / "incoming"
|
||||
@@ -60,9 +96,24 @@ def _resolve(raw: Path) -> Path:
|
||||
return raw if raw.is_absolute() else config.ROOT / raw
|
||||
|
||||
|
||||
def _classify(path: Path, incoming: Path) -> str:
|
||||
"""The type subdirectory `path` (already resolved, absolute) declares by
|
||||
where it sits under `incoming/`, or fail with the reason it doesn't."""
|
||||
def _shard_dir(today: Optional[datetime.date] = None) -> Path:
|
||||
"""`raw/<YYYY>/<MM>/`, computed from the accept date - never from a count,
|
||||
so it can never rebalance and break a `[^cite-id]` anchor (Gitea #67).
|
||||
"""
|
||||
d = today or datetime.date.today()
|
||||
return config.RAW_DIR / f"{d.year:04d}" / f"{d.month:02d}"
|
||||
|
||||
|
||||
def _validate_under_incoming(path: Path, incoming: Path) -> None:
|
||||
"""`path` must sit directly in `incoming/`, or exactly one level below it.
|
||||
|
||||
Unlike before #67, that one optional level carries no meaning any more -
|
||||
it is accepted and ignored, kept only so an old `incoming/<type>/` habit
|
||||
or script does not have to change to keep working (the MINOR condition
|
||||
named in Gitea #67 "Versionsteil"). Nesting deeper than that is still
|
||||
refused: it was never meaningful and silently accepting it would hide a
|
||||
typo'd path.
|
||||
"""
|
||||
try:
|
||||
rel = path.relative_to(incoming)
|
||||
except ValueError:
|
||||
@@ -70,37 +121,58 @@ def _classify(path: Path, incoming: Path) -> str:
|
||||
f"{rel_path(path)} is not under incoming/ - `raw accept` only promotes files "
|
||||
"from there. See raw/CONTRACT.md."
|
||||
)
|
||||
allowed = ", ".join(f"incoming/{s}/" for s in RAW_SUBDIRS)
|
||||
if len(rel.parts) < 2:
|
||||
fail(
|
||||
f"incoming/{rel} declares no type - place it inside one of {allowed} instead "
|
||||
"of directly in incoming/."
|
||||
)
|
||||
sub = rel.parts[0]
|
||||
if sub not in RAW_SUBDIRS:
|
||||
fail(f"incoming/{rel} lies under an unknown type directory 'incoming/{sub}/'. Allowed: {allowed}.")
|
||||
if len(rel.parts) < 1:
|
||||
fail(f"incoming/{rel} names no file.")
|
||||
if len(rel.parts) > 2:
|
||||
fail(f"incoming/{rel} is nested below its type directory - place it directly in incoming/{sub}/.")
|
||||
return sub
|
||||
fail(
|
||||
f"incoming/{rel} is nested more than one level below incoming/ - place it "
|
||||
"directly in incoming/, or in at most one subdirectory of it (the "
|
||||
"subdirectory itself is ignored, see raw/CONTRACT.md)."
|
||||
)
|
||||
|
||||
|
||||
def _occupied_stems(raw_sub_dir: Path) -> dict[str, Path]:
|
||||
"""Every name occupied at `raw/<type>/` level: file stems and bundle
|
||||
directory names alike, one level below `raw_sub_dir` only."""
|
||||
_YEAR_DIR = re.compile(r"\d{4}")
|
||||
|
||||
|
||||
def _occupied_stems(raw_dir: Path) -> dict[str, Path]:
|
||||
"""Every name occupied anywhere under `raw/`: file stems and bundle
|
||||
directory names alike, at whichever level directly holds files.
|
||||
|
||||
Distinguishes the pre-#67 flat layout (`raw/<name>/<file-or-bundle>`)
|
||||
from the #67 date shard (`raw/<YYYY>/<MM>/<file-or-bundle>`) structurally,
|
||||
by whether a top-level directory name is a 4-digit year - not from a
|
||||
hardcoded list of legacy type names, so this keeps working unchanged if a
|
||||
legacy directory is ever renamed by hand.
|
||||
"""
|
||||
occupied: dict[str, Path] = {}
|
||||
if raw_sub_dir.is_dir():
|
||||
for entry in raw_sub_dir.iterdir():
|
||||
occupied[entry.stem if entry.is_file() else entry.name] = entry
|
||||
if not raw_dir.is_dir():
|
||||
return occupied
|
||||
for top in raw_dir.iterdir():
|
||||
if not top.is_dir():
|
||||
continue # raw/CONTRACT.md
|
||||
if _YEAR_DIR.fullmatch(top.name):
|
||||
for month in top.iterdir():
|
||||
if month.is_dir():
|
||||
occupied.update(_leaf_group_entries(month))
|
||||
else:
|
||||
occupied.update(_leaf_group_entries(top))
|
||||
return occupied
|
||||
|
||||
|
||||
def _stem_collision_message(sub: str, claimed_name: str, holder: Path) -> str:
|
||||
def _leaf_group_entries(group_dir: Path) -> dict[str, Path]:
|
||||
"""The addressable entries directly inside one leaf-group directory
|
||||
(a legacy type dir, or one `raw/<YYYY>/<MM>/`): a lone file occupies its
|
||||
stem, a bundle directory occupies its own name."""
|
||||
return {entry.stem if entry.is_file() else entry.name: entry for entry in group_dir.iterdir()}
|
||||
|
||||
|
||||
def _stem_collision_message(claimed_name: str, holder: Path) -> str:
|
||||
example_target = holder
|
||||
if holder.is_dir():
|
||||
children = sorted(holder.iterdir())
|
||||
example_target = children[0] if children else holder
|
||||
return (
|
||||
f'{rel_path(holder)} already claims the stem "{claimed_name}" in raw/{sub}/.\n'
|
||||
f'{rel_path(holder)} already claims the stem "{claimed_name}" under raw/.\n'
|
||||
" These are two different intents and only you can tell them apart:\n"
|
||||
f" Same source, new edition -> tools/wikitool raw accept "
|
||||
f"--replaces {rel_path(example_target)} <incoming file>\n"
|
||||
@@ -110,10 +182,53 @@ def _stem_collision_message(sub: str, claimed_name: str, holder: Path) -> str:
|
||||
)
|
||||
|
||||
|
||||
def _replace(files: list[Path], replaces: Path, page: Optional[str], dry_run: bool) -> None:
|
||||
def _capture_choices() -> tuple[list[str], list[str]]:
|
||||
"""Allowed `--fidelity`/`--authority` values, straight from the schema
|
||||
(single source of truth) - `unknown` excluded, since it is backfill-only
|
||||
and neither flag may write it (Gitea #67)."""
|
||||
fidelity = [v for v in resolver.get_enum(_SOURCE_TYPE_PATH, "fidelity") if v != "unknown"]
|
||||
authority = [v for v in resolver.get_enum(_SOURCE_TYPE_PATH, "authority") if v != "unknown"]
|
||||
return fidelity, authority
|
||||
|
||||
|
||||
def _check_capture_value(field: str, value: Optional[str], allowed: list[str]) -> None:
|
||||
if value is None:
|
||||
return
|
||||
if value == "unknown" or value not in allowed:
|
||||
fail(
|
||||
f"--{field} must be one of: {', '.join(allowed)}. 'unknown' is backfill-only - only "
|
||||
"`wikitool touch` may write it, on a page predating this rule."
|
||||
)
|
||||
|
||||
|
||||
def _overwrite_capture_fields_on_page(page, fidelity: Optional[str], authority: Optional[str]) -> bool:
|
||||
"""Write `fidelity`/`authority` onto `page.frontmatter`, unconditionally -
|
||||
only called from `_replace`, the one sanctioned way to correct an
|
||||
already-set capture value (Gitea #67). Returns whether anything changed,
|
||||
so the caller only writes the page back when it needs to."""
|
||||
changed = False
|
||||
for field, value in (("fidelity", fidelity), ("authority", authority)):
|
||||
if value is not None and page.frontmatter.get(field) != value:
|
||||
page.frontmatter[field] = value
|
||||
changed = True
|
||||
return changed
|
||||
|
||||
|
||||
def _replace(
|
||||
files: list[Path],
|
||||
replaces: Path,
|
||||
page: Optional[str],
|
||||
fidelity: Optional[str],
|
||||
authority: Optional[str],
|
||||
dry_run: bool,
|
||||
) -> None:
|
||||
"""`raw accept --replaces <target> <incoming file>` - overwrite an
|
||||
existing raw/ file wholesale with a later edition, in place, with
|
||||
`raw_files:` on every source page left untouched (Gitea #64 decision 2).
|
||||
`--fidelity`/`--authority` are optional here, and are the one sanctioned
|
||||
way to correct an already-set capture value (Gitea #67) - passed, they
|
||||
overwrite the owning page's value; omitted, the page's capture fields are
|
||||
left exactly as they were.
|
||||
|
||||
Every check below runs before any write, so a failure leaves both the
|
||||
target and the incoming file exactly as they were - the invariant the
|
||||
@@ -124,24 +239,22 @@ def _replace(files: list[Path], replaces: Path, page: Optional[str], dry_run: bo
|
||||
if len(files) != 1:
|
||||
fail("--replaces takes exactly one incoming file - a replacement is one file for one file.")
|
||||
|
||||
fidelity_choices, authority_choices = _capture_choices()
|
||||
_check_capture_value("fidelity", fidelity, fidelity_choices)
|
||||
_check_capture_value("authority", authority, authority_choices)
|
||||
|
||||
incoming_path = _resolve(files[0])
|
||||
if not incoming_path.is_file():
|
||||
fail(f"{rel_path(incoming_path)} does not exist or is not a file.")
|
||||
incoming_sub = _classify(incoming_path, _incoming_dir())
|
||||
_validate_under_incoming(incoming_path, _incoming_dir())
|
||||
|
||||
target = _resolve(replaces)
|
||||
try:
|
||||
target_rel = target.relative_to(config.RAW_DIR)
|
||||
target.relative_to(config.RAW_DIR)
|
||||
except ValueError:
|
||||
fail(f"--replaces target {rel_path(target)} does not lie under raw/.")
|
||||
if not target.is_file():
|
||||
fail(f"--replaces target {rel_path(target)} does not exist or is not a file.")
|
||||
target_sub = target_rel.parts[0]
|
||||
if target_sub != incoming_sub:
|
||||
fail(
|
||||
f"--replaces target is under raw/{target_sub}/, but {rel_path(incoming_path)} is "
|
||||
f"under incoming/{incoming_sub}/ - a replacement stays within one type directory."
|
||||
)
|
||||
if target.name != incoming_path.name:
|
||||
fail(
|
||||
f"--replaces target {rel_path(target)} has a different filename than "
|
||||
@@ -157,13 +270,25 @@ def _replace(files: list[Path], replaces: Path, page: Optional[str], dry_run: bo
|
||||
f"({', '.join(owners)}). Resolve the multiple ownership first (see `wikitool sources coverage`)."
|
||||
)
|
||||
|
||||
owner_page = pages.get(owners[0]) if owners else None
|
||||
will_update_capture = owner_page is not None and _overwrite_capture_fields_on_page(
|
||||
owner_page, fidelity, authority
|
||||
)
|
||||
|
||||
if dry_run:
|
||||
typer.echo(f"[dry-run] would replace {rel_path(target)} with {rel_path(incoming_path)}. No files written.")
|
||||
if will_update_capture:
|
||||
typer.echo(
|
||||
f"[dry-run] would set fidelity={fidelity!r}, authority={authority!r} on '{owners[0]}'"
|
||||
)
|
||||
return
|
||||
|
||||
target.unlink()
|
||||
incoming_path.rename(target)
|
||||
|
||||
if will_update_capture:
|
||||
write_page(owner_page.path, owner_page.frontmatter, owner_page.body)
|
||||
|
||||
if not owners:
|
||||
success(
|
||||
f"Replaced {rel_path(target)} (previous version stays in git history). "
|
||||
@@ -188,7 +313,20 @@ def _replace(files: list[Path], replaces: Path, page: Optional[str], dry_run: bo
|
||||
def raw_accept_command(
|
||||
files: list[Path] = typer.Argument(
|
||||
...,
|
||||
help="One or more files under incoming/<type>/, all belonging to the same source",
|
||||
help="One or more files under incoming/, all belonging to the same source",
|
||||
),
|
||||
fidelity: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--fidelity",
|
||||
help="How faithful the capture is (see `wikitool types describe source`). Required unless "
|
||||
"--replaces is given, where it is the optional, sanctioned way to correct an already-set value",
|
||||
),
|
||||
authority: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--authority",
|
||||
help="What the material may claim about its subject (see `wikitool types describe source`). "
|
||||
"Required unless --replaces is given, where it is the optional, sanctioned way to correct "
|
||||
"an already-set value",
|
||||
),
|
||||
page: Optional[str] = typer.Option(
|
||||
None,
|
||||
@@ -205,7 +343,7 @@ def raw_accept_command(
|
||||
dry_run: bool = typer.Option(False, "--dry-run", help="List what would move without writing"),
|
||||
):
|
||||
"""Promote file(s) from incoming/ into raw/, computing the destination
|
||||
(type directory, bundle or not, bundle name) instead of taking it as an
|
||||
(date shard, bundle or not, bundle name) instead of taking it as an
|
||||
argument. See raw/CONTRACT.md "Getting a file in: incoming/"."""
|
||||
if not files:
|
||||
fail("Pass at least one file to promote.")
|
||||
@@ -213,27 +351,30 @@ def raw_accept_command(
|
||||
incoming = _incoming_dir()
|
||||
|
||||
if replaces is not None:
|
||||
_replace(files, replaces, page, dry_run)
|
||||
_replace(files, replaces, page, fidelity, authority, dry_run)
|
||||
return
|
||||
|
||||
fidelity_choices, authority_choices = _capture_choices()
|
||||
if fidelity is None or authority is None:
|
||||
missing = ", ".join(n for n, v in (("--fidelity", fidelity), ("--authority", authority)) if v is None)
|
||||
fail(
|
||||
f"{missing} required - raw accept does not guess how faithful a capture is or what it "
|
||||
f"may claim. --fidelity: {', '.join(fidelity_choices)}. --authority: {', '.join(authority_choices)}."
|
||||
)
|
||||
_check_capture_value("fidelity", fidelity, fidelity_choices)
|
||||
_check_capture_value("authority", authority, authority_choices)
|
||||
|
||||
resolved = [_resolve(f) for f in files]
|
||||
|
||||
for path in resolved:
|
||||
if not path.is_file():
|
||||
fail(f"{rel_path(path)} does not exist or is not a file.")
|
||||
|
||||
subs = {_classify(path, incoming) for path in resolved}
|
||||
if len(subs) > 1:
|
||||
allowed = ", ".join(sorted(f"incoming/{s}/" for s in subs))
|
||||
fail(f"All files in one `raw accept` call must share one type directory; got {allowed}.")
|
||||
sub = subs.pop()
|
||||
_validate_under_incoming(path, incoming)
|
||||
|
||||
names = [path.name for path in resolved]
|
||||
if len(names) != len(set(names)):
|
||||
fail("Two files share a filename; rename one before promoting.")
|
||||
|
||||
raw_sub_dir = config.RAW_DIR / sub
|
||||
|
||||
pages = None
|
||||
target_page = None
|
||||
existing_raw_paths: list[Path] = []
|
||||
@@ -256,12 +397,6 @@ def raw_accept_command(
|
||||
f"{', '.join(rel_path(p) for p in missing)}. Fix raw_files: (see `sources coverage`) "
|
||||
"before promoting more."
|
||||
)
|
||||
existing_subs = {p.relative_to(config.RAW_DIR).parts[0] for p in existing_raw_paths}
|
||||
if existing_subs != {sub}:
|
||||
fail(
|
||||
f"'{page}' already claims file(s) under {', '.join(sorted(f'raw/{s}/' for s in existing_subs))}, "
|
||||
f"not raw/{sub}/. A bundle is one type directory; promote separately."
|
||||
)
|
||||
|
||||
# A bundle directory forms once two or more files belong to the source
|
||||
# (Gitea #58 decision 3): from the second file on, never before. Whenever
|
||||
@@ -278,36 +413,46 @@ def raw_accept_command(
|
||||
)
|
||||
bundle_dir = parents.pop()
|
||||
elif total >= 2:
|
||||
primary = existing_raw_paths[0] if existing_raw_paths else resolved[0]
|
||||
bundle_dir = raw_sub_dir / primary.stem
|
||||
if existing_raw_paths:
|
||||
# Growing a bundle out of a single already-promoted file (Gitea
|
||||
# #67 decision): the bundle forms at that file's own parent
|
||||
# directory, never at today's shard - the file's capture date is
|
||||
# whatever it always was, and a bundle mixing an old and a new
|
||||
# shard would have no single correct address.
|
||||
primary = existing_raw_paths[0]
|
||||
bundle_dir = primary.parent / primary.stem
|
||||
else:
|
||||
primary = resolved[0]
|
||||
bundle_dir = _shard_dir() / primary.stem
|
||||
|
||||
moves: list[tuple[Path, Path]] = []
|
||||
for existing in existing_raw_paths:
|
||||
if bundle_dir is not None and existing.parent != bundle_dir:
|
||||
moves.append((existing, bundle_dir / existing.name))
|
||||
for new_path in resolved:
|
||||
dst = (bundle_dir / new_path.name) if bundle_dir is not None else (raw_sub_dir / new_path.name)
|
||||
dst = (bundle_dir / new_path.name) if bundle_dir is not None else (_shard_dir() / new_path.name)
|
||||
moves.append((new_path, dst))
|
||||
|
||||
for _src, dst in moves:
|
||||
if dst.exists():
|
||||
fail(f"Cannot promote: {rel_path(dst)} already exists.")
|
||||
|
||||
# Stem uniqueness at raw/<type>/ level (Gitea #64): the name this call is
|
||||
# about to claim there - the bundle's name, or the lone file's stem when no
|
||||
# bundle forms - must not already belong to something this call does not
|
||||
# itself own. "Owns" means: one of the page's already-registered raw files
|
||||
# (the pitfall from the module docstring - a single file growing into a
|
||||
# bundle of its own name momentarily still occupies that name), or, once a
|
||||
# bundle already has >=2 registered files, the bundle directory itself.
|
||||
# Stem uniqueness across raw/ (Gitea #64, widened by #67): the name this
|
||||
# call is about to claim - the bundle's name, or the lone file's stem when
|
||||
# no bundle forms - must not already belong to something this call does
|
||||
# not itself own. "Owns" means: one of the page's already-registered raw
|
||||
# files (the pitfall from the module docstring - a single file growing
|
||||
# into a bundle of its own name momentarily still occupies that name), or,
|
||||
# once a bundle already has >=2 registered files, the bundle directory
|
||||
# itself.
|
||||
claimed_name = bundle_dir.name if bundle_dir is not None else resolved[0].stem
|
||||
occupied = _occupied_stems(raw_sub_dir)
|
||||
occupied = _occupied_stems(config.RAW_DIR)
|
||||
owned = set(existing_raw_paths)
|
||||
if len(existing_raw_paths) >= 2:
|
||||
owned.add(bundle_dir)
|
||||
holder = occupied.get(claimed_name)
|
||||
if holder is not None and holder not in owned:
|
||||
fail(_stem_collision_message(sub, claimed_name, holder))
|
||||
fail(_stem_collision_message(claimed_name, holder))
|
||||
|
||||
moving_existing = [src for src, _dst in moves if src in existing_raw_paths]
|
||||
if moving_existing:
|
||||
@@ -325,6 +470,19 @@ def raw_accept_command(
|
||||
"Resolve the multiple ownership first (see `wikitool sources coverage`)."
|
||||
)
|
||||
|
||||
if target_page is not None:
|
||||
current_fidelity = target_page.frontmatter.get("fidelity")
|
||||
current_authority = target_page.frontmatter.get("authority")
|
||||
for field, value, current in (
|
||||
("fidelity", fidelity, current_fidelity),
|
||||
("authority", authority, current_authority),
|
||||
):
|
||||
if current not in (None, "") and current != value:
|
||||
fail(
|
||||
f"'{page}' already has {field}={current!r} - a capture field is fixed once. "
|
||||
"Use `raw accept --replaces` to correct it."
|
||||
)
|
||||
|
||||
if dry_run:
|
||||
for src, dst in moves:
|
||||
typer.echo(f"[dry-run] would move {rel_path(src)} -> {rel_path(dst)}")
|
||||
@@ -332,6 +490,7 @@ def raw_accept_command(
|
||||
moved_map = dict(moves)
|
||||
final = [moved_map.get(p, p) for p in existing_raw_paths] + [moved_map[p] for p in resolved]
|
||||
typer.echo(f"[dry-run] would set raw_files: on '{page}' to {[rel_path(p) for p in final]}")
|
||||
typer.echo(f"[dry-run] would set fidelity={fidelity!r}, authority={authority!r} on '{page}'")
|
||||
typer.echo(f"[dry-run] would move {len(moves)} file(s). No files written.")
|
||||
return
|
||||
|
||||
@@ -344,6 +503,8 @@ def raw_accept_command(
|
||||
if target_page is not None:
|
||||
final = [moved_map.get(p, p) for p in existing_raw_paths] + [moved_map[p] for p in resolved]
|
||||
target_page.frontmatter["raw_files"] = [rel_path(p) for p in final]
|
||||
target_page.frontmatter["fidelity"] = fidelity
|
||||
target_page.frontmatter["authority"] = authority
|
||||
write_page(target_page.path, target_page.frontmatter, target_page.body)
|
||||
success(
|
||||
f"Promoted {len(resolved)} file(s); updated raw_files: on '{page}' "
|
||||
@@ -352,7 +513,12 @@ def raw_accept_command(
|
||||
return
|
||||
|
||||
promoted = ", ".join(rel_path(moved_map[p]) for p in resolved)
|
||||
raw_files_arg = ",".join(rel_path(moved_map[p]) for p in resolved)
|
||||
success(
|
||||
f"Promoted {len(resolved)} file(s) to {promoted}. "
|
||||
"Run `wikitool new source --set raw_files=...` (or `touch --set` on an existing page) next."
|
||||
f"Promoted {len(resolved)} file(s) to {promoted}.\n"
|
||||
" Next:\n"
|
||||
" tools/wikitool new source --name \"<Title>\" \\\n"
|
||||
f" --set raw_files={raw_files_arg} \\\n"
|
||||
f" --set fidelity={fidelity} --set authority={authority} \\\n"
|
||||
" --set source_type=<category>"
|
||||
)
|
||||
|
||||
@@ -128,6 +128,23 @@ def _settable_or_fail(field: str, schema: Optional[Dict[str, Any]], type_path: s
|
||||
return properties[field]
|
||||
|
||||
|
||||
def _capture_field_or_fail(field: str, value: Any, frontmatter: Dict[str, Any]) -> None:
|
||||
"""Refuse to overwrite a capture field (Gitea #67) that already carries a
|
||||
value - fill-once, not a denylist entry: `UNSETTABLE` would also forbid
|
||||
the *first* write, which is exactly the write the backfill needs. A
|
||||
capture field is fixed at `raw accept`/`new source` time; the only
|
||||
sanctioned way to change an already-set value is a new edition of the
|
||||
raw material (`raw accept --replaces`), never a second `touch`.
|
||||
"""
|
||||
current = frontmatter.get(field)
|
||||
if current not in (None, "") and current != value:
|
||||
fail(
|
||||
f"`{field}` is a capture field: fixed once, at `raw accept`/`new source` time, and "
|
||||
f"already reads {current!r}. A corrected capture is a new edition of the source, not "
|
||||
f"a touch -> `wikitool raw accept --replaces <raw path> <incoming file>`."
|
||||
)
|
||||
|
||||
|
||||
def _apply_set(frontmatter: Dict[str, Any], field: str, value: Any) -> Optional[str]:
|
||||
if frontmatter.get(field) == value:
|
||||
return None
|
||||
@@ -233,6 +250,7 @@ def touch_command(
|
||||
|
||||
try:
|
||||
schema = resolver.get_schema(type_path, page.path)
|
||||
capture_fields = set(resolver.get_capture_fields(type_path, page.path))
|
||||
except ValueError as exc:
|
||||
fail(str(exc))
|
||||
|
||||
@@ -290,6 +308,8 @@ def touch_command(
|
||||
parsed = parse_set_fields(values, schema, flag=flag)
|
||||
for field, value in parsed.items():
|
||||
_settable_or_fail(field, schema, type_path)
|
||||
if field in capture_fields:
|
||||
_capture_field_or_fail(field, value, frontmatter)
|
||||
change = apply(frontmatter, field, value)
|
||||
touched.add(field)
|
||||
if change:
|
||||
|
||||
@@ -15,6 +15,7 @@ from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from collections import Counter
|
||||
|
||||
@@ -143,6 +144,59 @@ def unclassified_source_pages(pages: dict[str, Page]) -> list[dict]:
|
||||
]
|
||||
|
||||
|
||||
# Capture-field ceilings (Gitea #67, kb/CONTRACT.md § "Confidence against
|
||||
# source standing"): the weakest `authority`/`fidelity` among a page's cited
|
||||
# sources bounds how high its `confidence_base` may honestly sit. Stack
|
||||
# vocabulary, not instance configuration - `fidelity`/`authority` are defined
|
||||
# by `raw/CONTRACT.md`, not by `kb/CONVENTIONS.md`. `unknown` and every axis
|
||||
# value not listed here carry no ceiling: a backfilled "we don't know" is not
|
||||
# a claim about the source, and `normative`/`verbatim`/`published` are simply
|
||||
# not weaker than hand-set confidence gets to be.
|
||||
_AUTHORITY_CEILING = {"reporting": 0.8, "opinion": 0.6}
|
||||
_FIDELITY_CEILING = {"secondhand": 0.7, "nontextual": 0.7}
|
||||
|
||||
|
||||
def _capture_ceiling(source_pages: list[Page]) -> Optional[float]:
|
||||
"""The tightest ceiling implied by `source_pages`' capture fields, or
|
||||
None if none of them carry a value with a ceiling at all."""
|
||||
ceilings = [
|
||||
ceiling
|
||||
for src in source_pages
|
||||
for field_map, value in (
|
||||
(_AUTHORITY_CEILING, src.frontmatter.get("authority")),
|
||||
(_FIDELITY_CEILING, src.frontmatter.get("fidelity")),
|
||||
)
|
||||
if (ceiling := field_map.get(value)) is not None
|
||||
]
|
||||
return min(ceilings) if ceilings else None
|
||||
|
||||
|
||||
def confidence_exceeds_source_standing(pages: dict[str, Page]) -> list[dict]:
|
||||
"""Pages whose `confidence_base` sits above what their cited sources'
|
||||
capture standing can honestly carry.
|
||||
|
||||
Advisory, not a formula (kb/CONTRACT.md § "Confidence against source
|
||||
standing" has the reasoning): `confidence_base` stays a human judgment,
|
||||
and authority is a ceiling a page may sit under by independent
|
||||
verification, not a value a formula could compute outright.
|
||||
"""
|
||||
findings: list[dict] = []
|
||||
for title, page in sorted(pages.items()):
|
||||
if page.kind not in ("entity", "concept"):
|
||||
continue
|
||||
confidence_base = page.frontmatter.get("confidence_base")
|
||||
source_titles = page.frontmatter.get("sources") or []
|
||||
if confidence_base is None or not source_titles:
|
||||
continue
|
||||
source_pages = [
|
||||
pages[t] for t in source_titles if t in pages and pages[t].kind == "source"
|
||||
]
|
||||
ceiling = _capture_ceiling(source_pages)
|
||||
if ceiling is not None and confidence_base > ceiling:
|
||||
findings.append({"page": title, "confidence_base": confidence_base, "ceiling": ceiling})
|
||||
return findings
|
||||
|
||||
|
||||
def nested_pages(kb_dir: Path, pages: dict[str, Page]) -> list[dict]:
|
||||
"""Report form of `find_nested_pages`: `{"page", "at", "depth"}` per
|
||||
finding.
|
||||
@@ -491,6 +545,7 @@ def run_lint(kb_dir: Path) -> dict:
|
||||
"nested_pages": nested,
|
||||
"unsharded_collections": unsharded_collections(kb_dir, pages),
|
||||
"unclassified_source_pages": unclassified_source_pages(pages),
|
||||
"confidence_exceeds_source_standing": confidence_exceeds_source_standing(pages),
|
||||
"uncovered_raw_files": find_uncovered_raw_files(config.RAW_DIR, pages),
|
||||
"broken_raw_refs": find_broken_raw_refs(pages),
|
||||
"duplicate_raw_file_owners": find_duplicate_raw_file_owners(pages),
|
||||
@@ -590,6 +645,13 @@ def render_markdown(report: dict) -> str:
|
||||
lambda i: f"[[{i['page']}]] - `wikitool touch --set source_type=<value>` once its "
|
||||
"category is known",
|
||||
)
|
||||
_section(
|
||||
lines, "Confidence Above Source Standing (ceiling, not a formula) - recommendation, not an error",
|
||||
report.get("confidence_exceeds_source_standing", []),
|
||||
lambda i: f"[[{i['page']}]] confidence_base={i['confidence_base']} exceeds the "
|
||||
f"{i['ceiling']} ceiling its cited sources' fidelity/authority carry - "
|
||||
"kb/CONTRACT.md § \"Confidence against source standing\"",
|
||||
)
|
||||
_section(
|
||||
lines, "Uncovered Raw Files (no source page)", report["uncovered_raw_files"],
|
||||
lambda i: f"`{i}`",
|
||||
@@ -757,6 +819,13 @@ def default_report_path(report: dict) -> Path:
|
||||
# would penalise the honest "I don't know yet" that the slot exists to allow,
|
||||
# where the old silent `default: notes` hid the same uncertainty for free.
|
||||
#
|
||||
# `confidence_exceeds_source_standing` is advisory by construction, like
|
||||
# `unsharded_collections` above: `confidence_base` stays a hand-set judgment
|
||||
# call (kb/CONTRACT.md § Confidence), and a source's capture standing is a
|
||||
# ceiling a page may sit under by independent verification, not a value a
|
||||
# formula could compute outright - see kb/CONTRACT.md § "Confidence against
|
||||
# source standing" (Gitea #67).
|
||||
#
|
||||
# `malformed_edges` and `unbalanced_markers` are hard from the start: neither
|
||||
# describes an unconverted page, only a broken one.
|
||||
#
|
||||
|
||||
@@ -364,21 +364,15 @@ def test_page_type_specs_ship_as_templates_and_stack_types_do_not(repo, monkeypa
|
||||
assert "types/instruction.md.template" not in plan
|
||||
|
||||
|
||||
def test_plan_creates_empty_raw_subdirs_not_real_content(repo):
|
||||
def test_plan_creates_empty_raw_and_incoming_not_real_content(repo):
|
||||
"""Both flat since Gitea #67: `raw/` addresses a file by its accept date,
|
||||
never by a hand-picked type, so there is nothing left to seed per type."""
|
||||
plan = dist_cmd.build_plan()
|
||||
for sub in ("articles", "documents", "notes", "assets"):
|
||||
assert f"raw/{sub}/.gitkeep" in plan
|
||||
assert "raw/.gitkeep" in plan
|
||||
assert "incoming/.gitkeep" in plan
|
||||
assert not any("personal-note" in relative for relative in plan)
|
||||
|
||||
|
||||
def test_plan_creates_matching_incoming_subdirs(repo):
|
||||
"""The ingest inbox (Gitea #58) mirrors raw/'s type subdirectories one for
|
||||
one - both come from the same `RAW_SUBDIRS` tuple."""
|
||||
plan = dist_cmd.build_plan()
|
||||
for sub in ("articles", "documents", "notes", "assets"):
|
||||
assert f"incoming/{sub}/.gitkeep" in plan
|
||||
|
||||
|
||||
def test_plan_seeds_log_and_changes_from_templates(repo):
|
||||
plan = dist_cmd.build_plan()
|
||||
assert "Wiki Log" in plan["kb/log.md"].content
|
||||
@@ -496,8 +490,8 @@ def test_export_into_a_fresh_directory_works(repo, tmp_path):
|
||||
dist_cmd.run_export(target, dry_run=False)
|
||||
assert (target / "AGENTS.md").is_file()
|
||||
assert (target / "kb" / "entities" / "COLLECTION.md.template").is_file()
|
||||
assert (target / "raw" / "notes" / ".gitkeep").is_file()
|
||||
assert (target / "incoming" / "notes" / ".gitkeep").is_file()
|
||||
assert (target / "raw" / ".gitkeep").is_file()
|
||||
assert (target / "incoming" / ".gitkeep").is_file()
|
||||
|
||||
|
||||
def test_unbalanced_markers_fail_loudly(repo):
|
||||
|
||||
@@ -134,31 +134,6 @@ def test_legacy_type_regex_matches_pre_migration_form():
|
||||
assert not docs_verify.LEGACY_TYPE_RE.search("---\ntype: types/comparison.md\n---")
|
||||
|
||||
|
||||
def test_this_repos_raw_subdirs_are_documented():
|
||||
assert docs_verify.check_raw_subdirs() == []
|
||||
|
||||
|
||||
def test_raw_subdirs_mismatch_is_reported(tmp_path, monkeypatch):
|
||||
"""The regression this guards: `dist_cmd.RAW_SUBDIRS` and raw/CONTRACT.md's
|
||||
routing table are two places naming the same set (AGENTS.md invariant 8),
|
||||
so either one drifting from the other must be caught in both directions."""
|
||||
(tmp_path / "raw").mkdir()
|
||||
(tmp_path / "raw" / "CONTRACT.md").write_text(
|
||||
"| Directory | Holds |\n|---|---|\n| `articles/` | ... |\n| `videos/` | ... |\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setattr(config, "ROOT", tmp_path)
|
||||
monkeypatch.setattr(docs_verify.dist_cmd, "RAW_SUBDIRS", ("articles", "documents"))
|
||||
issues = docs_verify.check_raw_subdirs()
|
||||
assert any("missing `documents/`" in i for i in issues)
|
||||
assert any("lists `videos/`" in i for i in issues)
|
||||
|
||||
|
||||
def test_raw_subdirs_check_is_skipped_without_a_contract(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(config, "ROOT", tmp_path)
|
||||
assert docs_verify.check_raw_subdirs() == []
|
||||
|
||||
|
||||
def test_no_content_is_gitignored():
|
||||
"""The regression guard for the 2026-08-13 `.gitignore` rewrite: patterns
|
||||
like `*temp*` and `bin/` were silently excluding files under raw/, so the
|
||||
|
||||
@@ -892,3 +892,128 @@ def test_redundant_see_also_reaches_the_rendered_report_and_the_summary(kb_dir):
|
||||
summary = render_summary(report)
|
||||
assert "Redundant see-also" in summary
|
||||
assert "[[nearside]]" in summary and "depends-on" in summary
|
||||
|
||||
|
||||
# --- Confidence against source standing (Gitea #67) --------------------------
|
||||
|
||||
|
||||
def _write_capped_source(kb_dir, title, *, fidelity=None, authority=None, raw_path=None):
|
||||
"""A distinct, never-colliding raw_files: path per title by default - the
|
||||
fixture's own 'Source - Aurora' already claims raw/notes/Aurora.md, and
|
||||
two of these in one test must not claim the same path either."""
|
||||
if raw_path is None:
|
||||
raw_path = f"raw/notes/{title.replace(' ', '-')}.md"
|
||||
frontmatter = {
|
||||
"type": "types/source.md", "source_type": "notes", "author": "Torben",
|
||||
"raw_files": [raw_path], "date": "2026-09-01",
|
||||
"tags": [], "entities": [], "concepts": [], "summary": "Test source.",
|
||||
}
|
||||
if fidelity is not None:
|
||||
frontmatter["fidelity"] = fidelity
|
||||
if authority is not None:
|
||||
frontmatter["authority"] = authority
|
||||
write_page(kb_dir / "sources" / f"{title}.md", frontmatter, f"\n# {title}\n\n## Summary\n\nTest.\n")
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_flags_a_high_confidence_opinion_source(kb_dir):
|
||||
_write_capped_source(kb_dir, "Source - Weak", fidelity="secondhand", authority="opinion")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - Weak"],
|
||||
"confidence": 0.9, "confidence_base": 0.9},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert {"page": "Modbus", "confidence_base": 0.9, "ceiling": 0.6} in report["confidence_exceeds_source_standing"]
|
||||
# Advisory, not a hard error - unlike the fixture's own baseline issues,
|
||||
# this finding on its own must never appear in `hard_error_keys()`.
|
||||
assert "confidence_exceeds_source_standing" not in HARD_ERROR_KEYS
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_is_silent_at_or_under_the_ceiling(kb_dir):
|
||||
_write_capped_source(kb_dir, "Source - Weak", fidelity="secondhand", authority="opinion")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - Weak"],
|
||||
"confidence": 0.6, "confidence_base": 0.6},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert report["confidence_exceeds_source_standing"] == []
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_is_silent_on_normative_verbatim_sources(kb_dir):
|
||||
"""`normative`/`verbatim` carry no ceiling at all - a page may sit as
|
||||
confident as its own hand-set judgment allows."""
|
||||
_write_capped_source(kb_dir, "Source - Strong", fidelity="verbatim", authority="normative")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - Strong"],
|
||||
"confidence": 0.99, "confidence_base": 0.99},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert report["confidence_exceeds_source_standing"] == []
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_is_silent_when_capture_fields_are_unknown(kb_dir):
|
||||
"""A backfilled `unknown` is not a claim about the source - firing on it
|
||||
would report every page citing a pre-#67 source at once."""
|
||||
_write_capped_source(kb_dir, "Source - Backfilled", fidelity="unknown", authority="unknown")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - Backfilled"],
|
||||
"confidence": 0.99, "confidence_base": 0.99},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert report["confidence_exceeds_source_standing"] == []
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_is_silent_without_capture_fields_at_all(kb_dir):
|
||||
"""A source page predating Gitea #67 carries neither field yet - same
|
||||
silence as the explicit `unknown` case, not a finding by omission."""
|
||||
_write_capped_source(kb_dir, "Source - Predates 67")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - Predates 67"],
|
||||
"confidence": 0.99, "confidence_base": 0.99},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert report["confidence_exceeds_source_standing"] == []
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_takes_the_tightest_ceiling_among_several_sources(kb_dir):
|
||||
_write_capped_source(kb_dir, "Source - A", fidelity="published", authority="reporting")
|
||||
_write_capped_source(kb_dir, "Source - B", fidelity="secondhand", authority="opinion")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - A", "Source - B"],
|
||||
"confidence": 0.7, "confidence_base": 0.7},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert {"page": "Modbus", "confidence_base": 0.7, "ceiling": 0.6} in report["confidence_exceeds_source_standing"]
|
||||
|
||||
|
||||
def test_confidence_exceeds_source_standing_reaches_the_rendered_report_and_the_summary(kb_dir):
|
||||
_write_capped_source(kb_dir, "Source - Weak", fidelity="secondhand", authority="opinion")
|
||||
write_page(
|
||||
kb_dir / "concepts/protocols/Modbus.md",
|
||||
{"type": "types/concept.md", "concept_type": "protocol", "tags": [], "created": "2026-07-25",
|
||||
"modified": "2026-07-25", "related": [], "sources": ["Source - Weak"],
|
||||
"confidence": 0.9, "confidence_base": 0.9},
|
||||
"\n# Modbus\n\n## Definition\n\nx.\n",
|
||||
)
|
||||
report = run_lint(kb_dir)
|
||||
assert "Confidence Above Source Standing" in render_markdown(report)
|
||||
summary = render_summary(report)
|
||||
assert "Confidence Above Source Standing" in summary
|
||||
assert "[[Modbus]]" in summary
|
||||
|
||||
@@ -195,6 +195,7 @@ def test_new_source_prefixes_title_and_prefills_related_entities(monkeypatch, kb
|
||||
"new", "source", "--name", "gateway.example.net",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "raw_files=raw/notes/gateway.example.net.md",
|
||||
"--set", "fidelity=verbatim", "--set", "authority=reporting",
|
||||
"--set", "entities=aurora,Borealis",
|
||||
])
|
||||
assert result.exit_code == 0, result.output
|
||||
@@ -221,6 +222,7 @@ def test_new_source_rejects_missing_source_type(monkeypatch, kb_dir):
|
||||
result = _invoke_new(monkeypatch, kb_dir, [
|
||||
"new", "source", "--name", "no-source-type",
|
||||
"--set", "raw_files=raw/notes/gateway.example.net.md",
|
||||
"--set", "fidelity=verbatim", "--set", "authority=reporting",
|
||||
])
|
||||
assert result.exit_code != 0
|
||||
assert "source_type" in result.output
|
||||
@@ -229,6 +231,39 @@ def test_new_source_rejects_missing_source_type(monkeypatch, kb_dir):
|
||||
assert not list(kb_dir.glob("sources/**/Source - no-source-type.md"))
|
||||
|
||||
|
||||
def test_new_source_rejects_missing_capture_fields(monkeypatch, kb_dir):
|
||||
"""Gitea #67: `fidelity`/`authority` have no `default:` and are not in
|
||||
`required:` either (a required field would be boundary-crossing, see
|
||||
version-parts.md) - the refusal is enforced by `new` itself instead."""
|
||||
monkeypatch.setenv("WIKI_AUTHOR", "Torben")
|
||||
_fixture_raw_file(monkeypatch, kb_dir, "raw/notes/gateway.example.net.md")
|
||||
result = _invoke_new(monkeypatch, kb_dir, [
|
||||
"new", "source", "--name", "no-capture-fields",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "raw_files=raw/notes/gateway.example.net.md",
|
||||
])
|
||||
assert result.exit_code != 0
|
||||
assert "fidelity" in result.output
|
||||
assert "authority" in result.output
|
||||
assert not list(kb_dir.glob("sources/**/Source - no-capture-fields.md"))
|
||||
|
||||
|
||||
def test_new_source_rejects_unknown_as_a_capture_value(monkeypatch, kb_dir):
|
||||
"""`unknown` is backfill-only - only `wikitool touch` may write it, on a
|
||||
page predating this rule (Gitea #67)."""
|
||||
monkeypatch.setenv("WIKI_AUTHOR", "Torben")
|
||||
_fixture_raw_file(monkeypatch, kb_dir, "raw/notes/gateway.example.net.md")
|
||||
result = _invoke_new(monkeypatch, kb_dir, [
|
||||
"new", "source", "--name", "unknown-capture",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "raw_files=raw/notes/gateway.example.net.md",
|
||||
"--set", "fidelity=unknown", "--set", "authority=reporting",
|
||||
])
|
||||
assert result.exit_code != 0
|
||||
assert "unknown" in result.output
|
||||
assert not list(kb_dir.glob("sources/**/Source - unknown-capture.md"))
|
||||
|
||||
|
||||
def test_new_source_author_falls_back_to_git_config(monkeypatch, kb_dir):
|
||||
"""No WIKI_AUTHOR set - default_author() falls back to `git config
|
||||
user.name`, run with cwd=config.ROOT.
|
||||
@@ -251,6 +286,7 @@ def test_new_source_author_falls_back_to_git_config(monkeypatch, kb_dir):
|
||||
"new", "source", "--name", "git-config-author",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "raw_files=raw/notes/gateway.example.net.md",
|
||||
"--set", "fidelity=verbatim", "--set", "authority=reporting",
|
||||
])
|
||||
assert result.exit_code == 0, result.output
|
||||
fm, _body = read_page(kb_dir / "sources/notes/Source - git-config-author.md")
|
||||
@@ -363,6 +399,7 @@ def test_raw_files_error_points_at_the_comma_split(monkeypatch, kb_dir, raw_dir)
|
||||
"new", "source", "--name", "Split Path",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "raw_files=raw/notes/Versioning, CI-CD.md",
|
||||
"--set", "fidelity=verbatim", "--set", "authority=reporting",
|
||||
])
|
||||
assert result.exit_code == 1
|
||||
assert "splitting the value on commas" in result.output
|
||||
@@ -382,6 +419,7 @@ def test_source_page_accepts_a_raw_file_whose_name_has_a_comma(monkeypatch, kb_d
|
||||
"new", "source", "--name", "Comma Source",
|
||||
"--set", r"raw_files=raw/notes/Versioning\, CI-CD.md",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "fidelity=verbatim", "--set", "authority=reporting",
|
||||
])
|
||||
assert result.exit_code == 0, result.output
|
||||
frontmatter, _ = read_page(kb_dir / "sources/notes/Source - Comma Source.md")
|
||||
|
||||
@@ -402,6 +402,7 @@ def test_new_source_with_multiple_raw_files(kb_dir, raw_dir, monkeypatch):
|
||||
"new", "source", "--name", "Multi",
|
||||
"--set", "source_type=notes",
|
||||
"--set", "raw_files=raw/notes/Aurora.md,raw/notes/Second.md",
|
||||
"--set", "fidelity=verbatim", "--set", "authority=reporting",
|
||||
])
|
||||
assert result.exit_code == 0, result.output
|
||||
fm, _body = read_page(kb_dir / "sources/notes/Source - Multi.md")
|
||||
|
||||
+258
-110
@@ -1,3 +1,5 @@
|
||||
import datetime
|
||||
|
||||
import pytest
|
||||
import typer
|
||||
|
||||
@@ -8,75 +10,96 @@ from chemenu.provenance import uncovered_raw_files
|
||||
from chemenu.kb_scan import load_kb_pages
|
||||
|
||||
|
||||
def _shard() -> str:
|
||||
"""Today's `raw/<YYYY>/<MM>` shard, as a relative-path string - matches
|
||||
`raw_cmd._shard_dir()` without importing its internals, the same way the
|
||||
rest of the suite asserts against real dates (see test_touch.py)."""
|
||||
today = datetime.date.today()
|
||||
return f"raw/{today.year:04d}/{today.month:02d}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tree(kb_dir):
|
||||
"""kb_dir already repoints config.ROOT at tmp_path; add raw/ and
|
||||
incoming/ beside it, each with the four type subdirectories
|
||||
dist_cmd.RAW_SUBDIRS declares."""
|
||||
incoming/ beside it. raw/ keeps two pre-#67 legacy type directories (not
|
||||
all four - #67 stops treating them as a declared, exhaustive set), so
|
||||
tests can exercise both the legacy flat layout and the date shard.
|
||||
incoming/ is flat now (Gitea #67): no subdirectory is required for a
|
||||
promotion to succeed."""
|
||||
root = kb_dir.parent
|
||||
for sub in ("articles", "documents", "notes", "assets"):
|
||||
(root / "raw" / sub).mkdir(parents=True)
|
||||
(root / "incoming" / sub).mkdir(parents=True)
|
||||
(root / "raw" / "documents").mkdir(parents=True)
|
||||
(root / "raw" / "notes").mkdir(parents=True)
|
||||
(root / "incoming").mkdir(parents=True, exist_ok=True)
|
||||
return root
|
||||
|
||||
|
||||
def _accept(*files, page=None, replaces=None, dry_run=False):
|
||||
return raw_accept_command(files=list(files), page=page, replaces=replaces, dry_run=dry_run)
|
||||
|
||||
|
||||
def _write_source(kb_dir, title, raw_files):
|
||||
write_page(
|
||||
kb_dir / "sources" / f"{title}.md",
|
||||
{
|
||||
"type": "types/source.md", "source_type": "document", "author": "Torben",
|
||||
"raw_files": list(raw_files), "date": "2026-09-01",
|
||||
"tags": [], "entities": [], "concepts": [], "summary": "Test source.",
|
||||
},
|
||||
f"\n# {title}\n\n## Summary\n\nTest.\n",
|
||||
def _accept(*files, page=None, replaces=None, dry_run=False, fidelity="verbatim", authority="reporting"):
|
||||
return raw_accept_command(
|
||||
files=list(files), fidelity=fidelity, authority=authority,
|
||||
page=page, replaces=replaces, dry_run=dry_run,
|
||||
)
|
||||
|
||||
|
||||
def _write_source(kb_dir, title, raw_files, fidelity=None, authority=None):
|
||||
frontmatter = {
|
||||
"type": "types/source.md", "source_type": "document", "author": "Torben",
|
||||
"raw_files": list(raw_files), "date": "2026-09-01",
|
||||
"tags": [], "entities": [], "concepts": [], "summary": "Test source.",
|
||||
}
|
||||
if fidelity is not None:
|
||||
frontmatter["fidelity"] = fidelity
|
||||
if authority is not None:
|
||||
frontmatter["authority"] = authority
|
||||
write_page(kb_dir / "sources" / f"{title}.md", frontmatter, f"\n# {title}\n\n## Summary\n\nTest.\n")
|
||||
|
||||
|
||||
def test_single_file_needs_no_bundle(tree):
|
||||
src = tree / "incoming/documents/handbuch.pdf"
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"%PDF-1.4 fake\n")
|
||||
_accept(src)
|
||||
dst = tree / "raw/documents/handbuch.pdf"
|
||||
dst = tree / _shard() / "handbuch.pdf"
|
||||
assert dst.is_file()
|
||||
assert dst.read_bytes() == b"%PDF-1.4 fake\n"
|
||||
assert not src.exists()
|
||||
|
||||
|
||||
def test_two_files_bundle_under_the_first_files_stem(tree):
|
||||
pdf = tree / "incoming/documents/handbuch.pdf"
|
||||
md = tree / "incoming/documents/handbuch.md"
|
||||
pdf = tree / "incoming/handbuch.pdf"
|
||||
md = tree / "incoming/handbuch.md"
|
||||
pdf.write_bytes(b"pdf-bytes")
|
||||
md.write_text("# converted\n", encoding="utf-8")
|
||||
_accept(pdf, md)
|
||||
assert (tree / "raw/documents/handbuch/handbuch.pdf").read_bytes() == b"pdf-bytes"
|
||||
assert (tree / "raw/documents/handbuch/handbuch.md").read_text(encoding="utf-8") == "# converted\n"
|
||||
bundle = tree / _shard() / "handbuch"
|
||||
assert (bundle / "handbuch.pdf").read_bytes() == b"pdf-bytes"
|
||||
assert (bundle / "handbuch.md").read_text(encoding="utf-8") == "# converted\n"
|
||||
assert not pdf.exists() and not md.exists()
|
||||
|
||||
|
||||
def test_file_directly_in_incoming_is_rejected(tree):
|
||||
def test_file_directly_in_incoming_is_accepted(tree):
|
||||
"""Gitea #67: incoming/ is flat, so a file directly in it is the normal
|
||||
case now, not a rejection - unlike the pre-#67 behaviour this replaces."""
|
||||
src = tree / "incoming/stray.md"
|
||||
src.write_text("x\n", encoding="utf-8")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(src)
|
||||
assert src.exists()
|
||||
_accept(src)
|
||||
assert (tree / _shard() / "stray.md").is_file()
|
||||
assert not src.exists()
|
||||
|
||||
|
||||
def test_unknown_type_subdir_is_rejected(tree):
|
||||
def test_subdirectory_under_incoming_is_ignored_not_inspected(tree):
|
||||
"""Gitea #67: a subdirectory of incoming/ - old habit, old script - is
|
||||
tolerated and ignored rather than read as a type classification. This is
|
||||
the MINOR condition named in the issue's Versionsteil."""
|
||||
(tree / "incoming/videos").mkdir()
|
||||
src = tree / "incoming/videos/clip.mp4"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(src)
|
||||
assert src.exists()
|
||||
_accept(src)
|
||||
assert (tree / _shard() / "clip.mp4").is_file()
|
||||
assert not src.exists()
|
||||
|
||||
|
||||
def test_nested_too_deep_is_rejected(tree):
|
||||
nested = tree / "incoming/documents/sub"
|
||||
nested.mkdir()
|
||||
nested.mkdir(parents=True)
|
||||
src = nested / "deep.pdf"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
@@ -84,75 +107,116 @@ def test_nested_too_deep_is_rejected(tree):
|
||||
assert src.exists()
|
||||
|
||||
|
||||
def test_mixed_type_subdirs_in_one_call_is_rejected(tree):
|
||||
def test_files_from_different_ignored_subdirs_bundle_together(tree):
|
||||
"""No more per-call type agreement to enforce (Gitea #67): which ignored
|
||||
subdirectory each file happened to sit under is irrelevant now."""
|
||||
a = tree / "incoming/documents/a.pdf"
|
||||
b = tree / "incoming/notes/b.md"
|
||||
b = tree / "incoming/notes/a.md"
|
||||
a.parent.mkdir(parents=True)
|
||||
b.parent.mkdir(parents=True)
|
||||
a.write_bytes(b"a")
|
||||
b.write_text("b", encoding="utf-8")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(a, b)
|
||||
assert a.exists() and b.exists()
|
||||
_accept(a, b)
|
||||
bundle = tree / _shard() / "a"
|
||||
assert (bundle / "a.pdf").exists() and (bundle / "a.md").exists()
|
||||
|
||||
|
||||
def test_same_file_passed_twice_is_rejected(tree):
|
||||
src = tree / "incoming/documents/handbuch.pdf"
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(src, src)
|
||||
assert src.exists()
|
||||
|
||||
|
||||
def test_collision_with_existing_raw_file_is_rejected(tree):
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"already there")
|
||||
src = tree / "incoming/documents/handbuch.pdf"
|
||||
def test_collision_with_existing_file_at_same_shard_path_is_rejected(tree):
|
||||
dst_dir = tree / _shard()
|
||||
dst_dir.mkdir(parents=True)
|
||||
(dst_dir / "handbuch.pdf").write_bytes(b"already there")
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"new")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(src)
|
||||
assert src.exists()
|
||||
assert (tree / "raw/documents/handbuch.pdf").read_bytes() == b"already there"
|
||||
assert (dst_dir / "handbuch.pdf").read_bytes() == b"already there"
|
||||
|
||||
|
||||
def test_missing_file_is_rejected(tree):
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(tree / "incoming/documents/absent.pdf")
|
||||
_accept(tree / "incoming/absent.pdf")
|
||||
|
||||
|
||||
def test_missing_fidelity_is_rejected(tree):
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
raw_accept_command(files=[src], fidelity=None, authority="reporting", page=None, replaces=None, dry_run=False)
|
||||
assert src.exists()
|
||||
|
||||
|
||||
def test_missing_authority_is_rejected(tree):
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
raw_accept_command(files=[src], fidelity="verbatim", authority=None, page=None, replaces=None, dry_run=False)
|
||||
assert src.exists()
|
||||
|
||||
|
||||
def test_unknown_fidelity_is_rejected(tree):
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(src, fidelity="unknown")
|
||||
assert src.exists()
|
||||
|
||||
|
||||
def test_invalid_fidelity_value_is_rejected(tree):
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"x")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(src, fidelity="made-up")
|
||||
assert src.exists()
|
||||
|
||||
|
||||
def test_dry_run_moves_nothing(tree):
|
||||
src = tree / "incoming/documents/handbuch.pdf"
|
||||
src = tree / "incoming/handbuch.pdf"
|
||||
src.write_bytes(b"x")
|
||||
_accept(src, dry_run=True)
|
||||
assert src.exists()
|
||||
assert not (tree / "raw/documents/handbuch.pdf").exists()
|
||||
assert not (tree / _shard() / "handbuch.pdf").exists()
|
||||
|
||||
|
||||
def test_promoted_file_in_incoming_is_never_reported_uncovered(tree):
|
||||
"""AC: incoming/ is invisible to sources coverage / lint until accepted -
|
||||
both walk raw/ only."""
|
||||
(tree / "incoming/notes/not-yet-promoted.md").write_text("draft\n", encoding="utf-8")
|
||||
(tree / "incoming/not-yet-promoted.md").write_text("draft\n", encoding="utf-8")
|
||||
pages = load_kb_pages(tree / "kb")
|
||||
assert "raw/notes/not-yet-promoted.md" not in uncovered_raw_files(tree / "raw", pages)
|
||||
assert uncovered_raw_files(tree / "raw", pages) == []
|
||||
|
||||
|
||||
def test_without_page_flag_only_moves_and_prints_target(tree, capsys):
|
||||
src = tree / "incoming/notes/meeting.md"
|
||||
def test_without_page_flag_prints_target_and_new_source_followup(tree, capsys):
|
||||
src = tree / "incoming/meeting.md"
|
||||
src.write_text("notes\n", encoding="utf-8")
|
||||
_accept(src)
|
||||
_accept(src, fidelity="secondhand", authority="opinion")
|
||||
out = capsys.readouterr().out
|
||||
assert "raw/notes/meeting.md" in out
|
||||
assert f"{_shard()}/meeting.md" in out
|
||||
assert "new source" in out
|
||||
assert "fidelity=secondhand" in out
|
||||
assert "authority=opinion" in out
|
||||
|
||||
|
||||
def test_page_flag_extends_raw_files_for_a_single_new_file(tree):
|
||||
def test_page_flag_extends_raw_files_and_sets_capture_fields_for_a_single_new_file(tree):
|
||||
"""No growth case: the page already has >=1 file, so this always bundles -
|
||||
see test_growth_case below for the interesting path."""
|
||||
(tree / "raw/documents").mkdir(exist_ok=True)
|
||||
see test_growth_case below for the interesting path. The pre-existing page
|
||||
predates #67 and carries no capture fields yet, so this is also the first
|
||||
write of them."""
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.pdf"])
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"first")
|
||||
new_file = tree / "incoming/documents/handbuch-appendix.pdf"
|
||||
new_file = tree / "incoming/handbuch-appendix.pdf"
|
||||
new_file.write_bytes(b"second")
|
||||
|
||||
_accept(new_file, page="Source - Handbuch")
|
||||
_accept(new_file, page="Source - Handbuch", fidelity="published", authority="normative")
|
||||
|
||||
bundle = tree / "raw/documents/handbuch"
|
||||
assert (bundle / "handbuch.pdf").read_bytes() == b"first"
|
||||
@@ -163,12 +227,49 @@ def test_page_flag_extends_raw_files_for_a_single_new_file(tree):
|
||||
assert sorted(frontmatter["raw_files"]) == sorted(
|
||||
["raw/documents/handbuch/handbuch.pdf", "raw/documents/handbuch/handbuch-appendix.pdf"]
|
||||
)
|
||||
assert frontmatter["fidelity"] == "published"
|
||||
assert frontmatter["authority"] == "normative"
|
||||
|
||||
|
||||
def test_page_flag_rejects_when_capture_field_already_set_differently(tree):
|
||||
_write_source(
|
||||
tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.pdf"],
|
||||
fidelity="verbatim", authority="reporting",
|
||||
)
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"first")
|
||||
new_file = tree / "incoming/handbuch-appendix.pdf"
|
||||
new_file.write_bytes(b"second")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new_file, page="Source - Handbuch", fidelity="secondhand", authority="reporting")
|
||||
|
||||
assert new_file.exists()
|
||||
assert (tree / "raw/documents/handbuch.pdf").exists()
|
||||
assert not (tree / "raw/documents/handbuch").exists()
|
||||
frontmatter, _ = read_page(tree / "kb/sources/Source - Handbuch.md")
|
||||
assert frontmatter["fidelity"] == "verbatim"
|
||||
|
||||
|
||||
def test_page_flag_is_idempotent_when_capture_field_already_set_the_same(tree):
|
||||
_write_source(
|
||||
tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.pdf"],
|
||||
fidelity="verbatim", authority="reporting",
|
||||
)
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"first")
|
||||
new_file = tree / "incoming/handbuch-appendix.pdf"
|
||||
new_file.write_bytes(b"second")
|
||||
|
||||
_accept(new_file, page="Source - Handbuch", fidelity="verbatim", authority="reporting")
|
||||
|
||||
frontmatter, _ = read_page(tree / "kb/sources/Source - Handbuch.md")
|
||||
assert frontmatter["fidelity"] == "verbatim"
|
||||
assert frontmatter["authority"] == "reporting"
|
||||
|
||||
|
||||
def test_growth_case_no_broken_or_uncovered_refs_afterwards(tree):
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.pdf"])
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"first")
|
||||
new_file = tree / "incoming/documents/handbuch.md"
|
||||
new_file = tree / "incoming/handbuch.md"
|
||||
new_file.write_text("converted", encoding="utf-8")
|
||||
|
||||
_accept(new_file, page="Source - Handbuch")
|
||||
@@ -189,11 +290,12 @@ def test_adding_to_an_already_bundled_source_joins_the_existing_bundle(tree):
|
||||
_write_source(
|
||||
tree / "kb", "Source - Handbuch",
|
||||
["raw/documents/handbuch/handbuch.pdf", "raw/documents/handbuch/handbuch.md"],
|
||||
fidelity="verbatim", authority="reporting",
|
||||
)
|
||||
extra = tree / "incoming/documents/handbuch-notes.md"
|
||||
extra = tree / "incoming/handbuch-notes.md"
|
||||
extra.write_text("c", encoding="utf-8")
|
||||
|
||||
_accept(extra, page="Source - Handbuch")
|
||||
_accept(extra, page="Source - Handbuch", fidelity="verbatim", authority="reporting")
|
||||
|
||||
assert (tree / "raw/documents/handbuch/handbuch-notes.md").read_text(encoding="utf-8") == "c"
|
||||
# Nothing already-bundled was moved a second time.
|
||||
@@ -210,7 +312,7 @@ def test_growth_case_rejects_a_multi_owner_raw_file(tree):
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"shared")
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.pdf"])
|
||||
_write_source(tree / "kb", "Source - Also Handbuch", ["raw/documents/handbuch.pdf"])
|
||||
new_file = tree / "incoming/documents/handbuch.md"
|
||||
new_file = tree / "incoming/handbuch.md"
|
||||
new_file.write_text("x", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
@@ -222,7 +324,7 @@ def test_growth_case_rejects_a_multi_owner_raw_file(tree):
|
||||
|
||||
|
||||
def test_page_not_found_is_rejected(tree):
|
||||
new_file = tree / "incoming/documents/handbuch.md"
|
||||
new_file = tree / "incoming/handbuch.md"
|
||||
new_file.write_text("x", encoding="utf-8")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new_file, page="Source - Nonexistent")
|
||||
@@ -239,37 +341,29 @@ def test_page_with_no_raw_files_is_rejected(tree):
|
||||
},
|
||||
"\n# Source - Empty\n",
|
||||
)
|
||||
new_file = tree / "incoming/documents/handbuch.md"
|
||||
new_file = tree / "incoming/handbuch.md"
|
||||
new_file.write_text("x", encoding="utf-8")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new_file, page="Source - Empty")
|
||||
assert new_file.exists()
|
||||
|
||||
|
||||
def test_type_directory_mismatch_with_existing_raw_files_is_rejected(tree):
|
||||
(tree / "raw/notes/handbuch.md").write_bytes(b"first")
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/notes/handbuch.md"])
|
||||
new_file = tree / "incoming/documents/handbuch.pdf"
|
||||
new_file.write_bytes(b"second")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new_file, page="Source - Handbuch")
|
||||
assert new_file.exists()
|
||||
# --- Stem uniqueness across raw/ (Gitea #64, widened globally by #67) ---
|
||||
|
||||
|
||||
# --- Stem uniqueness at raw/<type>/ level (Gitea #64) ---
|
||||
|
||||
|
||||
def test_second_source_does_not_silently_join_an_existing_bundle(tree):
|
||||
"""The exact repro from #64: a bundle exists for one source, and a second,
|
||||
unrelated source's files (no --page) happen not to collide on filename -
|
||||
they must not slip into the existing bundle."""
|
||||
def test_second_source_does_not_silently_join_an_existing_legacy_bundle(tree):
|
||||
"""The exact repro from #64, now crossing the legacy/shard boundary: a
|
||||
bundle exists in the pre-#67 flat layout, and a second, unrelated source's
|
||||
files (no --page) land in today's date shard - they must still not slip
|
||||
into the existing bundle just because the two live in different
|
||||
directories now."""
|
||||
bundle = tree / "raw/documents/handbuch"
|
||||
bundle.mkdir()
|
||||
(bundle / "handbuch.pdf").write_bytes(b"pdf")
|
||||
(bundle / "handbuch.md").write_text("md", encoding="utf-8")
|
||||
|
||||
txt = tree / "incoming/documents/handbuch.txt"
|
||||
anhang = tree / "incoming/documents/anhang.md"
|
||||
txt = tree / "incoming/handbuch.txt"
|
||||
anhang = tree / "incoming/anhang.md"
|
||||
txt.write_text("txt", encoding="utf-8")
|
||||
anhang.write_text("anhang", encoding="utf-8")
|
||||
|
||||
@@ -280,11 +374,11 @@ def test_second_source_does_not_silently_join_an_existing_bundle(tree):
|
||||
assert sorted(p.name for p in bundle.iterdir()) == ["handbuch.md", "handbuch.pdf"]
|
||||
|
||||
|
||||
def test_flat_promote_rejected_when_stem_matches_an_existing_bundle(tree):
|
||||
def test_flat_promote_rejected_when_stem_matches_an_existing_legacy_bundle(tree):
|
||||
(tree / "raw/documents/handbuch").mkdir()
|
||||
(tree / "raw/documents/handbuch/handbuch.pdf").write_bytes(b"pdf")
|
||||
|
||||
txt = tree / "incoming/documents/handbuch.txt"
|
||||
txt = tree / "incoming/handbuch.txt"
|
||||
txt.write_text("txt", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
@@ -294,12 +388,12 @@ def test_flat_promote_rejected_when_stem_matches_an_existing_bundle(tree):
|
||||
assert (tree / "raw/documents/handbuch/handbuch.pdf").exists()
|
||||
|
||||
|
||||
def test_flat_promote_rejected_when_stem_matches_an_existing_flat_file(tree):
|
||||
def test_flat_promote_rejected_when_stem_matches_an_existing_legacy_flat_file(tree):
|
||||
"""Same stem, different extension: not caught by the old per-path
|
||||
dst.exists() check, which is the whole gap #64 closes."""
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"pdf")
|
||||
|
||||
md = tree / "incoming/documents/handbuch.md"
|
||||
md = tree / "incoming/handbuch.md"
|
||||
md.write_text("md", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
@@ -307,12 +401,26 @@ def test_flat_promote_rejected_when_stem_matches_an_existing_flat_file(tree):
|
||||
|
||||
assert md.exists()
|
||||
assert (tree / "raw/documents/handbuch.pdf").read_bytes() == b"pdf"
|
||||
assert not (tree / "raw/documents/handbuch.md").exists()
|
||||
|
||||
|
||||
def test_second_file_with_same_stem_in_same_shard_is_rejected(tree):
|
||||
"""Uniqueness also holds within one date shard, not only against the
|
||||
legacy layout."""
|
||||
first = tree / "incoming/foo.md"
|
||||
first.write_text("first", encoding="utf-8")
|
||||
_accept(first)
|
||||
assert (tree / _shard() / "foo.md").is_file()
|
||||
|
||||
second = tree / "incoming/foo.txt"
|
||||
second.write_text("second", encoding="utf-8")
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(second)
|
||||
assert second.exists()
|
||||
|
||||
|
||||
def test_stem_collision_message_names_both_routes_without_recommending_one(tree, capsys):
|
||||
(tree / "raw/documents/handbuch.pdf").write_bytes(b"pdf")
|
||||
md = tree / "incoming/documents/handbuch.md"
|
||||
md = tree / "incoming/handbuch.md"
|
||||
md.write_text("md", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
@@ -324,22 +432,58 @@ def test_stem_collision_message_names_both_routes_without_recommending_one(tree,
|
||||
assert "does not guess" in err
|
||||
|
||||
|
||||
# --- --replaces (Gitea #64 decision 2) ---
|
||||
# --- --replaces (Gitea #64 decision 2, --fidelity/--authority the correction
|
||||
# path for a capture field per Gitea #67) ---
|
||||
|
||||
|
||||
def test_replaces_swaps_bytes_and_leaves_raw_files_untouched(tree):
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old edition", encoding="utf-8")
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new edition", encoding="utf-8")
|
||||
|
||||
_accept(new, replaces=target)
|
||||
_accept(new, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "new edition"
|
||||
assert not new.exists()
|
||||
frontmatter, _ = read_page(tree / "kb/sources/Source - Handbuch.md")
|
||||
assert frontmatter["raw_files"] == ["raw/documents/handbuch.md"]
|
||||
assert "fidelity" not in frontmatter
|
||||
|
||||
|
||||
def test_replaces_with_fidelity_overwrites_the_owning_pages_capture_field(tree):
|
||||
"""The one sanctioned correction path (Gitea #67): --replaces may pass
|
||||
--fidelity/--authority to fix an already-set capture value, unlike
|
||||
`touch`, which fill-once refuses to."""
|
||||
_write_source(
|
||||
tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"],
|
||||
fidelity="secondhand", authority="opinion",
|
||||
)
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old edition", encoding="utf-8")
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new edition", encoding="utf-8")
|
||||
|
||||
_accept(new, replaces=target, fidelity="verbatim", authority=None)
|
||||
|
||||
frontmatter, _ = read_page(tree / "kb/sources/Source - Handbuch.md")
|
||||
assert frontmatter["fidelity"] == "verbatim"
|
||||
assert frontmatter["authority"] == "opinion"
|
||||
|
||||
|
||||
def test_replaces_rejects_unknown_fidelity(tree):
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new, replaces=target, fidelity="unknown", authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert new.exists()
|
||||
|
||||
|
||||
def test_replaces_reports_source_and_both_citing_pages(tree, capsys):
|
||||
@@ -365,10 +509,10 @@ def test_replaces_reports_source_and_both_citing_pages(tree, capsys):
|
||||
)
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
_accept(new, replaces=target)
|
||||
_accept(new, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "Source - Handbuch" in out
|
||||
@@ -379,10 +523,10 @@ def test_replaces_reports_source_and_both_citing_pages(tree, capsys):
|
||||
def test_replaces_succeeds_with_no_owner_and_reports_it(tree, capsys):
|
||||
target = tree / "raw/documents/orphan.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/documents/orphan.md"
|
||||
new = tree / "incoming/orphan.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
_accept(new, replaces=target)
|
||||
_accept(new, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "new"
|
||||
out = capsys.readouterr().out
|
||||
@@ -392,10 +536,10 @@ def test_replaces_succeeds_with_no_owner_and_reports_it(tree, capsys):
|
||||
def test_replaces_dry_run_writes_nothing(tree):
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
_accept(new, replaces=target, dry_run=True)
|
||||
_accept(new, replaces=target, dry_run=True, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert new.exists()
|
||||
@@ -404,13 +548,13 @@ def test_replaces_dry_run_writes_nothing(tree):
|
||||
def test_replaces_rejects_more_than_one_incoming_file(tree):
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
a = tree / "incoming/documents/handbuch.md"
|
||||
b = tree / "incoming/documents/extra.md"
|
||||
a = tree / "incoming/handbuch.md"
|
||||
b = tree / "incoming/extra.md"
|
||||
a.write_text("a", encoding="utf-8")
|
||||
b.write_text("b", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(a, b, replaces=target)
|
||||
_accept(a, b, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert a.exists() and b.exists()
|
||||
@@ -419,35 +563,39 @@ def test_replaces_rejects_more_than_one_incoming_file(tree):
|
||||
def test_replaces_rejects_filename_mismatch(tree):
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/documents/handbuch-v2.md"
|
||||
new = tree / "incoming/handbuch-v2.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new, replaces=target)
|
||||
_accept(new, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert new.exists()
|
||||
|
||||
|
||||
def test_replaces_rejects_type_directory_mismatch(tree):
|
||||
def test_replaces_across_legacy_directories_is_now_allowed(tree):
|
||||
"""Gitea #67 removes the type-directory-match check `--replaces` used to
|
||||
enforce: a subdirectory of incoming/ carries no meaning any more, so
|
||||
replacing a raw/notes/ file with an incoming file dropped under an
|
||||
unrelated incoming/documents/ works exactly like one dropped flat."""
|
||||
target = tree / "raw/notes/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new.parent.mkdir(parents=True)
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new, replaces=target)
|
||||
_accept(new, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert new.exists()
|
||||
assert target.read_text(encoding="utf-8") == "new"
|
||||
assert not new.exists()
|
||||
|
||||
|
||||
def test_replaces_rejects_nonexistent_target(tree):
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new, replaces=tree / "raw/documents/handbuch.md")
|
||||
_accept(new, replaces=tree / "raw/documents/handbuch.md", fidelity=None, authority=None)
|
||||
|
||||
assert new.exists()
|
||||
|
||||
@@ -457,11 +605,11 @@ def test_replaces_rejects_multi_owner_target(tree):
|
||||
target.write_text("old", encoding="utf-8")
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
|
||||
_write_source(tree / "kb", "Source - Also Handbuch", ["raw/documents/handbuch.md"])
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new, replaces=target)
|
||||
_accept(new, replaces=target, fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert new.exists()
|
||||
@@ -471,11 +619,11 @@ def test_replaces_rejects_combination_with_page(tree):
|
||||
_write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
|
||||
target = tree / "raw/documents/handbuch.md"
|
||||
target.write_text("old", encoding="utf-8")
|
||||
new = tree / "incoming/documents/handbuch.md"
|
||||
new = tree / "incoming/handbuch.md"
|
||||
new.write_text("new", encoding="utf-8")
|
||||
|
||||
with pytest.raises(typer.Exit):
|
||||
_accept(new, replaces=target, page="Source - Handbuch")
|
||||
_accept(new, replaces=target, page="Source - Handbuch", fidelity=None, authority=None)
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "old"
|
||||
assert new.exists()
|
||||
|
||||
@@ -230,3 +230,36 @@ def test_dry_run_covers_set_too(touch_wiki):
|
||||
before = path.read_text(encoding="utf-8")
|
||||
_touch(page_title="aurora", set_fields=["tags=nope"], dry_run=True)
|
||||
assert path.read_text(encoding="utf-8") == before
|
||||
|
||||
|
||||
# --- Capture fields: fill-once, not a UNSETTABLE denylist (Gitea #67) --------
|
||||
|
||||
|
||||
def test_capture_field_is_writable_while_absent(touch_wiki):
|
||||
"""The fixture 'Source - Aurora' predates #67 and carries no `fidelity:`
|
||||
yet - the backfill path `touch --set fidelity=unknown` (or a real value)
|
||||
must succeed exactly once."""
|
||||
_touch(page_title="Source - Aurora", set_fields=["fidelity=unknown"])
|
||||
frontmatter, _ = read_page(touch_wiki / "sources/notes/Source - Aurora.md")
|
||||
assert frontmatter["fidelity"] == "unknown"
|
||||
|
||||
|
||||
def test_capture_field_is_refused_once_already_set(touch_wiki, capsys):
|
||||
_touch(page_title="Source - Aurora", set_fields=["authority=reporting"])
|
||||
with pytest.raises(typer.Exit):
|
||||
_touch(page_title="Source - Aurora", set_fields=["authority=opinion"])
|
||||
out = capsys.readouterr().out
|
||||
assert "capture field" in out
|
||||
assert "raw accept --replaces" in out
|
||||
frontmatter, _ = read_page(touch_wiki / "sources/notes/Source - Aurora.md")
|
||||
assert frontmatter["authority"] == "reporting"
|
||||
|
||||
|
||||
def test_capture_field_set_to_the_same_value_again_is_a_no_op(touch_wiki):
|
||||
"""Not idempotent by accident: `_apply_set` already no-ops an unchanged
|
||||
value before the write, and the fill-once refusal only fires on an actual
|
||||
*change* - re-running the exact same backfill value must not error."""
|
||||
_touch(page_title="Source - Aurora", set_fields=["fidelity=verbatim"])
|
||||
_touch(page_title="Source - Aurora", set_fields=["fidelity=verbatim"])
|
||||
frontmatter, _ = read_page(touch_wiki / "sources/notes/Source - Aurora.md")
|
||||
assert frontmatter["fidelity"] == "verbatim"
|
||||
|
||||
@@ -64,6 +64,25 @@ def test_get_page_ref_fields_defaults_to_empty():
|
||||
assert resolver.get_page_ref_fields("types/type-spec.md") == []
|
||||
|
||||
|
||||
def test_get_capture_fields_reads_the_type_spec():
|
||||
"""`fidelity`/`authority` are fixed once, at capture time (Gitea #67) -
|
||||
`raw accept`, `new source` and `touch` all read the field list from here
|
||||
instead of hardcoding it three times over."""
|
||||
assert resolver.get_capture_fields("types/source.md") == ["fidelity", "authority"]
|
||||
|
||||
|
||||
def test_capture_fields_exist_in_the_type_schema():
|
||||
properties = resolver.get_schema("types/source.md")["properties"]
|
||||
for field in resolver.get_capture_fields("types/source.md"):
|
||||
assert field in properties, f"types/source.md declares unknown capture field {field}"
|
||||
assert "unknown" in properties[field]["enum"]
|
||||
assert "default" not in properties[field]
|
||||
|
||||
|
||||
def test_get_capture_fields_defaults_to_empty():
|
||||
assert resolver.get_capture_fields("types/entity.md") == []
|
||||
|
||||
|
||||
def test_get_layout_reads_entity_type_specs_own_layout_field():
|
||||
"""new_page.py's directory placement and index_build.py's section
|
||||
titles/order derive from here instead of hand-maintained
|
||||
|
||||
@@ -459,6 +459,28 @@ class TypeResolver:
|
||||
type_spec = self.load_type_spec(type_path, source_file)
|
||||
return list(type_spec['frontmatter'].get('page_ref_fields') or [])
|
||||
|
||||
def get_capture_fields(self, type_path: str, source_file: Path = None) -> list:
|
||||
"""Return the frontmatter fields that are fixed at capture time and
|
||||
never correctable afterwards except by re-capturing the source (e.g.
|
||||
`['fidelity', 'authority']` for a source page), as declared by the
|
||||
type-spec's own `capture_fields:` frontmatter (Gitea #67).
|
||||
|
||||
Three commands read this instead of hardcoding the field names
|
||||
separately (AGENTS.md invariant 8 - one rule, one place):
|
||||
`raw accept` and `new source` refuse to proceed without a value for
|
||||
each; `touch --set` writes one only while it is still absent
|
||||
(fill-once) and refuses once it holds a value, pointing at
|
||||
`raw accept --replaces` for a genuine correction.
|
||||
|
||||
Returns an empty list for a type that declares none.
|
||||
|
||||
Raises:
|
||||
ValueError: If the type path cannot be resolved (propagated from
|
||||
`load_type_spec`)
|
||||
"""
|
||||
type_spec = self.load_type_spec(type_path, source_file)
|
||||
return list(type_spec['frontmatter'].get('capture_fields') or [])
|
||||
|
||||
def list_type_specs(self) -> list:
|
||||
"""Return every type-spec document under types/ as a list of
|
||||
`(type_path, frontmatter)` tuples, sorted by path.
|
||||
|
||||
Reference in New Issue
Block a user