Add test for cancel propagation

/reviewed-by @kirr
/reviewed-on !14

nxdtest/__init__.py

@@ -111,6 +111,10 @@ def emit(*message):
   print(*message)
   sys.stdout.flush()
 
+# interval we will poll master periodically for test_alive.isAlive.
+# NOTE it is not e.g. one second not to overload master.
+_tmasterpoll = 5*time.minute
+
 @func
 def main():
     # testnode executes us giving URL to master results collecting instance and other details
@@ -221,7 +225,7 @@ def main():
         while 1:
             _, _rx = select(
                     ctx.done().recv,                # 0
-                    time.after(5*time.minute).recv, # 1   NOTE not e.g. one second not to overload master
+                    time.after(_tmasterpoll).recv,  # 1
             )
             if _ == 0:
                 break

nxdtest/nxdtest_test.py

@@ -24,7 +24,8 @@ import re
 import time
 from os.path import dirname
 from golang import chan, select, default, func, defer
-from golang import context, sync
+from golang import context, sync, time
+import psutil
 
 import pytest
 
@@ -166,3 +167,67 @@ TestCase('TEST_WITH_PROCLEAK', ['%s', 'AAA', 'BBB', 'CCC'])
     assert "AAA: terminating" in captured.out
     assert "BBB: terminating" in captured.out
     assert "CCC: terminating" in captured.out
+
+
+@pytest.fixture
+def distributor_with_cancelled_test(mocker):
+    """A distributor for a test result with one test result line named TEST1.
+
+    test_result.isAlive() will return False after 2 invocations, to simulate
+    a test_result that was cancelled by distributor.
+    """
+    def _retryRPC(func_id, args=()):
+        if func_id == 'getProtocolRevision':
+            return 1
+        assert False, ('unexpected RPC call', (func_id, args))
+    mocker.patch(
+        'erp5.util.taskdistribution.RPCRetry._retryRPC',
+        side_effect=_retryRPC)
+
+    test_result_line_proxy = mocker.patch(
+        'erp5.util.taskdistribution.TestResultLineProxy',
+        autospec=True)
+    type(test_result_line_proxy).name = mocker.PropertyMock(return_value='TEST1')
+
+    test_result_proxy = mocker.patch(
+        'erp5.util.taskdistribution.TestResultProxy',
+        autospec=True)
+    test_result_proxy.start.side_effect = [test_result_line_proxy, None]
+    test_result_proxy.isAlive.side_effect = [True, True, False]
+
+    mocked_createTestResult = mocker.patch(
+        'erp5.util.taskdistribution.TaskDistributor.createTestResult',
+        return_value=test_result_proxy)
+
+    yield
+
+    mocked_createTestResult.assert_called_once()
+    test_result_proxy.start.assert_called()
+    test_result_proxy.isAlive.assert_called()
+    test_result_line_proxy.stop.assert_called()
+
+
+# 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):
+    # 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.
+    mocker.patch('nxdtest._tmasterpoll', 0.1*time.second)
+
+    hang = "%s/testprog/hang" % (dirname(__file__),)
+    run_nxdtest(
+        """\
+TestCase('TEST1', ['%s'])
+""" % (hang),
+        argv=[
+            "nxdtest",
+            "--master_url", "http://localhost",
+        ],
+    )
+    captured = capsys.readouterr()
+    assert "TEST1" in captured.out
+    assert "# master asks to cancel test run" in captured.out
+    assert "# test run canceled" in captured.out
+    assert "hang: terminating" in captured.out
+    assert captured.err == ''

nxdtest/testprog/hang

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright (C) 2021  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
+# option) any later version, as published by the Free Software Foundation.
+#
+# You can also Link and Combine this program with other software covered by
+# the terms of any of the Free Software licenses or any of the Open Source
+# Initiative approved licenses and Convey the resulting work. Corresponding
+# source of such a combination shall include the source code for all other
+# software used.
+#
+# This program is distributed WITHOUT ANY WARRANTY; without even the implied
+# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+#
+# See COPYING file for full licensing terms.
+# See https://www.nexedi.com/licensing for rationale and options.
+"""Program hang helps to verify that nxdtest terminates processes when interrupted."""
+
+from __future__ import print_function, absolute_import
+
+import time
+from signal import signal, SIGTERM
+from setproctitle import setproctitle
+
+def main():
+    job = 'hang'
+    setproctitle(job)
+
+    def _(sig, frame):
+        print('%s: terminating' % job)
+        raise SystemExit
+    signal(SIGTERM, _)
+
+    while 1:
+        print('%s: hanging ...' % job)
+        time.sleep(1)
+
+
+if __name__ == '__main__':
+    main()

setup.py

@@ -15,7 +15,7 @@ setup(
     packages    = find_packages(),
     install_requires = ['erp5.util', 'six', 'pygolang', 'psutil'],
     extras_require = {
-                   'test': ['pytest', 'pytest-timeout', 'setproctitle'],
+                   'test': ['pytest', 'pytest-mock', 'pytest-timeout', 'setproctitle'],
     },
 
     entry_points= {'console_scripts': ['nxdtest = nxdtest:main']},