cheng
62 lines
3.5 KiB
Python
62 lines
3.5 KiB
Python
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from unittest import TestCase
|
|
from station.processes import ProcessSupervisor
|
|
from station.tasks import Cancelled
|
|
|
|
|
|
class ProcessesTest(TestCase):
|
|
def test_owned_process_cancel_and_exit(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
processes=ProcessSupervisor(directory)
|
|
try:
|
|
result=processes.run('echo',[sys.executable,'-c','print("completed")'],threading.Event())
|
|
self.assertTrue(result['ok']);self.assertIn('completed',result['detail'])
|
|
event=threading.Event();timer=threading.Timer(.2,event.set);timer.start()
|
|
with self.assertRaises(Cancelled):
|
|
processes.run('long',[sys.executable,'-c','import time;time.sleep(30)'],event)
|
|
self.assertFalse(processes.snapshot()['long']['running'])
|
|
finally: processes.close()
|
|
|
|
def test_each_run_has_its_own_log_slice_and_explicit_owner(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
processes=ProcessSupervisor(directory)
|
|
try:
|
|
processes.run('tool:89',[sys.executable,'-c','print("OLD_FAILURE")'],threading.Event())
|
|
result=processes.run('tool:89',[sys.executable,'-c','print("NEW_RESULT")'],threading.Event())
|
|
self.assertIn('NEW_RESULT',result['detail']);self.assertNotIn('OLD_FAILURE',result['detail'])
|
|
row=processes.snapshot()['tool:89']
|
|
self.assertEqual(row['owner'],'station');self.assertEqual(row['state'],'exited')
|
|
finally: processes.close()
|
|
|
|
def test_success_waits_for_orphaned_child_cleanup(self):
|
|
from pathlib import Path
|
|
from arm_application.process_lifecycle import _stat
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
processes=ProcessSupervisor(directory)
|
|
pidfile=Path(directory)/'pid'
|
|
source="import subprocess,sys,time;from pathlib import Path;p=subprocess.Popen([sys.executable,'-c','import time;time.sleep(120)'],start_new_session=True);Path(sys.argv[1]).write_text(str(p.pid))"
|
|
try:
|
|
result=processes.run('motion:89',[sys.executable,'-c',source,str(pidfile)],threading.Event(),timeout=12)
|
|
self.assertTrue(result['ok'],result)
|
|
child=int(pidfile.read_text());state=_stat(child)
|
|
self.assertTrue(state is None or state[2]=='Z','task reported success with a live orphan')
|
|
finally: processes.close()
|
|
|
|
def test_process_cleanup_phase_is_reported_before_resource_release(self):
|
|
from pathlib import Path
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
processes=ProcessSupervisor(directory)
|
|
ready=Path(directory)/'ready'
|
|
child="import signal,time,sys;from pathlib import Path;signal.signal(signal.SIGINT,lambda *a:(time.sleep(.4),sys.exit(0)));Path(sys.argv[1]).write_text('ready');time.sleep(20)"
|
|
source="import subprocess,sys,time;from pathlib import Path;subprocess.Popen([sys.executable,'-c',sys.argv[1],sys.argv[2]],start_new_session=True)\nwhile not Path(sys.argv[2]).exists():time.sleep(.01)"
|
|
phases=[]
|
|
try:
|
|
result=processes.run('motion:89',[sys.executable,'-c',source,child,str(ready)],threading.Event(),timeout=8,on_phase=phases.append)
|
|
self.assertTrue(result['ok'],result)
|
|
self.assertIn('收尾',phases)
|
|
self.assertFalse(processes.snapshot()['motion:89']['running'])
|
|
finally: processes.close()
|