fix: confirm SSH alias updates and keep key handling consistent
This commit is contained in:
@@ -0,0 +1,488 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Isolated ff ssh regressions: dummy keys and mock commands, no network access."""
|
||||
|
||||
import fcntl
|
||||
import json
|
||||
import os
|
||||
import pty
|
||||
import selectors
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import subprocess
|
||||
import tempfile
|
||||
import termios
|
||||
import time
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
REAL_SSH = shutil.which("ssh")
|
||||
SOURCE = Path(os.environ.get("FF_SSH_TEST_SOURCE", ROOT / "fish/functions/ff-ssh.fish"))
|
||||
MOCK = r'''#!/usr/bin/env python3
|
||||
import json, os, pathlib, signal, sys, time
|
||||
home = pathlib.Path(os.environ["HOME"])
|
||||
name = pathlib.Path(sys.argv[0]).name
|
||||
with (home / "calls.jsonl").open("a") as f:
|
||||
f.write(json.dumps([name, sys.argv[1:]]) + "\n")
|
||||
if name == "ssh":
|
||||
if "-G" in sys.argv:
|
||||
print((home / "resolved").read_text(), end="")
|
||||
sys.exit(int(os.environ.get("RESOLVE_STATUS", "0")))
|
||||
sys.exit(int(os.environ.get("VERIFY_STATUS", "0")))
|
||||
if name == "ssh-keygen":
|
||||
if os.environ.get("KEYGEN_FAIL"):
|
||||
sys.exit(1)
|
||||
key = pathlib.Path(sys.argv[sys.argv.index("-f") + 1])
|
||||
key.write_text("dummy private\n")
|
||||
key.with_name(key.name + ".pub").write_text("dummy public\n")
|
||||
sys.exit(0)
|
||||
mode = os.environ.get("COPY_MODE", "success")
|
||||
if mode == "cancel-zero":
|
||||
def cancel(*_):
|
||||
print("ERROR: Assertion failure: in filter_ids(): scratch_dir was not correctly set up", file=sys.stderr)
|
||||
print("WARNING: All keys were skipped because they already exist", file=sys.stderr)
|
||||
sys.exit(0)
|
||||
signal.signal(signal.SIGINT, cancel)
|
||||
(home / "ready").touch()
|
||||
time.sleep(15)
|
||||
if mode == "false-zero":
|
||||
print("ERROR: Assertion failure: in filter_ids(): scratch_dir was not correctly set up", file=sys.stderr)
|
||||
print("WARNING: All keys were skipped because they already exist", file=sys.stderr)
|
||||
if mode in ("failure", "timeout"):
|
||||
print("Connection timed out" if mode == "timeout" else "Permission denied", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
'''
|
||||
|
||||
|
||||
class SshTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory(prefix="ff-ssh-test-")
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.home = Path(self.temp.name)
|
||||
self.ssh = self.home / ".ssh"
|
||||
self.ssh.mkdir()
|
||||
bindir = self.home / "bin"
|
||||
bindir.mkdir()
|
||||
for name in ("ssh", "ssh-copy-id", "ssh-keygen"):
|
||||
path = bindir / name
|
||||
path.write_text(MOCK)
|
||||
path.chmod(0o755)
|
||||
self.env = dict(os.environ, HOME=str(self.home),
|
||||
XDG_CONFIG_HOME=str(self.home / ".config"),
|
||||
PATH=f"{bindir}:{os.environ['PATH']}", USER="test")
|
||||
self.resolve("~/.ssh/id_rsa", "~/.ssh/id_ed25519")
|
||||
|
||||
def resolve(self, *identities):
|
||||
(self.home / "resolved").write_text(
|
||||
"hostname machine.invalid\nuser testuser\nport 2222\nproxyjump none\nproxycommand none\n" +
|
||||
"".join(f"identityfile {key}\n" for key in identities))
|
||||
|
||||
def key(self, name):
|
||||
private = self.ssh / name
|
||||
private.write_text("dummy private\n")
|
||||
private.with_name(name + ".pub").write_text("dummy public\n")
|
||||
return private
|
||||
|
||||
def launch(self, args):
|
||||
return subprocess.Popen(
|
||||
["fish", "--no-config", "-c", 'source "$argv[1]"; ff_ssh $argv[2..-1]',
|
||||
str(SOURCE), *args], env=self.env, stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True)
|
||||
|
||||
def run_ff(self, *args):
|
||||
proc = self.launch(args)
|
||||
try:
|
||||
out, err = proc.communicate(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
os.killpg(proc.pid, signal.SIGKILL)
|
||||
proc.communicate()
|
||||
raise
|
||||
return proc.returncode, out + err
|
||||
|
||||
def calls(self, name):
|
||||
path = self.home / "calls.jsonl"
|
||||
return [args for cmd, args in map(json.loads, path.read_text().splitlines()) if cmd == name] if path.exists() else []
|
||||
|
||||
def assert_selected(self, private):
|
||||
copy = self.calls("ssh-copy-id")
|
||||
self.assertEqual(len(copy), 1)
|
||||
chosen = copy[0][copy[0].index("-i") + 1]
|
||||
self.assertEqual(str(private), chosen.removesuffix(".pub"))
|
||||
config = (self.ssh / "config.d/machine.invalid.conf").read_text()
|
||||
identities = [line.strip().split(maxsplit=1)[1].strip('"') for line in config.splitlines() if line.strip().startswith("IdentityFile ")]
|
||||
self.assertEqual(identities, [str(private).replace(str(self.home), "~", 1)])
|
||||
|
||||
def assert_no_host(self):
|
||||
self.assertFalse((self.ssh / "config.d/machine.invalid.conf").exists())
|
||||
|
||||
def test_default_matches_resolved_order_with_multiple_keys(self):
|
||||
selected = self.key("id_rsa")
|
||||
self.key("id_ed25519")
|
||||
status, output = self.run_ff("testuser@machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_selected(selected)
|
||||
|
||||
def test_nonstandard_default_key(self):
|
||||
selected = self.key("id_hx_new")
|
||||
status, output = self.run_ff("testuser@machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_selected(selected)
|
||||
|
||||
def test_explicit_private_and_public_paths(self):
|
||||
for suffix in ("", ".pub"):
|
||||
with self.subTest(suffix=suffix):
|
||||
self.setUp()
|
||||
self.key("id_rsa")
|
||||
selected = self.key("custom")
|
||||
self.resolve(str(selected) + suffix, "~/.ssh/id_rsa")
|
||||
status, output = self.run_ff("-i", str(selected) + suffix, "testuser@machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_selected(selected)
|
||||
|
||||
def test_identity_option_case_and_forms(self):
|
||||
for option in ("IdentityFile=", "identityfile "):
|
||||
with self.subTest(option=option):
|
||||
self.setUp()
|
||||
self.key("id_rsa")
|
||||
selected = self.key("custom")
|
||||
self.resolve(str(selected), "~/.ssh/id_rsa")
|
||||
status, output = self.run_ff("-o", option + str(selected), "testuser@machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_selected(selected)
|
||||
|
||||
def test_existing_config_identity(self):
|
||||
selected = self.key("custom")
|
||||
self.key("id_rsa")
|
||||
self.resolve(str(selected))
|
||||
status, output = self.run_ff("testuser@machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_selected(selected)
|
||||
|
||||
def test_invalid_explicit_identity_never_falls_back(self):
|
||||
self.key("id_rsa")
|
||||
# OpenSSH omits an unreadable -i from ssh -G, leaving defaults.
|
||||
self.resolve("~/.ssh/id_rsa")
|
||||
status, output = self.run_ff("-i", "/missing/key", "testuser@machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assert_no_host()
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
|
||||
def test_unavailable_config_identity_never_falls_back(self):
|
||||
self.key("id_rsa")
|
||||
self.resolve("~/.ssh/custom_missing")
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assert_no_host()
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
|
||||
def test_unpaired_default_key_is_not_overwritten(self):
|
||||
path = self.ssh / "id_ed25519"
|
||||
path.write_text("existing private")
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertEqual(path.read_text(), "existing private")
|
||||
self.assert_no_host()
|
||||
|
||||
def test_generation_then_copy_same_key(self):
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_selected(self.ssh / "id_ed25519")
|
||||
|
||||
def test_generation_failure_aborts(self):
|
||||
self.env["KEYGEN_FAIL"] = "1"
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assert_no_host()
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
|
||||
def test_copy_failure_and_timeout_abort(self):
|
||||
for mode in ("failure", "timeout"):
|
||||
with self.subTest(mode=mode):
|
||||
self.setUp()
|
||||
self.key("id_rsa")
|
||||
self.env["COPY_MODE"] = mode
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assert_no_host()
|
||||
|
||||
def test_false_success_requires_selected_key_verification(self):
|
||||
self.key("id_rsa")
|
||||
self.env.update(COPY_MODE="false-zero", VERIFY_STATUS="255")
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assert_no_host()
|
||||
|
||||
def test_ctrl_c_with_upstream_exit_zero_aborts(self):
|
||||
self.key("id_rsa")
|
||||
self.env["COPY_MODE"] = "cancel-zero"
|
||||
proc = self.launch(["machine.invalid"])
|
||||
try:
|
||||
deadline = time.monotonic() + 5
|
||||
while not (self.home / "ready").exists() and proc.poll() is None and time.monotonic() < deadline:
|
||||
time.sleep(.02)
|
||||
self.assertTrue((self.home / "ready").exists(), "mock copy must reach cancellable stage")
|
||||
os.killpg(proc.pid, signal.SIGINT)
|
||||
out, err = proc.communicate(timeout=5)
|
||||
self.assertNotEqual(proc.returncode, 0, out + err)
|
||||
self.assert_no_host()
|
||||
finally:
|
||||
if proc.poll() is None:
|
||||
os.killpg(proc.pid, signal.SIGKILL)
|
||||
proc.communicate()
|
||||
|
||||
def test_no_copy_never_generates_or_connects(self):
|
||||
status, output = self.run_ff("--no-copy", "machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assertTrue((self.ssh / "config.d/machine.invalid.conf").exists())
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
self.assertFalse(self.calls("ssh-keygen"))
|
||||
self.assertTrue(all("-G" in call for call in self.calls("ssh")))
|
||||
|
||||
def test_no_copy_public_identity_records_private(self):
|
||||
private = self.key("custom")
|
||||
self.resolve(str(private) + ".pub")
|
||||
status, output = self.run_ff("--no-copy", "-i", str(private) + ".pub", "machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
config = (self.ssh / "config.d/machine.invalid.conf").read_text()
|
||||
self.assertIn("~/.ssh/custom", config)
|
||||
self.assertNotIn("custom.pub", config)
|
||||
|
||||
def existing(self, location="config.d/machine.invalid.conf", shared=False):
|
||||
path = self.ssh / location
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
content = ("# original settings\nHost " + ("machine.invalid neighbor" if shared else "machine.invalid") +
|
||||
"\n HostName old.invalid\n User olduser\n Port 22\n"
|
||||
" IdentityFile ~/.ssh/old\n ForwardAgent no\n"
|
||||
"Host untouched\n HostName untouched.invalid\n")
|
||||
path.write_text(content)
|
||||
if location != "config":
|
||||
(self.ssh / "config").write_text(f"Include {self.ssh}/config.d/*\n")
|
||||
return path, content
|
||||
|
||||
def snapshot(self):
|
||||
return {str(p.relative_to(self.ssh)): p.read_bytes()
|
||||
for p in self.ssh.rglob("*") if p.is_file()}
|
||||
|
||||
def interactive(self, *args, answer="y", cancel_at=None):
|
||||
master, slave = pty.openpty()
|
||||
proc = subprocess.Popen(
|
||||
["fish", "--no-config", "-c", 'source "$argv[1]"; ff_ssh $argv[2..-1]',
|
||||
str(SOURCE), *args], env=self.env, stdin=slave,
|
||||
stdout=slave, stderr=slave, start_new_session=True,
|
||||
preexec_fn=lambda: fcntl.ioctl(slave, termios.TIOCSCTTY, 0))
|
||||
os.close(slave)
|
||||
output = b""
|
||||
selector = selectors.DefaultSelector()
|
||||
selector.register(master, selectors.EVENT_READ)
|
||||
|
||||
def until(prompt):
|
||||
nonlocal output
|
||||
deadline = time.monotonic() + 5
|
||||
while prompt.encode() not in output and time.monotonic() < deadline:
|
||||
for key, _ in selector.select(.1):
|
||||
chunk = os.read(key.fileobj, 65536)
|
||||
if not chunk:
|
||||
self.fail("process ended before prompt: " + output.decode(errors="replace"))
|
||||
output += chunk
|
||||
self.assertIn(prompt, output.decode(errors="replace"))
|
||||
|
||||
try:
|
||||
until("SSH 别名")
|
||||
time.sleep(.1) # let fish finish entering terminal input mode
|
||||
if cancel_at == "alias":
|
||||
os.write(master, b"\x04")
|
||||
else:
|
||||
os.write(master, b"\r")
|
||||
until("[y/N]")
|
||||
time.sleep(.1)
|
||||
if cancel_at == "interrupt":
|
||||
os.killpg(proc.pid, signal.SIGINT)
|
||||
elif cancel_at == "eof":
|
||||
os.write(master, b"\x04")
|
||||
else:
|
||||
os.write(master, (answer + "\r").encode())
|
||||
deadline = time.monotonic() + 10
|
||||
while time.monotonic() < deadline:
|
||||
ready = selector.select(.1)
|
||||
if ready:
|
||||
try:
|
||||
chunk = os.read(master, 65536)
|
||||
except OSError:
|
||||
break
|
||||
if not chunk:
|
||||
break
|
||||
output += chunk
|
||||
elif proc.poll() is not None:
|
||||
break
|
||||
proc.wait(timeout=1)
|
||||
return proc.returncode, output.decode(errors="replace")
|
||||
finally:
|
||||
selector.close()
|
||||
os.close(master)
|
||||
if proc.poll() is None:
|
||||
os.killpg(proc.pid, signal.SIGKILL)
|
||||
proc.communicate()
|
||||
|
||||
def assert_unchanged(self, before):
|
||||
self.assertEqual(self.snapshot(), before)
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
self.assertFalse(self.calls("ssh-keygen"))
|
||||
|
||||
def assert_backup(self, content):
|
||||
backups = list((self.ssh / "config.backups").glob("*.backup.*"))
|
||||
self.assertEqual(len(backups), 1)
|
||||
self.assertEqual(backups[0].read_text(), content)
|
||||
self.assertEqual((self.ssh / "config.backups").stat().st_mode & 0o777, 0o700)
|
||||
self.assertFalse(list((self.ssh / "config.d").glob("*.backup.*")))
|
||||
|
||||
def effective(self, host):
|
||||
# -G is local only; absolute Include plus -F avoid the real SSH config.
|
||||
result = subprocess.run([REAL_SSH, "-G", "-F", str(self.ssh / "config"), host],
|
||||
env=self.env, capture_output=True, text=True, timeout=5)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
return dict(line.split(" ", 1) for line in result.stdout.splitlines())
|
||||
|
||||
def test_existing_alias_is_actionable_failure_and_preserved(self):
|
||||
self.existing(location="config")
|
||||
before = self.snapshot()
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertIn("非交互", output)
|
||||
self.assert_unchanged(before)
|
||||
self.assertFalse(self.calls("ssh"))
|
||||
|
||||
def test_overwrite_accepts_only_explicit_yes(self):
|
||||
for answer in ("n", "", "sure", "yes please"):
|
||||
with self.subTest(answer=answer):
|
||||
self.setUp()
|
||||
self.existing(location="config")
|
||||
before = self.snapshot()
|
||||
status, output = self.interactive("machine.invalid", answer=answer)
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertIn("未确认覆盖", output)
|
||||
self.assert_unchanged(before)
|
||||
self.assertFalse(self.calls("ssh"))
|
||||
|
||||
def test_overwrite_eof_and_ctrl_c_preserve_everything(self):
|
||||
for cancel in ("eof", "interrupt", "alias"):
|
||||
with self.subTest(cancel=cancel):
|
||||
self.setUp()
|
||||
self.existing(location="config")
|
||||
before = self.snapshot()
|
||||
status, output = self.interactive("machine.invalid", cancel_at=cancel)
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertIn("取消", output)
|
||||
self.assert_unchanged(before)
|
||||
|
||||
def test_overwrite_yes_copies_then_backs_up_and_updates(self):
|
||||
for answer in ("y", "YES"):
|
||||
with self.subTest(answer=answer):
|
||||
self.setUp()
|
||||
selected = self.key("custom")
|
||||
self.resolve(str(selected))
|
||||
path, original = self.existing()
|
||||
status, output = self.interactive("newuser@machine.invalid", "-p", "2200", answer=answer)
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assertIn(str(path), output)
|
||||
self.assert_selected(selected)
|
||||
self.assert_backup(original)
|
||||
updated = path.read_text()
|
||||
for value in ("HostName machine.invalid", "User newuser", "Port 2200", "ForwardAgent no",
|
||||
"Host untouched\n HostName untouched.invalid"):
|
||||
self.assertIn(value, updated)
|
||||
self.assertNotIn("old.invalid", updated)
|
||||
self.assertNotIn("~/.ssh/old", updated)
|
||||
|
||||
def test_overwrite_no_copy_in_shared_nonmatching_filename(self):
|
||||
path, original = self.existing("config.d/20-shared.conf", shared=True)
|
||||
status, output = self.interactive("--no-copy", "machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_backup(original)
|
||||
self.assertFalse((self.ssh / "config.d/machine.invalid.conf").exists())
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
self.assertFalse(self.calls("ssh-keygen"))
|
||||
self.assertTrue(all("-G" in call for call in self.calls("ssh")))
|
||||
self.assertEqual(self.effective("machine.invalid")["hostname"], "machine.invalid")
|
||||
self.assertEqual(self.effective("neighbor")["hostname"], "old.invalid")
|
||||
self.assertEqual(self.effective("neighbor")["identityfile"], "~/.ssh/old")
|
||||
self.assertEqual(self.effective("untouched")["hostname"], "untouched.invalid")
|
||||
self.assertIn("Host neighbor !machine.invalid", path.read_text())
|
||||
|
||||
def test_overwrite_shared_wildcard_excludes_target(self):
|
||||
path, original = self.existing("config.d/20-shared.conf", shared=True)
|
||||
original = original.replace("machine.invalid neighbor", "machine.invalid neighbor *")
|
||||
path.write_text(original)
|
||||
status, output = self.interactive("--no-copy", "machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_backup(original)
|
||||
self.assertEqual(self.effective("machine.invalid")["hostname"], "machine.invalid")
|
||||
self.assertNotIn("~/.ssh/old", self.effective("machine.invalid")["identityfile"])
|
||||
self.assertEqual(self.effective("neighbor")["hostname"], "old.invalid")
|
||||
|
||||
def test_overwrite_main_config_preserves_neighbors(self):
|
||||
path, original = self.existing("config")
|
||||
status, output = self.interactive("--no-copy", "machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
self.assert_backup(original)
|
||||
self.assertIn("HostName machine.invalid", path.read_text())
|
||||
self.assertIn("Host untouched\n HostName untouched.invalid", path.read_text())
|
||||
self.assertFalse((self.ssh / "config.d/machine.invalid.conf").exists())
|
||||
|
||||
def test_overwrite_copy_or_verification_failure_preserves_original(self):
|
||||
for mode in ("failure", "false-zero"):
|
||||
with self.subTest(mode=mode):
|
||||
self.setUp()
|
||||
self.key("id_rsa")
|
||||
self.existing(location="config")
|
||||
before = self.snapshot()
|
||||
self.env.update(COPY_MODE=mode, VERIFY_STATUS="255")
|
||||
status, output = self.interactive("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertEqual(self.snapshot(), before)
|
||||
self.assertEqual(len(self.calls("ssh-copy-id")), 1)
|
||||
|
||||
def test_ambiguous_declarations_are_not_partially_updated(self):
|
||||
self.existing()
|
||||
(self.ssh / "config.d/20-second.conf").write_text("Host machine.invalid\n HostName second.invalid\n")
|
||||
before = self.snapshot()
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertIn("多个 Host", output)
|
||||
self.assert_unchanged(before)
|
||||
|
||||
def test_filename_collision_does_not_replace_other_alias(self):
|
||||
path = self.ssh / "config.d/machine.invalid.conf"
|
||||
path.parent.mkdir()
|
||||
path.write_text("Host other\n HostName other.invalid\n")
|
||||
before = self.snapshot()
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assertIn("目标文件已存在", output)
|
||||
self.assert_unchanged(before)
|
||||
|
||||
def test_connect_timeout_and_verification_identity_isolation(self):
|
||||
selected = self.key("id_rsa")
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertEqual(status, 0, output)
|
||||
copy = self.calls("ssh-copy-id")[0]
|
||||
self.assertIn("ConnectTimeout=10", copy)
|
||||
self.assertIn("ConnectionAttempts=1", copy)
|
||||
verify = [call for call in self.calls("ssh") if "-G" not in call]
|
||||
self.assertEqual(len(verify), 1)
|
||||
for option in ("ConnectTimeout=10", "ConnectionAttempts=1", "BatchMode=yes", "IdentitiesOnly=yes", "CertificateFile=none", "ControlPath=none", "PreferredAuthentications=publickey"):
|
||||
self.assertIn(option, verify[0])
|
||||
self.assertEqual(verify[0][verify[0].index("-F") + 1], "/dev/null")
|
||||
self.assertEqual(verify[0][verify[0].index("-i") + 1], str(selected))
|
||||
self.assertNotIn("StrictHostKeyChecking=no", verify[0])
|
||||
|
||||
def test_resolve_failure_aborts(self):
|
||||
self.key("id_rsa")
|
||||
self.env["RESOLVE_STATUS"] = "255"
|
||||
status, output = self.run_ff("machine.invalid")
|
||||
self.assertNotEqual(status, 0, output)
|
||||
self.assert_no_host()
|
||||
self.assertFalse(self.calls("ssh-copy-id"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user