56 خطوط
2.2 KiB
Python
56 خطوط
2.2 KiB
Python
"""Wait up to three minutes for the flashed device to answer ICMP."""
|
|
import argparse
|
|
import ipaddress
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def test_device(interface, host_ip, device_ip):
|
|
host = str(ipaddress.IPv4Address(host_ip))
|
|
device = str(ipaddress.IPv4Address(device_ip))
|
|
if host == device:
|
|
raise ValueError('Device IP must differ from the test host IP')
|
|
if not interface or interface.startswith('-'):
|
|
raise ValueError('Invalid network interface')
|
|
result = subprocess.run(
|
|
['ip', '-j', '-4', 'addr', 'show', 'dev', interface],
|
|
check=True, capture_output=True, text=True, timeout=5,
|
|
)
|
|
addresses = json.loads(result.stdout)
|
|
if not any(info.get('local') == host for entry in addresses for info in entry.get('addr_info', [])):
|
|
raise ValueError(f'Host IP {host} is not configured on {interface}')
|
|
result = subprocess.run(
|
|
['ip', '-j', '-4', 'route', 'get', device, 'from', host],
|
|
check=True, capture_output=True, text=True, timeout=5,
|
|
)
|
|
routes = json.loads(result.stdout)
|
|
if not routes or any(route.get('dev') != interface for route in routes):
|
|
raise ValueError(f'Route to {device} does not use {interface}')
|
|
print(f'Waiting up to 180 seconds: {host} ({interface}) -> {device}', flush=True)
|
|
# -c 1 exits on the first reply; -w bounds the entire ping run.
|
|
result = subprocess.run(
|
|
['ping', '-4', '-n', '-B', '-I', host, '-c', '1', '-i', '1', '-W', '1', '-w', '180', device],
|
|
timeout=180,
|
|
)
|
|
if result.returncode:
|
|
raise RuntimeError(f'FAIL: {device} did not pass the ping test (exit {result.returncode})')
|
|
print(f'PASS: device {device} answered ping', flush=True)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--interface', default='eno1')
|
|
parser.add_argument('--host-ip', default='192.168.75.15')
|
|
parser.add_argument('--device-ip', required=True)
|
|
args = parser.parse_args()
|
|
test_device(args.interface, args.host_ip, args.device_ip)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
except (ValueError, RuntimeError, OSError, subprocess.SubprocessError) as exc:
|
|
print(f'FAIL: {exc}', file=sys.stderr)
|
|
sys.exit(1)
|