138 lines
7.5 KiB
Python
138 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Actual CLI parsing/retry functions with isolated transcripts and fake clients."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SOURCE = Path(os.environ.get('TEST_CLI_SOURCE', ROOT/'bin/vpn-lemanapro.sh')).read_text()
|
|
|
|
def function(name):
|
|
start = SOURCE.index(name + '() {')
|
|
return SOURCE[start:SOURCE.index('\n}\n', start) + 3]
|
|
|
|
with tempfile.TemporaryDirectory(prefix='lemana-runtime-') as temporary:
|
|
folder = Path(temporary)
|
|
env = dict(os.environ, HOME=temporary, TEST_ROOT=temporary, TZ='Europe/Moscow')
|
|
functions = ['_emit', '_write_status', '_filter_output']
|
|
functions += [name for name in ('_parse_expiry', '_valid_ipv4', '_publish_connection') if name+'() {' in SOURCE]
|
|
setup = '\n'.join(function(name) for name in functions) + r'''
|
|
set -euo pipefail
|
|
JSON_MODE=true; DEBUG=false; MODE=keychain-auto
|
|
STATUS_DIR="$TEST_ROOT/state"; STATUS_FILE="$STATUS_DIR/status.json"
|
|
ATTEMPT_DIR="$TEST_ROOT/attempt"; mkdir -p "$ATTEMPT_DIR"
|
|
_log_connection_line() { :; }
|
|
_stop_connect_progress() { :; }
|
|
scutil() { :; }
|
|
'''
|
|
def transcript(lines):
|
|
result = subprocess.run(['bash', '-c', setup + '\n_filter_output'], input=lines,
|
|
env=env, capture_output=True, text=True, check=True, timeout=10)
|
|
return [json.loads(line) for line in result.stdout.splitlines()]
|
|
|
|
events = transcript('Configured as 10.2.3.4, with SSL connected\n')
|
|
assert len(events) == 1 and events[0]['event'] == 'connected', events
|
|
assert events[0]['ip'] == '10.2.3.4' and not events[0].get('expires')
|
|
assert json.loads((folder/'state/status.json').read_text())['state'] == 'connected'
|
|
assert transcript('Configured as 999.2.3.4\nConfigured as 1.2.3.4.5\n') == []
|
|
for stamp, expected in [
|
|
('Tue Jan 01 02:00:00 2030', '2029-12-31T23:00:00Z'),
|
|
('Tue, 01 Jan 2030 02:00:00 MSK', '2029-12-31T23:00:00Z'),
|
|
('Tue, 01 Jan 2030 02:00:00 +07', '2029-12-31T19:00:00Z'),
|
|
('Tue, 01 Jan 2030 02:00:00 -0330', '2030-01-01T05:30:00Z'),
|
|
('Tue, 01 Jan 2030 02:00:00 +05:30', '2029-12-31T20:30:00Z'),
|
|
]:
|
|
for before in (True, False):
|
|
ip = 'Configured as 10.2.3.4\n'
|
|
expiry = 'Session authentication will expire at ' + stamp + '\n'
|
|
events = transcript((expiry+ip if before else ip+expiry) + expiry + 'Session authentication will expire at invalid\n')
|
|
assert events[0]['event'] == 'connected' and events[-1]['expires'] == expected, events
|
|
assert len({item['started_at'] for item in events}) == 1
|
|
assert json.loads((folder/'state/status.json').read_text())['expires'] == expected
|
|
assert transcript('Session authentication will expire at Tue Jan 01 02:00:00 2030\n') == []
|
|
print('PASS status: IP alone, valid IPv4, expiry order/formats/timezones, invalid metadata cannot erase valid timer')
|
|
|
|
status_functions = '\n'.join(function(name) for name in ('_check_status', '_json_get'))
|
|
status_setup = setup + '\n' + status_functions + r'''
|
|
kill() { return "${TEST_PID_DEAD:-0}"; }
|
|
_module_status_json() { printf '{}'; }
|
|
_module_status_human() { :; }
|
|
date() {
|
|
if [[ "$*" == '+%s' ]]; then printf '1893456000\n'; else /bin/date "$@"; fi
|
|
}
|
|
'''
|
|
def status(value, **extra):
|
|
(folder/'state/status.json').write_text(json.dumps(dict(pid=123, **value)))
|
|
result = subprocess.run(['bash','-c',status_setup+'\n_check_status'], env=dict(env, **extra),
|
|
text=True, capture_output=True, check=True)
|
|
return json.loads(result.stdout)
|
|
for state in ('connecting','reconnecting','disconnecting'):
|
|
assert status(dict(state=state))['state'] == state
|
|
assert status(dict(state=state), TEST_PID_DEAD='1')['state'] == 'disconnected'
|
|
assert status(dict(state='connected', ip='10.2.3.4', expires='2030-01-01T02:00:00Z'))['remaining_sec'] == 7200
|
|
assert status(dict(state='connected', ip='10.2.3.4', expires='2029-12-31T23:00:00Z'))['remaining_sec'] == 0
|
|
assert status(dict(state='connected', ip='10.2.3.4', expires=''))['expires'] == ''
|
|
result = subprocess.run(['bash','-c',status_setup+'\nJSON_MODE=false; _check_status'], env=env,
|
|
text=True, capture_output=True, check=True)
|
|
assert 'срок неизвестен' in result.stdout
|
|
print('PASS status query: live phases, dead PID, UTC countdown outside UTC, expiry clamped, unknown duration')
|
|
|
|
runtime = '\n'.join(function(name) for name in ('_run_connections', '_recover_for_retry'))
|
|
client = folder/'client'
|
|
client.write_text('''#!/bin/sh
|
|
count=0
|
|
[ ! -f "$TEST_ROOT/launches" ] || read -r count < "$TEST_ROOT/launches"
|
|
count=$((count+1)); echo "$count" > "$TEST_ROOT/launches"
|
|
for last_arg do :; done
|
|
case "$last_arg" in *" --vpnc $LEMANA_VPN_AUTH_ATTEMPT") ;; *) exit 91 ;; esac
|
|
printf '%s\\n' "$LEMANA_VPN_AUTH_ATTEMPT" >> "$TEST_ROOT/attempts"
|
|
case "$TEST_CASE:$count" in
|
|
auth:*) echo LEMANA_VPN_AUTH_REJECTED; exit 2 ;;
|
|
connected:1) echo 'Configured as 10.2.3.4'; exit 0 ;;
|
|
connected:2) exit 2 ;;
|
|
*) exit 7 ;;
|
|
esac
|
|
''')
|
|
client.chmod(0o755)
|
|
setup += r'''
|
|
_start_connect_progress() { :; }
|
|
_show_connection_log_tail() { :; }
|
|
sleep() { echo "$1" >> "$TEST_ROOT/delays"; }
|
|
sudo() {
|
|
printf '%s\n' "$*" >> "$TEST_ROOT/helper-calls"
|
|
printf '%s\n' "${TEST_RECOVERY:-restored}"
|
|
}
|
|
DNS_CLEANUP=/fake/helper; SESSION_ID=11111111-2222-3333-4444-555555555555
|
|
OC_BIN="$TEST_ROOT/client"; OC_RUNTIME_XDG_CONFIG_HOME="$HOME"
|
|
display_mode=hidden; log_level=''; autofill_debug=0; autofill_disable=0; autofill_click=1; autofill_totp=1
|
|
dns_hook='/fake/helper --vpnc'; MODE="${TEST_MODE:-keychain-auto}"
|
|
'''
|
|
cases = [('crash', 'keychain-auto', 'restored', 6, [5,10,20,40,60]),
|
|
('crash', 'bitwarden-auto', 'restored', 6, [5,10,20,40,60]),
|
|
('crash', 'manual-keychain', 'restored', 1, []),
|
|
('connected', 'keychain-auto', 'restored', 2, [5]),
|
|
('auth', 'keychain-auto', 'restored', 1, []),
|
|
('crash', 'keychain-auto', 'cancelled', 1, []),
|
|
('crash', 'keychain-auto', 'error', 1, []),
|
|
('crash', 'keychain-auto', 'active', 1, [1]*29)]
|
|
for case, mode, recovery, launches, delays in cases:
|
|
for name in ('launches','delays','helper-calls','attempts'):
|
|
(folder/name).unlink(missing_ok=True)
|
|
result = subprocess.run(['bash','-c', setup+'\n'+runtime+'\n_run_connections'],
|
|
env=dict(env, TEST_CASE=case, TEST_MODE=mode, TEST_RECOVERY=recovery),
|
|
text=True, capture_output=True, timeout=10)
|
|
assert int((folder/'launches').read_text()) == launches, (case,mode,recovery,result.stdout,result.stderr)
|
|
attempts = (folder/'attempts').read_text().splitlines()
|
|
assert len(attempts) == launches and len(set(attempts)) == launches
|
|
actual = [int(s) for s in (folder/'delays').read_text().splitlines()] if (folder/'delays').exists() else []
|
|
assert actual == delays, (case,mode,recovery,actual)
|
|
events = [json.loads(line) for line in result.stdout.splitlines()]
|
|
if case == 'connected':
|
|
assert [e['attempt_id'] for e in events if e['event'] == 'connected'] == [attempts[0]]
|
|
if case == 'auth': assert any(e['event']=='auth_rejected' for e in events)
|
|
if recovery in ('error','active'): assert events[-1]['event']=='error'
|
|
if recovery == 'cancelled': assert result.returncode == 0
|
|
print('PASS retry: both automatic modes, manual/cancel/auth terminal, bounded backoff, DNS failures block new launches')
|