|
| 1 | +"""Interactive IPv4 suggestions: real prompt/CLI, mocked curl, no network.""" |
| 2 | +import os |
| 3 | +from pathlib import Path |
| 4 | +import tempfile |
| 5 | +import unittest |
| 6 | + |
| 7 | +from web_link_fixture import ROOT, terminal |
| 8 | + |
| 9 | + |
| 10 | +class PublicIPTests(unittest.TestCase): |
| 11 | + def setUp(self): |
| 12 | + self.temp = tempfile.TemporaryDirectory() |
| 13 | + self.addCleanup(self.temp.cleanup) |
| 14 | + self.root = Path(self.temp.name) |
| 15 | + self.calls = self.root / 'calls' |
| 16 | + self.env = dict(os.environ, PATH=f'{self.root}:{os.environ["PATH"]}', |
| 17 | + FIXTURE_ROOT=str(self.root), PRIMARY_STATUS='0', FALLBACK_STATUS='0') |
| 18 | + (self.root / 'curl').write_text('''#!/bin/bash |
| 19 | +printf '%s\\n' "$*" >>"$FIXTURE_ROOT/calls" |
| 20 | +case "${!#}" in |
| 21 | + https://api.ipify.org) cat "$FIXTURE_ROOT/primary"; exit "$PRIMARY_STATUS";; |
| 22 | + https://ipv4.icanhazip.com) cat "$FIXTURE_ROOT/fallback"; exit "$FALLBACK_STATUS";; |
| 23 | + *) exit 99;; |
| 24 | +esac |
| 25 | +''') |
| 26 | + (self.root / 'curl').chmod(0o755) |
| 27 | + (self.root / 'primary').write_text('8.8.8.8') |
| 28 | + (self.root / 'fallback').write_text('1.1.1.1\n') |
| 29 | + |
| 30 | + def run_prompt(self, ip='', code=None, tty=True): |
| 31 | + self.calls.unlink(missing_ok=True) |
| 32 | + script = 'source "$1"\n' + (code or '''prompt_install |
| 33 | +printf 'RESULT=%s|%s|%s\\n' "$DOMAIN" "$PUBLIC_IP" "$SOCKS" |
| 34 | +''') |
| 35 | + result, output = terminal(['bash', '-c', script, 'fixture', |
| 36 | + str(ROOT / 'telemt-web-manager.sh')], |
| 37 | + f'proxy.example.com\n{ip}\nn\n'.encode(), self.env, |
| 38 | + input_tty=tty) |
| 39 | + calls = self.calls.read_text().splitlines() if self.calls.exists() else [] |
| 40 | + for call in calls: |
| 41 | + self.assertEqual(call.split()[:-1], |
| 42 | + ['-4', '--noproxy', '*', '--connect-timeout', '2', |
| 43 | + '--max-time', '4', '--max-filesize', '64', '-fsS']) |
| 44 | + return result, output.decode(), [call.split()[-1] for call in calls] |
| 45 | + |
| 46 | + def test_primary_default_and_enter(self): |
| 47 | + result, output, calls = self.run_prompt() |
| 48 | + self.assertEqual(result, 0) |
| 49 | + self.assertIn('Detected public IPv4: 8.8.8.8', output) |
| 50 | + self.assertIn('Public IPv4 [8.8.8.8]:', output) |
| 51 | + self.assertIn('RESULT=proxy.example.com|8.8.8.8|', output) |
| 52 | + self.assertEqual(calls, ['https://api.ipify.org']) |
| 53 | + |
| 54 | + def test_fallback_only_after_primary_failure(self): |
| 55 | + self.env['PRIMARY_STATUS'] = '7' |
| 56 | + result, output, calls = self.run_prompt() |
| 57 | + self.assertEqual(result, 0) |
| 58 | + self.assertIn('RESULT=proxy.example.com|1.1.1.1|', output) |
| 59 | + self.assertEqual(calls, ['https://api.ipify.org', 'https://ipv4.icanhazip.com']) |
| 60 | + |
| 61 | + def test_both_fail_manual_input(self): |
| 62 | + self.env.update(PRIMARY_STATUS='7', FALLBACK_STATUS='28') |
| 63 | + result, output, calls = self.run_prompt('203.0.113.10') |
| 64 | + self.assertEqual(result, 0) |
| 65 | + self.assertIn('Could not detect public IPv4 automatically.', output) |
| 66 | + self.assertIn('Public IPv4 (must match DNS A):', output) |
| 67 | + self.assertIn('RESULT=proxy.example.com|203.0.113.10|', output) |
| 68 | + self.assertEqual(len(calls), 2) |
| 69 | + |
| 70 | + def test_invalid_or_nonpublic_response_rejected(self): |
| 71 | + for response in ('127.0.0.1', '10.0.0.1', '100.64.0.1', '169.254.1.1', |
| 72 | + '203.0.113.10', '224.0.0.1', '::1', '999.1.1.1', |
| 73 | + '8.8.8.8\nextra', '8.8.8.8\n\n', ' 8.8.8.8', '<html>error</html>'): |
| 74 | + with self.subTest(response=response): |
| 75 | + (self.root / 'primary').write_text(response) |
| 76 | + result, output, calls = self.run_prompt() |
| 77 | + self.assertEqual(result, 0) |
| 78 | + self.assertIn('RESULT=proxy.example.com|1.1.1.1|', output) |
| 79 | + self.assertEqual(len(calls), 2) |
| 80 | + |
| 81 | + def test_manual_override(self): |
| 82 | + result, output, _ = self.run_prompt('203.0.113.10') |
| 83 | + self.assertEqual(result, 0) |
| 84 | + self.assertIn('RESULT=proxy.example.com|203.0.113.10|', output) |
| 85 | + |
| 86 | + def test_noninteractive_does_not_detect(self): |
| 87 | + # Reach the first filesystem validation in real install_manager, then stop |
| 88 | + # before any host mutation. Platform/locking are outside this prompt test. |
| 89 | + code = '''STATE=$FIXTURE_ROOT/state |
| 90 | +preflight() { :; } |
| 91 | +take_lock() { :; } |
| 92 | +recover_update() { :; } |
| 93 | +helper() { |
| 94 | + [[ $1 == safe-path ]] || exit 99 |
| 95 | + printf 'RESULT=%s|%s\\n' "$DOMAIN" "$PUBLIC_IP" |
| 96 | + exit 0 |
| 97 | +} |
| 98 | +main --install --domain proxy.example.com --public-ip 203.0.113.10 |
| 99 | +''' |
| 100 | + result, output, calls = self.run_prompt(code=code, tty=False) |
| 101 | + self.assertEqual(result, 0) |
| 102 | + self.assertIn('RESULT=proxy.example.com|203.0.113.10', output) |
| 103 | + self.assertEqual(calls, []) |
| 104 | + |
| 105 | + |
| 106 | +if __name__ == '__main__': |
| 107 | + unittest.main() |
0 commit comments