Programmatically populate host_key in git-gate #335
@@ -11,13 +11,14 @@ Public entry point: `preflight_host_keys(manifest, headless=..., home_md=...)`.
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import cast
|
||||
|
||||
from .log import die, info
|
||||
from .manifest import Manifest
|
||||
from .yaml_subset import YamlSubsetError, parse_frontmatter, serialize_yaml_subset
|
||||
|
||||
|
||||
# Preferred key types, most secure first.
|
||||
@@ -81,79 +82,37 @@ def fetch_host_key(host: str, port: str) -> str:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Frontmatter text-level editing
|
||||
# Frontmatter editing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
|
|
||||
def _insert_host_key_in_frontmatter(fm_text: str, repo_name: str, host_key: str) -> str:
|
||||
"""Append `host_key: <key>` to the named repo entry in raw frontmatter text.
|
||||
def _add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) -> str:
|
||||
"""Return an updated copy of `file_text` with `host_key` set on the
|
||||
named repo entry in the YAML frontmatter.
|
||||
|
||||
Locates the `<repo_name>:` block-mapping key, detects its child
|
||||
indentation from the sibling keys already present, and inserts the
|
||||
new line after the last child. Returns the original text if the
|
||||
repo entry is not found or already has a `host_key:` line."""
|
||||
lines = fm_text.splitlines(keepends=True)
|
||||
|
||||
repo_pat = re.compile(r"^(\s+)" + re.escape(repo_name) + r"\s*:\s*(\r?\n)?$")
|
||||
repo_line_idx: int | None = None
|
||||
repo_indent: int | None = None
|
||||
for i, line in enumerate(lines):
|
||||
if repo_pat.match(line):
|
||||
repo_indent = len(line) - len(line.lstrip())
|
||||
repo_line_idx = i
|
||||
break
|
||||
|
||||
if repo_line_idx is None or repo_indent is None:
|
||||
return fm_text
|
||||
|
||||
child_indent: int | None = None
|
||||
last_child_idx = repo_line_idx
|
||||
|
||||
for i in range(repo_line_idx + 1, len(lines)):
|
||||
raw = lines[i].rstrip("\r\n")
|
||||
if not raw.strip():
|
||||
continue
|
||||
indent = len(raw) - len(raw.lstrip())
|
||||
if indent > repo_indent:
|
||||
child_indent = indent
|
||||
last_child_idx = i
|
||||
else:
|
||||
break
|
||||
|
||||
if child_indent is None:
|
||||
child_indent = repo_indent + 2
|
||||
|
||||
block = "".join(lines[repo_line_idx + 1 : last_child_idx + 1])
|
||||
if re.search(r"^\s*host_key\s*:", block, re.MULTILINE):
|
||||
return fm_text
|
||||
|
||||
new_line = " " * child_indent + f"host_key: {host_key}\n"
|
||||
lines.insert(last_child_idx + 1, new_line)
|
||||
return "".join(lines)
|
||||
|
||||
|
||||
def _update_frontmatter_in_file(file_text: str, repo_name: str, host_key: str) -> str:
|
||||
"""Apply `_insert_host_key_in_frontmatter` to the frontmatter section of
|
||||
a Markdown file, leaving the body unchanged. Returns the original text
|
||||
when no change is needed."""
|
||||
lines = file_text.splitlines(keepends=True)
|
||||
if not lines or lines[0].rstrip("\r\n") != "---":
|
||||
Parses the frontmatter into a dict, sets the key, and re-serializes.
|
||||
Returns the original text unchanged when: the file has no frontmatter,
|
||||
the git-gate.repos.<repo_name> entry is absent or already has a
|
||||
host_key, or the frontmatter cannot be parsed."""
|
||||
try:
|
||||
fm, body = parse_frontmatter(file_text)
|
||||
except YamlSubsetError:
|
||||
return file_text
|
||||
|
||||
close_idx: int | None = None
|
||||
for i in range(1, len(lines)):
|
||||
if lines[i].rstrip("\r\n") == "---":
|
||||
close_idx = i
|
||||
break
|
||||
if close_idx is None:
|
||||
git_gate = fm.get("git-gate")
|
||||
if not isinstance(git_gate, dict):
|
||||
return file_text
|
||||
repos = git_gate.get("repos")
|
||||
if not isinstance(repos, dict):
|
||||
return file_text
|
||||
repo = repos.get(repo_name)
|
||||
if not isinstance(repo, dict):
|
||||
return file_text
|
||||
if repo.get("host_key"):
|
||||
return file_text
|
||||
|
||||
fm_text = "".join(lines[1:close_idx])
|
||||
body_text = "".join(lines[close_idx:])
|
||||
updated_fm = _insert_host_key_in_frontmatter(fm_text, repo_name, host_key)
|
||||
if updated_fm == fm_text:
|
||||
return file_text
|
||||
return "---\n" + updated_fm + body_text
|
||||
cast(dict[str, object], repo)["host_key"] = host_key
|
||||
return f"---\n{serialize_yaml_subset(fm)}---\n{body}"
|
||||
|
||||
|
||||
def _find_and_update_bottle_file(
|
||||
@@ -168,28 +127,13 @@ def _find_and_update_bottle_file(
|
||||
if not bottles_dir.is_dir():
|
||||
return False
|
||||
|
||||
from .yaml_subset import YamlSubsetError, parse_frontmatter
|
||||
|
||||
for path in sorted(bottles_dir.glob("*.md")):
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
fm, _body = parse_frontmatter(text)
|
||||
except (OSError, UnicodeDecodeError, YamlSubsetError):
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
|
||||
git_gate = fm.get("git-gate")
|
||||
if not isinstance(git_gate, dict):
|
||||
continue
|
||||
repos = git_gate.get("repos")
|
||||
if not isinstance(repos, dict):
|
||||
continue
|
||||
repo = repos.get(repo_name)
|
||||
if not isinstance(repo, dict):
|
||||
continue
|
||||
if repo.get("host_key"):
|
||||
continue
|
||||
|
||||
updated = _update_frontmatter_in_file(text, repo_name, host_key)
|
||||
updated = _add_host_key_to_frontmatter(text, repo_name, host_key)
|
||||
if updated == text:
|
||||
continue
|
||||
path.write_text(updated, encoding="utf-8")
|
||||
@@ -292,7 +236,6 @@ def preflight_host_keys(
|
||||
__all__ = [
|
||||
"fetch_host_key",
|
||||
"preflight_host_keys",
|
||||
"_insert_host_key_in_frontmatter",
|
||||
"_add_host_key_to_frontmatter",
|
||||
"_find_and_update_bottle_file",
|
||||
"_update_frontmatter_in_file",
|
||||
]
|
||||
|
||||
@@ -21,6 +21,11 @@ Public API:
|
||||
For a Markdown file with YAML frontmatter delimited by `---`
|
||||
lines. Returns (frontmatter_dict, body_text).
|
||||
|
||||
serialize_yaml_subset(data) -> str
|
||||
Serialize a dict (as produced by parse_yaml_subset) back to
|
||||
block-style YAML text. The result ends with a newline and
|
||||
can be parsed back by parse_yaml_subset.
|
||||
|
||||
What we accept (block-style):
|
||||
|
||||
key: value # mapping entry, value is inline
|
||||
@@ -576,3 +581,105 @@ def parse_frontmatter(text: str) -> tuple[dict[str, object], str]:
|
||||
fm = parse_yaml_subset(fm_text)
|
||||
body = text[body_start:]
|
||||
return fm, body
|
||||
|
||||
|
||||
# --- Serializer -------------------------------------------------------------
|
||||
|
||||
|
||||
def _needs_quoting(s: str) -> bool:
|
||||
"""True when the string must be single-quoted to survive a round-trip."""
|
||||
if not s:
|
||||
return True
|
||||
if s in ("true", "false", "null", "~") or s in _RESERVED_BOOL_LIKE:
|
||||
return True
|
||||
if (
|
||||
_INT_RX.match(s)
|
||||
or _DATE_RX.match(s)
|
||||
or _OCTAL_RX.match(s)
|
||||
or _HEX_RX.match(s)
|
||||
or _FLOAT_RX.match(s)
|
||||
):
|
||||
return True
|
||||
# Characters that have special meaning at the start of a YAML value
|
||||
if s[0] in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">", "%", "@", "`"):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _yaml_scalar(v: object) -> str:
|
||||
"""Serialize a scalar Python value to its YAML text form."""
|
||||
if v is None:
|
||||
return "null"
|
||||
if isinstance(v, bool):
|
||||
return "true" if v else "false"
|
||||
if isinstance(v, int):
|
||||
return str(v)
|
||||
s = str(v)
|
||||
if _needs_quoting(s):
|
||||
return "'" + s.replace("'", "''") + "'"
|
||||
return s
|
||||
|
||||
|
||||
def _serialize_node(node: object, indent: int) -> list[str]:
|
||||
"""Return lines (without trailing newlines) for `node` at `indent`.
|
||||
|
||||
Called only for non-empty dicts and lists (the caller guards with
|
||||
`isinstance(val, (dict, list)) and val`), plus scalars at the leaf."""
|
||||
prefix = " " * indent
|
||||
if isinstance(node, dict):
|
||||
out: list[str] = []
|
||||
for key, val in node.items():
|
||||
if isinstance(val, (dict, list)) and val:
|
||||
out.append(f"{prefix}{key}:")
|
||||
out.extend(_serialize_node(val, indent + 2))
|
||||
else:
|
||||
scalar = (
|
||||
"{}" if isinstance(val, dict)
|
||||
else "[]" if isinstance(val, list)
|
||||
else _yaml_scalar(val)
|
||||
)
|
||||
out.append(f"{prefix}{key}: {scalar}")
|
||||
return out
|
||||
if isinstance(node, list):
|
||||
out = []
|
||||
for item in node:
|
||||
if isinstance(item, dict) and item:
|
||||
entries = list(item.items())
|
||||
first_key, first_val = entries[0]
|
||||
if isinstance(first_val, (dict, list)) and first_val:
|
||||
out.append(f"{prefix}- {first_key}:")
|
||||
out.extend(_serialize_node(first_val, indent + 4))
|
||||
else:
|
||||
scalar = (
|
||||
"{}" if isinstance(first_val, dict)
|
||||
else "[]" if isinstance(first_val, list)
|
||||
else _yaml_scalar(first_val)
|
||||
)
|
||||
out.append(f"{prefix}- {first_key}: {scalar}")
|
||||
cont = prefix + " "
|
||||
for key, val in entries[1:]:
|
||||
if isinstance(val, (dict, list)) and val:
|
||||
out.append(f"{cont}{key}:")
|
||||
out.extend(_serialize_node(val, indent + 4))
|
||||
else:
|
||||
scalar = (
|
||||
"{}" if isinstance(val, dict)
|
||||
else "[]" if isinstance(val, list)
|
||||
else _yaml_scalar(val)
|
||||
)
|
||||
out.append(f"{cont}{key}: {scalar}")
|
||||
else:
|
||||
out.append(f"{prefix}- {_yaml_scalar(item)}")
|
||||
return out
|
||||
return [_yaml_scalar(node)] # pragma: no cover
|
||||
|
||||
|
||||
def serialize_yaml_subset(data: dict[str, object]) -> str:
|
||||
"""Serialize `data` (as produced by parse_yaml_subset) to YAML text.
|
||||
|
||||
Produces block-style output with 2-space indentation. The result ends
|
||||
with a newline and can be parsed back by parse_yaml_subset. Keys are
|
||||
emitted in iteration order (insertion order in Python 3.7+)."""
|
||||
if not data:
|
||||
return ""
|
||||
return "\n".join(_serialize_node(data, 0)) + "\n"
|
||||
|
||||
@@ -8,10 +8,9 @@ from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from bot_bottle.git_gate_host_key import (
|
||||
_add_host_key_to_frontmatter,
|
||||
_find_and_update_bottle_file,
|
||||
_insert_host_key_in_frontmatter,
|
||||
_prompt_tty,
|
||||
_update_frontmatter_in_file,
|
||||
fetch_host_key,
|
||||
preflight_host_keys,
|
||||
)
|
||||
@@ -145,96 +144,11 @@ class TestPromptTty(unittest.TestCase):
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _insert_host_key_in_frontmatter
|
||||
# _add_host_key_to_frontmatter
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestInsertHostKeyInFrontmatter(unittest.TestCase):
|
||||
_FM = """\
|
||||
git-gate:
|
||||
repos:
|
||||
myrepo:
|
||||
url: ssh://git@host/org/repo.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
"""
|
||||
|
||||
def test_inserts_host_key_after_last_child(self) -> None:
|
||||
result = _insert_host_key_in_frontmatter(self._FM, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertIn(" host_key: ssh-ed25519 AAAA", result)
|
||||
|
||||
def test_no_change_when_repo_not_found(self) -> None:
|
||||
result = _insert_host_key_in_frontmatter(self._FM, "other", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(self._FM, result)
|
||||
|
||||
def test_no_change_when_host_key_already_present(self) -> None:
|
||||
fm = self._FM + " host_key: ssh-ed25519 EXISTING\n"
|
||||
result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 NEW")
|
||||
self.assertNotIn("NEW", result)
|
||||
self.assertEqual(fm, result)
|
||||
|
||||
def test_preserves_other_repos(self) -> None:
|
||||
fm = """\
|
||||
git-gate:
|
||||
repos:
|
||||
first:
|
||||
url: ssh://git@host/org/first.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
second:
|
||||
url: ssh://git@host/org/second.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
"""
|
||||
result = _insert_host_key_in_frontmatter(fm, "first", "ssh-ed25519 AAAA")
|
||||
self.assertIn("host_key: ssh-ed25519 AAAA", result)
|
||||
self.assertIn("second:", result)
|
||||
|
||||
def test_detects_child_indent_from_existing_keys(self) -> None:
|
||||
# 4-space indent (non-standard but valid)
|
||||
fm = """\
|
||||
git-gate:
|
||||
repos:
|
||||
myrepo:
|
||||
url: ssh://git@host/org/repo.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
"""
|
||||
result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertIn(" host_key: ssh-ed25519 AAAA", result)
|
||||
|
||||
def test_blank_lines_inside_block_do_not_end_scan(self) -> None:
|
||||
fm = """\
|
||||
git-gate:
|
||||
repos:
|
||||
myrepo:
|
||||
url: ssh://git@host/org/repo.git
|
||||
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
"""
|
||||
result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertIn("host_key: ssh-ed25519 AAAA", result)
|
||||
|
||||
def test_repo_with_no_children_uses_default_indent(self) -> None:
|
||||
# Edge case: repo entry has no child keys at all (empty block).
|
||||
fm = "git-gate:\n repos:\n myrepo:\n"
|
||||
result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
|
||||
# Should fall back to repo_indent+2 = 4
|
||||
self.assertIn(" host_key: ssh-ed25519 AAAA", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _update_frontmatter_in_file
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUpdateFrontmatterInFile(unittest.TestCase):
|
||||
class TestAddHostKeyToFrontmatter(unittest.TestCase):
|
||||
_FILE = """\
|
||||
---
|
||||
git-gate:
|
||||
@@ -248,23 +162,78 @@ git-gate:
|
||||
# Body text here
|
||||
"""
|
||||
|
||||
def test_updates_frontmatter_preserves_body(self) -> None:
|
||||
result = _update_frontmatter_in_file(self._FILE, "myrepo", "ssh-ed25519 AAAA")
|
||||
def test_inserts_host_key_at_repo_level(self) -> None:
|
||||
result = _add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertIn("host_key: ssh-ed25519 AAAA", result)
|
||||
# Must be a sibling of `url` and `key`, not nested inside `key`
|
||||
lines = result.splitlines()
|
||||
hk_line = next(l for l in lines if "host_key" in l)
|
||||
url_line = next(l for l in lines if "url:" in l)
|
||||
self.assertEqual(
|
||||
len(hk_line) - len(hk_line.lstrip()),
|
||||
len(url_line) - len(url_line.lstrip()),
|
||||
)
|
||||
|
||||
def test_preserves_body(self) -> None:
|
||||
result = _add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertIn("# Body text here", result)
|
||||
|
||||
def test_preserves_other_repos(self) -> None:
|
||||
text = """\
|
||||
---
|
||||
git-gate:
|
||||
repos:
|
||||
first:
|
||||
url: ssh://git@host/org/first.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
second:
|
||||
url: ssh://git@host/org/second.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
---
|
||||
"""
|
||||
result = _add_host_key_to_frontmatter(text, "first", "ssh-ed25519 AAAA")
|
||||
self.assertIn("host_key: ssh-ed25519 AAAA", result)
|
||||
self.assertIn("second:", result)
|
||||
|
||||
def test_no_change_when_host_key_already_present(self) -> None:
|
||||
text = self._FILE.replace(
|
||||
" path: /dev/null\n",
|
||||
" path: /dev/null\n host_key: ssh-ed25519 EXISTING\n",
|
||||
)
|
||||
result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 NEW")
|
||||
self.assertNotIn("NEW", result)
|
||||
|
||||
def test_no_change_when_repo_not_found(self) -> None:
|
||||
result = _add_host_key_to_frontmatter(self._FILE, "other", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(self._FILE, result)
|
||||
|
||||
def test_no_change_without_frontmatter(self) -> None:
|
||||
text = "No frontmatter here\n"
|
||||
result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(text, result)
|
||||
|
||||
def test_no_change_when_repo_not_in_file(self) -> None:
|
||||
result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(self._FILE, result)
|
||||
|
||||
def test_no_change_when_no_closing_delimiter(self) -> None:
|
||||
text = "---\ngit-gate:\n repos:\n myrepo:\n url: x\n"
|
||||
result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(text, result)
|
||||
|
||||
def test_no_change_when_git_gate_missing(self) -> None:
|
||||
text = "---\nenv:\n FOO: bar\n---\n"
|
||||
result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(text, result)
|
||||
|
||||
def test_no_change_when_repos_missing(self) -> None:
|
||||
text = "---\ngit-gate:\n foo: bar\n---\n"
|
||||
result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(text, result)
|
||||
|
||||
def test_no_change_when_repo_entry_not_a_dict(self) -> None:
|
||||
text = "---\ngit-gate:\n repos:\n - item\n---\n"
|
||||
result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
|
||||
self.assertEqual(text, result)
|
||||
|
||||
|
||||
@@ -372,9 +341,9 @@ git-gate:
|
||||
os.chmod(path, 0o644)
|
||||
|
||||
def test_skips_when_update_produces_no_change(self) -> None:
|
||||
# _update_frontmatter_in_file returns the original text unchanged
|
||||
# (e.g. because _insert_host_key_in_frontmatter can't locate the
|
||||
# entry in the raw text). Line 174 must `continue` rather than write.
|
||||
# _add_host_key_to_frontmatter returns the original text unchanged
|
||||
# (e.g. because the entry is absent from the parsed structure).
|
||||
# _find_and_update_bottle_file must continue rather than write.
|
||||
content = """\
|
||||
---
|
||||
git-gate:
|
||||
@@ -389,7 +358,7 @@ git-gate:
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
path = self._write_bottle(d, "dev", content)
|
||||
with patch(
|
||||
"bot_bottle.git_gate_host_key._update_frontmatter_in_file",
|
||||
"bot_bottle.git_gate_host_key._add_host_key_to_frontmatter",
|
||||
return_value=content, # no change
|
||||
):
|
||||
result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
|
||||
|
||||
@@ -6,7 +6,7 @@ import textwrap
|
||||
import unittest
|
||||
|
||||
from bot_bottle.yaml_subset import YamlSubsetError
|
||||
from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset
|
||||
from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset, serialize_yaml_subset
|
||||
|
||||
|
||||
def _y(s: str):
|
||||
@@ -457,5 +457,141 @@ class TestEdgeAndErrorBranches(unittest.TestCase):
|
||||
self.assertEqual(({}, ""), parse_frontmatter(""))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# serialize_yaml_subset
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSerializeYamlSubset(unittest.TestCase):
|
||||
def _roundtrip(self, text: str) -> None:
|
||||
"""Parse then serialize then re-parse; result must equal original parse."""
|
||||
parsed = parse_yaml_subset(text)
|
||||
serialized = serialize_yaml_subset(parsed)
|
||||
self.assertEqual(parsed, parse_yaml_subset(serialized))
|
||||
|
||||
# --- scalar values ---------------------------------------------------------
|
||||
|
||||
def test_string_bare(self) -> None:
|
||||
self.assertEqual("key: value\n", serialize_yaml_subset({"key": "value"}))
|
||||
|
||||
def test_string_empty_is_quoted(self) -> None:
|
||||
out = serialize_yaml_subset({"key": ""})
|
||||
self.assertIn("key: ''", out)
|
||||
self._roundtrip("key: ''\n")
|
||||
|
||||
def test_string_true_like_is_quoted(self) -> None:
|
||||
out = serialize_yaml_subset({"key": "true"})
|
||||
self.assertIn("'true'", out)
|
||||
self.assertEqual({"key": "true"}, parse_yaml_subset(out))
|
||||
|
||||
def test_string_int_like_is_quoted(self) -> None:
|
||||
out = serialize_yaml_subset({"key": "42"})
|
||||
self.assertIn("'42'", out)
|
||||
self.assertEqual({"key": "42"}, parse_yaml_subset(out))
|
||||
|
||||
def test_string_special_start_char_is_quoted(self) -> None:
|
||||
for ch in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">"):
|
||||
out = serialize_yaml_subset({"key": ch + "rest"})
|
||||
self.assertIn("'", out, msg=f"expected quoting for {ch!r}")
|
||||
|
||||
def test_none_emits_null(self) -> None:
|
||||
self.assertEqual("key: null\n", serialize_yaml_subset({"key": None}))
|
||||
|
||||
def test_bool_true(self) -> None:
|
||||
self.assertEqual("key: true\n", serialize_yaml_subset({"key": True}))
|
||||
|
||||
def test_bool_false(self) -> None:
|
||||
self.assertEqual("key: false\n", serialize_yaml_subset({"key": False}))
|
||||
|
||||
def test_int(self) -> None:
|
||||
self.assertEqual("key: 42\n", serialize_yaml_subset({"key": 42}))
|
||||
|
||||
# --- nested dict -----------------------------------------------------------
|
||||
|
||||
def test_nested_dict(self) -> None:
|
||||
data: dict[str, object] = {"outer": {"inner": "val"}}
|
||||
out = serialize_yaml_subset(data)
|
||||
self.assertIn("outer:", out)
|
||||
self.assertIn(" inner: val", out)
|
||||
self.assertEqual(data, parse_yaml_subset(out))
|
||||
|
||||
def test_empty_dict_value_inline(self) -> None:
|
||||
out = serialize_yaml_subset({"key": {}})
|
||||
self.assertIn("key: {}", out)
|
||||
|
||||
# --- lists -----------------------------------------------------------------
|
||||
|
||||
def test_list_of_scalars(self) -> None:
|
||||
data: dict[str, object] = {"items": ["a", "b", "c"]}
|
||||
out = serialize_yaml_subset(data)
|
||||
self.assertEqual(data, parse_yaml_subset(out))
|
||||
|
||||
def test_empty_list_value_inline(self) -> None:
|
||||
out = serialize_yaml_subset({"key": []})
|
||||
self.assertIn("key: []", out)
|
||||
|
||||
def test_list_of_mappings(self) -> None:
|
||||
data: dict[str, object] = {"items": [{"name": "alpha", "val": 1}, {"name": "beta", "val": 2}]}
|
||||
out = serialize_yaml_subset(data)
|
||||
self.assertEqual(data, parse_yaml_subset(out))
|
||||
|
||||
def test_list_item_mapping_with_nested_dict(self) -> None:
|
||||
data: dict[str, object] = {"items": [{"name": "x", "sub": {"k": "v"}}]}
|
||||
out = serialize_yaml_subset(data)
|
||||
self.assertEqual(data, parse_yaml_subset(out))
|
||||
|
||||
def test_list_item_first_key_is_nested_dict(self) -> None:
|
||||
# Covers the branch where the FIRST key of a mapping list item
|
||||
# has a nested dict/list value (not a scalar).
|
||||
data: dict[str, object] = {"items": [{"cfg": {"a": 1, "b": 2}, "name": "x"}]}
|
||||
out = serialize_yaml_subset(data)
|
||||
self.assertEqual(data, parse_yaml_subset(out))
|
||||
|
||||
def test_list_item_mapping_continuation_keys(self) -> None:
|
||||
data: dict[str, object] = {"items": [{"a": 1, "b": 2, "c": 3}]}
|
||||
out = serialize_yaml_subset(data)
|
||||
self.assertEqual(data, parse_yaml_subset(out))
|
||||
|
||||
# --- empty / trivial -------------------------------------------------------
|
||||
|
||||
def test_empty_dict(self) -> None:
|
||||
self.assertEqual("", serialize_yaml_subset({}))
|
||||
|
||||
# --- realistic bottle frontmatter round-trip --------------------------------
|
||||
|
||||
def test_bottle_frontmatter_round_trip(self) -> None:
|
||||
text = textwrap.dedent("""\
|
||||
git-gate:
|
||||
repos:
|
||||
myrepo:
|
||||
url: ssh://git@host/org/repo.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
""")
|
||||
self._roundtrip(text)
|
||||
|
||||
def test_host_key_inserted_at_repo_level(self) -> None:
|
||||
"""The canonical use-case: host_key lands as a sibling of url/key."""
|
||||
from typing import cast
|
||||
text = textwrap.dedent("""\
|
||||
git-gate:
|
||||
repos:
|
||||
myrepo:
|
||||
url: ssh://git@host/org/repo.git
|
||||
key:
|
||||
provider: static
|
||||
path: /dev/null
|
||||
""")
|
||||
data = parse_yaml_subset(text)
|
||||
repos = cast(dict[str, object], cast(dict[str, object], data["git-gate"])["repos"])
|
||||
cast(dict[str, object], repos["myrepo"])["host_key"] = "ssh-ed25519 AAAA"
|
||||
out = serialize_yaml_subset(data)
|
||||
parsed_back = parse_yaml_subset(out)
|
||||
repo = cast(dict[str, object], cast(dict[str, object], cast(dict[str, object], parsed_back["git-gate"])["repos"])["myrepo"])
|
||||
self.assertEqual("ssh-ed25519 AAAA", repo["host_key"])
|
||||
self.assertIn("provider", cast(dict[str, object], repo["key"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user
This is incorrect (
host_keyended up being persisted undergit-gate.repos.bot-bottle.keyinstead ofgit-gate.repos.bot-bottle), and the implementation is overly complicated: should instead parse the yaml into a nested dict, insert thehost_keyin the correct location in the nested dict, and then serialize it as yaml again.Confirmed — the bug is that
child_indentis overwritten on every line withindent > repo_indent, including grandchildren, so it ends up at the deepest nesting level (key.provider/key.pathindent) rather than the direct child level. That causeshost_keyto land inside thekey:block.Will replace
_insert_host_key_in_frontmatter(and_update_frontmatter_in_file) with a parse → mutate → re-serialize approach: useyaml_subset.parse_frontmatter()to get the dict, setfm["git-gate"]["repos"][repo_name]["host_key"], then serialize back with a newserialize_yaml_subset()added toyaml_subset.py. Since the project is stdlib-only, the serializer will handle the bounded shape (nested dicts and string scalars) that bot-bottle configs use.