From c117c45f0baf025621781123b66457dd15fe50d0 Mon Sep 17 00:00:00 2001 From: Quinten Steenhuis Date: Tue, 15 Sep 2026 10:53:54 -0400 Subject: [PATCH 1/5] Add deterministic YAML checker fixer --- README.md | 11 + scripts/fix_yaml_checker.py | 19 + src/dayamlchecker/fixer.py | 824 ++++++++++++++++++++++++++++ src/dayamlchecker/yaml_structure.py | 41 +- tests/test_fix_yaml_checker.py | 126 +++++ tests/test_yaml_structure_cli.py | 28 + 6 files changed, 1048 insertions(+), 1 deletion(-) create mode 100644 scripts/fix_yaml_checker.py create mode 100644 src/dayamlchecker/fixer.py create mode 100644 tests/test_fix_yaml_checker.py diff --git a/README.md b/README.md index c77a990..fa52417 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,17 @@ pip install . python3 -m dayamlchecker `find . -name "*.yml" -path "*/questions/*" snot -path "*/.venv/*" -not -path "*/build/*"` # i.e. a space separated list of files ``` +For the conservative, deterministic fixes supported by the checker, add +`--fix`. It writes only changes that validate after editing, then runs the +normal checker against the updated files. The fixes add missing question IDs, +expand yes/no shortcuts, label the first offending field on a multi-field +screen, and suffix later duplicate block IDs. A dry run is still available via +the standalone `scripts/fix_yaml_checker.py` tool. + +```bash +python3 -m dayamlchecker --fix path/to/interview.yml +``` + ## Suppressing checks You can suppress specific errors or warnings by their ID or finding class (`accessibility`, `style`, `translatability`, `general`). diff --git a/scripts/fix_yaml_checker.py b/scripts/fix_yaml_checker.py new file mode 100644 index 0000000..050902a --- /dev/null +++ b/scripts/fix_yaml_checker.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +"""Compatibility wrapper for the packaged deterministic YAML fixer.""" + +from dayamlchecker.fixer import FilePlan, TextEdit, apply_plan, main, plan_file, run +from dayamlchecker.fixer import _target_counts + +__all__ = [ + "FilePlan", + "TextEdit", + "_target_counts", + "apply_plan", + "main", + "plan_file", + "run", +] + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/dayamlchecker/fixer.py b/src/dayamlchecker/fixer.py new file mode 100644 index 0000000..f214c74 --- /dev/null +++ b/src/dayamlchecker/fixer.py @@ -0,0 +1,824 @@ +#!/usr/bin/env python3 +"""Apply conservative, formatting-preserving fixes for four DAYamlChecker rules. + +The fixer intentionally edits only YAML source lines that correspond to: + +* EG414: add an ID to a question block, using normalized question text; +* EA510: expand yes/no shortcuts into an explicit ``fields`` entry; +* EA502: label the first offending input field with the question text; +* EG104: suffix later duplicate block IDs until IDs are unique per YAML file. + +The default mode is a dry run. Pass ``--write`` to modify files. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from collections import Counter, defaultdict +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterable + +from ruamel.yaml import YAML + +from dayamlchecker.accessibility import ( + _extract_field_label, + _field_collects_user_input, +) +from dayamlchecker.messages import Finding +from dayamlchecker.yaml_structure import ( + ACCESSIBILITY_LINT_MODE, + _collect_yaml_files, + find_errors_from_string, +) + +TARGET_CODES = frozenset({"EG414", "EA510", "EA502", "EG104"}) +SHORTCUTS = ("yesno", "noyes", "yesnomaybe", "noyesmaybe") + + +@dataclass(frozen=True) +class TextEdit: + start: int + end: int + replacement: str + + +@dataclass +class FilePlan: + path: Path + edits: list[TextEdit] = field(default_factory=list) + counts: dict[str, int] = field(default_factory=dict) + skipped_reason: str | None = None + validation_error: str | None = None + + @property + def changed(self) -> bool: + return ( + bool(self.edits) and not self.skipped_reason and not self.validation_error + ) + + +@dataclass +class _EditAccumulator: + """Collect non-overlapping line edits for one YAML file.""" + + edits: list[TextEdit] = field(default_factory=list) + occupied_lines: set[int] = field(default_factory=set) + counts: Counter[str] = field(default_factory=Counter) + + def add(self, edit: TextEdit, code: str) -> None: + occupied = set(range(edit.start, max(edit.end, edit.start + 1))) + if edit.start in self.occupied_lines or ( + edit.end > edit.start and self.occupied_lines.intersection(occupied) + ): + raise ValueError(f"overlapping edits near line {edit.start + 1}") + self.edits.append(edit) + self.occupied_lines.update(occupied) + self.counts[code] += 1 + + +def _yaml_loader() -> YAML: + yaml = YAML(typ="rt") + # Duplicate block IDs are separate YAML documents, not duplicate YAML keys. + # Allowing duplicate keys here lets the fixer inspect a file without + # discarding the user's source before the checker reports it. + yaml.allow_duplicate_keys = True + return yaml + + +def _load_documents(text: str) -> tuple[list[Any] | None, str | None]: + try: + return list(_yaml_loader().load_all(text)), None + except Exception as exc: # ruamel uses several MarkedYAMLError subclasses + return None, str(exc) + + +def _newline_for(text: str) -> str: + return "\r\n" if "\r\n" in text else "\n" + + +def _line_ending(line: str, newline: str) -> str: + if line.endswith("\r\n"): + return "\r\n" + if line.endswith("\n"): + return "\n" + return "" + + +def _quote_yaml_string(value: str) -> str: + # JSON double-quoted strings are valid YAML scalars and safely handle + # colons, quotes, brackets, Mako syntax, and HTML in question text. + return json.dumps(value, ensure_ascii=False) + + +def _normalized_question(value: Any) -> str: + if not isinstance(value, str): + return "" + return re.sub(r"\s+", " ", value).strip() + + +def _is_truthy(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return value != 0 + if isinstance(value, str): + return value.strip().lower() in {"true", "yes", "1", "on"} + return False + + +def _split_inline_comment(line: str, start: int) -> tuple[str, str]: + """Split a YAML line after start into value text and a trailing comment.""" + quote: str | None = None + escaped = False + for index in range(start, len(line)): + char = line[index] + if quote == '"' and escaped: + escaped = False + continue + if quote == '"' and char == "\\": + escaped = True + continue + if quote and char == quote: + quote = None + continue + if not quote and char in {"'", '"'}: + quote = char + continue + if not quote and char == "#" and (index == start or line[index - 1].isspace()): + return line[start:index].rstrip(), line[index:].rstrip() + return line[start:].rstrip(), "" + + +def _key_column(document: Any, key: str) -> tuple[int, int] | None: + try: + line, column = document.lc.key(key) + except (AttributeError, KeyError, TypeError): + return None + return line, column + + +def _replace_scalar_line( + lines: list[str], + *, + line_number: int, + column: int, + value: str, + key: str, + newline: str, +) -> TextEdit | None: + """Replace a simple scalar value while retaining indentation and comments.""" + original = lines[line_number] + body = original.rstrip("\r\n") + colon = body.find(":", column) + if colon < 0: + return None + raw_value, comment = _split_inline_comment(body, colon + 1) + if raw_value.lstrip().startswith(("|", ">")): + return None + replacement_body = ( + body[: colon + 1] + + " " + + _quote_yaml_string(value) + + ((" " + comment) if comment else "") + ) + ending = _line_ending(original, newline) + return TextEdit( + start=line_number, + end=line_number + 1, + replacement=replacement_body + ending, + ) + + +def _replace_id_value( + lines: list[str], + *, + line_number: int, + column: int, + value: str, + newline: str, +) -> TextEdit | None: + """Rewrite a simple ID, including the common ``id: |`` form.""" + edit = _replace_scalar_line( + lines, + line_number=line_number, + column=column, + value=value, + key="id", + newline=newline, + ) + if edit is not None: + return edit + + original = lines[line_number] + body = original.rstrip("\r\n") + colon = body.find(":", column) + if colon < 0: + return None + raw_value = body[colon + 1 :].strip() + if not raw_value.startswith(("|", ">")): + return None + + content_line_number = line_number + 1 + if content_line_number >= len(lines): + return _insert_before_line( + lines, + line_number=content_line_number, + replacement_lines=[" " * (column + 2) + value], + newline=newline, + ) + + content_line = lines[content_line_number] + content_body = content_line.rstrip("\r\n") + content_indent = len(content_body) - len(content_body.lstrip()) + if content_indent <= column: + return _insert_before_line( + lines, + line_number=content_line_number, + replacement_lines=[" " * (column + 2) + value], + newline=newline, + ) + + # Do not guess how to rewrite a multi-line ID. The corpus uses one-line + # block scalars here, but preserving an unusual value is safer than + # silently changing its meaning. + if "\n" in str(value): + return None + replacement_body = content_body[:content_indent] + value + return TextEdit( + start=content_line_number, + end=content_line_number + 1, + replacement=replacement_body + _line_ending(content_line, newline), + ) + + +def _replace_mapping_key_line( + lines: list[str], + *, + line_number: int, + column: int, + value: str, + newline: str, +) -> TextEdit | None: + """Replace a mapping key such as ``no label`` with a quoted label.""" + original = lines[line_number] + body = original.rstrip("\r\n") + colon = body.find(":", column) + if colon < 0: + return None + replacement_body = ( + body[:column] + _quote_yaml_string(value) + body[colon:] + ).rstrip() + ending = _line_ending(original, newline) + return TextEdit( + start=line_number, + end=line_number + 1, + replacement=replacement_body + ending, + ) + + +def _replace_full_line( + lines: list[str], + *, + line_number: int, + replacement_lines: Iterable[str], + newline: str, +) -> TextEdit: + original = lines[line_number] + ending = _line_ending(original, newline) + replacement = newline.join(replacement_lines) + ending + return TextEdit(line_number, line_number + 1, replacement) + + +def _insert_before_line( + lines: list[str], + *, + line_number: int, + replacement_lines: Iterable[str], + newline: str, +) -> TextEdit: + replacement = newline.join(replacement_lines) + newline + return TextEdit(line_number, line_number, replacement) + + +def _field_items(document: Any) -> list[Any]: + fields = document.get("fields") if isinstance(document, dict) else None + if isinstance(fields, dict): + return [fields] + if isinstance(fields, list): + return [item for item in fields if isinstance(item, dict)] + return [] + + +def _is_ea502_offending(field_item: dict[str, Any]) -> bool: + has_no_label = _is_truthy(field_item.get("no label")) + explicit_label = str(field_item.get("label") or "") + inferred_label = _extract_field_label(field_item) + label_is_blank = "label" in field_item and not explicit_label.strip() + missing_label = not inferred_label.strip() + return has_no_label or label_is_blank or missing_label + + +def _first_ea502_field(document: Any, target_lines: set[int]) -> Any | None: + fields = _field_items(document) + labelable_fields = [ + item + for item in fields + if _field_collects_user_input(item) or "no label" in item + ] + if len(labelable_fields) <= 1: + return None + first_field = labelable_fields[0] + if not _is_ea502_offending(first_field): + return None + if getattr(getattr(first_field, "lc", None), "line", None) not in target_lines: + return None + return first_field + + +def _unique_id(base: str, *, reserved: set[str], used: set[str]) -> str: + if base and base not in reserved and base not in used: + return base + number = 2 + while f"{base} {number}" in reserved or f"{base} {number}" in used: + number += 1 + return f"{base} {number}" + + +def _document_start_matches(document: Any, target_lines: set[int]) -> bool: + """Match checker locations that point at a document's opening line.""" + start = getattr(getattr(document, "lc", None), "line", None) + if start is None: + return False + # Depending on the document's first key and surrounding ``---`` marker, + # the checker may report the marker, the first key, or the adjacent line. + return bool({start - 1, start, start + 1}.intersection(target_lines)) + + +def fix_missing_question_id( + document: Any, + *, + lines: list[str], + newline: str, + target_lines: set[int], + reserved_ids: set[str], + used_ids: set[str], + edits: _EditAccumulator, +) -> None: + """Fix EG414 by deriving a unique ID from the question text.""" + if not isinstance(document, dict): + return + + current_id = ( + str(document.get("id")).strip() if isinstance(document.get("id"), str) else "" + ) + question = _normalized_question(document.get("question")) + question_key = _key_column(document, "question") + if current_id or not question or question_key is None: + return + if question_key[0] not in target_lines: + return + + final_id = _unique_id(question, reserved=reserved_ids, used=used_ids) + used_ids.add(final_id) + id_key = _key_column(document, "id") + if id_key is not None: + edit = _replace_id_value( + lines, + line_number=id_key[0], + column=id_key[1], + value=final_id, + newline=newline, + ) + if edit is None: + raise ValueError(f"could not safely rewrite id near line {id_key[0] + 1}") + edits.add(edit, "EG414") + return + + question_line = lines[question_key[0]] + indent = question_line[: len(question_line) - len(question_line.lstrip())] + edits.add( + _insert_before_line( + lines, + line_number=question_key[0], + replacement_lines=[f"{indent}id: {_quote_yaml_string(final_id)}"], + newline=newline, + ), + "EG414", + ) + + +def fix_duplicate_id( + document: Any, + *, + lines: list[str], + newline: str, + target_lines: set[int], + reserved_ids: set[str], + used_ids: set[str], + edits: _EditAccumulator, +) -> None: + """Fix EG104 by suffixing each later duplicate ID once.""" + if not isinstance(document, dict): + return + + current_id = ( + str(document.get("id")).strip() if isinstance(document.get("id"), str) else "" + ) + if not current_id: + return + + duplicate_is_target = current_id in used_ids and _document_start_matches( + document, target_lines + ) + if not duplicate_is_target: + used_ids.add(current_id) + return + + final_id = _unique_id(current_id, reserved=reserved_ids, used=used_ids) + used_ids.add(final_id) + id_key = _key_column(document, "id") + if id_key is None: + raise ValueError("duplicate ID has no source location") + edit = _replace_id_value( + lines, + line_number=id_key[0], + column=id_key[1], + value=final_id, + newline=newline, + ) + if edit is None: + raise ValueError(f"could not safely rewrite id near line {id_key[0] + 1}") + edits.add(edit, "EG104") + + +def fix_yesno_shortcut( + document: Any, + *, + lines: list[str], + newline: str, + target_lines: set[int], + edits: _EditAccumulator, +) -> None: + """Fix EA510 by expanding one yes/no shortcut into an explicit field.""" + if not isinstance(document, dict) or "fields" in document: + # Merging a shortcut with an existing fields block needs semantic + # review; duplicate ``fields`` keys would be worse than the finding. + return + + for shortcut in SHORTCUTS: + if shortcut not in document: + continue + key_location = _key_column(document, shortcut) + value = document.get(shortcut) + if ( + key_location is None + or not isinstance(value, str) + or not value.strip() + or key_location[0] not in target_lines + ): + continue + + datatype = ( + "yesnomaybe" if shortcut in {"yesnomaybe", "noyesmaybe"} else "yesnoradio" + ) + original_line = lines[key_location[0]].rstrip("\r\n") + colon = original_line.find(":", key_location[1]) + _, comment = _split_inline_comment(original_line, colon + 1) + key_indent = original_line[: key_location[1]] + field_value = ( + value.strip() + if re.fullmatch(r"[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)*", value.strip()) + else _quote_yaml_string(value.strip()) + ) + fields_line = f"{key_indent}fields:" + if comment: + fields_line += f" {comment}" + edits.add( + _replace_full_line( + lines, + line_number=key_location[0], + replacement_lines=[ + fields_line, + f"{key_indent} - no label: {field_value}", + f"{key_indent} datatype: {datatype}", + ], + newline=newline, + ), + "EA510", + ) + + +def fix_missing_field_label( + document: Any, + *, + lines: list[str], + newline: str, + target_lines: set[int], + edits: _EditAccumulator, +) -> None: + """Fix EA502 by labeling only the first offending input field.""" + if not isinstance(document, dict): + return + + field_item = _first_ea502_field(document, target_lines) + if field_item is None: + return + question = _normalized_question(document.get("question")) + if not question: + return + + first_key = next(iter(field_item), None) + if first_key == "": + key_location = _key_column(field_item, first_key) + if key_location is None: + return + edit = _replace_mapping_key_line( + lines, + line_number=key_location[0], + column=key_location[1], + value=question, + newline=newline, + ) + if edit is not None: + edits.add(edit, "EA502") + return + + if "no label" in field_item: + key_location = _key_column(field_item, "no label") + if key_location is None: + return + edit = _replace_mapping_key_line( + lines, + line_number=key_location[0], + column=key_location[1], + value=question, + newline=newline, + ) + if edit is not None: + edits.add(edit, "EA502") + return + + if "label" in field_item: + key_location = _key_column(field_item, "label") + if key_location is None: + return + edit = _replace_scalar_line( + lines, + line_number=key_location[0], + column=key_location[1], + value=question, + key="label", + newline=newline, + ) + if edit is not None: + edits.add(edit, "EA502") + return + + if first_key is None: + return + key_location = _key_column(field_item, first_key) + if key_location is None: + return + field_line = lines[key_location[0]] + indent = field_line[: key_location[1]] + edits.add( + _insert_before_line( + lines, + line_number=key_location[0] + 1, + replacement_lines=[f"{indent}label: {_quote_yaml_string(question)}"], + newline=newline, + ), + "EA502", + ) + + +def _target_findings(text: str, path: Path) -> list[Finding]: + return [ + finding + for finding in find_errors_from_string( + text, + input_file=str(path), + lint_mode=ACCESSIBILITY_LINT_MODE, + ) + if finding.code in TARGET_CODES + ] + + +def _target_counts(text: str, path: Path) -> Counter[str]: + return Counter(finding.code for finding in _target_findings(text, path)) + + +def plan_file(path: Path) -> FilePlan: + plan = FilePlan(path=path) + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + plan.skipped_reason = f"could not read file: {exc}" + return plan + + documents, parse_error = _load_documents(text) + if documents is None: + plan.skipped_reason = f"YAML parse failed: {parse_error}" + return plan + + target_findings = _target_findings(text, path) + target_lines_by_code: dict[str, set[int]] = defaultdict(set) + for finding in target_findings: + if finding.line_number is not None: + target_lines_by_code[finding.code].add(finding.line_number - 1) + + lines = text.splitlines(keepends=True) + newline = _newline_for(text) + edits = _EditAccumulator() + + # Preserve all original IDs whenever possible, then generate IDs for + # missing-question blocks and suffix later duplicate IDs. + reserved_ids = { + str(document.get("id")).strip() + for document in documents + if isinstance(document, dict) + and isinstance(document.get("id"), str) + and str(document.get("id")).strip() + } + used_ids: set[str] = set() + + try: + for document in documents: + fix_missing_question_id( + document, + lines=lines, + newline=newline, + target_lines=target_lines_by_code["EG414"], + reserved_ids=reserved_ids, + used_ids=used_ids, + edits=edits, + ) + fix_duplicate_id( + document, + lines=lines, + newline=newline, + target_lines=target_lines_by_code["EG104"], + reserved_ids=reserved_ids, + used_ids=used_ids, + edits=edits, + ) + fix_yesno_shortcut( + document, + lines=lines, + newline=newline, + target_lines=target_lines_by_code["EA510"], + edits=edits, + ) + fix_missing_field_label( + document, + lines=lines, + newline=newline, + target_lines=target_lines_by_code["EA502"], + edits=edits, + ) + except ValueError as exc: + plan.skipped_reason = str(exc) + return plan + + plan.counts = dict(edits.counts) + if not edits.edits: + return plan + + # Apply bottom-up so line locations from ruamel remain valid. + new_lines = list(lines) + for edit in sorted( + edits.edits, key=lambda item: (item.start, item.end), reverse=True + ): + new_lines[edit.start : edit.end] = edit.replacement.splitlines(keepends=True) + candidate = "".join(new_lines) + + # Never write a candidate that no longer parses. Also ensure the three + # deterministic families are gone after the edit; EA502 may intentionally + # remain when a screen had multiple no-label fields. + _, candidate_parse_error = _load_documents(candidate) + if candidate_parse_error: + plan.validation_error = ( + f"candidate YAML does not parse: {candidate_parse_error}" + ) + return plan + candidate_counts = _target_counts(candidate, path) + for code in ("EG414", "EA510", "EG104"): + if candidate_counts[code] > 0: + plan.validation_error = ( + f"candidate still has {candidate_counts[code]} {code} finding(s)" + ) + return plan + original_counts = _target_counts(text, path) + if candidate_counts["EA502"] > original_counts["EA502"]: + plan.validation_error = ( + f"candidate increased EA502 from {original_counts['EA502']} " + f"to {candidate_counts['EA502']}" + ) + return plan + plan.edits = edits.edits + return plan + + +def apply_plan(plan: FilePlan, *, write: bool) -> None: + if not plan.changed or not write: + return + text = plan.path.read_text(encoding="utf-8") + lines = text.splitlines(keepends=True) + for edit in sorted( + plan.edits, key=lambda item: (item.start, item.end), reverse=True + ): + lines[edit.start : edit.end] = edit.replacement.splitlines(keepends=True) + plan.path.write_text("".join(lines), encoding="utf-8") + + +def run( + paths: list[Path], *, write: bool, include_default_ignores: bool +) -> dict[str, Any]: + yaml_files = _collect_yaml_files( + paths, include_default_ignores=include_default_ignores + ) + plans = [plan_file(path) for path in yaml_files] + for plan in plans: + apply_plan(plan, write=write) + + counts: Counter[str] = Counter() + for plan in plans: + counts.update(plan.counts) + result = { + "mode": "write" if write else "dry-run", + "yaml_files": len(yaml_files), + "files_with_changes": sum(plan.changed for plan in plans), + "files_skipped": sum(plan.skipped_reason is not None for plan in plans), + "files_rejected": sum(plan.validation_error is not None for plan in plans), + "changes_by_code": dict(sorted(counts.items())), + "plans": [ + { + "file": str(plan.path), + "changes": plan.counts, + "skipped_reason": plan.skipped_reason, + "validation_error": plan.validation_error, + } + for plan in plans + if plan.changed or plan.skipped_reason or plan.validation_error + ], + } + return result + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "files", + nargs="+", + type=Path, + help="YAML files or directories to scan recursively", + ) + parser.add_argument( + "--write", + action="store_true", + help="Write safe plans to disk (default: dry run)", + ) + parser.add_argument( + "--check-all", + action="store_true", + help="Include default-ignored directories during recursive search", + ) + parser.add_argument( + "--report", + type=Path, + default=None, + help="Write the run summary as JSON", + ) + args = parser.parse_args(argv) + result = run( + args.files, + write=args.write, + include_default_ignores=not args.check_all, + ) + if args.report is not None: + args.report.parent.mkdir(parents=True, exist_ok=True) + args.report.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") + + print(f"Mode: {result['mode']}") + print(f"YAML files scanned: {result['yaml_files']}") + print(f"Files with changes: {result['files_with_changes']}") + print(f"Files skipped: {result['files_skipped']}") + print(f"Files rejected by validation: {result['files_rejected']}") + print(f"Changes by rule: {result['changes_by_code']}") + for plan in result["plans"]: + if plan["skipped_reason"]: + print(f"SKIP {plan['file']}: {plan['skipped_reason']}", file=sys.stderr) + if plan["validation_error"]: + print( + f"REJECT {plan['file']}: {plan['validation_error']}", + file=sys.stderr, + ) + return 1 if result["files_skipped"] or result["files_rejected"] else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/dayamlchecker/yaml_structure.py b/src/dayamlchecker/yaml_structure.py index 6210576..eac8f0e 100644 --- a/src/dayamlchecker/yaml_structure.py +++ b/src/dayamlchecker/yaml_structure.py @@ -2708,6 +2708,14 @@ def main(argv: Optional[list[str]] = None) -> int: "(.git*, .github*, build, dist, node_modules, sources)" ), ) + parser.add_argument( + "--fix", + action="store_true", + help=( + "Apply safe, deterministic YAML fixes before checking; " + "the remaining findings are still reported" + ), + ) parser.add_argument( "--no-wcag", dest="wcag", @@ -2901,6 +2909,37 @@ def main(argv: Optional[list[str]] = None) -> int: ) return 1 + fix_had_problem = False + if args.fix and yaml_files: + from dayamlchecker.fixer import run as run_fixes + + fix_result = run_fixes( + yaml_files, + write=True, + include_default_ignores=not args.check_all, + ) + fix_had_problem = bool( + fix_result["files_skipped"] or fix_result["files_rejected"] + ) + print( + "Fix mode: scanned {yaml_files} YAML files; wrote changes in " + "{files_with_changes}; skipped {files_skipped}; rejected " + "{files_rejected}.".format(**fix_result) + ) + if fix_result["changes_by_code"]: + print(f"Fixes by rule: {fix_result['changes_by_code']}") + for plan in fix_result["plans"]: + if plan["skipped_reason"]: + print( + f"Fix skipped {plan['file']}: {plan['skipped_reason']}", + file=sys.stderr, + ) + if plan["validation_error"]: + print( + f"Fix rejected {plan['file']}: {plan['validation_error']}", + file=sys.stderr, + ) + from dayamlchecker.messages import print_github_annotation all_findings = [] @@ -2948,7 +2987,7 @@ def main(argv: Optional[list[str]] = None) -> int: if not _finding_matches_suppression(f, cli_suppressed_codes) ] - had_error = False + had_error = fix_had_problem warning_count = sum(1 for f in all_findings if f.severity == "warning") if args.format == "github": diff --git a/tests/test_fix_yaml_checker.py b/tests/test_fix_yaml_checker.py new file mode 100644 index 0000000..f1cf13b --- /dev/null +++ b/tests/test_fix_yaml_checker.py @@ -0,0 +1,126 @@ +import tempfile +import unittest +from pathlib import Path + +from dayamlchecker.fixer import _target_counts, apply_plan, plan_file + + +class TestYAMLCheckerFixer(unittest.TestCase): + def test_fixer_handles_ids_shortcuts_labels_and_duplicate_ids(self) -> None: + source = """--- +question: What is your name? +yesno: user_agrees +--- +question: What is your name? +fields: + - no label: first_name + - no label: last_name +--- +id: duplicate +question: First duplicate +--- +id: duplicate +question: Second duplicate +""" + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "interview.yml" + path.write_text(source, encoding="utf-8") + + before = _target_counts(source, path) + plan = plan_file(path) + self.assertIsNone(plan.skipped_reason) + self.assertIsNone(plan.validation_error) + self.assertEqual( + plan.counts, {"EA502": 1, "EA510": 1, "EG104": 1, "EG414": 2} + ) + + apply_plan(plan, write=True) + result = path.read_text(encoding="utf-8") + after = _target_counts(result, path) + + self.assertEqual(before["EG414"], 2) + self.assertEqual(before["EA510"], 1) + self.assertEqual(before["EG104"], 1) + self.assertEqual(after["EG414"], 0) + self.assertEqual(after["EA510"], 0) + self.assertEqual(after["EG104"], 0) + self.assertLess(after["EA502"], before["EA502"]) + self.assertIn('id: "What is your name?"', result) + self.assertIn('id: "What is your name? 2"', result) + self.assertIn('id: "duplicate 2"', result) + self.assertIn('"What is your name?": first_name', result) + + second_plan = plan_file(path) + self.assertFalse(second_plan.changed) + self.assertEqual(second_plan.counts, {}) + + def test_fixer_preserves_yesnomaybe_semantics(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "yesnomaybe.yml" + path.write_text( + "question: Continue?\nyesnomaybe: continue\n", encoding="utf-8" + ) + + plan = plan_file(path) + self.assertIsNone(plan.validation_error) + apply_plan(plan, write=True) + result = path.read_text(encoding="utf-8") + + self.assertIn("datatype: yesnomaybe", result) + self.assertNotIn("yesnomaybe:", result) + self.assertEqual(_target_counts(result, path)["EA510"], 0) + + def test_fixer_labels_blank_mapping_key_in_place(self) -> None: + source = """question: Child support? +fields: + - "": pays_child_support + datatype: yesnoradio + - Amount: child_support_amount +""" + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "blank-label.yml" + path.write_text(source, encoding="utf-8") + + plan = plan_file(path) + self.assertIsNone(plan.validation_error) + apply_plan(plan, write=True) + result = path.read_text(encoding="utf-8") + + self.assertIn(' - "Child support?": pays_child_support', result) + self.assertNotIn("pays_child_support ", result) + self.assertNotIn("label:", result) + self.assertEqual(_target_counts(result, path)["EA502"], 0) + self.assertFalse(plan_file(path).changed) + + def test_fixer_rewrites_block_scalar_duplicate_id(self) -> None: + source = """--- +id: | + same screen +question: First +--- +id: | + same screen +question: Second +""" + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "block-id.yml" + path.write_text(source, encoding="utf-8") + + plan = plan_file(path) + self.assertIsNone(plan.validation_error) + apply_plan(plan, write=True) + result = path.read_text(encoding="utf-8") + + self.assertIn(" same screen 2\n", result) + self.assertEqual(_target_counts(result, path)["EG104"], 0) + + def test_dry_run_does_not_modify_source(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "dry-run.yml" + source = "question: What is your name?\n" + path.write_text(source, encoding="utf-8") + + plan = plan_file(path) + apply_plan(plan, write=False) + + self.assertEqual(path.read_text(encoding="utf-8"), source) diff --git a/tests/test_yaml_structure_cli.py b/tests/test_yaml_structure_cli.py index fabc00e..40a344b 100644 --- a/tests/test_yaml_structure_cli.py +++ b/tests/test_yaml_structure_cli.py @@ -419,6 +419,34 @@ def fake_run_url_check(**kwargs): assert called is False +def test_main_fix_mode_writes_safe_fixes_before_checking(tmp_path, capsys): + interview = tmp_path / "interview.yml" + interview.write_text( + "---\nquestion: What is your name?\nfields:\n - Name: user_name\n" + "---\nquestion: What is your age?\nfields:\n - Age: user_age\n", + encoding="utf-8", + ) + + assert ( + main( + [ + "--fix", + "--no-url-check", + "--no-docx-accessibility", + str(interview), + ] + ) + == 0 + ) + + result = interview.read_text(encoding="utf-8") + assert 'id: "What is your name?"' in result + assert 'id: "What is your age?"' in result + assert ( + "Fix mode: scanned 1 YAML files; wrote changes in 1" in capsys.readouterr().out + ) + + def test_main_fails_on_url_checker_errors(monkeypatch, capsys): with TemporaryDirectory() as tmp: root = Path(tmp) From 80104b9dcbb481a695557f800e15660db466d2d6 Mon Sep 17 00:00:00 2001 From: Quinten Steenhuis Date: Tue, 15 Sep 2026 14:46:48 -0400 Subject: [PATCH 2/5] Fix correctness defects in the deterministic YAML fixer Review of the fixer found several edits that could write invalid or semantically changed YAML, and a validation gap that let them through. Validation now compares every rule's finding count before and after the edit rather than only the four targeted codes, and rejects a candidate that introduces any new finding or that applies an edit without reducing that rule's count. The permissive loader used for planning tolerates duplicate keys, so it could not see a botched edit on its own. Individual fixes: * only expand a yes/no shortcut when a screen has exactly one, so two shortcuts no longer produce two top-level `fields` keys; * decide the multi-line ID guard from the block scalar's source lines instead of the single-line replacement, which corrupted `id: >` values spanning two lines; * treat `no label` as a variable shorthand only when its value is a non-empty string, and replace a boolean or empty modifier with the label it is missing; * pad a sibling `label:` to the key's column instead of copying the `- ` sequence marker, which added a new list item; * label the first *offending* field rather than the first labelable one, and skip `code` fields the way the checker does; * skip a screen whose question text is already in use as a field label, keeping repeated `--fix` runs idempotent now that later offending fields are reachable; * parse with the checker's tab expansion and translate columns back to the raw line, so tab-indented files are fixable rather than skipped; * honour `--no-wcag` and `--suppress` via a new `FixOptions`, so `--fix` never rewrites source for a rule the run would not report; and * populate `plan.counts` only after validation passes, so `changes_by_code` no longer reports edits that were rejected. A file the fixer cannot safely rewrite is a limitation of the fixer, not a finding in the user's interview, so it no longer fails the run. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LNbgp4BzZY9JmUHjWjnWT1 --- README.md | 6 + src/dayamlchecker/fixer.py | 269 +++++++++++++++++++++++----- src/dayamlchecker/yaml_structure.py | 21 ++- tests/test_fix_yaml_checker.py | 178 +++++++++++++++++- 4 files changed, 423 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index fa52417..ac2d775 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,12 @@ expand yes/no shortcuts, label the first offending field on a multi-field screen, and suffix later duplicate block IDs. A dry run is still available via the standalone `scripts/fix_yaml_checker.py` tool. +`--fix` respects `--no-wcag` and `--suppress`: it never rewrites source for a +rule the run would not report. A candidate edit is written only when it parses, +introduces no new finding of any rule, and actually removes the finding it was +made for; anything else is reported on stderr and left alone. A file the fixer +cannot safely rewrite is not itself an error, so it does not fail the run. + ```bash python3 -m dayamlchecker --fix path/to/interview.yml ``` diff --git a/src/dayamlchecker/fixer.py b/src/dayamlchecker/fixer.py index f214c74..2b488dd 100644 --- a/src/dayamlchecker/fixer.py +++ b/src/dayamlchecker/fixer.py @@ -31,14 +31,31 @@ from dayamlchecker.messages import Finding from dayamlchecker.yaml_structure import ( ACCESSIBILITY_LINT_MODE, + RuntimeOptions, _collect_yaml_files, + _finding_matches_suppression, find_errors_from_string, + fix_tabs, ) TARGET_CODES = frozenset({"EG414", "EA510", "EA502", "EG104"}) SHORTCUTS = ("yesno", "noyes", "yesnomaybe", "noyesmaybe") +@dataclass(frozen=True) +class FixOptions: + """The lint settings the fixer must agree with. + + ``--fix`` runs inside the checker, so it has to see exactly the findings + the checker will report: rewriting source for a rule the user disabled + with ``--no-wcag`` or ``--suppress`` is never what they asked for. + """ + + lint_mode: str = ACCESSIBILITY_LINT_MODE + suppressed_codes: frozenset[str] = frozenset() + runtime_options: RuntimeOptions | None = None + + @dataclass(frozen=True) class TextEdit: start: int @@ -89,9 +106,31 @@ def _yaml_loader() -> YAML: return yaml +def _normalize_for_parsing(text: str) -> str: + """Expand tabs the way ``find_errors_from_string`` does before parsing. + + Without this the fixer rejects files the checker reads happily. Tabs are + one character in the source but two columns after expansion, so every + column ruamel reports is translated back by :func:`_raw_column`. + """ + return fix_tabs.sub(" ", text) + + +def _raw_column(raw_line: str, column: int) -> int: + """Map a column in the tab-expanded line back onto the raw line.""" + if "\t" not in raw_line: + return column + expanded = 0 + for index, char in enumerate(raw_line): + if expanded >= column: + return index + expanded += 2 if char == "\t" else 1 + return len(raw_line) + + def _load_documents(text: str) -> tuple[list[Any] | None, str | None]: try: - return list(_yaml_loader().load_all(text)), None + return list(_yaml_loader().load_all(_normalize_for_parsing(text))), None except Exception as exc: # ruamel uses several MarkedYAMLError subclasses return None, str(exc) @@ -153,11 +192,15 @@ def _split_inline_comment(line: str, start: int) -> tuple[str, str]: return line[start:].rstrip(), "" -def _key_column(document: Any, key: str) -> tuple[int, int] | None: +def _key_column( + document: Any, key: str, lines: list[str] | None = None +) -> tuple[int, int] | None: try: line, column = document.lc.key(key) except (AttributeError, KeyError, TypeError): return None + if lines is not None and 0 <= line < len(lines): + column = _raw_column(lines[line], column) return line, column @@ -193,6 +236,37 @@ def _replace_scalar_line( ) +def _replace_key_and_value_line( + lines: list[str], + *, + line_number: int, + column: int, + key: str, + value: str, + newline: str, +) -> TextEdit | None: + """Replace a whole ``key: value`` pair, keeping indentation and comments.""" + original = lines[line_number] + body = original.rstrip("\r\n") + colon = body.find(":", column) + if colon < 0: + return None + raw_value, comment = _split_inline_comment(body, colon + 1) + if raw_value.lstrip().startswith(("|", ">")): + return None + replacement_body = ( + body[:column] + + f"{key}: " + + _quote_yaml_string(value) + + ((" " + comment) if comment else "") + ) + return TextEdit( + start=line_number, + end=line_number + 1, + replacement=replacement_body + _line_ending(original, newline), + ) + + def _replace_id_value( lines: list[str], *, @@ -244,8 +318,16 @@ def _replace_id_value( # Do not guess how to rewrite a multi-line ID. The corpus uses one-line # block scalars here, but preserving an unusual value is safer than - # silently changing its meaning. - if "\n" in str(value): + # silently changing its meaning. ``value`` is single-line by construction, + # so only the block scalar's own source lines can answer this: a folded + # ``id: >`` spanning two lines would otherwise keep its trailing lines and + # fold them onto the replacement. + for following in lines[content_line_number + 1 :]: + following_body = following.rstrip("\r\n") + if not following_body.strip(): + continue + if len(following_body) - len(following_body.lstrip()) < content_indent: + break return None replacement_body = content_body[:content_indent] + value return TextEdit( @@ -331,8 +413,30 @@ def _first_ea502_field(document: Any, target_lines: set[int]) -> Any | None: ] if len(labelable_fields) <= 1: return None - first_field = labelable_fields[0] - if not _is_ea502_offending(first_field): + # Only one field can sensibly carry the question text, so a screen whose + # question text is already in use was labeled by an earlier run: taking a + # second field would duplicate the label and would make repeated ``--fix`` + # runs keep eating into the fields left for human review. + question = _normalized_question(document.get("question")) + if question and any( + _extract_field_label(item).strip() == question + or str(item.get("label") or "").strip() == question + for item in labelable_fields + ): + return None + + # ``_check_multifield_no_label_usage`` counts ``code`` fields towards the + # screen's field total but never reports them, so skip them here too and + # take the first field the checker would actually flag. + first_field = next( + ( + item + for item in labelable_fields + if "code" not in item and _is_ea502_offending(item) + ), + None, + ) + if first_field is None: return None if getattr(getattr(first_field, "lc", None), "line", None) not in target_lines: return None @@ -376,7 +480,7 @@ def fix_missing_question_id( str(document.get("id")).strip() if isinstance(document.get("id"), str) else "" ) question = _normalized_question(document.get("question")) - question_key = _key_column(document, "question") + question_key = _key_column(document, "question", lines) if current_id or not question or question_key is None: return if question_key[0] not in target_lines: @@ -384,7 +488,7 @@ def fix_missing_question_id( final_id = _unique_id(question, reserved=reserved_ids, used=used_ids) used_ids.add(final_id) - id_key = _key_column(document, "id") + id_key = _key_column(document, "id", lines) if id_key is not None: edit = _replace_id_value( lines, @@ -440,7 +544,7 @@ def fix_duplicate_id( final_id = _unique_id(current_id, reserved=reserved_ids, used=used_ids) used_ids.add(final_id) - id_key = _key_column(document, "id") + id_key = _key_column(document, "id", lines) if id_key is None: raise ValueError("duplicate ID has no source location") edit = _replace_id_value( @@ -469,10 +573,15 @@ def fix_yesno_shortcut( # review; duplicate ``fields`` keys would be worse than the finding. return - for shortcut in SHORTCUTS: - if shortcut not in document: - continue - key_location = _key_column(document, shortcut) + present = [shortcut for shortcut in SHORTCUTS if shortcut in document] + if len(present) != 1: + # One replacement per shortcut would write one ``fields`` key per + # shortcut; docassemble keeps only the last, and the checker reports + # the duplicate key. Combining them needs semantic review. + return + + for shortcut in present: + key_location = _key_column(document, shortcut, lines) value = document.get(shortcut) if ( key_location is None @@ -533,7 +642,7 @@ def fix_missing_field_label( first_key = next(iter(field_item), None) if first_key == "": - key_location = _key_column(field_item, first_key) + key_location = _key_column(field_item, first_key, lines) if key_location is None: return edit = _replace_mapping_key_line( @@ -548,22 +657,39 @@ def fix_missing_field_label( return if "no label" in field_item: - key_location = _key_column(field_item, "no label") + key_location = _key_column(field_item, "no label", lines) if key_location is None: return - edit = _replace_mapping_key_line( - lines, - line_number=key_location[0], - column=key_location[1], - value=question, - newline=newline, - ) + no_label_value = field_item.get("no label") + if isinstance(no_label_value, str) and no_label_value.strip(): + # ``no label: some_variable`` names the field: turning the key + # into the label keeps the variable where it is. + edit = _replace_mapping_key_line( + lines, + line_number=key_location[0], + column=key_location[1], + value=question, + newline=newline, + ) + else: + # ``no label: true`` (or a bare ``no label:``) is a modifier, not + # a variable name. Rewriting the key would make the boolean the + # field's value; replace the modifier with the missing label + # instead, which is the change the rule is asking for. + edit = _replace_key_and_value_line( + lines, + line_number=key_location[0], + column=key_location[1], + key="label", + value=question, + newline=newline, + ) if edit is not None: edits.add(edit, "EA502") return if "label" in field_item: - key_location = _key_column(field_item, "label") + key_location = _key_column(field_item, "label", lines) if key_location is None: return edit = _replace_scalar_line( @@ -580,11 +706,14 @@ def fix_missing_field_label( if first_key is None: return - key_location = _key_column(field_item, first_key) + key_location = _key_column(field_item, first_key, lines) if key_location is None: return + # The key column sits past any ``- `` sequence marker, so copying the + # prefix verbatim would start a new list item instead of adding a sibling + # property. Blank out the marker but keep the original whitespace bytes. field_line = lines[key_location[0]] - indent = field_line[: key_location[1]] + indent = re.sub(r"\S", " ", field_line[: key_location[1]]) edits.add( _insert_before_line( lines, @@ -596,23 +725,46 @@ def fix_missing_field_label( ) -def _target_findings(text: str, path: Path) -> list[Finding]: +def _all_findings( + text: str, path: Path, options: FixOptions | None = None +) -> list[Finding]: + options = options or FixOptions() return [ finding for finding in find_errors_from_string( text, input_file=str(path), - lint_mode=ACCESSIBILITY_LINT_MODE, + lint_mode=options.lint_mode, + runtime_options=options.runtime_options, ) + if not _finding_matches_suppression(finding, options.suppressed_codes) + ] + + +def _all_counts( + text: str, path: Path, options: FixOptions | None = None +) -> Counter[str]: + return Counter(finding.code for finding in _all_findings(text, path, options)) + + +def _target_findings( + text: str, path: Path, options: FixOptions | None = None +) -> list[Finding]: + return [ + finding + for finding in _all_findings(text, path, options) if finding.code in TARGET_CODES ] -def _target_counts(text: str, path: Path) -> Counter[str]: - return Counter(finding.code for finding in _target_findings(text, path)) +def _target_counts( + text: str, path: Path, options: FixOptions | None = None +) -> Counter[str]: + return Counter(finding.code for finding in _target_findings(text, path, options)) -def plan_file(path: Path) -> FilePlan: +def plan_file(path: Path, options: FixOptions | None = None) -> FilePlan: + options = options or FixOptions() plan = FilePlan(path=path) try: text = path.read_text(encoding="utf-8") @@ -625,7 +777,7 @@ def plan_file(path: Path) -> FilePlan: plan.skipped_reason = f"YAML parse failed: {parse_error}" return plan - target_findings = _target_findings(text, path) + target_findings = _target_findings(text, path, options) target_lines_by_code: dict[str, set[int]] = defaultdict(set) for finding in target_findings: if finding.line_number is not None: @@ -684,7 +836,6 @@ def plan_file(path: Path) -> FilePlan: plan.skipped_reason = str(exc) return plan - plan.counts = dict(edits.counts) if not edits.edits: return plan @@ -696,30 +847,53 @@ def plan_file(path: Path) -> FilePlan: new_lines[edit.start : edit.end] = edit.replacement.splitlines(keepends=True) candidate = "".join(new_lines) - # Never write a candidate that no longer parses. Also ensure the three - # deterministic families are gone after the edit; EA502 may intentionally - # remain when a screen had multiple no-label fields. + # Never write a candidate that no longer parses. _, candidate_parse_error = _load_documents(candidate) if candidate_parse_error: plan.validation_error = ( f"candidate YAML does not parse: {candidate_parse_error}" ) return plan - candidate_counts = _target_counts(candidate, path) + + # Compare every rule, not just the four the fixer targets. The loader + # above tolerates duplicate keys and ignores rules outside TARGET_CODES, + # so a botched edit can round-trip cleanly here while the checker reports + # a brand new error against the file the user is left with. + original_counts = _all_counts(text, path, options) + candidate_counts = _all_counts(candidate, path, options) + introduced = sorted( + code + for code in candidate_counts + if candidate_counts[code] > original_counts[code] + ) + if introduced: + plan.validation_error = "candidate introduces new findings: " + ", ".join( + f"{code} ({original_counts[code]} -> {candidate_counts[code]})" + for code in introduced + ) + return plan + + # The three deterministic families must be gone after the edit; EA502 may + # intentionally remain when a screen had several unlabeled fields. for code in ("EG414", "EA510", "EG104"): if candidate_counts[code] > 0: plan.validation_error = ( f"candidate still has {candidate_counts[code]} {code} finding(s)" ) return plan - original_counts = _target_counts(text, path) - if candidate_counts["EA502"] > original_counts["EA502"]: - plan.validation_error = ( - f"candidate increased EA502 from {original_counts['EA502']} " - f"to {candidate_counts['EA502']}" - ) - return plan + + # An edit that removes nothing is not a fix, however well it parses. + for code, applied in sorted(edits.counts.items()): + if candidate_counts[code] >= original_counts[code]: + plan.validation_error = ( + f"candidate applied {applied} {code} edit(s) without reducing " + f"{code} findings ({original_counts[code]} -> " + f"{candidate_counts[code]})" + ) + return plan + plan.edits = edits.edits + plan.counts = dict(edits.counts) return plan @@ -736,12 +910,17 @@ def apply_plan(plan: FilePlan, *, write: bool) -> None: def run( - paths: list[Path], *, write: bool, include_default_ignores: bool + paths: list[Path], + *, + write: bool, + include_default_ignores: bool, + options: FixOptions | None = None, ) -> dict[str, Any]: + options = options or FixOptions() yaml_files = _collect_yaml_files( paths, include_default_ignores=include_default_ignores ) - plans = [plan_file(path) for path in yaml_files] + plans = [plan_file(path, options) for path in yaml_files] for plan in plans: apply_plan(plan, write=write) diff --git a/src/dayamlchecker/yaml_structure.py b/src/dayamlchecker/yaml_structure.py index eac8f0e..8b818eb 100644 --- a/src/dayamlchecker/yaml_structure.py +++ b/src/dayamlchecker/yaml_structure.py @@ -2909,17 +2909,25 @@ def main(argv: Optional[list[str]] = None) -> int: ) return 1 - fix_had_problem = False if args.fix and yaml_files: + from dayamlchecker.fixer import FixOptions from dayamlchecker.fixer import run as run_fixes + # The fixer must see the same findings this run will report, or it + # rewrites source for rules the user turned off. fix_result = run_fixes( yaml_files, write=True, include_default_ignores=not args.check_all, - ) - fix_had_problem = bool( - fix_result["files_skipped"] or fix_result["files_rejected"] + options=FixOptions( + lint_mode=lint_mode, + suppressed_codes=( + _parse_suppression_codes(",".join(args.suppress)) + if args.suppress + else frozenset() + ), + runtime_options=runtime_options, + ), ) print( "Fix mode: scanned {yaml_files} YAML files; wrote changes in " @@ -2987,7 +2995,10 @@ def main(argv: Optional[list[str]] = None) -> int: if not _finding_matches_suppression(f, cli_suppressed_codes) ] - had_error = fix_had_problem + # A file the fixer could not rewrite is a limitation of the fixer, not a + # finding in the user's interview: it is reported on stderr above and must + # not by itself fail the run. + had_error = False warning_count = sum(1 for f in all_findings if f.severity == "warning") if args.format == "github": diff --git a/tests/test_fix_yaml_checker.py b/tests/test_fix_yaml_checker.py index f1cf13b..0d02d36 100644 --- a/tests/test_fix_yaml_checker.py +++ b/tests/test_fix_yaml_checker.py @@ -2,7 +2,13 @@ import unittest from pathlib import Path -from dayamlchecker.fixer import _target_counts, apply_plan, plan_file +from dayamlchecker.fixer import ( + FixOptions, + _target_counts, + apply_plan, + plan_file, +) +from dayamlchecker.yaml_structure import DEFAULT_LINT_MODE class TestYAMLCheckerFixer(unittest.TestCase): @@ -124,3 +130,173 @@ def test_dry_run_does_not_modify_source(self) -> None: apply_plan(plan, write=False) self.assertEqual(path.read_text(encoding="utf-8"), source) + + def _fixed(self, source: str, name: str = "interview.yml") -> tuple[str, object]: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / name + path.write_text(source, encoding="utf-8") + plan = plan_file(path) + apply_plan(plan, write=True) + return path.read_text(encoding="utf-8"), plan + + def test_fixer_leaves_screens_with_two_shortcuts_alone(self) -> None: + source = "id: two\nquestion: Agree?\nyesno: a\nnoyes: b\n" + result, plan = self._fixed(source, "two-shortcuts.yml") + + # One replacement per shortcut would write two ``fields`` keys. + self.assertEqual(result, source) + self.assertEqual(result.count("fields:"), 0) + self.assertEqual(plan.counts, {}) + + def test_fixer_adds_label_as_a_sibling_of_a_sequence_field(self) -> None: + source = ( + "id: seq\nquestion: |\n What is your name?\n" + "fields:\n - field: user_name\n - Second thing: other_var\n" + ) + result, _ = self._fixed(source, "sequence-field.yml") + + self.assertIn(' - field: user_name\n label: "What is your name?"\n', result) + # A copied ``- `` prefix would add a third field instead of a label. + self.assertEqual(result.count(" - "), 2) + self.assertEqual(_target_counts(result, Path("sequence-field.yml"))["EA502"], 0) + + def test_fixer_replaces_boolean_no_label_with_a_label(self) -> None: + source = ( + "id: bool\nquestion: |\n Do you agree?\n" + "fields:\n - no label: true\n field: user_agrees\n" + " - Second thing: other_var\n" + ) + result, _ = self._fixed(source, "bool-no-label.yml") + + # ``true`` is a modifier, not the field's variable name. + self.assertIn(' - label: "Do you agree?"\n field: user_agrees\n', result) + self.assertNotIn('"Do you agree?": true', result) + self.assertNotIn("no label", result) + + def test_fixer_labels_the_first_offending_field_not_the_first_field(self) -> None: + source = ( + "id: later\nquestion: Income?\n" + "fields:\n - Employer: employer_name\n - no label: income_amount\n" + ) + result, plan = self._fixed(source, "later-offender.yml") + + self.assertEqual(plan.counts, {"EA502": 1}) + self.assertIn(' - "Income?": income_amount\n', result) + + def test_fixer_skips_code_fields_like_the_checker(self) -> None: + source = ( + "id: code\nquestion: Income?\n" + "fields:\n - no label: choices\n code: options\n" + " - no label: income_amount\n" + ) + result, plan = self._fixed(source, "code-field.yml") + + # The checker never reports the ``code`` field, so neither may the fixer. + self.assertIn(" - no label: choices\n code: options\n", result) + self.assertIn(' - "Income?": income_amount\n', result) + self.assertEqual(plan.counts, {"EA502": 1}) + + def test_fixer_leaves_folded_multiline_ids_alone(self) -> None: + source = ( + "id: >\n same\n screen\nquestion: First\n---\n" + "id: >\n same\n screen\nquestion: Second\n" + ) + result, plan = self._fixed(source, "folded-id.yml") + + # Rewriting only the first content line would fold the remaining lines + # onto the new ID, producing "same screen 2 screen". + self.assertEqual(result, source) + self.assertIsNotNone(plan.skipped_reason) + + def test_fixer_plans_tab_indented_files(self) -> None: + source = ( + "id: tabbed\nquestion: Agree?\nfields:\n\t- no label: a\n\t- Second: b\n" + ) + result, plan = self._fixed(source, "tabbed.yml") + + # The checker expands tabs before parsing; the fixer must agree, and + # must translate the expanded columns back onto the raw line. + self.assertIsNone(plan.skipped_reason) + self.assertIn('\t- "Agree?": a\n', result) + self.assertIn("\t- Second: b\n", result) + + def test_fixer_honors_lint_mode_and_suppressions(self) -> None: + source = "id: flag\nquestion: Agree?\nyesno: user_agrees\n" + with tempfile.TemporaryDirectory() as directory: + for name, options in ( + ("no-wcag.yml", FixOptions(lint_mode=DEFAULT_LINT_MODE)), + ("suppressed.yml", FixOptions(suppressed_codes=frozenset({"EA510"}))), + ): + path = Path(directory) / name + path.write_text(source, encoding="utf-8") + plan = plan_file(path, options) + apply_plan(plan, write=True) + + # The rule is turned off for this run, so nothing to rewrite. + self.assertEqual(path.read_text(encoding="utf-8"), source, name) + self.assertEqual(plan.counts, {}, name) + + def test_fixer_rejects_an_edit_that_removes_no_finding(self) -> None: + from dayamlchecker import fixer + + source = ( + "id: noop\nquestion: |\n What is your name?\n" + "fields:\n - field: user_name\n - Second thing: other_var\n" + ) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "noop.yml" + path.write_text(source, encoding="utf-8") + + # Recreate the pre-fix behaviour of copying the ``- `` prefix. + original = fixer._insert_before_line + + def _sibling_becomes_a_new_item( + lines, *, line_number, replacement_lines, newline + ): + return original( + lines, + line_number=line_number, + replacement_lines=[ + line.replace(" ", " - ", 1) for line in replacement_lines + ], + newline=newline, + ) + + fixer._insert_before_line = _sibling_becomes_a_new_item + try: + plan = plan_file(path) + finally: + fixer._insert_before_line = original + + self.assertIsNotNone(plan.validation_error) + self.assertFalse(plan.changed) + # Counts must describe what was written, not what was attempted. + self.assertEqual(plan.counts, {}) + apply_plan(plan, write=True) + self.assertEqual(path.read_text(encoding="utf-8"), source) + + def test_fixer_is_idempotent_on_screens_with_several_bare_fields(self) -> None: + source = ( + "id: assets\nquestion: |\n Is anyone holding assets for you?\n" + "fields:\n - no label: anyone_holds\n - no label: anyone_holds_describe\n" + ) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "repeat.yml" + path.write_text(source, encoding="utf-8") + + first = plan_file(path) + apply_plan(first, write=True) + after_first = path.read_text(encoding="utf-8") + self.assertEqual(first.counts, {"EA502": 1}) + self.assertIn( + ' - "Is anyone holding assets for you?": anyone_holds\n', after_first + ) + + # The second field is intentionally left for human review, so a + # repeat run must not spend the same question text on it. + second = plan_file(path) + apply_plan(second, write=True) + + self.assertFalse(second.changed) + self.assertEqual(second.counts, {}) + self.assertEqual(path.read_text(encoding="utf-8"), after_first) From f464233ab1aee77a052dc453fec9a0c605533650 Mon Sep 17 00:00:00 2001 From: Quinten Steenhuis Date: Tue, 15 Sep 2026 15:18:51 -0400 Subject: [PATCH 3/5] Normalize generated block IDs to lowercase alphanumerics Question text went into an ID verbatim apart from whitespace collapsing, so IDs picked up punctuation, smart quotes and, in four places across the corpus, a whole Mako expression: `id: "${city_only_address}"`. `_normalized_id` now keeps only letters, digits and spaces and lowercases the result, so the uniqueness pass sees two questions that differ only in case or punctuation as the collision they are. Apostrophes are dropped in place rather than separated on, so "didn't" becomes "didnt" instead of "didn t", while every other mark becomes a space and keeps `city_only_address` three readable words. `isalnum` is Unicode-aware, so accented question text survives. Only generated IDs are normalized. EA502 labels still use the question text as written, and EG104 keeps the author's own ID text and appends only the suffix. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LNbgp4BzZY9JmUHjWjnWT1 --- src/dayamlchecker/fixer.py | 22 ++++++++++++++++- tests/test_fix_yaml_checker.py | 42 ++++++++++++++++++++++++++++++-- tests/test_yaml_structure_cli.py | 4 +-- 3 files changed, 63 insertions(+), 5 deletions(-) diff --git a/src/dayamlchecker/fixer.py b/src/dayamlchecker/fixer.py index 2b488dd..099d3b8 100644 --- a/src/dayamlchecker/fixer.py +++ b/src/dayamlchecker/fixer.py @@ -159,6 +159,24 @@ def _normalized_question(value: Any) -> str: return re.sub(r"\s+", " ", value).strip() +def _normalized_id(value: Any) -> str: + """Normalize question text into a block ID. + + An ID only has to be unique and readable, so nothing but letters, digits + and spaces survives: punctuation, Mako delimiters and smart quotes all + make an ID awkward to quote and to reference. Lowercasing means two + questions that differ only in case or punctuation collide here rather than + producing two IDs that read identically. + + Apostrophes are dropped rather than separated on, so ``didn't`` becomes + ``didnt`` instead of ``didn t``; every other mark becomes a space, so + ``city_only_address`` stays three readable words. + """ + text = re.sub(r"['\u2018\u2019\u02bc]", "", _normalized_question(value)) + text = "".join(char if char.isalnum() else " " for char in text) + return re.sub(r"\s+", " ", text).strip().lower() + + def _is_truthy(value: Any) -> bool: if isinstance(value, bool): return value @@ -486,7 +504,9 @@ def fix_missing_question_id( if question_key[0] not in target_lines: return - final_id = _unique_id(question, reserved=reserved_ids, used=used_ids) + final_id = _unique_id( + _normalized_id(document.get("question")), reserved=reserved_ids, used=used_ids + ) used_ids.add(final_id) id_key = _key_column(document, "id", lines) if id_key is not None: diff --git a/tests/test_fix_yaml_checker.py b/tests/test_fix_yaml_checker.py index 0d02d36..904b379 100644 --- a/tests/test_fix_yaml_checker.py +++ b/tests/test_fix_yaml_checker.py @@ -51,8 +51,8 @@ def test_fixer_handles_ids_shortcuts_labels_and_duplicate_ids(self) -> None: self.assertEqual(after["EA510"], 0) self.assertEqual(after["EG104"], 0) self.assertLess(after["EA502"], before["EA502"]) - self.assertIn('id: "What is your name?"', result) - self.assertIn('id: "What is your name? 2"', result) + self.assertIn('id: "what is your name"', result) + self.assertIn('id: "what is your name 2"', result) self.assertIn('id: "duplicate 2"', result) self.assertIn('"What is your name?": first_name', result) @@ -300,3 +300,41 @@ def test_fixer_is_idempotent_on_screens_with_several_bare_fields(self) -> None: self.assertFalse(second.changed) self.assertEqual(second.counts, {}) self.assertEqual(path.read_text(encoding="utf-8"), after_first) + + def test_generated_ids_are_alphanumeric_lowercase(self) -> None: + source = ( + "---\nquestion: What is your name?\n" + "---\nquestion: |\n WHAT is your NAME!?\n" + "---\nquestion: Does ${ users[0] } agree?\n" + ) + result, plan = self._fixed(source, "ids.yml") + + self.assertEqual(plan.counts, {"EG414": 3}) + self.assertIn('id: "what is your name"\n', result) + # Punctuation and case no longer hide a collision from the uniqueness + # pass, so the second screen is suffixed rather than reading the same. + self.assertIn('id: "what is your name 2"\n', result) + self.assertIn('id: "does users 0 agree"\n', result) + + def test_generated_ids_drop_apostrophes_but_separate_on_punctuation(self) -> None: + source = ( + "---\nquestion: We didn\u2019t find a matching court\n" + "---\nquestion: ${city_only_address}\n" + ) + result, _ = self._fixed(source, "punct.yml") + + self.assertIn('id: "we didnt find a matching court"\n', result) + self.assertIn('id: "city only address"\n', result) + + def test_duplicate_id_fix_keeps_the_author_s_own_text(self) -> None: + source = ( + "---\nid: My Screen?\nquestion: First\n" + "---\nid: My Screen?\nquestion: Second\n" + ) + result, plan = self._fixed(source, "dup.yml") + + # EG104 only has to make an existing ID unique; rewriting the author's + # own text beyond the suffix is not the fixer's call. + self.assertEqual(plan.counts, {"EG104": 1}) + self.assertIn("id: My Screen?\n", result) + self.assertIn('id: "My Screen? 2"\n', result) diff --git a/tests/test_yaml_structure_cli.py b/tests/test_yaml_structure_cli.py index 40a344b..b091921 100644 --- a/tests/test_yaml_structure_cli.py +++ b/tests/test_yaml_structure_cli.py @@ -440,8 +440,8 @@ def test_main_fix_mode_writes_safe_fixes_before_checking(tmp_path, capsys): ) result = interview.read_text(encoding="utf-8") - assert 'id: "What is your name?"' in result - assert 'id: "What is your age?"' in result + assert 'id: "what is your name"' in result + assert 'id: "what is your age"' in result assert ( "Fix mode: scanned 1 YAML files; wrote changes in 1" in capsys.readouterr().out ) From 1cfbd8ec2bb69d42f1fa2e370f99adf020393cf0 Mon Sep 17 00:00:00 2001 From: Quinten Steenhuis Date: Tue, 15 Sep 2026 15:37:45 -0400 Subject: [PATCH 4/5] Keep Mako line directives in EA502 labels by using the long form Question text carrying ``% if``/``% for`` cannot survive the one-line ``"question": variable`` shorthand: collapsing it puts the directives mid-line, where Mako does not evaluate them, so the applicant sees the literal source. Two screens in the corpus hit this, and one of them flattened both branches of an if/else into a single label. There is no correct flattening -- the text is genuinely conditional -- so the fix is to stop flattening. When the question contains a ``%`` directive the field moves to the ``label:``/``field:`` form, where a literal block scalar keeps every directive on its own line exactly as written. Inline ``${ }`` still uses the shorthand, since it evaluates fine mid-line. The sibling-insertion fallback declines for these questions rather than inserting a block scalar beside a key whose layout it has not inspected, leaving the screen for a human the way the rule already does elsewhere. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LNbgp4BzZY9JmUHjWjnWT1 --- src/dayamlchecker/fixer.py | 155 +++++++++++++++++++++++++++++---- tests/test_fix_yaml_checker.py | 61 +++++++++++++ 2 files changed, 200 insertions(+), 16 deletions(-) diff --git a/src/dayamlchecker/fixer.py b/src/dayamlchecker/fixer.py index 099d3b8..b0c678d 100644 --- a/src/dayamlchecker/fixer.py +++ b/src/dayamlchecker/fixer.py @@ -159,6 +159,62 @@ def _normalized_question(value: Any) -> str: return re.sub(r"\s+", " ", value).strip() +# Mako honours ``% if``/``% for`` only at the start of a line, so question +# text using them cannot survive being collapsed into a one-line label. +_MAKO_DIRECTIVE = re.compile(r"^[ \t]*%", re.MULTILINE) + + +def _question_needs_block_label(value: Any) -> bool: + return isinstance(value, str) and bool(_MAKO_DIRECTIVE.search(value)) + + +def _question_block_lines(value: Any) -> list[str]: + """The question's own lines, trailing blanks dropped, Mako left as written.""" + return str(value).rstrip().splitlines() + + +def _block_label_edit( + lines: list[str], + *, + line_number: int, + column: int, + question: Any, + variable: str | None, + newline: str, +) -> TextEdit | None: + """Rewrite a field's first line as a multi-line ``label:`` block. + + The one-line ``"question": variable`` shorthand cannot hold a line break, + so a question carrying Mako line directives has to move to the + ``label:``/``field:`` form, where a literal block scalar keeps the + directives on their own lines exactly as the author wrote them. + """ + body = lines[line_number].rstrip("\r\n") + colon = body.find(":", column) + if colon < 0: + return None + _, comment = _split_inline_comment(body, colon + 1) + question_lines = _question_block_lines(question) + if not question_lines: + return None + + marker = body[:column] + key_indent = " " * column + content_indent = " " * (column + 2) + replacement = [marker + "label: |" + ((" " + comment) if comment else "")] + replacement += [ + content_indent + line if line.strip() else "" for line in question_lines + ] + if variable is not None: + replacement.append(key_indent + "field: " + variable) + ending = _line_ending(lines[line_number], newline) + return TextEdit( + start=line_number, + end=line_number + 1, + replacement=newline.join(replacement) + ending, + ) + + def _normalized_id(value: Any) -> str: """Normalize question text into a block ID. @@ -660,16 +716,22 @@ def fix_missing_field_label( if not question: return + raw_question = document.get("question") + needs_block = _question_needs_block_label(raw_question) + first_key = next(iter(field_item), None) if first_key == "": key_location = _key_column(field_item, first_key, lines) if key_location is None: return - edit = _replace_mapping_key_line( + edit = _shorthand_label_edit( lines, - line_number=key_location[0], - column=key_location[1], - value=question, + key_location=key_location, + field_item=field_item, + key=first_key, + question=question, + raw_question=raw_question, + needs_block=needs_block, newline=newline, ) if edit is not None: @@ -684,11 +746,14 @@ def fix_missing_field_label( if isinstance(no_label_value, str) and no_label_value.strip(): # ``no label: some_variable`` names the field: turning the key # into the label keeps the variable where it is. - edit = _replace_mapping_key_line( + edit = _shorthand_label_edit( lines, - line_number=key_location[0], - column=key_location[1], - value=question, + key_location=key_location, + field_item=field_item, + key="no label", + question=question, + raw_question=raw_question, + needs_block=needs_block, newline=newline, ) else: @@ -696,13 +761,24 @@ def fix_missing_field_label( # a variable name. Rewriting the key would make the boolean the # field's value; replace the modifier with the missing label # instead, which is the change the rule is asking for. - edit = _replace_key_and_value_line( - lines, - line_number=key_location[0], - column=key_location[1], - key="label", - value=question, - newline=newline, + edit = ( + _block_label_edit( + lines, + line_number=key_location[0], + column=key_location[1], + question=raw_question, + variable=None, + newline=newline, + ) + if needs_block + else _replace_key_and_value_line( + lines, + line_number=key_location[0], + column=key_location[1], + key="label", + value=question, + newline=newline, + ) ) if edit is not None: edits.add(edit, "EA502") @@ -712,6 +788,18 @@ def fix_missing_field_label( key_location = _key_column(field_item, "label", lines) if key_location is None: return + if needs_block: + edit = _block_label_edit( + lines, + line_number=key_location[0], + column=key_location[1], + question=raw_question, + variable=None, + newline=newline, + ) + if edit is not None: + edits.add(edit, "EA502") + return edit = _replace_scalar_line( lines, line_number=key_location[0], @@ -724,7 +812,9 @@ def fix_missing_field_label( edits.add(edit, "EA502") return - if first_key is None: + if first_key is None or needs_block: + # A sibling ``label:`` would have to be a block scalar inserted next to + # a key whose own layout we have not inspected; leave it for a human. return key_location = _key_column(field_item, first_key, lines) if key_location is None: @@ -767,6 +857,39 @@ def _all_counts( return Counter(finding.code for finding in _all_findings(text, path, options)) +def _shorthand_label_edit( + lines: list[str], + *, + key_location: tuple[int, int], + field_item: dict[str, Any], + key: str, + question: str, + raw_question: Any, + needs_block: bool, + newline: str, +) -> TextEdit | None: + """Label a ``: variable`` field, in block form when Mako requires it.""" + if not needs_block: + return _replace_mapping_key_line( + lines, + line_number=key_location[0], + column=key_location[1], + value=question, + newline=newline, + ) + variable = field_item.get(key) + if not isinstance(variable, str) or not variable.strip(): + return None + return _block_label_edit( + lines, + line_number=key_location[0], + column=key_location[1], + question=raw_question, + variable=variable.strip(), + newline=newline, + ) + + def _target_findings( text: str, path: Path, options: FixOptions | None = None ) -> list[Finding]: diff --git a/tests/test_fix_yaml_checker.py b/tests/test_fix_yaml_checker.py index 904b379..abfedc0 100644 --- a/tests/test_fix_yaml_checker.py +++ b/tests/test_fix_yaml_checker.py @@ -338,3 +338,64 @@ def test_duplicate_id_fix_keeps_the_author_s_own_text(self) -> None: self.assertEqual(plan.counts, {"EG104": 1}) self.assertIn("id: My Screen?\n", result) self.assertIn('id: "My Screen? 2"\n', result) + + def test_mako_directives_force_a_block_label(self) -> None: + source = ( + "id: guardian\n" + "question: |\n" + " % if filled_by_attorney:\n" + " Does ${ users[0] } want to be the guardian?\n" + " % else:\n" + " Do you want to be the guardian?\n" + " % endif\n" + "fields:\n" + " - no label: wants_guardianship\n" + " datatype: yesnoradio\n" + " - Something else: other_var\n" + ) + result, plan = self._fixed(source, "mako.yml") + + # `% if` only works at the start of a line, so the one-line + # `"question": variable` shorthand cannot carry it; the long form can. + self.assertEqual(plan.counts, {"EA502": 1}) + self.assertIn( + " - label: |\n" + " % if filled_by_attorney:\n" + " Does ${ users[0] } want to be the guardian?\n" + " % else:\n" + " Do you want to be the guardian?\n" + " % endif\n" + " field: wants_guardianship\n" + " datatype: yesnoradio\n", + result, + ) + self.assertNotIn('"% if', result) + + from ruamel.yaml import YAML + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "parsed.yml" + path.write_text(result, encoding="utf-8") + doc = YAML().load(path) + # The label must be the question text verbatim, not a flattened copy. + self.assertEqual(doc["fields"][0]["label"], doc["question"]) + self.assertEqual(doc["fields"][0]["field"], "wants_guardianship") + self.assertFalse(plan_file(path).changed) + + def test_inline_mako_expressions_still_use_the_shorthand(self) -> None: + source = ( + "id: inline\n" + "question: |\n" + " What is ${ other_parties[0].familiar() }'s address?\n" + "fields:\n" + " - no label: other_address\n" + " - Something else: other_var\n" + ) + result, _ = self._fixed(source, "inline-mako.yml") + + # ``${ }`` evaluates fine mid-line, so nothing needs to move. + self.assertIn( + ' - "What is ${ other_parties[0].familiar() }\'s address?": other_address\n', + result, + ) + self.assertNotIn("label: |", result) From 0ccb0e68600a5e4a424a42a7a39f188d9d317905 Mon Sep 17 00:00:00 2001 From: Quinten Steenhuis Date: Tue, 15 Sep 2026 22:11:23 -0400 Subject: [PATCH 5/5] Type the fixer test helper so mypy passes over tests CI runs mypy across the whole tree, not just src, and the `_fixed` helper added with the regression tests returned `tuple[str, object]`. Every `plan.counts` and `plan.skipped_reason` on its result was then an attribute access on `object`, which is 8 errors in CI and none locally under `mypy src`. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LNbgp4BzZY9JmUHjWjnWT1 --- tests/test_fix_yaml_checker.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_fix_yaml_checker.py b/tests/test_fix_yaml_checker.py index abfedc0..8734fee 100644 --- a/tests/test_fix_yaml_checker.py +++ b/tests/test_fix_yaml_checker.py @@ -3,6 +3,7 @@ from pathlib import Path from dayamlchecker.fixer import ( + FilePlan, FixOptions, _target_counts, apply_plan, @@ -131,7 +132,7 @@ def test_dry_run_does_not_modify_source(self) -> None: self.assertEqual(path.read_text(encoding="utf-8"), source) - def _fixed(self, source: str, name: str = "interview.yml") -> tuple[str, object]: + def _fixed(self, source: str, name: str = "interview.yml") -> tuple[str, FilePlan]: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / name path.write_text(source, encoding="utf-8")