diff --git a/AGENTS.md b/AGENTS.md index c415933e..518c5586 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -107,6 +107,11 @@ src/morphobricks_preprocessing/ description.py the model (Description, Condition, Channel, Repeat, Member, Unit, Mask) built from a reader-accepted document; description_from_bytes -> DescriptionError description_write.py the canonical writer (Spec experiment-v1 9.1) and section digests (9.3) + description_transition.py the R_* comparison of a candidate with the published file (4.4a, 4.6, 4.9), + Reconfirmation (the user's recorded acceptance of a changed ID) + description_file.py read_description(path): symlink refusal (R_FILE_SYMLINK), bytes -> Description + run/description_file.py publish_description: lock, base SHA-256, strict re-read, transitions, + atomic write, append-only morphobricks.records.jsonl (Spec experiment-v1 4.6, 10) image_type.py image type (P4): FLUORESCENCE/RGB, review_image_type (suggestion from TiffMetadata.rgb_samples and the confirmed roles, whether RGB is possible via the brick's check_rgb), image_type_issues (rgb_not_possible) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cef1330..abb7a144 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,19 @@ are in [docs/plan.md](docs/plan.md) ยง12. Upgrading from v0.3: see the ## Unreleased +- **Experiment-description publication (Stage 1, PR 5).** `description_file.read_description(path)` + reads and validates a `morphobricks.toml` (a symbolic link is `R_FILE_SYMLINK`, never followed); + `description_transition` compares a candidate with the published file and reports every `R_*` + that applies (`R_CONFIRMED_ID_CHANGED`, `R_USED_ID_DROPPED`, `R_USED_ID_REUSED`, + `R_CONFIRMED_ON_CHANGED`), accepting a changed ID only with the user's recorded + `Reconfirmation`; `run.description_file.publish_description` does it under the output folder's + lock: base SHA-256 or `absent` (`R_DRAFT_BASE_MISSING`, `R_PUBLISHED_FILE_CHANGED`), canonical + bytes re-read by the strict reader, atomic write, and the re-confirmations appended to the + append-only `morphobricks.records.jsonl` (old ID, new ID, date, SHA-256 before and after). + All twelve Spec transitions and scenarios P09, P16-P20 and P22 pass. A new `confirmed_on` beside + a changed image count alone is refused (new FOVs do not reopen a confirmation, section 10); the + Spec's validator accepts it. Copying the records into the run manifest is a later pull request. + Not used by any run, notebook or CLI command yet. - **Experiment-description model, canonical writer and section digests (Stage 1, PR 4).** New modules `description` (the `Description` model; `description_from_bytes` reads with the strict reader and raises `DescriptionError` with every issue of the first failing phase) and diff --git a/pyproject.toml b/pyproject.toml index bd553997..efd2eec7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -175,6 +175,8 @@ baseline = "tools/code_rules_baseline.json" "morphobricks_preprocessing.description_identity" = "pure" "morphobricks_preprocessing.description" = "pure" "morphobricks_preprocessing.description_write" = "pure" +"morphobricks_preprocessing.description_transition" = "pure" +"morphobricks_preprocessing.description_file" = "io" "morphobricks_preprocessing.description_read" = "pure" "morphobricks_preprocessing.description_semantics" = "pure" "morphobricks_preprocessing.description_walk" = "pure" @@ -238,6 +240,7 @@ baseline = "tools/code_rules_baseline.json" "morphobricks_preprocessing.run.mask_step" = "orchestration" "morphobricks_preprocessing.run.planned" = "orchestration" "morphobricks_preprocessing.run.design_file" = "io" +"morphobricks_preprocessing.run.description_file" = "io" "morphobricks_preprocessing.run.lock" = "io" "morphobricks_preprocessing.run.modes" = "pure" "morphobricks_preprocessing.run.recorded" = "pure" diff --git a/src/morphobricks_preprocessing/description_file.py b/src/morphobricks_preprocessing/description_file.py new file mode 100644 index 00000000..4978c7fc --- /dev/null +++ b/src/morphobricks_preprocessing/description_file.py @@ -0,0 +1,91 @@ +"""Reading ``morphobricks.toml`` from disk: the entry point of every reader (experiment-v1 2.1). + +``read_description(path)`` is the one way a consumer gets a :class:`.description.Description` +from a file. A symbolic link is refused (``R_FILE_SYMLINK``), never followed, whatever it points +to; the bytes are read once and the same bytes are hashed and parsed. Everything else is the +strict reader's: nothing is converted, defaulted or repaired. +""" + +from __future__ import annotations + +import errno +import hashlib +import os +import stat +from pathlib import Path + +from .description import Description, DescriptionError, description_from_bytes +from .description_report import DescriptionIssue +from .description_rules import Refusal + +DESCRIPTION_FILENAME = "morphobricks.toml" +RECORDS_FILENAME = "morphobricks.records.jsonl" +FILE_PHASE = 0 +"""The ``phase`` of a ``DescriptionError`` about the file itself, before any byte is parsed.""" + + +def symlink_error(path: Path) -> DescriptionError: + """Return the refusal of a symbolic link (``R_FILE_SYMLINK``) at ``path``.""" + issue = DescriptionIssue(Refusal.R_FILE_SYMLINK, path.name, path.as_posix()) + return DescriptionError((issue,), FILE_PHASE) + + +def read_file_bytes(path: str | os.PathLike) -> bytes | None: + """Return the bytes of the file at ``path``, or None when there is no such file. + + Raises + ------ + DescriptionError + ``R_FILE_SYMLINK`` when ``path`` is a symbolic link (also one that is dangling or that + appears between the check and the open). + OSError + When the file exists and cannot be read, or is not a regular file; nothing is guessed. + """ + path = Path(path) + if path.is_symlink(): + raise symlink_error(path) + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + handle = os.open(path, flags) + except FileNotFoundError: + return None + except OSError as error: + if error.errno == errno.ELOOP: + raise symlink_error(path) from error + raise + with os.fdopen(handle, "rb") as stream: + if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): + raise OSError(errno.EINVAL, "not a regular file", path.as_posix()) + return stream.read() + + +def sha256_hex(data: bytes) -> str: + """Return the SHA-256 of ``data`` in lower-case hex (the file's provenance identity).""" + return hashlib.sha256(data).hexdigest() + + +def read_description(path: str | os.PathLike) -> Description: + """Read and validate a ``morphobricks.toml`` and return its model. + + Parameters + ---------- + path + The file. A missing file is a ``FileNotFoundError``, not an empty description. + + Returns + ------- + Description + The model of a file that passed phases 1 to 5 of the strict reader. + + Raises + ------ + DescriptionError + With every issue of the first failing phase; for a symbolic link, the one issue + ``R_FILE_SYMLINK`` (``phase`` 0). + FileNotFoundError + When there is no file. + """ + data = read_file_bytes(path) + if data is None: + raise FileNotFoundError(errno.ENOENT, "no experiment description", str(path)) + return description_from_bytes(data) diff --git a/src/morphobricks_preprocessing/description_report.py b/src/morphobricks_preprocessing/description_report.py index 047bc173..51cd922b 100644 --- a/src/morphobricks_preprocessing/description_report.py +++ b/src/morphobricks_preprocessing/description_report.py @@ -11,7 +11,7 @@ from collections.abc import Mapping from dataclasses import dataclass -from .description_rules import Code +from .description_rules import Code, Refusal BARE_KEY = re.compile(r"[A-Za-z0-9_-]+") @@ -26,7 +26,7 @@ def key_path(base: str, key: str) -> str: class DescriptionIssue: """One error or warning: its code, table path and the offending value (None if absent).""" - code: Code + code: Code | Refusal path: str value: object = None diff --git a/src/morphobricks_preprocessing/description_rules.py b/src/morphobricks_preprocessing/description_rules.py index cb85b35b..96f28dd2 100644 --- a/src/morphobricks_preprocessing/description_rules.py +++ b/src/morphobricks_preprocessing/description_rules.py @@ -65,6 +65,22 @@ class Code(StrEnum): W_REFERENCE_UNPAIRED = "W_REFERENCE_UNPAIRED" +class Refusal(StrEnum): + """The ``R_*`` refusals of the file and of publication (experiment-v1 section 8). + + Only the ones the file, the symbolic-link rule and the comparison of a candidate with the + published file raise are here; the physical checks against the folder are another module's. + """ + + R_FILE_SYMLINK = "R_FILE_SYMLINK" + R_CONFIRMED_ID_CHANGED = "R_CONFIRMED_ID_CHANGED" + R_USED_ID_REUSED = "R_USED_ID_REUSED" + R_USED_ID_DROPPED = "R_USED_ID_DROPPED" + R_CONFIRMED_ON_CHANGED = "R_CONFIRMED_ON_CHANGED" + R_PUBLISHED_FILE_CHANGED = "R_PUBLISHED_FILE_CHANGED" + R_DRAFT_BASE_MISSING = "R_DRAFT_BASE_MISSING" + + @dataclass(frozen=True) class TableSpec: """The closed set of fields of one table, in the order of the Spec's table. diff --git a/src/morphobricks_preprocessing/description_transition.py b/src/morphobricks_preprocessing/description_transition.py new file mode 100644 index 00000000..e387ae11 --- /dev/null +++ b/src/morphobricks_preprocessing/description_transition.py @@ -0,0 +1,250 @@ +"""The ``R_*`` refusals of a candidate description against the published one (experiment-v1). + +Sections 4.4a, 4.6, 4.9 and 10: a candidate is compared with the file it replaces. The result is +the whole set of refusals that apply (never the first), each naming the kind of ID and the IDs. +A user's recorded re-confirmation (:class:`Reconfirmation`) is the only thing that accepts a +changed entity, and only the exact change it names; nothing else is waived, and no ID, entity or +date is repaired or guessed. Pure: it compares two models and touches no file. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import date + +from .description import USED_KINDS, Description, Repeat +from .description_rules import Refusal + +CONDITIONS, CHANNELS, REPEATS, TECHNICAL = USED_KINDS +DATE = re.compile(r"\d{4}-\d{2}-\d{2}") + + +@dataclass(frozen=True, order=True) +class Reconfirmation: + """The user's recorded confirmation of one change of a confirmed ID (section 4.6). + + ``old_id`` is the ID the published file holds; ``new_id`` is the ID that replaces it (a + renamed or corrected ID), or None when the same ID loses or is reassigned something it + held (a removed member, folder or position). ``confirmed_on`` is the calendar date of the + user's action: it is a user fact and has no default. + """ + + kind: str + old_id: str + new_id: str | None + confirmed_on: str + + def __post_init__(self) -> None: + if self.kind not in USED_KINDS: + raise ValueError(f"a re-confirmation is of one of {USED_KINDS}, not {self.kind!r}") + if not DATE.fullmatch(self.confirmed_on): + raise ValueError(f"confirmed_on is YYYY-MM-DD, not {self.confirmed_on!r}") + date.fromisoformat(self.confirmed_on) # a real calendar day, else ValueError + + +@dataclass(frozen=True, order=True) +class TransitionIssue: + """One refusal: its code, the kind of ID and the IDs concerned. + + For a changed ID ``old_id`` is the published ID and ``new_id`` the ID that replaces it (None + when the same ID changed in place); for the ``used_ids`` refusals ``old_id`` is the ID. + """ + + code: Refusal + kind: str + old_id: str + new_id: str | None = None + + @property + def change(self) -> tuple[str, str, str | None]: + """Return the (kind, old ID, new ID) a re-confirmation must name to accept it.""" + return (self.kind, self.old_id, self.new_id) + + +def _declared(description: Description) -> dict[str, set[str]]: + """Return the IDs each kind declares in the file.""" + units = { + unit + for repeat in description.repeats.values() + for member in repeat.members.values() + for unit in member.technical_repeats + } + return { + CONDITIONS: set(description.conditions), + CHANNELS: set(description.channels), + REPEATS: set(description.repeats), + TECHNICAL: units, + } + + +def _member_folders(description: Description) -> dict[str, frozenset[tuple[str, str]]]: + """Return, per repeat ID, the member folders ``(condition ID, source repeat)``.""" + return { + key: frozenset((cond, member.source_repeat) for cond, member in repeat.members.items()) + for key, repeat in description.repeats.items() + } + + +def _repeat_folders_by_condition(description: Description) -> dict[str, set[str]]: + """Return, per condition ID, the source repeat folders it has over all repeats.""" + out: dict[str, set[str]] = {key: set() for key in description.conditions} + for repeat in description.repeats.values(): + for cond, member in repeat.members.items(): + out.setdefault(cond, set()).add(member.source_repeat) + return out + + +def _technical(description: Description) -> dict[str, tuple[set[str], set[tuple[str, str]]]]: + """Return, per technical repeat ID, its source folders and the member folders declaring it.""" + out: dict[str, tuple[set[str], set[tuple[str, str]]]] = {} + for repeat in description.repeats.values(): + for cond, member in repeat.members.items(): + for tid, unit in member.technical_repeats.items(): + folders, members = out.setdefault(tid, (set(), set())) + folders.add(unit.source_folder) + members.add((cond, member.source_repeat)) + return out + + +def _positions(description: Description) -> dict[str, set[tuple[str, str, str, int]]]: + """Return, per channel ID, its positions ``(condition, source repeat, technical, index)``.""" + out: dict[str, set[tuple[str, str, str, int]]] = {} + for repeat in description.repeats.values(): + for cond, member in repeat.members.items(): + for tid, unit in member.technical_repeats.items(): + for index, channel in enumerate(unit.channels): + out.setdefault(channel, set()).add((cond, member.source_repeat, tid, index)) + return out + + +def _evidence(description: Description, channel: str) -> frozenset[str]: + return frozenset(description.channels[channel].header_names or ()) + + +def _used(old: Description, new: Description) -> list[TransitionIssue]: + """Section 4.9: an ID dropped from ``used_ids``, or a retired ID declared again.""" + out = [] + old_declared, new_declared = _declared(old), _declared(new) + for kind in USED_KINDS: + old_used, new_used = set(old.used_ids[kind]), set(new.used_ids[kind]) + out += [TransitionIssue(Refusal.R_USED_ID_DROPPED, kind, i) for i in old_used - new_used] + retired = old_used - old_declared[kind] + out += [ + TransitionIssue(Refusal.R_USED_ID_REUSED, kind, i) for i in retired & new_declared[kind] + ] + return out + + +def _renamed(old: Description, new: Description) -> list[TransitionIssue]: + """Section 4.6 (a), (b) and (b2): the same entity under another ID.""" + changed = Refusal.R_CONFIRMED_ID_CHANGED + out = [] + old_by_folders = {folders: key for key, folders in _member_folders(old).items()} + for key, folders in _member_folders(new).items(): + if folders in old_by_folders and old_by_folders[folders] != key: + out.append(TransitionIssue(changed, REPEATS, old_by_folders[folders], key)) + old_positions, new_positions = _positions(old), _positions(new) + for channel, positions in old_positions.items(): + if channel in new.channels: + continue + for other, other_positions in new_positions.items(): + same_evidence = _evidence(old, channel) == _evidence(new, other) + if other not in old.channels and other_positions == positions and same_evidence: + out.append(TransitionIssue(changed, CHANNELS, channel, other)) + old_folder = {c.source_folder: key for key, c in old.conditions.items()} + for key, condition in new.conditions.items(): + folder = condition.source_folder + if folder in old_folder and old_folder[folder] != key: + out.append(TransitionIssue(changed, CONDITIONS, old_folder[folder], key)) + old_unit = {f: tid for tid, (folders, _) in _technical(old).items() for f in folders} + for tid, (folders, _) in _technical(new).items(): + out += [ + TransitionIssue(changed, TECHNICAL, old_unit[f], tid) + for f in sorted(folders) + if f in old_unit and old_unit[f] != tid + ] + return out + + +def _lost(old: Description, new: Description) -> list[TransitionIssue]: + """Section 4.6 (c): an ID declared in both files whose entity lost or reassigned something.""" + changed = Refusal.R_CONFIRMED_ID_CHANGED + out = [] + old_members, new_members = _member_folders(old), _member_folders(new) + out += [ + TransitionIssue(changed, REPEATS, key) + for key in old_members.keys() & new_members.keys() + if not old_members[key] <= new_members[key] + ] + old_by_condition = _repeat_folders_by_condition(old) + new_by_condition = _repeat_folders_by_condition(new) + for key in old.conditions.keys() & new.conditions.keys(): + moved = old.conditions[key].source_folder != new.conditions[key].source_folder + if moved or not old_by_condition[key] <= new_by_condition[key]: + out.append(TransitionIssue(changed, CONDITIONS, key)) + old_unit, new_unit = _technical(old), _technical(new) + for tid in old_unit.keys() & new_unit.keys(): + if old_unit[tid][0] != new_unit[tid][0] or not old_unit[tid][1] <= new_unit[tid][1]: + out.append(TransitionIssue(changed, TECHNICAL, tid)) + old_positions, new_positions = _positions(old), _positions(new) + for key in old.channels.keys() & new.channels.keys(): + lost_position = not old_positions.get(key, set()) <= new_positions.get(key, set()) + if lost_position or not _evidence(old, key) <= _evidence(new, key): + out.append(TransitionIssue(changed, CHANNELS, key)) + return out + + +def _layout(repeat: Repeat) -> frozenset: + """Return what reopens a repeat's confirmation (section 10): members, units, channels, type. + + The image count is not part of it: new FOVs in an existing unit do not reopen it. + """ + return frozenset( + (cond, member.source_repeat, tid, unit.source_folder, unit.channels, unit.image_type) + for cond, member in repeat.members.items() + for tid, unit in member.technical_repeats.items() + ) + + +def _dates(old: Description, new: Description) -> list[TransitionIssue]: + """Section 4.4a: ``confirmed_on`` of a repeat that was not reopened must not change.""" + return [ + TransitionIssue(Refusal.R_CONFIRMED_ON_CHANGED, REPEATS, key) + for key in old.repeats.keys() & new.repeats.keys() + if old.repeats[key].confirmed_on != new.repeats[key].confirmed_on + and _layout(old.repeats[key]) == _layout(new.repeats[key]) + ] + + +def transition_issues(old: Description, new: Description) -> tuple[TransitionIssue, ...]: + """Return every refusal that applies to ``new`` offered over the published ``old``. + + The set is complete (section 4.6, "publication reports a set"), without duplicates and in + a fixed order. An empty result means the candidate is accepted, before any re-confirmation + is considered: the changed-entity refusals are accepted by :func:`unwaived`. + """ + found = _used(old, new) + _renamed(old, new) + _lost(old, new) + _dates(old, new) + return tuple(sorted(set(found))) + + +def unwaived( + issues: tuple[TransitionIssue, ...], confirmations: tuple[Reconfirmation, ...] +) -> tuple[TransitionIssue, ...]: + """Return the issues the re-confirmations do not accept. + + A re-confirmation accepts a changed entity (``R_CONFIRMED_ID_CHANGED``) only when it names + the same kind, old ID and new ID. It accepts no other refusal. + """ + named = {(c.kind, c.old_id, c.new_id) for c in confirmations} + return tuple( + i for i in issues if i.code != Refusal.R_CONFIRMED_ID_CHANGED or i.change not in named + ) + + +def unused( + issues: tuple[TransitionIssue, ...], confirmations: tuple[Reconfirmation, ...] +) -> tuple[Reconfirmation, ...]: + """Return the re-confirmations that accept no change of the candidate.""" + needed = {i.change for i in issues if i.code == Refusal.R_CONFIRMED_ID_CHANGED} + return tuple(c for c in confirmations if (c.kind, c.old_id, c.new_id) not in needed) diff --git a/src/morphobricks_preprocessing/run/description_file.py b/src/morphobricks_preprocessing/run/description_file.py new file mode 100644 index 00000000..08b5c239 --- /dev/null +++ b/src/morphobricks_preprocessing/run/description_file.py @@ -0,0 +1,234 @@ +"""Publishing ``morphobricks.toml`` under the output folder's lock (experiment-v1 sections 4.6, 10). + +``publish_description`` is the only writer of the published file. Under +:func:`.lock.output_lock` it refuses a symbolic link, checks that the file on disk is still the +base the draft was derived from (the SHA-256 of its exact bytes, or ``absent``), writes the +candidate with the canonical writer, reads those exact bytes back with the strict reader, +compares the candidate with the published file (:mod:`..description_transition`), and only then +replaces the file atomically. Every refusal is reported as a set and writes nothing. + +The user's re-confirmations that the candidate needed are appended, in the same publication, to +the append-only ``morphobricks.records.jsonl`` next to the file (one JSON object per line: old +ID, new ID, date and the SHA-256 of the file before and after). The file is never rewritten or +truncated; a later run copies it unchanged into its manifest. If the append fails after the file +was replaced, the previous file is put back, so the publication is whole or absent (a process +killed in the instant between the two steps can leave a published file without its records, and +the records file is then the one that cannot be trusted to be complete). +""" + +from __future__ import annotations + +import json +import os +import re +from dataclasses import dataclass +from pathlib import Path + +from .._io import write_text_atomic +from ..description import Description, DescriptionError, description_from_bytes +from ..description_file import ( + DESCRIPTION_FILENAME, + RECORDS_FILENAME, + read_file_bytes, + sha256_hex, +) +from ..description_report import DescriptionIssue, key_path +from ..description_rules import Refusal +from ..description_transition import ( + Reconfirmation, + TransitionIssue, + transition_issues, + unused, + unwaived, +) +from ..description_write import canonical_bytes +from .lock import output_lock + +ABSENT = "absent" +"""The recorded base of a draft derived when no ``morphobricks.toml`` existed.""" +BASE = re.compile(r"[0-9a-f]{64}") + + +class PublicationRefused(Exception): + """Raised when a candidate is not published; ``issues`` is every refusal that applies.""" + + def __init__(self, issues: tuple[DescriptionIssue, ...]) -> None: + self.issues = issues + self.codes = frozenset(issue.code for issue in issues) + shown = "; ".join(f"{i.code.value} at {i.path}" for i in issues[:6]) + super().__init__(f"the description was not published: {shown}") + + +class RecordsFileError(OSError): + """Raised when ``morphobricks.records.jsonl`` cannot be appended to as it stands.""" + + +@dataclass(frozen=True) +class Publication: + """A published description: the file, its SHA-256, the base it replaced, the records.""" + + path: Path + sha256: str + base_sha256: str + records: tuple[dict, ...] + + +def _refuse(code: Refusal, path: str, value: object = None) -> PublicationRefused: + return PublicationRefused((DescriptionIssue(code, path, value),)) + + +def _transition_refusal(issue: TransitionIssue) -> DescriptionIssue: + """Name the table of a refusal: ``used_ids.`` for a registry rule, else the ID's own.""" + if issue.code in (Refusal.R_USED_ID_DROPPED, Refusal.R_USED_ID_REUSED): + return DescriptionIssue(issue.code, f"used_ids.{issue.kind}", issue.old_id) + return DescriptionIssue(issue.code, key_path(issue.kind, issue.old_id), issue.new_id) + + +def _check_base(base: str | None, current: bytes | None) -> str: + """Return the current file's identity (hex SHA-256 or ``absent``) after checking the base.""" + if base is None or not (base == ABSENT or BASE.fullmatch(base)): + raise _refuse(Refusal.R_DRAFT_BASE_MISSING, "base_sha256", base) + now = ABSENT if current is None else sha256_hex(current) + if now != base: + raise _refuse(Refusal.R_PUBLISHED_FILE_CHANGED, DESCRIPTION_FILENAME, now) + return now + + +def _record(item: Reconfirmation, old_sha: str, new_sha: str) -> dict: + return { + "type": "reconfirmation", + "id_kind": item.kind, + "old_id": item.old_id, + "new_id": item.new_id, + "confirmed_on": item.confirmed_on, + "old_sha256": old_sha, + "new_sha256": new_sha, + } + + +def _order(item: Reconfirmation) -> tuple[str, str, str, str]: + return (item.kind, item.old_id, item.new_id or "", item.confirmed_on) + + +def read_records(output_folder: str | os.PathLike) -> tuple[dict, ...]: + """Return the records written so far, in order (empty when there is no records file). + + Raises + ------ + DescriptionError + ``R_FILE_SYMLINK`` when the records file is a symbolic link. + RecordsFileError + When a line is not a JSON object or the last line is torn (no final newline). + """ + data = read_file_bytes(Path(output_folder) / RECORDS_FILENAME) + if not data: + return () + if not data.endswith(b"\n"): + raise RecordsFileError(f"{RECORDS_FILENAME} ends in the middle of a line") + try: + rows = tuple(json.loads(line) for line in data.decode("utf-8").splitlines()) + except ValueError as error: + raise RecordsFileError(f"{RECORDS_FILENAME} holds a line that is not JSON") from error + if not all(isinstance(row, dict) for row in rows): + raise RecordsFileError(f"{RECORDS_FILENAME} holds a line that is not an object") + return rows + + +def _append_records(folder: Path, records: tuple[dict, ...]) -> None: + """Append the records as lines; the file is created if absent and never rewritten.""" + if not records: + return + read_records(folder) # refuses a symbolic link and a torn or foreign file before writing + lines = "".join(json.dumps(r, ensure_ascii=False, sort_keys=True) + "\n" for r in records) + flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) + handle = os.open(folder / RECORDS_FILENAME, flags, 0o644) + with os.fdopen(handle, "ab") as stream: + stream.write(lines.encode("utf-8")) + stream.flush() + os.fsync(stream.fileno()) + + +def _compare( + published: bytes | None, + candidate: Description, + confirmations: tuple[Reconfirmation, ...], +) -> None: + """Refuse every ``R_*`` that applies, or a re-confirmation the candidate does not need.""" + old = description_from_bytes(published) if published is not None else None + issues = transition_issues(old, candidate) if old is not None else () + left = unwaived(issues, confirmations) + if left: + raise PublicationRefused(tuple(_transition_refusal(i) for i in left)) + if unused(issues, confirmations): + names = ", ".join(f"{c.kind} {c.old_id}" for c in unused(issues, confirmations)) + raise ValueError(f"a re-confirmation names a change the candidate does not make: {names}") + + +def publish_description( + output_folder: str | os.PathLike, + candidate: Description, + *, + base_sha256: str | None, + confirmations: tuple[Reconfirmation, ...] = (), +) -> Publication: + """Publish ``candidate`` as ``morphobricks.toml``, or refuse and write nothing. + + Parameters + ---------- + output_folder + The output folder (it must exist). + candidate + The description to publish. ``confirmed_on`` is written as it holds it. + base_sha256 + The SHA-256 of the published bytes the draft was derived from, or ``"absent"``; None + (a draft that recorded none) is ``R_DRAFT_BASE_MISSING``. + confirmations + The user's recorded re-confirmations of changed IDs; each must be needed. + + Returns + ------- + Publication + What was written. + + Raises + ------ + OutputLockedError + When a run holds the lock. + PublicationRefused + With every applicable refusal: ``R_FILE_SYMLINK`` (the file or the records file), + ``R_DRAFT_BASE_MISSING``, ``R_PUBLISHED_FILE_CHANGED``, or the changed-ID, used-ID and + ``confirmed_on`` refusals. + DescriptionError + When the candidate's canonical bytes, or the published file, is not a valid description. + ValueError + When a re-confirmation is not needed by the candidate. + """ + folder = Path(output_folder) + path = folder / DESCRIPTION_FILENAME + with output_lock(folder): + try: + published = read_file_bytes(path) + read_records(folder) # a symbolic or torn records file refuses before anything + except DescriptionError as error: + raise PublicationRefused(error.issues) from error + base = _check_base(base_sha256, published) + data = canonical_bytes(candidate) + description_from_bytes(data) # the exact bytes to be written pass the strict reader + _compare(published, candidate, confirmations) + new_sha = sha256_hex(data) + records = tuple(_record(c, base, new_sha) for c in sorted(confirmations, key=_order)) + write_text_atomic(path, data.decode("utf-8")) + try: + _append_records(folder, records) + except BaseException: + _restore(path, published) + raise + return Publication(path, new_sha, base, records) + + +def _restore(path: Path, published: bytes | None) -> None: + """Put the previous published file back (or remove the new one when there was none).""" + if published is None: + path.unlink(missing_ok=True) + else: + write_text_atomic(path, published.decode("utf-8")) diff --git a/tests/test_description_file.py b/tests/test_description_file.py new file mode 100644 index 00000000..a7b20ad7 --- /dev/null +++ b/tests/test_description_file.py @@ -0,0 +1,268 @@ +"""``read_description`` and ``publish_description``: symbolic links, the lock, the records file. + +The Spec scenarios (P09, P16-P20, P22) are in ``test_spec_publication.py``; this file holds the +mechanics around them: a symbolic link is never followed or replaced, a held lock stops a +publication, the records file is append-only, and a failed publication leaves the folder as it was. +""" + +from __future__ import annotations + +import os +from dataclasses import replace +from pathlib import Path + +import pytest + +from morphobricks_preprocessing.description import DescriptionError, description_from_bytes +from morphobricks_preprocessing.description_file import ( + DESCRIPTION_FILENAME, + FILE_PHASE, + RECORDS_FILENAME, + read_description, + sha256_hex, +) +from morphobricks_preprocessing.description_rules import Refusal +from morphobricks_preprocessing.description_transition import Reconfirmation +from morphobricks_preprocessing.description_write import canonical_bytes +from morphobricks_preprocessing.run import description_file as publication +from morphobricks_preprocessing.run.description_file import ( + ABSENT, + PublicationRefused, + RecordsFileError, + publish_description, + read_records, +) +from morphobricks_preprocessing.run.lock import LOCK_FILENAME, OutputLockedError, output_lock +from spec_pin import fixtures + +BY_NAME = {item.name.split("_")[0]: item for item in [*fixtures("valid"), *fixtures("transitions")]} +V01 = description_from_bytes(BY_NAME["v01"].data) +DAY = "2026-11-07" +RENAMES = ( + Reconfirmation("repeats", "rep1", "Day1", DAY), + Reconfirmation("repeats", "rep2", "Day2", DAY), +) + + +def _link(target: Path, link: Path) -> None: + try: + link.symlink_to(target) + except (OSError, NotImplementedError): + pytest.skip("this system does not allow symbolic links") + + +def _snapshot(folder: Path) -> dict[str, bytes]: + return {p.name: p.read_bytes() for p in folder.iterdir() if not p.is_symlink()} + + +def test_p09_a_symlinked_description_is_refused_by_the_reader(tmp_path): + real = tmp_path / "elsewhere.toml" + real.write_bytes(BY_NAME["v01"].data) + _link(real, tmp_path / DESCRIPTION_FILENAME) + with pytest.raises(DescriptionError) as refusal: + read_description(tmp_path / DESCRIPTION_FILENAME) + assert [i.code for i in refusal.value.issues] == [Refusal.R_FILE_SYMLINK] + assert refusal.value.phase == FILE_PHASE + assert read_description(real) == V01 # the same bytes, not through a link + + +def test_p09_a_dangling_symlink_is_refused_not_missing(tmp_path): + _link(tmp_path / "nowhere.toml", tmp_path / DESCRIPTION_FILENAME) + with pytest.raises(DescriptionError) as refusal: + read_description(tmp_path / DESCRIPTION_FILENAME) + assert refusal.value.issues[0].code == Refusal.R_FILE_SYMLINK + + +def test_p09_publication_refuses_a_symlink_and_neither_follows_nor_replaces_it(tmp_path): + real = tmp_path / "elsewhere.toml" + real.write_bytes(BY_NAME["v01"].data) + link = tmp_path / DESCRIPTION_FILENAME + _link(real, link) + before = _snapshot(tmp_path) + with pytest.raises(PublicationRefused) as refusal: + publish_description( + tmp_path, + description_from_bytes(BY_NAME["v16"].data), + base_sha256=sha256_hex(real.read_bytes()), + ) + assert refusal.value.codes == {Refusal.R_FILE_SYMLINK} + assert link.is_symlink() and _snapshot(tmp_path) == before + assert not (tmp_path / LOCK_FILENAME).exists() + + +def test_a_missing_file_is_not_an_empty_description(tmp_path): + with pytest.raises(FileNotFoundError): + read_description(tmp_path / DESCRIPTION_FILENAME) + + +def test_the_reader_reports_every_issue_of_the_first_failing_phase(tmp_path): + (tmp_path / DESCRIPTION_FILENAME).write_bytes( + next(f for f in fixtures("invalid") if f.name.startswith("i59")).data + ) + with pytest.raises(DescriptionError) as refusal: + read_description(tmp_path / DESCRIPTION_FILENAME) + assert refusal.value.issues and refusal.value.phase == 4 + + +def test_a_held_lock_stops_publication_and_nothing_is_written(tmp_path): + with output_lock(tmp_path), pytest.raises(OutputLockedError): + publish_description(tmp_path, V01, base_sha256=ABSENT) + assert list(tmp_path.iterdir()) == [] + + +def test_the_lock_is_released_after_a_publication_and_after_a_refusal(tmp_path): + publish_description(tmp_path, V01, base_sha256=ABSENT) + assert not (tmp_path / LOCK_FILENAME).exists() + with pytest.raises(PublicationRefused): + publish_description(tmp_path, V01, base_sha256=ABSENT) + assert not (tmp_path / LOCK_FILENAME).exists() + + +def test_the_published_file_is_replaced_atomically_and_leaves_no_temporary_file(tmp_path): + publish_description(tmp_path, V01, base_sha256=ABSENT) + assert sorted(p.name for p in tmp_path.iterdir()) == [DESCRIPTION_FILENAME] + + +def _published(tmp_path: Path) -> str: + return publish_description(tmp_path, V01, base_sha256=ABSENT).sha256 + + +def test_a_second_publication_appends_after_the_first_records(tmp_path): + base = _published(tmp_path) + t01 = description_from_bytes(BY_NAME["t01"].data) + first = publish_description(tmp_path, t01, base_sha256=base, confirmations=RENAMES) + path = tmp_path / RECORDS_FILENAME + written = path.read_bytes() + channel = replace(t01.channels["DNA"], display_name="Nuclei") + named = replace(t01, channels={**t01.channels, "DNA": channel}) + publish_description(tmp_path, named, base_sha256=first.sha256) + assert path.read_bytes() == written # no new record: none was needed + assert read_records(tmp_path) == first.records + + +def test_a_torn_records_file_refuses_before_anything_is_written(tmp_path): + base = _published(tmp_path) + (tmp_path / RECORDS_FILENAME).write_bytes(b'{"type": "reconfirmation"') + before = _snapshot(tmp_path) + with pytest.raises(RecordsFileError): + publish_description( + tmp_path, + description_from_bytes(BY_NAME["t01"].data), + base_sha256=base, + confirmations=RENAMES, + ) + assert _snapshot(tmp_path) == before + + +@pytest.mark.parametrize("content", [b"not json\n", b"[1, 2]\n"]) +def test_a_records_file_that_is_not_lines_of_objects_is_refused(tmp_path, content): + (tmp_path / RECORDS_FILENAME).write_bytes(content) + with pytest.raises(RecordsFileError): + read_records(tmp_path) + + +def test_a_symlinked_records_file_is_refused(tmp_path): + real = tmp_path / "log.jsonl" + real.write_bytes(b"") + _link(real, tmp_path / RECORDS_FILENAME) + with pytest.raises(PublicationRefused) as refusal: + publish_description(tmp_path, V01, base_sha256=ABSENT) + assert refusal.value.codes == {Refusal.R_FILE_SYMLINK} + assert not (tmp_path / DESCRIPTION_FILENAME).exists() + + +def test_no_records_file_reads_as_no_records(tmp_path): + assert read_records(tmp_path) == () + + +def test_a_failed_append_puts_the_previous_file_back(tmp_path, monkeypatch): + base = _published(tmp_path) + before = _snapshot(tmp_path) + + def broken(folder, records): + raise OSError("disk full") + + monkeypatch.setattr(publication, "_append_records", broken) + with pytest.raises(OSError, match="disk full"): + publish_description( + tmp_path, + description_from_bytes(BY_NAME["t01"].data), + base_sha256=base, + confirmations=RENAMES, + ) + assert _snapshot(tmp_path) == before and not (tmp_path / LOCK_FILENAME).exists() + + +def test_a_failed_append_removes_a_first_publication(tmp_path, monkeypatch): + def broken(folder, records): + raise OSError("disk full") + + monkeypatch.setattr(publication, "_append_records", broken) + with pytest.raises(OSError, match="disk full"): + publish_description(tmp_path, V01, base_sha256=ABSENT) + assert list(tmp_path.iterdir()) == [] + + +def test_a_reconfirmation_the_candidate_does_not_need_is_an_error_and_writes_nothing(tmp_path): + base = _published(tmp_path) + before = _snapshot(tmp_path) + with pytest.raises(ValueError, match="does not make"): + publish_description( + tmp_path, + V01, + base_sha256=base, + confirmations=(Reconfirmation("repeats", "rep1", "Day1", DAY),), + ) + assert _snapshot(tmp_path) == before + empty = tmp_path / "new" + empty.mkdir() + with pytest.raises(ValueError, match="does not make"): # nothing published yet: nothing to name + publish_description(empty, V01, base_sha256=ABSENT, confirmations=RENAMES) + assert list(empty.iterdir()) == [] + assert not (tmp_path / RECORDS_FILENAME).exists() + + +def test_a_candidate_the_reader_would_refuse_is_not_published(tmp_path): + bad = replace(V01, conditions={}) + with pytest.raises(DescriptionError) as refusal: + publish_description(tmp_path, bad, base_sha256=ABSENT) + assert refusal.value.issues + assert list(tmp_path.iterdir()) == [] + + +def test_an_unconfirmed_repeat_is_not_published(tmp_path): + repeat = replace(V01.repeats["rep1"], confirmed=False) + with pytest.raises(DescriptionError): + publish_description( + tmp_path, replace(V01, repeats={**V01.repeats, "rep1": repeat}), base_sha256=ABSENT + ) + assert list(tmp_path.iterdir()) == [] + + +def test_a_published_file_that_is_not_valid_is_not_a_base(tmp_path): + (tmp_path / DESCRIPTION_FILENAME).write_bytes(b"schema = 1\n") + base = sha256_hex(b"schema = 1\n") + with pytest.raises(DescriptionError): + publish_description(tmp_path, V01, base_sha256=base) + assert (tmp_path / DESCRIPTION_FILENAME).read_bytes() == b"schema = 1\n" + + +def test_confirmed_on_is_never_taken_from_the_clock(tmp_path): + publish_description(tmp_path, V01, base_sha256=ABSENT) + text = (tmp_path / DESCRIPTION_FILENAME).read_text(encoding="utf-8") + assert text.count('confirmed_on = "2026-11-06"') == 2 + + +def test_the_records_are_written_with_the_same_hashes_as_the_files(tmp_path): + base = _published(tmp_path) + done = publish_description( + tmp_path, + description_from_bytes(BY_NAME["t01"].data), + base_sha256=base, + confirmations=RENAMES, + ) + assert done.sha256 == sha256_hex((tmp_path / DESCRIPTION_FILENAME).read_bytes()) + assert {r["old_sha256"] for r in read_records(tmp_path)} == {base} + assert {r["new_sha256"] for r in read_records(tmp_path)} == {done.sha256} + assert canonical_bytes(V01) != (tmp_path / DESCRIPTION_FILENAME).read_bytes() + assert os.path.getsize(tmp_path / RECORDS_FILENAME) > 0 diff --git a/tests/test_spec_publication.py b/tests/test_spec_publication.py new file mode 100644 index 00000000..004f4c7e --- /dev/null +++ b/tests/test_spec_publication.py @@ -0,0 +1,205 @@ +"""Publication scenarios P16-P20 and P22 of the conformance plan, on the Spec fixtures. + +Each uses a valid fixture as the published file in a real output folder and offers a candidate +through ``publish_description`` (lock, base SHA-256, strict re-read, transition checks). A +refusal reports every applicable ``R_*`` and writes nothing. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import replace +from pathlib import Path + +import pytest + +from morphobricks_preprocessing.description import description_from_bytes +from morphobricks_preprocessing.description_file import ( + DESCRIPTION_FILENAME, + RECORDS_FILENAME, + read_description, +) +from morphobricks_preprocessing.description_rules import Refusal +from morphobricks_preprocessing.description_transition import Reconfirmation +from morphobricks_preprocessing.description_write import canonical_bytes, section_digests +from morphobricks_preprocessing.run.description_file import ( + ABSENT, + PublicationRefused, + publish_description, + read_records, +) +from spec_pin import fixtures + +BY_NAME = {item.name.split("_")[0]: item for item in [*fixtures("valid"), *fixtures("transitions")]} +DAY = "2026-11-07" + + +def _model(name: str): + return description_from_bytes(BY_NAME[name].data) + + +def _sha(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _publish_as_found(folder: Path, name: str) -> str: + """Put a fixture's exact bytes in the folder as the published file; return its SHA-256.""" + data = BY_NAME[name].data + (folder / DESCRIPTION_FILENAME).write_bytes(data) + return _sha(data) + + +def _refused(folder: Path, candidate, base, confirmations=()) -> frozenset[Refusal]: + before = sorted((p.name, p.read_bytes()) for p in folder.iterdir()) + with pytest.raises(PublicationRefused) as refusal: + publish_description(folder, candidate, base_sha256=base, confirmations=confirmations) + assert sorted((p.name, p.read_bytes()) for p in folder.iterdir()) == before # nothing written + return frozenset(refusal.value.codes) + + +@pytest.fixture +def folder(tmp_path): + return tmp_path + + +def test_p16_a_changed_id_is_refused_and_t02_also_drops_a_used_id(folder): + base = _publish_as_found(folder, "v01") + assert _refused(folder, _model("t01"), base) == {Refusal.R_CONFIRMED_ID_CHANGED} + assert _refused(folder, _model("t02"), base) == { + Refusal.R_CONFIRMED_ID_CHANGED, + Refusal.R_USED_ID_DROPPED, + } + + +def test_p16_t01_is_accepted_only_with_a_recorded_reconfirmation(folder): + base = _publish_as_found(folder, "v01") + one = Reconfirmation("repeats", "rep1", "Day1", DAY) + assert _refused(folder, _model("t01"), base, (one,)) == {Refusal.R_CONFIRMED_ID_CHANGED} + both = (one, Reconfirmation("repeats", "rep2", "Day2", DAY)) + done = publish_description(folder, _model("t01"), base_sha256=base, confirmations=both) + assert read_description(folder / DESCRIPTION_FILENAME) == _model("t01") + assert done.base_sha256 == base and done.sha256 == _sha(canonical_bytes(_model("t01"))) + rows = read_records(folder) + assert rows == done.records and len(rows) == 2 + assert rows[0] == { + "type": "reconfirmation", + "id_kind": "repeats", + "old_id": "rep1", + "new_id": "Day1", + "confirmed_on": DAY, + "old_sha256": base, + "new_sha256": done.sha256, + } + assert [(r["old_id"], r["new_id"]) for r in rows] == [("rep1", "Day1"), ("rep2", "Day2")] + + +def test_p16_a_display_name_change_is_accepted_and_invalidates_nothing(folder): + base = _publish_as_found(folder, "v01") + done = publish_description(folder, _model("v16"), base_sha256=base) + assert done.records == () and not (folder / RECORDS_FILENAME).exists() + published = read_description(folder / DESCRIPTION_FILENAME) + assert section_digests(published) == section_digests(_model("v01")) + + +def test_p17_a_second_draft_from_the_same_base_is_refused_and_not_merged(folder): + base = _publish_as_found(folder, "v01") + first = publish_description(folder, _model("v21"), base_sha256=base) + on_disk = (folder / DESCRIPTION_FILENAME).read_bytes() + assert on_disk == canonical_bytes(_model("v21")) + assert _refused(folder, _model("v16"), base) == {Refusal.R_PUBLISHED_FILE_CHANGED} + assert (folder / DESCRIPTION_FILENAME).read_bytes() == on_disk # still the first one's + # Re-derived from the file now published, the same draft is accepted. + second = publish_description(folder, _model("v16"), base_sha256=first.sha256) + assert second.base_sha256 == first.sha256 + assert read_description(folder / DESCRIPTION_FILENAME) == _model("v16") + + +def test_p18_publishing_over_nothing_with_the_absent_base_is_accepted(folder): + done = publish_description(folder, _model("v01"), base_sha256=ABSENT) + assert done.base_sha256 == ABSENT + assert (folder / DESCRIPTION_FILENAME).read_bytes() == canonical_bytes(_model("v01")) + assert not (folder / RECORDS_FILENAME).exists() + + +def test_p18_another_file_appears_before_publication(tmp_path): + (tmp_path / DESCRIPTION_FILENAME).write_bytes(BY_NAME["v03"].data) # appeared meanwhile + assert _refused(tmp_path, _model("v01"), ABSENT) == {Refusal.R_PUBLISHED_FILE_CHANGED} + + +@pytest.mark.parametrize("base", [None, "", "ABSENT", "0" * 63, "G" * 64, _sha(b"x").upper()]) +def test_p18_a_draft_without_a_usable_base_is_refused_and_nothing_is_published(folder, base): + assert _refused(folder, _model("v01"), base) == {Refusal.R_DRAFT_BASE_MISSING} + _publish_as_found(folder, "v01") + assert _refused(folder, _model("v16"), base) == {Refusal.R_DRAFT_BASE_MISSING} + + +def test_p19_a_retired_id_declared_again_is_refused(folder): + base = _publish_as_found(folder, "v25") + assert _refused(folder, _model("t03"), base) == {Refusal.R_USED_ID_REUSED} + + +def test_p19_also_when_the_returning_folder_is_the_one_the_retired_id_had(folder): + base = _publish_as_found(folder, "v25") + candidate = _model("t03") + back = replace(candidate.conditions["DrugA"], source_folder="DrugA") + candidate = replace(candidate, conditions={**candidate.conditions, "DrugA": back}) + assert candidate.conditions["DrugA"].source_folder == "DrugA" + assert _refused(folder, candidate, base) == {Refusal.R_USED_ID_REUSED} + + +def test_p20_a_dropped_used_id_is_refused(folder): + base = _publish_as_found(folder, "v25") + assert _refused(folder, _model("t04"), base) == {Refusal.R_USED_ID_DROPPED} + + +def _without_drug_a(old): + repeats = { + key: replace(repeat, members={c: m for c, m in repeat.members.items() if c != "DrugA"}) + for key, repeat in old.repeats.items() + } + conditions = {c: v for c, v in old.conditions.items() if c != "DrugA"} + return replace(old, conditions=conditions, repeats=repeats) # used_ids still lists DrugA + + +def test_p20_removing_the_last_member_of_a_condition_needs_the_recorded_confirmation(folder): + base = _publish_as_found(folder, "v01") + candidate = _without_drug_a(_model("v01")) + assert "DrugA" in candidate.used_ids["conditions"] + assert _refused(folder, candidate, base) == {Refusal.R_CONFIRMED_ID_CHANGED} + # Every entity that lost a member folder or a position is a changed entity (4.6 (c)). + lost = [("repeats", "rep1"), ("repeats", "rep2"), ("channels", "ACTIN"), ("channels", "DNA")] + lost.append(("technical_repeats", "single")) + removals = tuple(Reconfirmation(kind, i, None, DAY) for kind, i in lost) + done = publish_description(folder, candidate, base_sha256=base, confirmations=removals) + published = read_description(folder / DESCRIPTION_FILENAME) + assert "DrugA" not in published.conditions and "DrugA" in published.used_ids["conditions"] + assert len(done.records) == 5 and {r["new_id"] for r in done.records} == {None} + + +def test_p22_confirmed_on_stays_byte_for_byte_through_an_unrelated_publication(folder): + base = _publish_as_found(folder, "v01") + publish_description(folder, _model("v16"), base_sha256=base) + published = (folder / DESCRIPTION_FILENAME).read_text(encoding="utf-8") + assert published.count('confirmed_on = "2026-11-06"') == 2 + assert DAY not in published + + +def test_p22_a_new_date_on_an_unchanged_repeat_is_refused(folder): + base = _publish_as_found(folder, "v01") + assert _refused(folder, _model("t05"), base) == {Refusal.R_CONFIRMED_ON_CHANGED} + + +def test_p22_after_a_reopening_and_a_new_confirmation_the_new_date_is_accepted(folder): + base = _publish_as_found(folder, "v01") + candidate = _model("t06") # a new member and a new folder join: the layout changed + repeats = dict(candidate.repeats) + repeats["rep1"] = replace(repeats["rep1"], confirmed_on=DAY) + candidate = replace(candidate, repeats=repeats) + publish_description(folder, candidate, base_sha256=base) + published = read_description(folder / DESCRIPTION_FILENAME) + assert published.repeats["rep1"].confirmed_on == DAY + + +def test_the_candidate_is_published_as_the_canonical_bytes_it_was_checked_as(folder): + publish_description(folder, _model("v18"), base_sha256=ABSENT) + assert (folder / DESCRIPTION_FILENAME).read_bytes() == canonical_bytes(_model("v01")) diff --git a/tests/test_spec_transitions.py b/tests/test_spec_transitions.py new file mode 100644 index 00000000..be458adc --- /dev/null +++ b/tests/test_spec_transitions.py @@ -0,0 +1,158 @@ +"""The transition fixtures t01-t12 and the Spec's oracle, for the ``R_*`` comparison. + +Each transition file is offered over its named predecessor and must give exactly the listed set +of ``R_*`` codes (or none); the oracle (``refval.transition``) must agree on every ordered pair +of valid and transition files, which also covers the used-ID and ``confirmed_on`` rules in +combinations the 12 fixtures do not reach. + +One deliberate difference: the oracle takes a changed ``image_count`` as reopening a repeat's +confirmation (so it accepts a new ``confirmed_on`` beside it). Section 10 says new FOVs in an +existing unit do not reopen it, so the reader refuses that date change (``R_CONFIRMED_ON_CHANGED``); +the pairs are compared with every ``image_count`` set to 1 and the difference has its own test. +""" + +from __future__ import annotations + +import re +import tomllib +from dataclasses import replace + +import pytest + +from morphobricks_preprocessing.description import description_from_bytes +from morphobricks_preprocessing.description_rules import Refusal +from morphobricks_preprocessing.description_transition import ( + Reconfirmation, + transition_issues, + unused, + unwaived, +) +from spec_oracle import refval +from spec_pin import fixtures +from spec_text import section + +VALID = {item.name.split("_")[0]: item for item in fixtures("valid")} +TRANSITIONS = fixtures("transitions") +ALL = [*fixtures("valid"), *TRANSITIONS] +PREDECESSOR = re.compile(r"predecessor valid/(v\d+)") + + +def _counts_equal(doc: dict) -> dict: + """Return the oracle's model with every member's image count set to 1.""" + for repeat in doc["repeats"].values(): + for member in repeat["members"].values(): + member["image_count"] = 1 + return doc + + +def _codes(old, new) -> set[str]: + return {issue.code.value for issue in transition_issues(old, new)} + + +def _predecessor(item) -> str: + return PREDECESSOR.search(item.header["Expected"]).group(1) + + +@pytest.mark.parametrize("item", TRANSITIONS, ids=lambda item: item.name) +def test_a_transition_gives_exactly_its_listed_codes(item): + old = description_from_bytes(VALID[_predecessor(item)].data) + listed = {c for c in item.expected_codes if c.startswith("R_")} + assert _codes(old, description_from_bytes(item.data)) == listed + + +def test_the_codes_are_the_refusals_of_the_text(): + text = "\n".join(section("8")) + assert {code.value for code in Refusal} <= set(re.findall(r"`(R_[A-Z_]+)`", text)) + + +def test_nothing_is_refused_when_the_candidate_is_the_published_file(): + for item in ALL: + model = description_from_bytes(item.data) + assert transition_issues(model, model) == () + + +@pytest.mark.parametrize("old", ALL, ids=lambda item: item.name.split("_")[0]) +def test_the_oracle_agrees_on_every_ordered_pair(old): + oracle = refval() + old_model = description_from_bytes(old.data) + old_doc = _counts_equal(oracle.model(tomllib.loads(old.data.decode("utf-8")))) + differences = [] + for new in ALL: + new_doc = _counts_equal(oracle.model(tomllib.loads(new.data.decode("utf-8")))) + ours = _codes(old_model, description_from_bytes(new.data)) + theirs = oracle.transition(old_doc, new_doc) + if ours != theirs: + differences.append((new.name, sorted(ours), sorted(theirs))) + assert differences == [] + + +def test_a_reconfirmation_accepts_only_the_change_it_names(): + old = description_from_bytes(VALID["v01"].data) + by_name = {item.name.split("_")[0]: item for item in TRANSITIONS} + renamed = transition_issues(old, description_from_bytes(by_name["t01"].data)) + day = Reconfirmation("repeats", "rep1", "Day1", "2026-11-07") + assert unwaived(renamed, (day,)) # Day2 is still refused + both = (day, Reconfirmation("repeats", "rep2", "Day2", "2026-11-07")) + assert unwaived(renamed, both) == () and unused(renamed, both) == () + # Another kind, another old ID or no new ID names another change. + for wrong in ( + Reconfirmation("conditions", "rep1", "Day1", "2026-11-07"), + Reconfirmation("repeats", "rep1", None, "2026-11-07"), + Reconfirmation("repeats", "rep1", "Day9", "2026-11-07"), + ): + assert len(unwaived(renamed, (wrong, both[1]))) == 1 + assert unused(renamed, (wrong,)) == (wrong,) + + +def test_a_reconfirmation_never_accepts_a_used_id_or_a_date_refusal(): + old = description_from_bytes(VALID["v01"].data) + by_name = {item.name.split("_")[0]: item for item in TRANSITIONS} + issues = transition_issues(old, description_from_bytes(by_name["t02"].data)) + assert {i.code for i in issues} == {Refusal.R_CONFIRMED_ID_CHANGED, Refusal.R_USED_ID_DROPPED} + left = unwaived(issues, (Reconfirmation("channels", "ACTIN", "Actin", "2026-11-07"),)) + assert {i.code for i in left} == {Refusal.R_USED_ID_DROPPED} + dated = transition_issues(old, description_from_bytes(by_name["t05"].data)) + assert unwaived(dated, (Reconfirmation("repeats", "rep1", None, "2026-11-07"),)) == dated + + +@pytest.mark.parametrize( + "kwargs", + [ + {"kind": "masks", "old_id": "a", "new_id": None, "confirmed_on": "2026-11-07"}, + {"kind": "repeats", "old_id": "a", "new_id": None, "confirmed_on": "2026-13-07"}, + {"kind": "repeats", "old_id": "a", "new_id": None, "confirmed_on": "20261107"}, + {"kind": "repeats", "old_id": "a", "new_id": None, "confirmed_on": "2026-02-30"}, + ], +) +def test_a_reconfirmation_has_a_known_kind_and_a_real_date(kwargs): + with pytest.raises(ValueError): + Reconfirmation(**kwargs) + + +def test_new_fov_counts_do_not_reopen_a_confirmation_so_a_new_date_is_refused(): + old = description_from_bytes(VALID["v01"].data) + repeat = old.repeats["rep1"] + member = repeat.members["CTRL"] + more = replace(member, image_count=member.image_count + 5) + members = {**repeat.members, "CTRL": more} + same_day = replace(repeat, members=members) + later = replace(repeat, members=members, confirmed_on="2026-11-07") + assert _codes(old, replace(old, repeats={**old.repeats, "rep1": same_day})) == set() + assert _codes(old, replace(old, repeats={**old.repeats, "rep1": later})) == { + "R_CONFIRMED_ON_CHANGED" + } + + +def test_a_reopened_confirmation_may_take_a_new_date(): + old = description_from_bytes(VALID["v01"].data) + repeat = old.repeats["rep1"] + member = repeat.members["CTRL"] + unit = member.technical_repeats["single"] + swapped = replace(unit, channels=tuple(reversed(unit.channels))) + reopened = replace( + repeat, + confirmed_on="2026-11-07", + members={**repeat.members, "CTRL": replace(member, technical_repeats={"single": swapped})}, + ) + codes = _codes(old, replace(old, repeats={**old.repeats, "rep1": reopened})) + assert "R_CONFIRMED_ON_CHANGED" not in codes