1"""Fixture readiness relay tests; no network or production credentials."""
2import importlib.util
3from pathlib import Path
4import unittest
5from unittest.mock import patch, MagicMock
6
7spec = importlib.util.spec_from_file_location("fixture_ssh", Path(__file__).with_name("ssh.py"))
8wrapper = importlib.util.module_from_spec(spec)
9spec.loader.exec_module(wrapper)
10
11
12class ControlTests(unittest.TestCase):
13 def test_control_failure_is_not_a_port(self):
14 conn = MagicMock()
15 conn.__enter__.return_value = conn
16 conn.makefile.return_value.readline.return_value = b'{"id":1,"error":"listener limit"}\n'
17 with patch.object(wrapper.socket, "socket", return_value=conn):
18 with self.assertRaises(RuntimeError):
19 wrapper.control({"id": 1, "op": "udp", "port": 1234})
20
21 def test_control_returns_fixture_port(self):
22 conn = MagicMock()
23 conn.__enter__.return_value = conn
24 conn.makefile.return_value.readline.return_value = b'{"id":1,"result":4321}\n'
25 with patch.object(wrapper.socket, "socket", return_value=conn):
26 self.assertEqual(wrapper.control({"id": 1, "op": "udp", "port": 1234}), 4321)
27
28
29class MainTests(unittest.TestCase):
30 BOOTSTRAP = ["-T", "--", "remote", "'vev' '_broker-mux-quic-bootstrap'"]
31
32 def run_main(self, readiness, rewritten=4321):
33 process = MagicMock()
34 process.stdout.readline.return_value = readiness
35 process.stdout.read1.return_value = b""
36 process.wait.return_value = 0
37 process.poll.return_value = 0
38 out = MagicMock()
39 with patch.object(wrapper.sys, "argv", ["ssh"] + self.BOOTSTRAP), \
40 patch.object(wrapper.subprocess, "Popen", return_value=process), \
41 patch.object(wrapper, "control", return_value=rewritten) as control, \
42 patch.object(wrapper.sys, "stdout", MagicMock(buffer=out)):
43 code = wrapper.main()
44 return code, control, out
45
46 def test_rewrites_only_the_port(self):
47 code, control, out = self.run_main(b'{"port":1234,"fingerprint":"f"}\n')
48 self.assertEqual(code, 0)
49 control.assert_called_once_with({"id": 1, "op": "udp", "port": 1234})
50 self.assertEqual(out.write.call_args_list[0].args[0], b'{"port":4321,"fingerprint":"f"}\n')
51
52 def test_rejects_invalid_readiness(self):
53 for line in (b'{"port":1}', b"x" * 4097 + b"\n", b'{"port":0}\n', b'{"port":65536}\n', b'{"port":"22"}\n'):
54 with self.subTest(line=line[:20]):
55 with self.assertRaises(RuntimeError):
56 self.run_main(line)
57
58 def test_other_commands_exec_ssh_unchanged(self):
59 with patch.object(wrapper.sys, "argv", ["ssh", "remote", "true"]), \
60 patch.object(wrapper.os, "execv", side_effect=SystemExit(0)) as execv:
61 with self.assertRaises(SystemExit):
62 wrapper.main()
63 execv.assert_called_once_with("/usr/bin/ssh-real", ["/usr/bin/ssh-real", "-p", "2222", "-o", "HostKeyAlias=remote", "remote", "true"])
64
65
66if __name__ == "__main__":
67 unittest.main()