diff --git a/src/dvc_task/proc/manager.py b/src/dvc_task/proc/manager.py index 2f1ae21..29e4e9d 100644 --- a/src/dvc_task/proc/manager.py +++ b/src/dvc_task/proc/manager.py @@ -147,6 +147,10 @@ def handle_closed_process(): else: raise ProcessLookupError + def interrupt(self, name: str): + """Send interrupt signal to specified named process""" + self.send_signal(name, signal.SIGINT) + def terminate(self, name: str): """Terminate the specified named process.""" self.send_signal(name, signal.SIGTERM) diff --git a/tests/proc/test_manager.py b/tests/proc/test_manager.py index e8e68df..56c0066 100644 --- a/tests/proc/test_manager.py +++ b/tests/proc/test_manager.py @@ -61,43 +61,32 @@ def side_effect(*args): process_manager.send_signal("nonexists", signal.SIGTERM) -def test_kill( - mocker: MockerFixture, - process_manager: ProcessManager, - finished_process: str, - running_process: str, -): - """Kill signal should be sent.""" - mock_kill = mocker.patch("os.kill") - process_manager.kill(running_process) - if sys.platform == "win32": - mock_kill.assert_called_once_with(PID_RUNNING, signal.SIGTERM) - else: - mock_kill.assert_called_once_with( - PID_RUNNING, signal.SIGKILL # pylint: disable=no-member - ) - - mock_kill.reset_mock() - with pytest.raises(ProcessLookupError): - process_manager.kill(finished_process) - mock_kill.assert_not_called() - - -def test_terminate( +if sys.platform == "win32": + SIGKILL = signal.SIGTERM +else: + SIGKILL = (signal.SIGKILL,) # pylint: disable=no-member + + +@pytest.mark.parametrize( + "method, sig", + [ + ("kill", SIGKILL), + ("terminate", signal.SIGTERM), + ("interrupt", signal.SIGINT), + ], +) +def test_kill_commands( mocker: MockerFixture, process_manager: ProcessManager, - running_process: str, - finished_process: str, + method: str, + sig: signal.Signals, ): - """Terminate signal should be sent.""" - mock_kill = mocker.patch("os.kill") - process_manager.terminate(running_process) - mock_kill.assert_called_once_with(PID_RUNNING, signal.SIGTERM) - - mock_kill.reset_mock() - with pytest.raises(ProcessLookupError): - process_manager.terminate(finished_process) - mock_kill.assert_not_called() + """Test shortcut for different signals.""" + name = "process" + mock_kill = mocker.patch.object(process_manager, "send_signal") + func = getattr(process_manager, method) + func(name) + mock_kill.assert_called_once_with(name, sig) def test_remove(