Coverage for plugin/scripts/_rhiza_yaml.py: 100%

97 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-15 14:46 +0000

1#!/usr/bin/env python3 

2"""Minimal, dependency-free YAML reader/writer for rhiza template files. 

3 

4The bundled scripts (`status.py`, `validate.py`, `sync.py`, ...) are stdlib-only 

5ports of the `rhiza` CLI's commands, so they can run inside this plugin without 

6the CLI (or PyYAML) installed. They read `.rhiza/template.yml`, 

7`.rhiza/template.lock`, and the upstream `.rhiza/template-bundles.yml`, and 

8`sync.py` writes `.rhiza/template.lock`. 

9 

10`load_yaml` parses the subset of YAML those files use: nested mappings, block 

11and inline (`[a, b]`) sequences, inline (`{source: x, dest: y}`) mappings, block 

12scalars (`key: |`), quoted/bare scalars, and `#` comments. When PyYAML *is* 

13importable we defer to it (same "stdlib works, third-party enhances" posture as 

14the tomllib/tomli fallback), so hand-authored configs using constructs 

15this parser doesn't cover still load correctly. 

16 

17**Two readers means they must agree, and they did not.** PyYAML applies YAML 1.1 

18implicit resolution where this parser applies something close to YAML 1.2 core, so 

19the same file produced different answers depending only on whether PyYAML happened to 

20be importable — `ref: 1.20` read as the float `1.2`, `strategy: no` as `False`, 

21`0755` as octal 493, a timestamp as a `datetime`, and a `|` block keeping its trailing 

22newline. `ref` selects the template tag a sync pulls from, so that one silently 

23changed which release a repo tracked. `_build_loader` normalises the PyYAML path down 

24to this parser's rules; PyYAML is still what handles *structure* (anchors, flow 

25collections, quoting), which is the reason for deferring to it. 

26 

27What parity guarantees: for a well-formed mapping document, both readers return equal 

28data. For a malformed one they differ by design — this parser is lenient, PyYAML is 

29strict — but both fail as `ValueError`, the error every caller guards against. 

30 

31`dump_yaml` emits the flat top-level scalar/sequence subset the lock file uses, 

32matching PyYAML's `default_flow_style=False, sort_keys=False` layout (zero-indent 

33list items, `[]` for empty lists, single-quoted values where a bare token would 

34be re-read as a non-string) so a lock this module writes round-trips through all 

35three readers (this parser, PyYAML, and the rhiza CLI). 

36 

37The built-in parser deliberately does NOT handle anchors, aliases, or multiple 

38documents — none of which appear in rhiza template files. 

39""" 

40 

41from __future__ import annotations 

42 

43import re 

44import sys 

45from pathlib import Path 

46from typing import Any 

47 

48sys.path.insert(0, str(Path(__file__).resolve().parent)) 

49from _rhiza_yaml_parse import parse_subset, scalar # noqa: E402 

50 

51try: 

52 import yaml as _pyyaml 

53except ModuleNotFoundError: # pragma: no cover - the runtime case; tests install PyYAML 

54 _pyyaml = None 

55 

56# YAML 1.1 timestamp shapes PyYAML resolves to datetime; we must quote these on 

57# output (and never coerce them on input) to keep values like `synced_at` strings. 

58_TIMESTAMP = re.compile( 

59 r"^\d{4}-\d{1,2}-\d{1,2}([Tt ]\d{1,2}:\d{1,2}:\d{1,2}(\.\d+)?([Zz]|[+-]\d{1,2}(:\d{1,2})?)?)?$" 

60) 

61 

62# --- keeping the two readers in agreement ------------------------------------ 

63# 

64# `load_yaml` has two implementations behind it, and they used to disagree. PyYAML 

65# applies YAML **1.1** implicit resolution; the subset parser below applies something 

66# close to YAML 1.2 core. On one realistic pointer file that produced four different 

67# answers, and one of them was not cosmetic: 

68# 

69# ref: 1.20 -> "1.20" (subset) vs 1.2 (PyYAML, float) 

70# 

71# `ref` selects the template tag to sync from, so the same repo would resolve to a 

72# different release depending on whether PyYAML happened to be importable. The others: 

73# `synced_at` became a datetime, `strategy: no` became False, and `0755` was read as 

74# octal 493 rather than 755. 

75# 

76# The subset parser is the reference implementation — these files are configuration 

77# whose scalars are strings unless they are plainly a bool, an int or null — so the 

78# PyYAML path is normalised down to match it, rather than the reverse. Structure 

79# (anchors, flow collections, block scalars, quoting) still comes from PyYAML, which is 

80# the reason for deferring to it at all. 

81_DROPPED_TAGS = frozenset({"tag:yaml.org,2002:float", "tag:yaml.org,2002:timestamp"}) 

82# YAML 1.2 core booleans only: `yes`/`no`/`on`/`off`/`y`/`n` stay strings. 

83_STRICT_BOOL = re.compile(r"^(?:true|True|TRUE|false|False|FALSE)$") 

84# Decimal only: no octal, no hex, no underscores — `int(s)` is what the subset does. 

85_STRICT_INT = re.compile(r"^[-+]?[0-9]+$") 

86 

87 

88def _build_loader() -> Any: 

89 """Return a PyYAML loader whose scalars resolve the way `scalar` does. 

90 

91 Implemented by editing the implicit-resolver table rather than post-processing the 

92 result, because post-processing cannot tell an unquoted ``true`` from a quoted 

93 ``"true"`` — by then both are the same Python string, and coercing the second 

94 would trade one disagreement for another. 

95 """ 

96 if _pyyaml is None: # pragma: no cover - guarded by the caller 

97 return None 

98 

99 class _RhizaLoader(_pyyaml.SafeLoader): # type: ignore[misc] 

100 """SafeLoader with YAML 1.1's scalar surprises removed.""" 

101 

102 table: dict[str, list[Any]] = {} 

103 for char, mappings in _pyyaml.SafeLoader.yaml_implicit_resolvers.items(): 

104 kept = [] 

105 for tag, regexp in mappings: 

106 if tag in _DROPPED_TAGS: 

107 continue 

108 if tag == "tag:yaml.org,2002:bool": 

109 regexp = _STRICT_BOOL 

110 elif tag == "tag:yaml.org,2002:int": 

111 regexp = _STRICT_INT 

112 kept.append((tag, regexp)) 

113 table[char] = kept 

114 _RhizaLoader.yaml_implicit_resolvers = table 

115 

116 # Resolving the tag is only half of it: PyYAML's int *constructor* still reads a 

117 # leading zero as octal, so `0755` came back as 493 even once the resolver was 

118 # restricted to decimal. `_STRICT_INT` has already guaranteed the token is plain 

119 # decimal, so `int()` is both safe here and exactly what `scalar` does. 

120 _RhizaLoader.add_constructor( 

121 "tag:yaml.org,2002:int", 

122 lambda loader, node: int(loader.construct_scalar(node)), 

123 ) 

124 

125 # A fifth disagreement, found by running the existing suite with PyYAML installed: 

126 # `key: |` clips to one trailing newline per the YAML spec, while the subset 

127 # parser's `_parse_block_scalar` strips. Reconciled toward the subset parser for 

128 # consistency with the rows above — and only for block styles, since stripping every 

129 # string would destroy deliberate whitespace in a quoted one. 

130 def _construct_str(loader: Any, node: Any) -> str: 

131 value: str = loader.construct_scalar(node) 

132 return value.strip() if node.style in ("|", ">") else value 

133 

134 _RhizaLoader.add_constructor("tag:yaml.org,2002:str", _construct_str) 

135 return _RhizaLoader 

136 

137 

138def load_yaml(path: Path) -> dict[str, Any]: 

139 """Load a rhiza template/lock/bundles file into a plain dict. 

140 

141 Prefers PyYAML when available; otherwise falls back to the built-in 

142 subset parser. A file whose top level is empty yields ``{}``. Raises 

143 ``ValueError`` when the document's top level is not a mapping, mirroring 

144 how the CLI treats a malformed config. 

145 """ 

146 text = path.read_text(encoding="utf-8", errors="ignore") 

147 if _pyyaml is not None: 

148 try: 

149 data = _pyyaml.load(text, Loader=_build_loader()) # nosec B506 - SafeLoader subclass 

150 except _pyyaml.YAMLError as exc: 

151 # Every caller guards `load_yaml` with `except (OSError, ValueError)`, which 

152 # is the module's contract. PyYAML's YAMLError is neither, so on a damaged 

153 # lock it escaped all eight of them — and `stage_synced` stopped degrading 

154 # to "stage the pointer only", which is a safety property, not a nicety. 

155 raise ValueError(f"could not parse YAML: {exc}") from exc 

156 if data is None: 

157 return {} 

158 if not isinstance(data, dict): 

159 raise ValueError("top-level YAML is not a mapping") 

160 return data 

161 return parse_subset(text) 

162 

163 

164def dump_yaml(data: dict[str, Any], path: Path) -> None: 

165 """Write *data* to *path* as YAML, matching PyYAML's block layout. 

166 

167 Only the flat subset the lock file uses is supported: top-level keys whose 

168 values are scalars, ``None``, or lists of scalars. Nested mappings are not 

169 emitted (the lock has none). The output re-reads identically via this 

170 parser, PyYAML, and the rhiza CLI. 

171 """ 

172 path.write_text(dumps_yaml(data), encoding="utf-8") 

173 

174 

175def dumps_yaml(data: dict[str, Any]) -> str: 

176 """Serialise *data* to a YAML string (see :func:`dump_yaml`).""" 

177 lines: list[str] = [] 

178 for key, value in data.items(): 

179 if isinstance(value, list): 

180 if not value: 

181 lines.append(f"{key}: []") 

182 else: 

183 lines.append(f"{key}:") 

184 lines.extend(f"- {_emit_scalar(item)}" for item in value) 

185 else: 

186 lines.append(f"{key}: {_emit_scalar(value)}") 

187 return "\n".join(lines) + "\n" if lines else "" 

188 

189 

190def _emit_scalar(value: Any) -> str: 

191 """Render a scalar for output, quoting when a bare token would misparse.""" 

192 if value is None: 

193 return "null" 

194 if isinstance(value, bool): 

195 return "true" if value else "false" 

196 if isinstance(value, int): 

197 return str(value) 

198 text = str(value) 

199 if _needs_quote(text): 

200 return "'" + text.replace("'", "''") + "'" 

201 return text 

202 

203 

204def _misreads_as_non_string(text: str) -> bool: 

205 """Would *text*, emitted bare, be read back as something other than this string?""" 

206 # `scalar` is the reference reader, so a value it does not return unchanged would come 

207 # back as a bool, int, None, list or flow-map. The float and timestamp shapes are 

208 # PyYAML's YAML 1.1 surprises, which `scalar` deliberately leaves as strings. 

209 return scalar(text) != text or bool(_TIMESTAMP.match(text)) or _is_float(text) 

210 

211 

212def _starts_with_yaml_syntax(text: str) -> bool: 

213 """Does *text* open with a character YAML reads as structure rather than content?""" 

214 # Indicators, anchors, tags, quotes, comments and flow punctuation, all of which 

215 # change the meaning of the line when they lead it. 

216 if text[0] in "!&*?|>%@`\"'#[]{},": 

217 return True 

218 # `-` and `:` are only structural as a bare token or when followed by a space: 

219 # `-item` and `a:b` are ordinary scalars, `- item` and `a: b` are not. 

220 return text[0] in "-:" and (len(text) == 1 or text[1] == " ") 

221 

222 

223def _needs_quote(text: str) -> bool: 

224 """Return True when *text* must be single-quoted to survive a round-trip.""" 

225 if text == "": 

226 return True 

227 if _misreads_as_non_string(text) or _starts_with_yaml_syntax(text): 

228 return True 

229 return text != text.strip() or ": " in text or text.endswith(":") or "\n" in text 

230 

231 

232def _is_float(text: str) -> bool: 

233 """Return True when *text* parses as a float (and so needs quoting).""" 

234 try: 

235 float(text) 

236 except ValueError: 

237 return False 

238 return True 

239 

240 

241def as_list(value: Any) -> list[str]: 

242 """Normalise a scalar/None/list config field into a list of strings.""" 

243 if value is None: 

244 return [] 

245 if isinstance(value, list): 

246 return [str(x) for x in value] 

247 if isinstance(value, str): 

248 parts = value.split("\\n") if "\\n" in value and "\n" not in value else value.split("\n") 

249 return [p.strip() for p in parts if p.strip()] 

250 return [str(value)]