Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature: Adds new CLI parameter entrypoint #12

Merged
merged 16 commits into from
Dec 11, 2022
Merged
48 changes: 36 additions & 12 deletions pytest_watcher/watcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,10 @@
import sys
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Sequence, Tuple
from typing import Sequence

from watchdog import events
from watchdog.observers import Observer
Expand All @@ -14,6 +15,15 @@
trigger = None


@dataclass
class ParsedArguments:
path: Path
now: bool
delay: float
runner: str
runner_args: Sequence[str]


def emit_trigger():
"""
Emits trigger to run pytest
Expand Down Expand Up @@ -53,11 +63,11 @@ def process_event(self, event: events.FileSystemEvent) -> None:
emit_trigger()


def _run_pytest(args) -> None:
subprocess.run(["pytest", *args])
def _invoke_runner(runner: str, args: Sequence[str]) -> None:
subprocess.run([runner, *args])


def _parse_arguments(args: Sequence[str]) -> Tuple[Path, bool, float, Sequence[str]]:
def _parse_arguments(args: Sequence[str]) -> ParsedArguments:
parser = argparse.ArgumentParser(
prog="pytest_watcher",
description="""
Expand All @@ -74,18 +84,30 @@ 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(
"--runner",
type=str,
default="pytest",
help="Use another executable to run the tests.",
)

namespace, pytest_args = parser.parse_known_args(args)
namespace, runner_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,
runner=namespace.runner,
runner_args=runner_args,
)


def _run_main_loop(delay: float, pytest_args: Sequence[str]) -> None:
def _run_main_loop(*, runner: str, runner_args: Sequence[str], delay: float) -> None:
global trigger

now = datetime.now()
if trigger and now - trigger > timedelta(seconds=delay):
_run_pytest(pytest_args)
_invoke_runner(runner, runner_args)

with trigger_lock:
trigger = None
Expand All @@ -94,21 +116,23 @@ 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:])

event_handler = EventHandler()

observer = Observer()

observer.schedule(event_handler, path_to_watch, recursive=True)
observer.schedule(event_handler, args.path, recursive=True)
observer.start()

if now:
if args.now:
emit_trigger()

try:
while True:
_run_main_loop(delay, pytest_args)
_run_main_loop(
runner=args.runner, runner_args=args.runner_args, delay=args.delay
)
finally:
observer.stop()
observer.join()
73 changes: 54 additions & 19 deletions tests/test_pytest_watcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -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", "runner_args", "runner"),
[
(["/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, [], "pytest"),
(["/home/", "--lf", "--nf", "-x"], "/home", False, 0.5, ["--lf", "--nf", "-x"], "pytest"),
(["/home/", "--lf", "--now", "--nf", "-x"], "/home", True, 0.5, ["--lf", "--nf", "-x"], "pytest"),
(["/home/", "--now", "--lf", "--nf", "-x"], "/home", True, 0.5, ["--lf", "--nf", "-x"], "pytest"),
([".", "--lf", "--nf", "-x"], ".", False, 0.5, ["--lf", "--nf", "-x"], "pytest"),
([".", "--delay=0.2", "--lf", "--nf", "-x"], ".", False, 0.2, ["--lf", "--nf", "-x"], "pytest"),
([".", "--lf", "--nf", "--delay=0.3", "-x"], ".", False, 0.3, ["--lf", "--nf", "-x"], "pytest"),
(["/home/", "--runner", "tox"], "/home", False, 0.5, [], "tox"),
(["/home/", "--runner", "'make test'"], "/home", False, 0.5, [], "'make test'"),
(["/home/", "--runner", "make", "test"], "/home", False, 0.5, ["test"], "make"),
],
)
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, runner_args, runner):
_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.runner == runner
assert _arguments.runner_args == runner_args

# fmt: on

Expand All @@ -127,7 +131,7 @@ def test_run_main_loop_no_trigger(
):
watcher.trigger = None

watcher._run_main_loop(5, ["--lf"])
watcher._run_main_loop(runner="pytest", runner_args=["--lf"], delay=5)

mock_subprocess_run.assert_not_called()
mock_time_sleep.assert_called_once_with(5)
Expand All @@ -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(runner="pytest", runner_args=["--lf"], delay=5)

mock_subprocess_run.assert_not_called()
mock_time_sleep.assert_called_once_with(5)
Expand All @@ -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(runner="pytest", runner_args=["--lf"], delay=5)

mock_subprocess_run.assert_called_once_with(["pytest", "--lf"])
mock_time_sleep.assert_called_once_with(5)
Expand Down Expand Up @@ -185,7 +189,9 @@ 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(
runner="pytest", runner_args=["--lf", "--nf"], delay=0.5
)


def test_run_now(
Expand All @@ -208,4 +214,33 @@ 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(
runner="pytest", runner_args=["--lf", "--nf"], delay=0.5
)


@pytest.mark.parametrize("runner", [("tox"), ("'make test'")])
def test_invoke_runner(
mocker: MockerFixture,
mock_observer: MagicMock,
mock_emit_trigger: MagicMock,
mock_run_main_loop: MagicMock,
runner: str,
):
args = ["ptw", ".", "--lf", "--nf", "--now", "--runner", runner]

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(
runner=runner, runner_args=["--lf", "--nf"], delay=0.5
)
2 changes: 1 addition & 1 deletion tox.ini
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tox]
isolated_build = True
envlist = py36,py37,py38,py39,py310,py311
envlist = py37,py38,py39,py310,py311

[tox:.package]
basepython = python3
Expand Down