Threaded version 0.7

component-source/tests/test_worker.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Exercise production worker with real sockets/PTYS and deliberately sparse PLC calls."""
+import ctypes as C
+import os,pty,select,socket,struct,subprocess,tempfile,time,threading
+from pathlib import Path
+root=Path(__file__).resolve().parents[1]
+with tempfile.TemporaryDirectory() as td:
+ so=Path(td)/'worker.so'
+ subprocess.run(['cc','-shared','-fPIC','-std=c11','-Wall','-Wextra','-Werror','-O2','-pthread',
+  *[str(root/'src'/f) for f in ['atom_mb.c','atom_rtu.c','atom_server.c','atom_bridge.c','atom_worker.c']],'-o',str(so)],check=True)
+ lib=C.CDLL(str(so));U=C.c_uint32;W=C.c_uint16;B=C.c_uint8
+ lib.ab_call.argtypes=[U,U,U,W,B,B,W,W,C.POINTER(W),U,C.POINTER(U),U,C.c_uint64,U,U,U];lib.ab_call.restype=U
+ def call(cmd,h=0,ip=0,port=0,f=3,addr=0,q=1,values=(),ct=100,rt=100,retry=10,unit=1):
+  data=(W*2000)(*values);status=(U*24)()
+  result=lib.ab_call(cmd,h,ip,port,unit,f,addr,q,data,2000,status,24,0,ct,rt,retry)
+  return result,list(status),list(data)
+ def until(fn,seconds=1):
+  end=time.monotonic()+seconds
+  while time.monotonic()<end:
+   v=fn()
+   if v:return v
+   time.sleep(.002)
+  raise AssertionError('condition timed out')
+ def configure(watchdog=2000,cpu=-1):
+  r,s,_=call(40,addr=cpu+1,ct=1,rt=watchdog);assert r==0,(r,s)
+ def client(port):
+  r,s,_=call(1,ip=0x7f000001,port=port);assert not r;h=s[0]
+  assert not call(9,h,rt=100)[0];assert not call(3,h)[0]
+  until(lambda:call(5,h)[1][1]==3)
+  return h
+ def recv_exact(peer,count):
+  result=b''
+  while len(result)<count:
+   v=peer.recv(count-len(result));assert v;result+=v
+  return result
+ def server():
+  r,s,_=call(30,ip=0x7f000001,port=port);assert not r;h=s[0]
+  for t in range(4):assert not call(35,h,f=t,q=8,values=[0]*8)[0]
+  assert not call(32,h)[0]
+  until(lambda:call(34,h)[1][1])
+  return h
+ probe=socket.socket();probe.bind(('127.0.0.1',0));port=probe.getsockname()[1];probe.close()
+ try:
+  configure();sh=server();ch=client(port)
+  assert not call(6,ch,q=2)[0]
+  # The reply deadline is 100 ms; the PLC deliberately does nothing for 250 ms.
+  time.sleep(.25)
+  r,s,d=call(7,ch);assert not r and s[3] and not s[4] and d[:2]==[0,0],s
+  assert call(6,ch)[0]==2 # Completion cannot be overwritten before acknowledgment.
+  r,m,_=call(12,ch);assert not r and m[21]<100 and m[22]>=100,m
+  assert call(10,ch)[0]==0
+  print('PASS worker completes TCP before timeout with PLC absent for 250 ms; result retained and timed')
+  # Remote write must survive a stale PLC push before the PLC pulls its result.
+  peer=socket.create_connection(('127.0.0.1',port));peer.settimeout(1)
+  pdu=struct.pack('>BHH',6,0,4321);peer.sendall(struct.pack('>HHHB',1,0,len(pdu)+1,1)+pdu)
+  assert recv_exact(peer,12)[7:]==pdu
+  assert call(35,sh,f=2,q=8,values=[0]*8)[0]==0
+  assert call(36,sh,f=2,q=8)[2][0]==4321
+  # After acknowledgment a deliberate application override is permitted.
+  call(35,sh,f=2,q=8,values=[42]*8);assert call(36,sh,f=2,q=8)[2][0]==42
+  assert call(35,sh,f=2,addr=1,q=8,values=[0]*8)[0]==1 # explicit recreate for remap
+  peer.close();call(2,ch);call(31,sh)
+  print('PASS autonomous TCP slave and lossless remote-write handoff')
+  # Real serial PTY, again no PLC ticks after submission.
+  master,slave=pty.openpty();path=os.ttyname(slave);os.close(slave)
+  r,s,_=call(20,ip=9600,unit=0,f=8,addr=1,q=len(path),ct=0,values=list(path.encode()));assert not r;h=s[0]
+  call(9,h,rt=100);call(3,h);until(lambda:call(5,h)[1][1]==3)
+  call(6,h,unit=1,f=3,q=1)
+  assert select.select([master],[],[],1)[0];request=os.read(master,256);assert len(request)==8
+  def crc(p):
+   c=65535
+   for b in p:
+    c^=b
+    for _ in range(8):c=(c>>1)^(0xa001 if c&1 else 0)
+   return c
+  p=bytes([1,3,2,0,77]);os.write(master,p+struct.pack('<H',crc(p)))
+  time.sleep(.25);r,s,d=call(7,h);assert not r and s[3] and d[0]==77,s
+  call(2,h);os.close(master)
+  print('PASS RTU completion while PLC does not scan for 250 ms')
+  # Watchdog cancels an in-flight write without replaying it on resume.
+  configure(100)
+  listener=socket.socket();listener.bind(('127.0.0.1',0));listener.listen();listener.settimeout(1)
+  h=client(listener.getsockname()[1]);peer,_=listener.accept();peer.settimeout(1)
+  call(11,h,rt=1000);assert not call(6,h,f=16,q=1,values=[9])[0]
+  packet=recv_exact(peer,15);assert packet[7]==16
+  time.sleep(.16)
+  # command 12 is read-only and does not renew the scan lease.
+  r,s,_=call(12,h);assert not r and s[1]==0 and s[5]==10 and s[9]==1,s
+  assert peer.recv(1)==b''
+  call(5,h);peer2,_=listener.accept();peer2.settimeout(.03)
+  try:assert not peer2.recv(1), 'write replayed'
+  except socket.timeout:pass
+  call(2,h);peer.close();peer2.close();listener.close()
+  print('PASS scan watchdog closes transport, preserves uncertain write, resumes without replay')
+  # Server closes its listener when scanning ceases and recovers on refresh.
+  configure(100);sh=server();time.sleep(.16)
+  probe=socket.socket();probe.settimeout(.1);assert probe.connect_ex(('127.0.0.1',port))!=0;probe.close()
+  call(34,sh);until(lambda:call(34,sh)[1][1]);call(31,sh)
+  print('PASS server watchdog and restart')
+  # CPU selection, failure reporting, and restoring inherited mask.
+  configure(cpu=min(os.sched_getaffinity(0)))
+  r,s,_=call(41);assert not r and s[3]==min(os.sched_getaffinity(0))+1
+  assert call(40,addr=65535,ct=1,rt=2000)[0]==1
+  configure()
+  failures=[]
+  def reader():
+   for _ in range(200):
+    if call(41)[0]:failures.append('snapshot failure')
+  threads=[threading.Thread(target=reader) for _ in range(4)]
+  for t in threads:t.start()
+  for t in threads:t.join()
+  assert not failures
+  print('PASS configurable affinity, invalid arguments, concurrent snapshot access')
+  for _ in range(20):
+   lib.ab_shutdown();assert call(41)[1][1]==0;configure();assert call(41)[1][1]==1
+  print('PASS repeated joined shutdown/restart')
+ finally:lib.ab_shutdown()