Feature: Adds new CLI parameter entrypoint

pytest_watcher/watcher.py

@@ -5,7 +5,8 @@
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Sequence, Tuple
+from typing import Sequence
+from dataclasses import dataclass
 
 from watchdog import events
 from watchdog.observers import Observer
@@ -14,6 +15,15 @@
 trigger = None
 
 
+@dataclass
+class ParsedArguments:
+    path: Path
+    now: bool
+    delay: float
+    entrypoint: str
+    pytest_args: Sequence[str]
+
+
 def emit_trigger():
     """
     Emits trigger to run pytest
@@ -57,7 +67,11 @@ def _run_pytest(args) -> None:
     subprocess.run(["pytest", *args])
 
 
-def _parse_arguments(args: Sequence[str]) -> Tuple[Path, bool, float, Sequence[str]]:
+def _run_entrypoint(entrypoint) -> None:
+    subprocess.run([entrypoint])
+
+
+def _parse_arguments(args: Sequence[str]) -> ParsedArguments:
     parser = argparse.ArgumentParser(
         prog="pytest_watcher",
         description="""
@@ -74,18 +88,33 @@ def _parse_arguments(args: Sequence[str]) -> Tuple[Path, bool, float, Sequence[s
         default=0.5,
         help="Watcher delay in seconds (default 0.5)",
     )
+    parser.add_argument(
+        "--entrypoint",
+        type=str,
+        default=None,
+        help="Use another executable to run the tests.",
+    )
 
     namespace, pytest_args = parser.parse_known_args(args)
 
-    return namespace.path, namespace.now, namespace.delay, pytest_args
+    return ParsedArguments(
+        path=namespace.path,
+        now=namespace.now,
+        delay=namespace.delay,
+        entrypoint=namespace.entrypoint,
+        pytest_args=pytest_args,
+    )
 
 
-def _run_main_loop(delay: float, pytest_args: Sequence[str]) -> None:
+def _run_main_loop(delay, pytest_args, entrypoint) -> None:
     global trigger
 
     now = datetime.now()
     if trigger and now - trigger > timedelta(seconds=delay):
-        _run_pytest(pytest_args)
+        if not entrypoint:
+            _run_pytest(pytest_args)
+        else:
+            _run_entrypoint(entrypoint)
 
         with trigger_lock:
             trigger = None
@@ -94,7 +123,9 @@ def _run_main_loop(delay: float, pytest_args: Sequence[str]) -> None:
 
 
 def run():
-    path_to_watch, now, delay, pytest_args = _parse_arguments(sys.argv[1:])
+    args = _parse_arguments(sys.argv[1:])
+    path_to_watch = args.path
+    now = args.now
 
     event_handler = EventHandler()
 
@@ -108,7 +139,7 @@ def run():
 
     try:
         while True:
-            _run_main_loop(delay, pytest_args)
+            _run_main_loop(args.delay, args.pytest_args, args.entrypoint)
     finally:
         observer.stop()
         observer.join()

tests/test_pytest_watcher.py

@@ -100,24 +100,28 @@ def test_emit_trigger():
 # fmt: off
 
 @pytest.mark.parametrize(
-    ("sys_args", "path_to_watch", "now", "delay", "pytest_args"),
+    ("sys_args", "path_to_watch", "now", "delay", "pytest_args", "entrypoint"),
     [
-        (["/home/"], "/home", False, 0.5, []),
-        (["/home/", "--lf", "--nf", "-x"], "/home", False, 0.5, ["--lf", "--nf", "-x"]),
-        (["/home/", "--lf", "--now", "--nf", "-x"], "/home", True, 0.5, ["--lf", "--nf", "-x"]),
-        (["/home/", "--now", "--lf", "--nf", "-x"], "/home", True, 0.5, ["--lf", "--nf", "-x"]),
-        ([".", "--lf", "--nf", "-x"], ".", False, 0.5, ["--lf", "--nf", "-x"]),
-        ([".", "--delay=0.2", "--lf", "--nf", "-x"], ".", False, 0.2, ["--lf", "--nf", "-x"]),
-        ([".", "--lf", "--nf", "--delay=0.3", "-x"], ".", False, 0.3, ["--lf", "--nf", "-x"]),
+        (["/home/"], "/home", False, 0.5, [], None),
+        (["/home/", "--lf", "--nf", "-x"], "/home", False, 0.5, ["--lf", "--nf", "-x"], None),
+        (["/home/", "--lf", "--now", "--nf", "-x"], "/home", True, 0.5, ["--lf", "--nf", "-x"], None),
+        (["/home/", "--now", "--lf", "--nf", "-x"], "/home", True, 0.5, ["--lf", "--nf", "-x"], None),
+        ([".", "--lf", "--nf", "-x"], ".", False, 0.5, ["--lf", "--nf", "-x"], None),
+        ([".", "--delay=0.2", "--lf", "--nf", "-x"], ".", False, 0.2, ["--lf", "--nf", "-x"], None),
+        ([".", "--lf", "--nf", "--delay=0.3", "-x"], ".", False, 0.3, ["--lf", "--nf", "-x"], None),
+        (["/home/", "--entrypoint", "tox"], "/home", False, 0.5, [], "tox"),
+        (["/home/", "--entrypoint", "make test"], "/home", False, 0.5, [], "make test"),
+
     ],
 )
-def test_parse_arguments(sys_args, path_to_watch, now, delay, pytest_args):
-    _path, _now, _delay, _pytest_args = watcher._parse_arguments(sys_args)
+def test_parse_arguments(sys_args, path_to_watch, now, delay, pytest_args, entrypoint):
+    _arguments = watcher._parse_arguments(sys_args)
 
-    assert str(_path) == path_to_watch
-    assert _now == now
-    assert _delay == delay
-    assert _pytest_args == pytest_args
+    assert str(_arguments.path) == path_to_watch
+    assert _arguments.now == now
+    assert _arguments.delay == delay
+    assert _arguments.entrypoint == entrypoint
+    assert _arguments.pytest_args == pytest_args
 
 # fmt: on
 
@@ -127,7 +131,7 @@ def test_run_main_loop_no_trigger(
 ):
     watcher.trigger = None
 
-    watcher._run_main_loop(5, ["--lf"])
+    watcher._run_main_loop(5, ["--lf"], None)
 
     mock_subprocess_run.assert_not_called()
     mock_time_sleep.assert_called_once_with(5)
@@ -142,7 +146,7 @@ def test_run_main_loop_trigger_fresh(
     watcher.trigger = datetime(2020, 1, 1, 0, 0, 0)
 
     with freeze_time("2020-01-01 00:00:04"):
-        watcher._run_main_loop(5, ["--lf"])
+        watcher._run_main_loop(5, ["--lf"], None)
 
     mock_subprocess_run.assert_not_called()
     mock_time_sleep.assert_called_once_with(5)
@@ -157,7 +161,7 @@ def test_run_main_loop_trigger(
     watcher.trigger = datetime(2020, 1, 1, 0, 0, 0)
 
     with freeze_time("2020-01-01 00:00:06"):
-        watcher._run_main_loop(5, ["--lf"])
+        watcher._run_main_loop(5, ["--lf"], None)
 
     mock_subprocess_run.assert_called_once_with(["pytest", "--lf"])
     mock_time_sleep.assert_called_once_with(5)
@@ -185,7 +189,7 @@ def test_run(
 
     mock_emit_trigger.assert_not_called()
 
-    mock_run_main_loop.assert_called_once_with(0.5, ["--lf", "--nf"])
+    mock_run_main_loop.assert_called_once_with(0.5, ["--lf", "--nf"], None)
 
 
 def test_run_now(
@@ -208,4 +212,29 @@ def test_run_now(
 
     mock_emit_trigger.assert_called_once_with()
 
-    mock_run_main_loop.assert_called_once_with(0.5, ["--lf", "--nf"])
+    mock_run_main_loop.assert_called_once_with(0.5, ["--lf", "--nf"], None)
+
+
+@pytest.mark.parametrize("entrypoint", [("tox"), ("'make test'")])
+def test_run_entrypoint(
+    mocker: MockerFixture,
+    mock_observer: MagicMock,
+    mock_emit_trigger: MagicMock,
+    mock_run_main_loop: MagicMock,
+    entrypoint: str,
+):
+    args = ["ptw", ".", "--lf", "--nf", "--now", "--entrypoint", entrypoint]
+
+    mocker.patch.object(sys, "argv", args)
+
+    with pytest.raises(InterruptedError):
+        watcher.run()
+
+    mock_observer.assert_called_once_with()
+    observer_instance = mock_observer.return_value
+    observer_instance.schedule.assert_called_once()
+    observer_instance.start.assert_called_once()
+
+    mock_emit_trigger.assert_called_once_with()
+
+    mock_run_main_loop.assert_called_once_with(0.5, ["--lf", "--nf"], entrypoint)