Skip to content

Commit

Permalink
tests: Fix broken SIGINT test
Browse files Browse the repository at this point in the history
The old test would send the interrupt signal before the script had
a chance to enter the integration loop, and thus the custom signal
handler was never tested. The new version waits for the subprocess
to write it's ready for the interrupt signal, and the traceback is
checked to confirm the signal was re-raised by the script interface.
  • Loading branch information
jngrad committed Apr 24, 2023
1 parent 62069c7 commit 9c3a00f
Show file tree
Hide file tree
Showing 2 changed files with 46 additions and 8 deletions.
50 changes: 43 additions & 7 deletions testsuite/python/sigint.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,19 +22,55 @@
import time
import sys
import pathlib
import os


class SigintTest(ut.TestCase):

def setUp(self):
script = str(pathlib.Path(__file__).parent / 'sigint_child.py')
self.process = subprocess.Popen([sys.executable, script])
script = str(pathlib.Path(__file__).parent / 'sigint_child.py')

def check_signal_handling(self, process, sig):
# send signal
process.send_signal(sig)
# capture stderr and return code (negative of signum)
stdout, stderr = process.communicate(input=None, timeout=6.)
assert stdout is None
traceback = stderr.decode()
return_code = process.poll()
signum = -return_code
self.assertEqual(signum, sig.value)
if sig == signal.Signals.SIGTERM:
self.assertEqual(traceback, "")
elif sig == signal.Signals.SIGINT:
self.assertIn(" self.integrator.run(", traceback)
self.assertTrue(traceback.endswith(
" in handle_sigint\n signal.raise_signal(signal.Signals.SIGINT)\nKeyboardInterrupt\n"))

def test_signal_handling(self):
self.process.send_signal(signal.Signals.SIGINT)
# Wait for the signal to arrive and one integration step to be finished
time.sleep(1)
self.assertIsNotNone(self.process.poll())
signals = [signal.Signals.SIGINT, signal.Signals.SIGTERM]
processes = []
# open asynchronous processes with non-blocking read access on stderr
for _ in range(len(signals)):
process = subprocess.Popen([sys.executable, self.script],
stderr=subprocess.PIPE)
os.set_blocking(process.stderr.fileno(), False)
processes.append(process)

# wait for the script to reach the integration loop
time.sleep(0.5)
for process, sig in zip(processes, signals):
tick = time.time()
while True:
message = process.stderr.readline().decode()
if message == "start of integration loop\n":
# wait for the script to enter the integrator run method
time.sleep(0.1)
# send signal and check process behavior
self.check_signal_handling(process, sig)
break
tock = time.time()
assert tock - tick < 8., "subprocess timed out"
time.sleep(0.1)


if __name__ == '__main__':
Expand Down
4 changes: 3 additions & 1 deletion testsuite/python/sigint_child.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import sys
import numpy as np
import espressomd

Expand All @@ -26,5 +27,6 @@
for i in range(100):
system.part.add(pos=np.random.random() * system.box_l)

print("start of integration loop", file=sys.stderr)
while True:
system.integrator.run(1000)
system.integrator.run(100000)

0 comments on commit 9c3a00f

Please sign in to comment.