#!/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')