Coverage for integrations / service_tools / _test_echo_worker.py: 0.0%

1"""Tiny test worker — no GPU, just echoes requests. 

2 

3Spawned via the centralized dispatcher: 

4 python -m integrations.service_tools.gpu_worker \\ 

5 integrations.service_tools._test_echo_worker 

6 

7The dispatcher picks up `_load` and `_synthesize` by convention. 

8""" 

9 

10 

11def _load(): 

12 # Simulate a noisy library printing to stdout during load. 

13 # This must NOT corrupt the JSON protocol channel. A correctly 

14 # isolated worker redirects fd 1 and sys.stdout to stderr before 

15 # calling _load(), so these writes go to the log stream. 

16 import os, sys 

17 print('[noise] library init message on sys.stdout') 

18 sys.stdout.write('[noise] raw sys.stdout.write\n') 

19 try: 

20 os.write(1, b'[noise] raw os.write to fd 1\n') 

21 except OSError: 

22 pass 

23 return {'message': 'loaded'} 

24 

25 

26def _synthesize(state, req: dict) -> dict: 

27 op = req.get('op', 'echo') 

28 if op == 'crash': 

29 # Simulate an uncatchable crash 

30 import os 

31 os._exit(137) 

32 if op == 'raise': 

33 raise RuntimeError('simulated handler failure') 

34 if op == 'noisy_echo': 

35 # Handler also prints garbage before returning — protocol must survive. 

36 print('[noise] handler stdout print') 

37 import sys 

38 sys.stdout.write('[noise] handler raw write\n') 

39 return {'echo': req, 'noisy': True} 

40 if op == 'sleep': 

41 import time as _t 

42 _t.sleep(float(req.get('sleep_s', 0))) 

43 return {'slept': req.get('sleep_s', 0)} 

44 if op == 'args': 

45 import sys as _s 

46 return {'argv': _s.argv[1:]} 

47 return {'echo': req, 'state': state}