Skip to content

Commit

Permalink
Add a new method interrupt to send SIGINT to tasks
Browse files Browse the repository at this point in the history
fix: #100
Required from dvc to stop the checkpoint during running.

1. Add a new method interrupt to send SIGINT to tasks
2. Modify the unit test for this new method.
  • Loading branch information
karajan1001 committed Dec 5, 2022
1 parent 01b5843 commit fe484ae
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 34 deletions.
4 changes: 4 additions & 0 deletions src/dvc_task/proc/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
57 changes: 23 additions & 34 deletions tests/proc/test_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down

0 comments on commit fe484ae

Please sign in to comment.