60 lines
2.8 KiB
Python
60 lines
2.8 KiB
Python
import socket, struct, subprocess, threading, time, sys
|
|
from pathlib import Path
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
DRIVER=ROOT/'tests'/(sys.argv[1] if len(sys.argv)>1 else 'driver')
|
|
def exact(s,n):
|
|
b=b''
|
|
while len(b)<n:
|
|
part=s.recv(n-len(b))
|
|
if not part: raise EOFError
|
|
b+=part
|
|
return b
|
|
def run(mode, function=3):
|
|
listener=socket.socket(); listener.bind(('127.0.0.1',0)); listener.listen(); listener.settimeout(2)
|
|
issues=[]
|
|
def server():
|
|
first=True
|
|
try:
|
|
while True:
|
|
try: s,_=listener.accept()
|
|
except (socket.timeout,OSError): break
|
|
with s:
|
|
s.settimeout(2)
|
|
while True:
|
|
try:
|
|
h=exact(s,7); p=exact(s,struct.unpack('>H',h[4:6])[0]-1)
|
|
except (EOFError,ConnectionError,socket.timeout): break
|
|
if first and mode in ('close','timeout','bad_tid','bad_length','write_drop'):
|
|
first=False
|
|
if mode=='timeout': time.sleep(.12)
|
|
elif mode=='bad_tid': s.sendall(b'\xff\xff'+h[2:4]+b'\x00\x07'+h[6:7]+b'\x03\x04\x00\x2a\x00\x2b')
|
|
elif mode=='bad_length': s.sendall(h[:4]+b'\xff\xff'+h[6:7])
|
|
break
|
|
if first and mode=='exception': reply=bytes([p[0]|128,2]); first=False
|
|
elif function==16: reply=p[:5]
|
|
elif function==1: reply=b'\x01\x01\x03'
|
|
else: reply=b'\x03\x04\x00\x2a\x00\x2b'
|
|
packet=h[:4]+struct.pack('>H',len(reply)+1)+h[6:7]+reply
|
|
if mode=='fragment':
|
|
for b in packet: s.sendall(bytes([b])); time.sleep(.001)
|
|
else: s.sendall(packet)
|
|
except Exception as e: issues.append(e)
|
|
worker=threading.Thread(target=server); worker.start()
|
|
try:
|
|
p=subprocess.run([str(DRIVER),str(listener.getsockname()[1]),str(function),'2'],capture_output=True,text=True,timeout=4)
|
|
assert p.returncode==0,(mode,p.stdout,p.stderr)
|
|
ok,err,reconnect,uncertain,last,a,b=map(int,p.stdout.split())
|
|
assert ok==2
|
|
if function==3: assert (a,b)==(42,43)
|
|
if function==1: assert (a,b)==(1,1)
|
|
if mode in ('close','timeout','bad_tid','bad_length','write_drop'): assert err>=1 and reconnect>=1
|
|
if mode=='exception': assert err==1 and reconnect==0 and last==9
|
|
if mode=='write_drop': assert uncertain==1
|
|
print('PASS',mode,p.stdout.strip())
|
|
finally:
|
|
listener.close(); worker.join(3)
|
|
assert not issues,issues
|
|
for mode in ('normal','fragment','close','timeout','bad_tid','bad_length','exception'): run(mode)
|
|
run('write_drop',16)
|
|
run('coils',1)
|