1
+ import pathlib
1
2
import time
2
3
import os
3
4
import sys
@@ -23,13 +24,15 @@ def test_monitor_processes(create_plain_run_offline: tuple[Run, dict]):
23
24
@pytest .mark .executor
24
25
def test_abort_all_processes (create_plain_run : tuple [Run , dict ]) -> None :
25
26
_run , _ = create_plain_run
26
- start_time = time .time ()
27
27
with tempfile .NamedTemporaryFile (suffix = ".py" ) as temp_f :
28
28
with open (temp_f .name , "w" ) as out_f :
29
29
out_f .writelines ([
30
30
"import time\n " ,
31
- "while True:\n "
32
- " time.sleep(5)\n "
31
+ "count = 0\n "
32
+ "while True:\n " ,
33
+ " print(count)\n "
34
+ " time.sleep(1)\n "
35
+ " count += 1"
33
36
])
34
37
35
38
for i in range (1 , 3 ):
@@ -39,9 +42,20 @@ def test_abort_all_processes(create_plain_run: tuple[Run, dict]) -> None:
39
42
time .sleep (3 )
40
43
41
44
_run .kill_all_processes ()
42
- end_time = time .time ()
43
45
44
- assert end_time - start_time < 10 , f"{ end_time - start_time } >= 10"
46
+ # Check the Python process did not error
47
+ _out_err = pathlib .Path .cwd ().glob ("*process_*.err" )
48
+ for file in _out_err :
49
+ with file .open () as in_f :
50
+ assert not in_f .readlines ()
51
+
52
+ # Now check the counter in the process was terminated
53
+ # just beyond the sleep time
54
+ _out_files = pathlib .Path .cwd ().glob ("*process_*.out" )
55
+ for file in _out_files :
56
+ with file .open () as in_f :
57
+ assert (lines := in_f .readlines ())
58
+ assert int (lines [0 ].strip ()) < 4
45
59
46
60
47
61
def test_processes_cwd (create_plain_run : dict [Run , dict ]) -> None :
0 commit comments