Cancel test run on SIGINT/SIGTERM

In addition to canceling test run is master tells us to do so, also cancel the
run if interrupted or terminated.

With the following sample .nxdtest

    TestCase('sleep', ['sleep', '10'])

before the patch it does not react to CTRL+C:

    $ nxdtest
    ...
    >>> sleep
    $ sleep 10
    ^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C# ran 0 test cases.   <-- no reaction to CTRL+C, finishes after 10 seconds
    Traceback (most recent call last):
      File "/home/kirr/src/tools/go/py3.venv2/bin/nxdtest", line 33, in <module>
        sys.exit(load_entry_point('nxdtest', 'console_scripts', 'nxdtest')())
      File "/home/kirr/src/tools/go/py3.venv2/lib/python3.9/site-packages/decorator.py", line 232, in fun
        return caller(func, *(extras + args), **kw)
      File "/home/kirr/src/tools/go/pygolang-master/golang/__init__.py", line 103, in _
        return f(*argv, **kw)
      File "/home/kirr/src/wendelin/nxdtest/nxdtest/__init__.py", line 339, in main
        wg.wait()
    KeyboardInterrupt

after the patch:

    $ nxdtest
    ...
    >>> sleep
    $ sleep 10
    ^C# Interrupt					<-- prompt reaction to CTRL+C
    # stopping due to cancel
    # leaked pid=188877 'sleep' ['sleep', '10']
    error   sleep   1.030s  # 1t 1e 0f 0s
    # test run canceled
    # ran 1 test case:  1·error

Needs pygolang!17 to work.

/reviewed-by @jerome
/reviewed-on !16

nxdtest/__init__.py

 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright (C) 2018-2021  Nexedi SA and Contributors.
+# Copyright (C) 2018-2022  Nexedi SA and Contributors.
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
 # it under the terms of the GNU General Public License version 3, or (at your
@@ -62,8 +62,9 @@ import os, sys, argparse, logging, traceback, re, pwd, socket
 from errno import ESRCH, EPERM
 from os.path import dirname
 import six
-from golang import b, defer, func, select, default
-from golang import errors, context, sync, time
+from golang import b, chan, defer, func, go, select, default
+from golang import errors, context, os as gos, sync, syscall, time
+from golang.os import signal
 import psutil
 
 # trun.py is a helper via which we run tests.
@@ -219,12 +220,30 @@ def main():
       bstdout = sys.stdout.buffer
       bstderr = sys.stderr.buffer
 
-    # setup context that is canceled when/if test_result is canceled on master
+
+    # setup main context that is canceled on SIGINT/SIGTERM
     # we will use this context as the base for all spawned jobs
     ctx, cancel = context.with_cancel(context.background())
+    sigq = chan(1, dtype=gos.Signal)
+    signal.Notify(sigq, syscall.SIGINT, syscall.SIGTERM)
+    def _():
+        signal.Stop(sigq)
+        sigq.close()
+    defer(_)
+    def _(cancel):
+        sig, ok = sigq.recv_()
+        if not ok:
+            return
+        emit("# %s" % sig)
+        cancel()
+    go(_, cancel)
+    defer(cancel)
+
+    # adjust ctx to be also canceled when/if test_result is canceled on master
+    ctx, cancel = context.with_cancel(ctx)
     cancelWG = sync.WorkGroup(ctx)
     @func
-    def _(ctx):
+    def _(ctx, cancel):
         defer(cancel)
         while 1:
             _, _rx = select(
@@ -237,7 +256,7 @@ def main():
             if not test_result.isAlive():
                 emit("# master asks to cancel test run")
                 break
-    cancelWG.go(_)
+    cancelWG.go(_, cancel)
     defer(cancelWG.wait)
     defer(cancel)
 

nxdtest/nxdtest_test.py

 # -*- coding: utf-8 -*-
-# Copyright (C) 2020-2021  Nexedi SA and Contributors.
+# Copyright (C) 2020-2022  Nexedi SA and Contributors.
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
 # it under the terms of the GNU General Public License version 3, or (at your
@@ -24,18 +24,18 @@ import os
 import pwd
 import sys
 import re
-import time
 import tempfile
+import signal
 import shutil
-import subprocess
-from os.path import dirname, exists, devnull
-from golang import chan, select, default, func, defer
+from subprocess import Popen, PIPE
+from os.path import dirname, exists
+from golang import chan, select, default, func, defer, b
 from golang import context, sync, time
-import psutil
 
 import pytest
 
-from nxdtest import main, trun
+import nxdtest
+from nxdtest import trun
 
 
 @pytest.fixture
@@ -64,7 +64,7 @@ def run_nxdtest(tmpdir):
             @func
             def _(ctx):
                 defer(done.close)
-                main()
+                nxdtest.main()
             wg.go(_)
 
             while 1:
@@ -338,7 +338,7 @@ def distributor_with_cancelled_test(mocker):
 
 # verify that nxdtest cancels test run when master reports that test_result is no longer alive.
 @pytest.mark.timeout(timeout=10)
-def test_cancel_from_master(run_nxdtest, capsys, tmp_path, distributor_with_cancelled_test, mocker):
+def test_cancel_from_master(run_nxdtest, capsys, distributor_with_cancelled_test, mocker):
     # nxdtest polls every 5 minutes, but in test we don't want to wait so long.
     # set master poll interval to small, but enough time for spawned hang to
     # setup its signal handler.
@@ -360,3 +360,57 @@ TestCase('TEST1', ['%s'])
     assert "# test run canceled" in captured.out
     assert "hang: terminating" in captured.out
     assert captured.err == ''
+
+
+# verify that nxdtest cancels test run on SIGINT/SIGTERM.
+#@pytest.mark.timeout(timeout=10)
+@pytest.mark.timeout(timeout=3)
+@pytest.mark.parametrize('sig', [(signal.SIGINT, "Interrupt"), (signal.SIGTERM, "Terminate")])
+@func
+def test_cancel_from_signal(tmpdir, sig):
+    hang = "%s/testprog/hang" % (dirname(__file__),)
+
+    with tmpdir.as_cwd():
+        with open(".nxdtest", "w") as f:
+            f.write("""\
+TestCase('TEST1', ['%s'])
+""" % hang)
+
+        proc = Popen([sys.executable, "%s/__init__.py" % dirname(nxdtest.__file__)], stdout=PIPE)
+        def _():
+            proc.terminate()
+            if proc.poll() is None:
+                time.sleep(1)
+                proc.kill()
+            proc.wait()
+        defer(_)
+
+        # procreadline reads next line from proc stdout.
+        outv = []
+        def procreadline():
+            l = proc.stdout.readline()
+            if len(l) != 0: # EOF
+                outv.append(l)
+            return l
+
+        # wait for hang to start and setup its signal handler
+        while 1:
+            l = procreadline()
+            if not l:
+                raise AssertionError("did not got 'hanging'")
+            if b"hanging" in l:
+                break
+
+        # send SIGINT/SIGTERM to proc and wait for it to complete
+        signo, sigmsg = sig
+        proc.send_signal(signo)
+        while 1:
+            if not procreadline():
+                break
+
+        out = b''.join(outv)
+
+        assert b"TEST1" in out
+        assert b("# %s" % sigmsg) in out
+        assert b"# test run canceled" in out
+        assert b"hang: terminating" in out

setup.py

@@ -13,7 +13,7 @@ setup(
     keywords    = 'Nexedi testing infrastructure tool tox',
 
     packages    = find_packages(),
-    install_requires = ['erp5.util', 'six', 'pygolang', 'psutil', 'python-prctl'],
+    install_requires = ['erp5.util', 'six', 'pygolang >= 0.1', 'psutil', 'python-prctl'],
     extras_require = {
                    'test': ['pytest', 'pytest-mock', 'pytest-timeout', 'setproctitle'],
     },