Coverage for plugin/scripts/_rhiza_yaml_parse.py: 100%
140 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 14:46 +0000
1#!/usr/bin/env python3
2"""The hand-rolled YAML subset parser — `load_yaml`'s fallback when PyYAML is absent.
4Split from `_rhiza_yaml` because the two answer different questions: that module owns the
5*public* read/write API and the emitter, while everything here exists only to read a file
6with no third-party parser available. Keeping them together meant one module was
7simultaneously a reader, a writer, and a parser.
9**This is the reference implementation, not the fallback's poor cousin.** PyYAML applies
10YAML **1.1** implicit resolution; this applies something close to YAML 1.2 core, and where
11they disagreed it is PyYAML that was normalised down to match — see `_rhiza_yaml`'s
12`_build_loader`. So the coercion rules here (`scalar` and friends) define what a rhiza
13config *means*: scalars are strings unless they are plainly a bool, an int or null.
15The subset covered: nested mappings, block and inline (`[a, b]`) sequences, inline
16(`{source: x, dest: y}`) mappings, block scalars (`key: |`), quoted/bare scalars, and `#`
17comments. Deliberately **not** anchors, aliases, or multiple documents — none of which
18appear in rhiza template files.
20Lenient by design. A malformed line is skipped rather than raised on, because the CLI
21reader behaves that way and a damaged lock must degrade rather than crash.
22"""
24from __future__ import annotations
26import re
27from typing import Any
29_BLOCK_SCALAR_INDICATORS = {"|", ">", "|-", ">-", "|+", ">+"}
32def parse_subset(text: str) -> dict[str, Any]:
33 """Parse the nested scalar/list/mapping YAML subset rhiza files use."""
34 lines = text.splitlines()
35 value, _ = _parse_map(lines, _next_content(lines, 0), 0)
36 return value
39def _strip_comment(value: str) -> str:
40 """Drop a trailing ``# comment`` that sits outside any quotes."""
41 quote: str | None = None
42 for i, ch in enumerate(value):
43 if quote:
44 if ch == quote:
45 quote = None
46 elif ch in ("'", '"'):
47 quote = ch
48 elif ch == "#" and (i == 0 or value[i - 1] in " \t"):
49 return value[:i]
50 return value
53def _split_flow(inner: str) -> list[str]:
54 """Split the body of an inline ``[a, b, c]`` list on top-level commas."""
55 items: list[str] = []
56 buf = ""
57 quote: str | None = None
58 for ch in inner:
59 if quote:
60 buf += ch
61 if ch == quote:
62 quote = None
63 elif ch in ("'", '"'):
64 quote = ch
65 buf += ch
66 elif ch == ",":
67 items.append(buf)
68 buf = ""
69 else:
70 buf += ch
71 if buf.strip():
72 items.append(buf)
73 return items
76def _flow_map(inner: str) -> dict[str, Any]:
77 """Parse the body of an inline ``{a: x, b: y}`` mapping into a dict."""
78 result: dict[str, Any] = {}
79 for part in _split_flow(inner):
80 key, sep, rest = part.partition(":")
81 if sep:
82 result[key.strip()] = scalar(rest.strip())
83 return result
86def _is_quoted(s: str) -> bool:
87 """Is *s* wrapped in a matching pair of quotes? *s* must be non-empty.
89 A lone quote character satisfies this — ``s[0]`` and ``s[-1]`` are then the same
90 character — which is deliberate: it is malformed YAML either way, and
91 :func:`_needs_quote` relies on the resulting round-trip mismatch to quote it.
92 """
93 return (s[0] == '"' and s[-1] == '"') or (s[0] == "'" and s[-1] == "'")
96def _flow_list(body: str) -> list[Any]:
97 """Parse the already-stripped body of an inline ``[a, b, c]`` list."""
98 return [scalar(x) for x in _split_flow(body)] if body else []
101def _plain_scalar(s: str) -> Any:
102 """Coerce an unquoted, non-flow token: null, bool, int, else the string itself."""
103 low = s.lower()
104 if low in ("null", "~"):
105 return None
106 if low in ("true", "false"):
107 return low == "true"
108 try:
109 return int(s)
110 except ValueError:
111 return s
114def scalar(raw: str) -> Any:
115 """Coerce a scalar token to str/int/bool/None/list/dict, honouring quotes.
117 Shape first — quoted, then the two flow collections — and only then the unquoted
118 keyword/int/string coercions, which :func:`_plain_scalar` owns. The order matters:
119 a quoted ``'true'`` is the string, not the boolean.
120 """
121 s = raw.strip()
122 if not s:
123 return None
124 if _is_quoted(s):
125 return s[1:-1]
126 if s.startswith("[") and s.endswith("]"):
127 return _flow_list(s[1:-1].strip())
128 if s.startswith("{") and s.endswith("}"):
129 return _flow_map(s[1:-1].strip())
130 return _plain_scalar(s)
133def _indent_of(line: str) -> int:
134 """Return the number of leading spaces on *line*."""
135 return len(line) - len(line.lstrip(" "))
138def _next_content(lines: list[str], i: int) -> int:
139 """Return the index of the next non-blank, non-comment line at or after *i*."""
140 while i < len(lines):
141 stripped = lines[i].strip()
142 if stripped and not stripped.startswith("#"):
143 return i
144 i += 1
145 return len(lines)
148def _parse_map(lines: list[str], i: int, indent: int) -> tuple[dict[str, Any], int]:
149 """Parse a block mapping whose keys sit at *indent*, returning it and the next index."""
150 data: dict[str, Any] = {}
151 while True:
152 i = _next_content(lines, i)
153 if i >= len(lines) or _indent_of(lines[i]) < indent:
154 break
155 stripped = lines[i].strip()
156 if stripped.startswith("- ") or stripped == "-":
157 break # a sequence at this level is not part of a mapping
158 if ":" not in stripped:
159 i += 1 # tolerate a stray non-mapping line, as the CLI reader does
160 continue
161 key, _, rest = stripped.partition(":")
162 key = key.strip()
163 rest = _strip_comment(rest).strip()
164 i += 1
165 if rest in _BLOCK_SCALAR_INDICATORS:
166 data[key], i = _parse_block_scalar(lines, i, indent)
167 elif rest == "":
168 data[key], i = _parse_child(lines, i, indent)
169 else:
170 data[key] = scalar(rest)
171 return data, i
174def _parse_child(lines: list[str], i: int, parent_indent: int) -> tuple[Any, int]:
175 """Parse the value introduced by a bare ``key:`` line, or ``None`` when absent."""
176 j = _next_content(lines, i)
177 if j >= len(lines):
178 return None, i
179 child_indent = _indent_of(lines[j])
180 child = lines[j].strip()
181 is_seq = child.startswith("- ") or child == "-"
182 # Block sequences may sit at the parent's indent (zero-indent style); block
183 # mappings must be strictly deeper.
184 if is_seq and child_indent >= parent_indent:
185 return _parse_seq(lines, j, child_indent)
186 if not is_seq and child_indent > parent_indent:
187 return _parse_map(lines, j, child_indent)
188 return None, i
191def _parse_seq(lines: list[str], i: int, indent: int) -> tuple[list[Any], int]:
192 """Parse a block sequence whose ``- `` items sit at *indent*."""
193 items: list[Any] = []
194 while True:
195 i = _next_content(lines, i)
196 if i >= len(lines) or _indent_of(lines[i]) < indent:
197 break
198 stripped = lines[i].strip()
199 if not (stripped.startswith("- ") or stripped == "-"):
200 break
201 item = "" if stripped == "-" else stripped[2:]
202 item = _strip_comment(item).strip()
203 if item and item[0] not in "[{'\"" and re.match(r"[^:\s]+:(\s|$)", item):
204 # A block mapping under this item: reparse from the "- " column.
205 lines[i] = lines[i].replace("- ", " ", 1)
206 value, i = _parse_map(lines, i, _indent_of(lines[i]))
207 items.append(value)
208 else:
209 items.append(scalar(item))
210 i += 1
211 return items, i
214def _parse_block_scalar(lines: list[str], i: int, parent_indent: int) -> tuple[str, int]:
215 """Consume the indented body of a ``key: |`` block scalar into a string."""
216 body: list[str] = []
217 while i < len(lines):
218 line = lines[i]
219 if line.strip() and _indent_of(line) <= parent_indent:
220 break
221 body.append(line.strip())
222 i += 1
223 return "\n".join(body).strip(), i