feat(proxy-client): добавлен мастер установки системной службы
- Зачем: установка с нуля и переход из ручного Xray одной командой. - Что: план с подтверждением или --yes, скрытый ввод ссылки, проверка Xray, перенос настроек, запуск службы и переключение домашнего профиля; безопасное продолжение после сбоя и сохранение существующих настроек. - Проверка: cross-review Claude и исправление находок; по 108 тестов на Python 3.8 и 3.14 под PTY, 18 Bash-проверок. После аудита уточнена фикстура повторного запуска и отдельно проверена на обеих версиях Python. Настоящие службы и настройки машины не изменялись.
This commit is contained in:
ddadmin committed
1 parent
45938cde98
commit
203e515479
5 files changed
+1250
-56
No files matched your search
@@ -0,0 +1,520 @@
|
||||
"""Интерактивная установка через CLI; системные действия подменены."""
|
||||
import json
|
||||
import hashlib
|
||||
import os
|
||||
import pty
|
||||
import select
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
CLI = Path(__file__).resolve().parents[1] / 'proxy-client/.local/bin/proxy-client'
|
||||
URL = ('vless://11111111-2222-4333-8444-555555555555@cdn.example:443'
|
||||
'?type=xhttp&security=tls&encryption=none&sni=cdn.example&fp=chrome&path=%2FSECRET')
|
||||
CONFIG = '[proxy]\ntype=xray\nname=Test\naddress=127.0.0.1\nport=12334\nno_proxy_extra=.example\n'
|
||||
|
||||
|
||||
class ServiceSetupTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
temporary = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(temporary.cleanup)
|
||||
self.root = Path(temporary.name)
|
||||
self.config = self.root / 'config/proxy-client'
|
||||
self.config.mkdir(parents=True)
|
||||
self.fake = self.root / 'bin'
|
||||
self.fake.mkdir()
|
||||
self.log = self.root / 'actions'
|
||||
self.stage = self.root / 'system'
|
||||
self.env = dict(os.environ, HOME=str(self.root / 'home'),
|
||||
XDG_CONFIG_HOME=str(self.root / 'config'),
|
||||
XDG_STATE_HOME=str(self.root / 'state'),
|
||||
PYTHONPATH=str(self.fake), TEST_ROOT=str(self.root),
|
||||
PATH=str(self.fake) + os.pathsep + os.defpath)
|
||||
(self.fake / 'python3').symlink_to(sys.executable)
|
||||
(self.fake / 'sitecustomize.py').write_text('''
|
||||
import os, sys, pathlib, getpass, grp, types, socket, urllib.request, io
|
||||
OriginalPath = type(pathlib.Path())
|
||||
class StagedPath:
|
||||
home = staticmethod(OriginalPath.home)
|
||||
def __new__(cls, *parts, **kwargs):
|
||||
if parts == ('/',):
|
||||
parts = (os.environ['TEST_ROOT'] + '/system',)
|
||||
return OriginalPath(*parts, **kwargs)
|
||||
pathlib.Path = StagedPath
|
||||
os.geteuid = lambda: 0 if os.environ.get('TEST_PRIVILEGED') else 1000
|
||||
sys.stdin.isatty = lambda: not os.environ.get('TEST_PRIVILEGED') and not os.environ.get('TEST_NO_TTY')
|
||||
if not os.environ.get('TEST_REAL_GETPASS'):
|
||||
getpass.getpass = lambda prompt: input(prompt)
|
||||
grp.getgrnam = lambda name: types.SimpleNamespace(gr_gid=os.getgid())
|
||||
def chown(path, uid, gid):
|
||||
assert str(path).startswith(os.environ['TEST_ROOT'] + '/system/')
|
||||
os.chown = chown
|
||||
class LocalSocket:
|
||||
def __init__(self, *args, **kwargs): pass
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *args): pass
|
||||
def close(self): pass
|
||||
def setsockopt(self, *args): pass
|
||||
def bind(self, address):
|
||||
busy = os.environ.get('TEST_PORT_BUSY')
|
||||
if busy == 'manual':
|
||||
busy = os.path.exists(os.environ['TEST_ROOT'] + '/manual-active')
|
||||
if busy: raise OSError('busy')
|
||||
socket.socket = LocalSocket
|
||||
def connect(*args, **kwargs):
|
||||
if not os.path.exists(os.environ['TEST_ROOT'] + '/system/active'):
|
||||
raise OSError('not ready')
|
||||
return LocalSocket()
|
||||
socket.create_connection = connect
|
||||
original_readlink = os.readlink
|
||||
original_read_bytes = OriginalPath.read_bytes
|
||||
def readlink(path, *args, **kwargs):
|
||||
if str(path) == '/proc/4321/exe':
|
||||
if not os.path.exists(os.environ['TEST_ROOT'] + '/manual-active'): raise FileNotFoundError()
|
||||
return os.environ['TEST_ROOT'] + '/bin/xray'
|
||||
return original_readlink(path, *args, **kwargs)
|
||||
def read_bytes(path):
|
||||
if str(path) == '/proc/4322/cmdline':
|
||||
if not os.path.exists(os.environ['TEST_ROOT'] + '/manual-active'): raise FileNotFoundError()
|
||||
return b'proxy-client start'
|
||||
return original_read_bytes(path)
|
||||
os.readlink = readlink
|
||||
OriginalPath.read_bytes = read_bytes
|
||||
def kill(pid, signum):
|
||||
assert not os.environ.get('TEST_PRIVILEGED'), 'нельзя останавливать пользовательский процесс от root'
|
||||
assert pid == 4322, 'чужой PID'
|
||||
os.unlink(os.environ['TEST_ROOT'] + '/manual-active')
|
||||
with open(os.environ['TEST_ROOT'] + '/actions', 'a') as stream:
|
||||
stream.write('["manual-stop"]\\n')
|
||||
os.kill = kill
|
||||
def urlopen(url, **kwargs):
|
||||
if os.environ.get('TEST_DOWNLOAD_FAIL'): raise OSError('download failed')
|
||||
with open(os.environ['TEST_ROOT'] + '/actions', 'a') as stream:
|
||||
stream.write('["download"]\\n')
|
||||
suffix = 'digest' if url.endswith('.dgst') else 'archive'
|
||||
return io.BytesIO(open(os.environ['TEST_ROOT'] + '/' + suffix, 'rb').read())
|
||||
urllib.request.urlopen = urlopen
|
||||
''')
|
||||
self.executable('sudo', '''#!/usr/bin/env python3
|
||||
import json, os, sys
|
||||
args = sys.argv[1:]
|
||||
with open(os.environ['TEST_ROOT'] + '/actions', 'a') as stream:
|
||||
stream.write(json.dumps(['sudo'] + args) + '\\n')
|
||||
if os.environ.get('TEST_FAIL_SUDO'): sys.exit(23)
|
||||
if args[0] == '--': args.pop(0)
|
||||
os.environ['TEST_PRIVILEGED'] = '1'
|
||||
os.environ['HOME'] = os.environ['TEST_ROOT'] + '/root-home'
|
||||
os.environ.pop('XDG_CONFIG_HOME', None)
|
||||
os.environ.pop('XDG_STATE_HOME', None)
|
||||
os.execvpe(args[0], args, os.environ)
|
||||
''')
|
||||
self.executable('systemd-sysusers', '#!/bin/sh\nexit 0\n')
|
||||
self.executable('systemctl', '''#!/usr/bin/env python3
|
||||
import json, os, sys
|
||||
from pathlib import Path
|
||||
root = Path(os.environ['TEST_ROOT'])
|
||||
with (root / 'actions').open('a') as stream:
|
||||
stream.write(json.dumps(['systemctl'] + sys.argv[1:]) + '\\n')
|
||||
action = sys.argv[1]
|
||||
active = root / 'system/active'
|
||||
if action in ('start', 'restart') or (action == 'enable' and '--now' in sys.argv):
|
||||
if os.environ.get('TEST_FAIL_START'): sys.exit(1)
|
||||
if not os.environ.get('TEST_NOT_READY'): active.touch()
|
||||
if os.environ.get('TEST_CHANGE_LINK_AT_START'):
|
||||
link = root / 'config/proxy-client/default.url'
|
||||
link.write_text(link.read_text().replace('cdn.example:443', 'after-start.example:443'))
|
||||
if (action == 'stop' or (action == 'disable' and '--now' in sys.argv)) and active.exists(): active.unlink()
|
||||
if action == 'is-active': sys.exit(0 if active.exists() else 1)
|
||||
''')
|
||||
self.executable('xray', '#!/bin/sh\n'
|
||||
'[ "$1" != version ] || echo "Xray 26.10.10 (fake)"\nexit 0\n')
|
||||
|
||||
def executable(self, name, text):
|
||||
path = self.fake / name
|
||||
path.write_text(text)
|
||||
path.chmod(0o755)
|
||||
|
||||
def ready_profile(self):
|
||||
(self.config / 'config').write_text(CONFIG)
|
||||
(self.config / 'default.url').write_text(URL + '\n')
|
||||
(self.config / 'default.url').chmod(0o600)
|
||||
|
||||
def running_profile(self):
|
||||
self.ready_profile()
|
||||
state = self.root / 'state/proxy-client'
|
||||
state.mkdir(parents=True, exist_ok=True)
|
||||
(state / 'state.json').write_text(json.dumps({
|
||||
'supervisor_pid': 4322, 'xray_pid': 4321, 'xray_path': str(self.fake / 'xray'),
|
||||
'address': '127.0.0.1', 'port': 12334}))
|
||||
(self.root / 'manual-active').touch()
|
||||
|
||||
def cli(self, answers='', *args):
|
||||
return subprocess.run([str(CLI), 'service', 'install', *args], input=answers,
|
||||
env=self.env, capture_output=True, text=True, timeout=20)
|
||||
|
||||
def test_declining_plan_changes_nothing(self):
|
||||
self.ready_profile()
|
||||
result = self.cli('n\n')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertIn('отменена', result.stdout)
|
||||
self.assertFalse(self.log.exists())
|
||||
self.assertFalse(self.stage.exists())
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
|
||||
def test_fresh_setup_collects_settings_without_writing_before_confirmation(self):
|
||||
result = self.cli(URL + '\n\nn\n')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertIn('12334', result.stdout)
|
||||
self.assertIn('отменена', result.stdout)
|
||||
self.assertNotIn(URL, result.stdout + result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
self.assertEqual([], list(self.config.iterdir()))
|
||||
|
||||
def test_dangling_profile_symlink_is_reported_before_secret_input(self):
|
||||
(self.config / 'config').write_text(CONFIG)
|
||||
(self.config / 'default.url').symlink_to(self.root / 'missing.url')
|
||||
result = self.cli(URL + '\ny\n')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('битый симлинк', result.stderr)
|
||||
self.assertNotIn('ввод скрыт', result.stdout)
|
||||
self.assertNotIn('rm -- ', result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
self.assertFalse((self.root / 'state/proxy-client/service-setup.json').exists())
|
||||
|
||||
def test_fresh_setup_installs_starts_and_configures_shell_profile(self):
|
||||
result = self.cli(URL + '\n\ny\n')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
settings = self.stage / 'etc/proxy-client'
|
||||
self.assertIn('type = xray', (settings / 'config').read_text())
|
||||
self.assertEqual(URL + '\n', (settings / 'default.url').read_text())
|
||||
self.assertEqual(0o640, (settings / 'default.url').stat().st_mode & 0o777)
|
||||
home_config = (self.config / 'config').read_text()
|
||||
self.assertIn('type = external', home_config)
|
||||
self.assertIn('address = 127.0.0.1', home_config)
|
||||
self.assertIn('port = 12334', home_config)
|
||||
self.assertTrue((self.stage / 'active').exists())
|
||||
self.assertNotIn(URL, result.stdout + result.stderr + self.log.read_text())
|
||||
|
||||
def test_readiness_check_does_not_request_sudo(self):
|
||||
self.ready_profile()
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
actions = [json.loads(line) for line in self.log.read_text().splitlines()]
|
||||
self.assertIn(['systemctl', 'is-active', '--quiet', 'proxy-client.service'], actions)
|
||||
self.assertFalse(any(action[0] == 'sudo' and 'is-active' in action for action in actions))
|
||||
|
||||
def test_yes_accepts_file_without_a_terminal(self):
|
||||
self.env['TEST_NO_TTY'] = '1'
|
||||
profile = self.root / 'connection.url'
|
||||
profile.write_text(URL + '\n')
|
||||
profile.chmod(0o600)
|
||||
result = self.cli('', '--auto-yes', '--file', str(profile), '--port', '12335')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertNotIn('Продолжить?', result.stdout)
|
||||
self.assertIn('port = 12335', (self.config / 'config').read_text())
|
||||
self.assertNotIn(URL, result.stdout + result.stderr + self.log.read_text())
|
||||
|
||||
def test_no_terminal_requires_an_explicit_installation_mode(self):
|
||||
self.running_profile()
|
||||
self.env['TEST_NO_TTY'] = '1'
|
||||
result = self.cli('')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('--yes', result.stderr)
|
||||
self.assertIn('--install-only', result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
self.assertFalse(self.stage.exists())
|
||||
self.assertTrue((self.root / 'manual-active').exists())
|
||||
|
||||
def test_update_keeps_system_settings_without_home_profile(self):
|
||||
self.ready_profile()
|
||||
first = self.cli('y\n')
|
||||
self.assertEqual(0, first.returncode, first.stderr)
|
||||
settings = self.stage / 'etc/proxy-client'
|
||||
before = {name: (settings / name).read_bytes() for name in ('config', 'default.url')}
|
||||
for path in self.config.iterdir():
|
||||
path.unlink()
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertIn('обнов', result.stdout.lower())
|
||||
for name, content in before.items():
|
||||
self.assertEqual(content, (settings / name).read_bytes())
|
||||
self.assertFalse((self.config / 'config').exists())
|
||||
|
||||
def test_migration_stops_only_own_process_after_files_are_installed(self):
|
||||
self.running_profile()
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertFalse((self.root / 'manual-active').exists())
|
||||
actions = [json.loads(line) for line in self.log.read_text().splitlines()]
|
||||
self.assertLess(actions.index(['systemctl', 'daemon-reload']), actions.index(['manual-stop']))
|
||||
self.assertLess(actions.index(['manual-stop']),
|
||||
actions.index(['systemctl', 'enable', '--now', 'proxy-client.service']))
|
||||
backups = list(self.config.glob('config.before-service.*'))
|
||||
self.assertEqual(1, len(backups))
|
||||
self.assertEqual(CONFIG, backups[0].read_text())
|
||||
self.assertIn('type = external', (self.config / 'config').read_text())
|
||||
|
||||
def test_install_only_can_be_completed_by_the_wizard(self):
|
||||
self.running_profile()
|
||||
first = self.cli('', '--install-only')
|
||||
self.assertEqual(0, first.returncode, first.stderr)
|
||||
self.env['TEST_PORT_BUSY'] = 'manual'
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertFalse((self.root / 'manual-active').exists())
|
||||
self.assertTrue((self.stage / 'active').exists())
|
||||
self.assertIn('type = external', (self.config / 'config').read_text())
|
||||
|
||||
def test_installed_different_profile_is_not_replaced_or_stopped(self):
|
||||
self.running_profile()
|
||||
first = self.cli('', '--install-only')
|
||||
self.assertEqual(0, first.returncode, first.stderr)
|
||||
system_config = self.stage / 'etc/proxy-client/config'
|
||||
changed = system_config.read_text().replace('12334', '12335')
|
||||
system_config.write_text(changed)
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode, result.stderr)
|
||||
self.assertTrue((self.root / 'manual-active').exists())
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
self.assertEqual(changed, system_config.read_text())
|
||||
self.assertNotIn('["systemctl", "stop"', self.log.read_text())
|
||||
snapshot = self.root / 'state/proxy-client/service-setup.json'
|
||||
self.assertTrue(snapshot.exists())
|
||||
self.assertIn('rm -- ' + str(snapshot), result.stderr)
|
||||
self.assertIn('/etc/proxy-client', result.stderr)
|
||||
|
||||
def test_start_failure_preserves_home_and_explains_recovery(self):
|
||||
self.running_profile()
|
||||
self.env['TEST_FAIL_START'] = '1'
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
self.assertIn('proxy-client start', result.stderr)
|
||||
self.assertIn('disable', self.log.read_text())
|
||||
self.assertEqual([], list(self.config.glob('config.before-service.*')))
|
||||
|
||||
def test_missing_xray_is_downloaded_and_verified_before_installation(self):
|
||||
binary = (self.fake / 'xray').read_bytes()
|
||||
(self.fake / 'xray').unlink()
|
||||
with zipfile.ZipFile(str(self.root / 'archive'), 'w') as archive:
|
||||
archive.writestr('xray', binary)
|
||||
digest = hashlib.sha256((self.root / 'archive').read_bytes()).hexdigest()
|
||||
(self.root / 'digest').write_text('SHA2-256= ' + digest + '\n')
|
||||
self.ready_profile()
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertEqual(binary, (self.stage / 'usr/local/lib/proxy-client/xray').read_bytes())
|
||||
actions = [json.loads(line) for line in self.log.read_text().splitlines()]
|
||||
self.assertEqual(['download'], actions[0])
|
||||
self.assertFalse((self.root / 'home/.local/bin/xray').exists())
|
||||
|
||||
def test_retry_after_failed_start_completes_the_transition(self):
|
||||
self.running_profile()
|
||||
self.env['TEST_FAIL_START'] = '1'
|
||||
self.assertEqual(1, self.cli('', '--yes').returncode)
|
||||
self.env.pop('TEST_FAIL_START')
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertIn('type = external', (self.config / 'config').read_text())
|
||||
|
||||
def test_file_option_cannot_be_silently_ignored_in_install_only_mode(self):
|
||||
self.ready_profile()
|
||||
result = self.cli('', '--install-only', '--file', str(self.config / 'default.url'))
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('--file', result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_update_downloads_missing_binary_without_asking_for_profile(self):
|
||||
self.ready_profile()
|
||||
self.assertEqual(0, self.cli('', '--yes').returncode)
|
||||
binary = (self.fake / 'xray').read_bytes()
|
||||
(self.fake / 'xray').unlink()
|
||||
(self.stage / 'usr/local/lib/proxy-client/xray').unlink()
|
||||
with zipfile.ZipFile(str(self.root / 'archive'), 'w') as archive:
|
||||
archive.writestr('xray', binary)
|
||||
digest = hashlib.sha256((self.root / 'archive').read_bytes()).hexdigest()
|
||||
(self.root / 'digest').write_text('SHA2-256= ' + digest + '\n')
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertEqual(binary, (self.stage / 'usr/local/lib/proxy-client/xray').read_bytes())
|
||||
|
||||
def test_preparation_failures_do_not_stop_manual_process(self):
|
||||
self.running_profile()
|
||||
self.env['TEST_FAIL_SUDO'] = '1'
|
||||
failed = self.cli('', '--yes')
|
||||
self.assertEqual(23, failed.returncode, failed.stderr)
|
||||
self.assertTrue((self.root / 'manual-active').exists())
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
pending = self.root / 'state/proxy-client/service-setup.json'
|
||||
self.assertEqual(0o600, pending.stat().st_mode & 0o777)
|
||||
self.env.pop('TEST_FAIL_SUDO')
|
||||
(self.fake / 'xray').unlink()
|
||||
self.env['TEST_DOWNLOAD_FAIL'] = '1'
|
||||
failed = self.cli('', '--yes')
|
||||
self.assertEqual(1, failed.returncode)
|
||||
self.assertTrue((self.root / 'manual-active').exists())
|
||||
self.assertFalse(self.stage.exists())
|
||||
|
||||
def test_retry_from_scratch_uses_private_snapshot_without_new_secret_input(self):
|
||||
self.env['TEST_FAIL_START'] = '1'
|
||||
failed = self.cli(URL + '\n\ny\n')
|
||||
self.assertEqual(1, failed.returncode, failed.stderr)
|
||||
self.assertFalse((self.config / 'config').exists())
|
||||
self.env.pop('TEST_FAIL_START')
|
||||
self.env['TEST_NO_TTY'] = '1'
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertIn('type = external', (self.config / 'config').read_text())
|
||||
self.assertFalse((self.root / 'state/proxy-client/service-setup.json').exists())
|
||||
|
||||
def test_changed_link_blocks_stale_snapshot_and_reset_uses_new_settings(self):
|
||||
self.ready_profile()
|
||||
self.env['TEST_FAIL_SUDO'] = '1'
|
||||
self.assertEqual(23, self.cli('', '--yes').returncode)
|
||||
self.env.pop('TEST_FAIL_SUDO')
|
||||
changed = URL.replace('cdn.example:443', 'new.example:443')
|
||||
(self.config / 'default.url').write_text(changed + '\n')
|
||||
before = self.log.read_bytes()
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode, result.stderr)
|
||||
self.assertIn('ссылки изменился', result.stderr)
|
||||
snapshot = self.root / 'state/proxy-client/service-setup.json'
|
||||
self.assertIn('rm -- ' + str(snapshot), result.stderr)
|
||||
self.assertFalse(self.stage.exists())
|
||||
self.assertEqual(before, self.log.read_bytes())
|
||||
snapshot.unlink() # Пользователь явно сбросил снимок по подсказке CLI.
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertEqual(changed + '\n', (self.stage / 'etc/proxy-client/default.url').read_text())
|
||||
|
||||
def test_link_edit_after_service_start_does_not_abort_the_confirmed_transition(self):
|
||||
self.running_profile()
|
||||
self.env['TEST_CHANGE_LINK_AT_START'] = '1'
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(0, result.returncode, result.stderr)
|
||||
self.assertIn('type = external', (self.config / 'config').read_text())
|
||||
self.assertIn('after-start.example', (self.config / 'default.url').read_text())
|
||||
self.assertEqual(URL + '\n', (self.stage / 'etc/proxy-client/default.url').read_text())
|
||||
self.assertFalse((self.root / 'state/proxy-client/service-setup.json').exists())
|
||||
|
||||
def test_pending_override_explains_how_to_reset_the_snapshot(self):
|
||||
self.ready_profile()
|
||||
self.env['TEST_FAIL_SUDO'] = '1'
|
||||
self.assertEqual(23, self.cli('', '--yes').returncode)
|
||||
result = self.cli('', '--yes', '--port', '12335')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('rm -- ', result.stderr)
|
||||
|
||||
def test_retry_refuses_changed_system_settings_before_stopping_manual_process(self):
|
||||
self.running_profile()
|
||||
self.env['TEST_FAIL_START'] = '1'
|
||||
self.assertEqual(1, self.cli('', '--yes').returncode)
|
||||
self.env.pop('TEST_FAIL_START')
|
||||
self.running_profile()
|
||||
config = self.stage / 'etc/proxy-client/config'
|
||||
changed = config.read_text().replace('12334', '12335')
|
||||
config.write_text(changed)
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('отличаются от снимка', result.stderr)
|
||||
self.assertTrue((self.root / 'manual-active').exists())
|
||||
self.assertEqual(changed, config.read_text())
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
|
||||
def test_bad_url_and_invalid_port_do_not_install(self):
|
||||
result = self.cli('not-a-vless-link\n')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertNotIn('not-a-vless-link', result.stdout + result.stderr)
|
||||
result = self.cli('', '--yes', '--port', '0')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertFalse(self.log.exists())
|
||||
self.assertFalse(self.stage.exists())
|
||||
|
||||
def test_foreign_port_is_never_stopped(self):
|
||||
self.ready_profile()
|
||||
self.env['TEST_PORT_BUSY'] = '1'
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('занят', result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
|
||||
def test_corrupt_transition_snapshot_gets_clear_error(self):
|
||||
self.ready_profile()
|
||||
pending = self.root / 'state/proxy-client/service-setup.json'
|
||||
pending.parent.mkdir(parents=True)
|
||||
pending.write_text('{}')
|
||||
pending.chmod(0o600)
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('снимок', result.stderr)
|
||||
self.assertNotIn('Traceback', result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_retry_does_not_accept_a_foreign_listener_as_the_service(self):
|
||||
self.ready_profile()
|
||||
self.env['TEST_FAIL_START'] = '1'
|
||||
self.assertEqual(1, self.cli('', '--yes').returncode)
|
||||
self.env.pop('TEST_FAIL_START')
|
||||
self.env['TEST_PORT_BUSY'] = '1'
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertEqual(CONFIG, (self.config / 'config').read_text())
|
||||
|
||||
def test_malformed_config_does_not_expose_its_contents(self):
|
||||
(self.config / 'config').write_text('PRIVATE_SENTINEL_not_ini\n')
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertNotIn('PRIVATE_SENTINEL', result.stdout + result.stderr)
|
||||
self.assertNotIn('Traceback', result.stderr)
|
||||
self.assertFalse(self.log.exists())
|
||||
|
||||
def test_retry_keeps_custom_file_in_manual_recovery_command(self):
|
||||
self.running_profile()
|
||||
custom = self.root / 'custom.url'
|
||||
(self.config / 'default.url').rename(custom)
|
||||
self.env['TEST_FAIL_START'] = '1'
|
||||
self.assertEqual(1, self.cli('', '--yes', '--file', str(custom)).returncode)
|
||||
self.running_profile()
|
||||
(self.config / 'default.url').unlink()
|
||||
result = self.cli('', '--yes')
|
||||
self.assertEqual(1, result.returncode)
|
||||
self.assertIn('proxy-client start -f ' + str(custom), result.stderr)
|
||||
|
||||
def test_secret_input_is_not_echoed_in_a_real_terminal(self):
|
||||
master, slave = pty.openpty()
|
||||
self.addCleanup(os.close, master)
|
||||
process = subprocess.Popen([str(CLI), 'service', 'install'], stdin=slave,
|
||||
stdout=slave, stderr=slave, start_new_session=True,
|
||||
env=dict(self.env, TEST_REAL_GETPASS='1'))
|
||||
os.close(slave)
|
||||
output = bytearray()
|
||||
def until(prompt):
|
||||
deadline = time.monotonic() + 10
|
||||
while prompt.encode() not in output:
|
||||
if time.monotonic() > deadline:
|
||||
self.fail('не появился запрос ввода: ' + prompt)
|
||||
if select.select([master], [], [], 0.1)[0]:
|
||||
output.extend(os.read(master, 4096))
|
||||
try:
|
||||
until('ввод скрыт')
|
||||
os.write(master, (URL + '\n').encode())
|
||||
until('Локальный порт')
|
||||
os.write(master, b'\n')
|
||||
until('Продолжить?')
|
||||
os.write(master, b'n\n')
|
||||
self.assertEqual(0, process.wait(timeout=10))
|
||||
self.assertNotIn(URL.encode(), output)
|
||||
self.assertFalse(self.log.exists())
|
||||
finally:
|
||||
if process.poll() is None:
|
||||
process.kill()
|
||||
process.wait()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user