refactor: consolidate read_tty_line into bot_bottle/util.py
tracker-policy-pr / check-pr (pull_request) Successful in 10s
test / integration-docker (pull_request) Successful in 26s
test / unit (pull_request) Successful in 36s
lint / lint (push) Failing after 42s
test / stage-firecracker-inputs (pull_request) Successful in 2s
test / build-infra (pull_request) Successful in 3m57s
test / integration-firecracker (pull_request) Successful in 1m55s
test / coverage (pull_request) Successful in 2m14s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 10s
test / integration-docker (pull_request) Successful in 26s
test / unit (pull_request) Successful in 36s
lint / lint (push) Failing after 42s
test / stage-firecracker-inputs (pull_request) Successful in 2s
test / build-infra (pull_request) Successful in 3m57s
test / integration-firecracker (pull_request) Successful in 1m55s
test / coverage (pull_request) Successful in 2m14s
test / publish-infra (pull_request) Has been skipped
Remove the private _read_tty_line duplicate from backend/__init__.py and the local definition from cli/_common.py. Both now import from the shared bot_bottle.util module.
This commit is contained in:
@@ -46,6 +46,7 @@ from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provi
|
||||
from ..egress import EgressPlan
|
||||
from ..git_gate import GitGatePlan
|
||||
from ..log import die, info, warn
|
||||
from ..util import read_tty_line
|
||||
from ..manifest import Manifest, ManifestIndex
|
||||
from ..supervise import SupervisePlan
|
||||
from ..util import expand_tilde
|
||||
@@ -668,15 +669,6 @@ def get_bottle_backend(
|
||||
return backends[resolved]
|
||||
|
||||
|
||||
def _read_tty_line() -> str:
|
||||
"""Read a line from /dev/tty, falling back to stdin."""
|
||||
try:
|
||||
with open("/dev/tty", "r", encoding="utf-8") as tty:
|
||||
return tty.readline().rstrip("\n")
|
||||
except OSError:
|
||||
return sys.stdin.readline().rstrip("\n")
|
||||
|
||||
|
||||
def _platform_vm_suggestion() -> str:
|
||||
"""Platform-appropriate VM backend name for install suggestions."""
|
||||
return "macos-container" if sys.platform == "darwin" else "firecracker"
|
||||
@@ -733,7 +725,7 @@ def _auto_select_backend() -> str:
|
||||
"bot-bottle: choice [i/d/q]: "
|
||||
)
|
||||
sys.stderr.flush()
|
||||
reply = _read_tty_line().strip().lower()
|
||||
reply = read_tty_line().strip().lower()
|
||||
if reply == "d":
|
||||
return "docker"
|
||||
if reply == "i":
|
||||
|
||||
@@ -3,18 +3,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from ..util import read_tty_line
|
||||
|
||||
PROG = "cli.py"
|
||||
USER_CWD = os.getcwd()
|
||||
REPO_DIR = str(Path(__file__).resolve().parent.parent.parent)
|
||||
|
||||
|
||||
def read_tty_line() -> str:
|
||||
"""Mirror `IFS= read -r REPLY </dev/tty`. Falls back to stdin."""
|
||||
try:
|
||||
with open("/dev/tty", "r", encoding="utf-8") as tty:
|
||||
return tty.readline().rstrip("\n")
|
||||
except OSError:
|
||||
return sys.stdin.readline().rstrip("\n")
|
||||
|
||||
@@ -7,6 +7,7 @@ from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def is_ip_literal(value: str) -> bool:
|
||||
@@ -17,6 +18,15 @@ def is_ip_literal(value: str) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def read_tty_line() -> str:
|
||||
"""Mirror `IFS= read -r REPLY </dev/tty`. Falls back to stdin."""
|
||||
try:
|
||||
with open("/dev/tty", "r", encoding="utf-8") as tty:
|
||||
return tty.readline().rstrip("\n")
|
||||
except OSError:
|
||||
return sys.stdin.readline().rstrip("\n")
|
||||
|
||||
|
||||
def expand_tilde(path: str) -> str:
|
||||
"""Expand a leading '~' to $HOME. Leaves paths without a leading
|
||||
tilde unchanged. Falls back to the empty string if $HOME is unset
|
||||
|
||||
@@ -66,7 +66,7 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
"macos-container": _FakeBackend("macos-container", False),
|
||||
"docker": _FakeBackend("docker", True),
|
||||
}), \
|
||||
patch.object(backend_mod, "_read_tty_line", return_value="d"):
|
||||
patch.object(backend_mod, "read_tty_line", return_value="d"):
|
||||
b = get_bottle_backend()
|
||||
self.assertEqual("docker", b.name)
|
||||
|
||||
@@ -136,7 +136,7 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
"macos-container": _FakeBackend("macos-container", False),
|
||||
"docker": _FakeBackend("docker", True),
|
||||
}), \
|
||||
patch.object(backend_mod, "_read_tty_line", return_value="q"), \
|
||||
patch.object(backend_mod, "read_tty_line", return_value="q"), \
|
||||
patch.object(backend_mod, "die", side_effect=SystemExit("die")):
|
||||
with self.assertRaises(SystemExit):
|
||||
get_bottle_backend()
|
||||
@@ -159,7 +159,7 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
"macos-container": _FakeBackend("macos-container", False),
|
||||
"docker": _FakeBackend("docker", True),
|
||||
}), \
|
||||
patch.object(backend_mod, "_read_tty_line", return_value="i"), \
|
||||
patch.object(backend_mod, "read_tty_line", return_value="i"), \
|
||||
patch.object(backend_mod, "_print_vm_install_instructions") as mock_inst, \
|
||||
patch.object(backend_mod, "die", side_effect=SystemExit("die")):
|
||||
with self.assertRaises(SystemExit):
|
||||
@@ -168,26 +168,26 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
|
||||
|
||||
class TestReadTtyLine(unittest.TestCase):
|
||||
"""Unit tests for the _read_tty_line helper."""
|
||||
"""Unit tests for the shared read_tty_line helper in bot_bottle.util."""
|
||||
|
||||
def test_reads_from_dev_tty(self):
|
||||
from unittest.mock import mock_open
|
||||
from bot_bottle.backend import _read_tty_line
|
||||
from bot_bottle.util import read_tty_line
|
||||
|
||||
m = mock_open(read_data="hello\n")
|
||||
with patch("builtins.open", m):
|
||||
result = _read_tty_line()
|
||||
result = read_tty_line()
|
||||
self.assertEqual("hello", result)
|
||||
m.assert_called_once_with("/dev/tty", "r", encoding="utf-8")
|
||||
|
||||
def test_falls_back_to_stdin_on_oserror(self):
|
||||
import io
|
||||
from bot_bottle.backend import _read_tty_line
|
||||
from bot_bottle.util import read_tty_line
|
||||
import sys as _sys
|
||||
|
||||
with patch("builtins.open", side_effect=OSError("no tty")), \
|
||||
patch.object(_sys, "stdin", io.StringIO("world\n")):
|
||||
result = _read_tty_line()
|
||||
result = read_tty_line()
|
||||
self.assertEqual("world", result)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user