import json, urllib.request, urllib.parse, subprocess, threading, time, sys
base=sys.argv[1] if len(sys.argv)>1 else 'http://127.0.0.1:8080'
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
def api(path,body=None,method=None):
req=urllib.request.Request(base+path,data=None if body is None else json.dumps(body).encode(),headers={'Content-Type':'application/json'},method=method)
with opener.open(req,timeout=15) as r:return json.load(r)
p=subprocess.Popen(['tests/targets/bin/activity'],stdin=subprocess.PIPE,stdout=subprocess.PIPE,text=True)
lease=None;response=None
try:
pid,peer,address,size=p.stdout.readline().split();pid=int(pid);peer=int(peer);address=int(address,16);size=int(size)
lease=api('/api/space/leases',{'density':3});token=lease['token']
response=opener.open(base+'/api/space/events?'+urllib.parse.urlencode({'token':token}),timeout=20)
frames=[]
def consume():
try:
for line in response:
if line.startswith(b'data: '):
value=json.loads(line[6:]);
if isinstance(value,dict) and 'memory' in value:frames.append(value)
except (OSError,ValueError):pass
thread=threading.Thread(target=consume,daemon=True);thread.start()
deadline=time.monotonic()+12
while time.monotonic()<deadline:
status=api('/api/space/status')
if any(str(status.get(sensor,'')).startswith('unavailable') for sensor in ('cpu','ipc','memory')):raise AssertionError(status)
snapshot=api('/api/space/snapshot')
if any(n['identity']['pid']==pid for n in snapshot['nodes']):break
time.sleep(.3)
assert status['memory']=='idle',status
selected=next(n['identity'] for n in snapshot['nodes'] if n['identity']['pid']==pid)
api('/api/space/leases',{'token':token,'density':3,'selected_process':selected})
time.sleep(1)
status=api('/api/space/status')
assert status['cpu']=='observing' and status['ipc']=='observing' and status['memory']=='sampling',status
p.stdin.write('go\n');p.stdin.flush();time.sleep(6)
memory=[m for f in frames for m in f['memory'] if m['process_id']['pid']==pid and address<=int(m['page'],16)<address+size]
assert memory,'No memory samples in known fixture mapping'
assert all(m['process_id']['pid']==pid for f in frames for m in f['memory'])
api('/api/space/leases',{'token':token,'density':3,'selected_process':None})
time.sleep(.5)
assert api('/api/space/status')['memory']=='idle'
ipc=[e for f in frames for e in f['ipc'] if e['process_id']['pid']==pid and e['write'] and e['bytes']>0]
from collections import defaultdict
sends=defaultdict(lambda:[0,0]);receives=defaultdict(lambda:[0,0])
for f in frames:
for e in f['ipc']:
if e['process_id']['pid']==pid and e['write']:sends[e['resource']][0]+=e['bytes'];sends[e['resource']][1]+=e['count']
if e['process_id']['pid']==peer and not e['write']:receives[e['resource']][0]+=e['bytes'];receives[e['resource']][1]+=e['count']
pipe_sends=[v for k,v in sends.items() if k.startswith('pipe:') and v==[256,1]]
socket_sends=[v for k,v in sends.items() if k.startswith('socket:')]
socket_receives=[v for k,v in receives.items() if k.startswith('socket:')]
cpu=[e for f in frames for e in f.get('cpu',[]) if e['process_id']['pid']==pid]
assert cpu and sum(e['runtime_ns'] for e in cpu)>500_000_000,cpu
assert any(e['running_threads']>0 and e['cpus'] for e in cpu),cpu
assert any(e['running_threads']==0 for e in cpu),cpu
assert pipe_sends and len(socket_sends)==3 and all(v==[384,2] for v in socket_sends),sends
assert len(socket_receives)==3 and all(v==[384,2] for v in socket_receives),receives
assert any(k.startswith('pipe:') and v==[256,1] for k,v in receives.items()),receives
print('Live sensors passed:',len(memory),'memory bins; scheduler runtime/current CPU; pipe/UNIX/TCP/UDP send and receive; exact bytes/counts; MSG_PEEK and failed send excluded')
finally:
if lease:api('/api/space/leases',{'token':lease['token']},'DELETE')
if p.poll() is None:p.terminate()
p.wait(timeout=5)