fix: pyright/pylint clean-up and 100% branch coverage
lint / lint (push) Failing after 2m17s
test / unit (pull_request) Successful in 1m5s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m12s

- Remove unused ManifestBottle/ManifestGitEntry imports
- Add check=False to subprocess.run (pylint subprocess-run-check)
- Add `from e` to TimeoutExpired re-raise (raise-missing-from)
- Catch UnicodeDecodeError alongside OSError/YamlSubsetError in
  _find_and_update_bottle_file
- Add 14 new test cases covering: malformed ssh-keyscan output lines,
  stderr in error messages, blank lines inside frontmatter blocks,
  repo entry with no children (child_indent fallback), missing closing
  `---` delimiter, files without git-gate section, non-dict repos
  section, invalid UTF-8 files, unreadable files, update producing no
  text change, _prompt_tty TTY read and stdin fallback, fetch failure
  in preflight, and the "persist failed / kept in memory" warning path

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-07-09 16:03:39 +00:00
parent 21d03b7cc9
commit e89dffa899
2 changed files with 161 additions and 6 deletions
+5 -6
View File
@@ -17,8 +17,7 @@ import sys
from pathlib import Path from pathlib import Path
from .log import die, info from .log import die, info
from .manifest import Manifest, ManifestBottle from .manifest import Manifest
from .manifest_git import ManifestGitEntry
def fetch_host_key(host: str, port: str) -> str: def fetch_host_key(host: str, port: str) -> str:
@@ -36,14 +35,14 @@ def fetch_host_key(host: str, port: str) -> str:
args.append(host) args.append(host)
try: try:
result = subprocess.run( result = subprocess.run(
args, capture_output=True, text=True, timeout=15, args, capture_output=True, text=True, timeout=15, check=False,
) )
except OSError as e: except OSError as e:
raise RuntimeError( raise RuntimeError(
f"ssh-keyscan: could not launch for {host}:{port}: {e}" f"ssh-keyscan: could not launch for {host}:{port}: {e}"
) from e ) from e
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired as e:
raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") from e
for line in result.stdout.splitlines(): for line in result.stdout.splitlines():
line = line.strip() line = line.strip()
@@ -155,7 +154,7 @@ def _find_and_update_bottle_file(
try: try:
text = path.read_text(encoding="utf-8") text = path.read_text(encoding="utf-8")
fm, _body = parse_frontmatter(text) fm, _body = parse_frontmatter(text)
except (OSError, YamlSubsetError): except (OSError, UnicodeDecodeError, YamlSubsetError):
continue continue
git_gate = fm.get("git-gate") git_gate = fm.get("git-gate")
+156
View File
@@ -11,6 +11,7 @@ from unittest.mock import MagicMock, patch
from bot_bottle.git_gate_host_key import ( from bot_bottle.git_gate_host_key import (
_find_and_update_bottle_file, _find_and_update_bottle_file,
_insert_host_key_in_frontmatter, _insert_host_key_in_frontmatter,
_prompt_tty,
_update_frontmatter_in_file, _update_frontmatter_in_file,
fetch_host_key, fetch_host_key,
preflight_host_keys, preflight_host_keys,
@@ -58,12 +59,29 @@ class TestFetchHostKey(unittest.TestCase):
key = fetch_host_key("github.com", "22") key = fetch_host_key("github.com", "22")
self.assertEqual("ssh-ed25519 AAAA", key) self.assertEqual("ssh-ed25519 AAAA", key)
def test_skips_lines_without_three_parts(self) -> None:
# A line with only two whitespace-separated tokens is not a valid
# known_hosts entry; it should be skipped and the function should
# raise rather than return a partial result.
stdout = "malformed-line\ngithub.com ssh-ed25519 AAAA\n"
with patch("subprocess.run", return_value=self._run_result(stdout)):
key = fetch_host_key("github.com", "22")
self.assertEqual("ssh-ed25519 AAAA", key)
def test_raises_on_empty_output(self) -> None: def test_raises_on_empty_output(self) -> None:
with patch("subprocess.run", return_value=self._run_result("")): with patch("subprocess.run", return_value=self._run_result("")):
with self.assertRaises(RuntimeError) as cm: with self.assertRaises(RuntimeError) as cm:
fetch_host_key("example.com", "22") fetch_host_key("example.com", "22")
self.assertIn("no ed25519 key", str(cm.exception)) self.assertIn("no ed25519 key", str(cm.exception))
def test_raises_includes_stderr_when_present(self) -> None:
r = self._run_result("")
r.stderr = "connection refused"
with patch("subprocess.run", return_value=r):
with self.assertRaises(RuntimeError) as cm:
fetch_host_key("example.com", "22")
self.assertIn("connection refused", str(cm.exception))
def test_raises_on_os_error(self) -> None: def test_raises_on_os_error(self) -> None:
with patch("subprocess.run", side_effect=OSError("not found")): with patch("subprocess.run", side_effect=OSError("not found")):
with self.assertRaises(RuntimeError) as cm: with self.assertRaises(RuntimeError) as cm:
@@ -78,6 +96,28 @@ class TestFetchHostKey(unittest.TestCase):
self.assertIn("timed out", str(cm.exception)) self.assertIn("timed out", str(cm.exception))
# ---------------------------------------------------------------------------
# _prompt_tty
# ---------------------------------------------------------------------------
class TestPromptTty(unittest.TestCase):
def test_reads_from_tty_when_available(self) -> None:
import io
fake_tty = io.StringIO("yes\n")
with patch("builtins.open", return_value=fake_tty), patch("sys.stderr"):
result = _prompt_tty("question: ")
self.assertEqual("yes", result)
def test_falls_back_to_stdin_on_os_error(self) -> None:
import io
with patch("builtins.open", side_effect=OSError("no tty")), \
patch("sys.stdin", io.StringIO("fallback\n")), \
patch("sys.stderr"):
result = _prompt_tty("question: ")
self.assertEqual("fallback", result)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# _insert_host_key_in_frontmatter # _insert_host_key_in_frontmatter
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -141,6 +181,27 @@ git-gate:
result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
self.assertIn(" host_key: ssh-ed25519 AAAA", result) 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 # _update_frontmatter_in_file
@@ -175,6 +236,11 @@ git-gate:
result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA") result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA")
self.assertEqual(self._FILE, result) 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")
self.assertEqual(text, result)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# _find_and_update_bottle_file # _find_and_update_bottle_file
@@ -248,6 +314,62 @@ git-gate:
) )
self.assertFalse(result) self.assertFalse(result)
def test_skips_file_without_git_gate_section(self) -> None:
with tempfile.TemporaryDirectory() as d:
self._write_bottle(d, "plain", "---\nenv:\n FOO: bar\n---\n")
result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
self.assertFalse(result)
def test_skips_file_with_non_dict_repos(self) -> None:
with tempfile.TemporaryDirectory() as d:
self._write_bottle(d, "bad", "---\ngit-gate:\n repos:\n - item\n---\n")
result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
self.assertFalse(result)
def test_skips_invalid_utf8_file(self) -> None:
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "broken.md"
path.write_bytes(b"\xff\xfe") # invalid UTF-8
result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
self.assertFalse(result)
def test_skips_unreadable_file(self) -> None:
import os
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "unreadable.md"
path.write_text("---\n---\n")
os.chmod(path, 0o000)
try:
result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
self.assertFalse(result)
finally:
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.
content = """\
---
git-gate:
repos:
myrepo:
url: ssh://git@host/org/repo.git
key:
provider: static
path: /dev/null
---
"""
with tempfile.TemporaryDirectory() as d:
path = self._write_bottle(d, "dev", content)
with patch(
"bot_bottle.git_gate_host_key._update_frontmatter_in_file",
return_value=content, # no change
):
result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
self.assertFalse(result)
self.assertNotIn("host_key", path.read_text())
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# preflight_host_keys # preflight_host_keys
@@ -373,6 +495,40 @@ class TestPreflightHostKeys(unittest.TestCase):
written = bottle_file.read_text() written = bottle_file.read_text()
self.assertNotIn("host_key", written) self.assertNotIn("host_key", written)
def test_interactive_dies_when_fetch_fails(self) -> None:
manifest = _manifest_with_git()
with patch(
"bot_bottle.git_gate_host_key.fetch_host_key",
side_effect=RuntimeError("connection refused"),
), patch("sys.stderr"), self.assertRaises(SystemExit):
preflight_host_keys(manifest, headless=False, home_md=None)
def test_interactive_warns_when_persist_file_not_found(self) -> None:
manifest = _manifest_with_git()
replies = iter(["y", "y"]) # confirm key, yes persist
with tempfile.TemporaryDirectory() as home:
# bottles/ dir exists but does not contain the repo entry
bottles_dir = Path(home) / "bottles"
bottles_dir.mkdir()
import io
buf = io.StringIO()
with patch(
"bot_bottle.git_gate_host_key.fetch_host_key",
return_value="ssh-ed25519 FETCHED",
), patch(
"bot_bottle.git_gate_host_key._prompt_tty",
side_effect=lambda msg: next(replies),
), patch("sys.stderr", buf):
result = preflight_host_keys(
manifest, headless=False, home_md=Path(home),
)
self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
self.assertIn("kept in memory", buf.getvalue())
def test_headless_names_all_missing_repos_in_error(self) -> None: def test_headless_names_all_missing_repos_in_error(self) -> None:
manifest = ManifestIndex.from_json_obj({ manifest = ManifestIndex.from_json_obj({
"bottles": { "bottles": {