diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..d5572d0 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,2 @@ +[run] +source=pytest_watch diff --git a/.gitignore b/.gitignore index 28582d3..a363305 100644 --- a/.gitignore +++ b/.gitignore @@ -9,8 +9,14 @@ env .cache *.so *.py[cod] +.eggs +.coverage +.coverage.* +.pytest_cache +.python-version # OS-specific files .DS_Store Desktop.ini Thumbs.db +coverage.xml diff --git a/.travis.yml b/.travis.yml new file mode 100644 index 0000000..c639c4d --- /dev/null +++ b/.travis.yml @@ -0,0 +1,22 @@ +language: python +python: + - 2.7 + - pypy2.7-5.8.0 + - 3.4 + - 3.5.4 + - 3.6 + - pypy3.5-5.8.0 + +os: + - linux + +cache: pip + +install: + - pip install -e ".[qa]" + +script: +- python -m pytest --cov=pytest_watch --cov-report=term-missing -v pytest_watch + +after_success: + - codecov --token=$CODECOV_TOKEN diff --git a/README.md b/README.md index c5987cb..db1b98f 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,7 @@ Options: --ignore Ignore directory from being watched and during collection (multi-allowed). --ext Comma-separated list of file extensions that can - trigger a new test run when changed (default: .py). + trigger a new test run when changed [default: .py]. Use --ext=* to allow any file (including .pyc). --config Load configuration from `file` instead of trying to locate one of the implicit configuration files. @@ -129,7 +129,7 @@ Options: --pdb Start the interactive Python debugger on errors. This also enables --wait to prevent pdb interruption. --spool Re-run after a delay (in milliseconds), allowing for - more file system events to queue up (default: 200 ms). + more file system events to queue up [default: 200]. -p --poll Use polling instead of OS events (useful in VMs). -v --verbose Increase verbosity of the output. -q --quiet Decrease verbosity of the output (precedence over -v). diff --git a/appveyor.yml b/appveyor.yml new file mode 100644 index 0000000..e2acd60 --- /dev/null +++ b/appveyor.yml @@ -0,0 +1,4 @@ +version: 1.0.{build} + +build_script: + - cmd: python setup.py test diff --git a/pytest_watch/__main__.py b/pytest_watch/__main__.py index 0fc492e..cb2dd72 100644 --- a/pytest_watch/__main__.py +++ b/pytest_watch/__main__.py @@ -8,12 +8,16 @@ :license: MIT, see LICENSE for more details. """ -import os -import sys +def run_cli(): + import os + import sys -if __name__ == '__main__': sys.path.append(os.path.dirname(__file__)) from pytest_watch.command import main main() + + +if __name__ == '__main__': + run_cli() diff --git a/pytest_watch/command.py b/pytest_watch/command.py index f87a248..f0475da 100644 --- a/pytest_watch/command.py +++ b/pytest_watch/command.py @@ -11,7 +11,7 @@ --ignore Ignore directory from being watched and during collection (multi-allowed). --ext Comma-separated list of file extensions that can - trigger a new test run when changed (default: .py). + trigger a new test run when changed [default: .py]. Use --ext=* to allow any file (including .pyc). --config Load configuration from `file` instead of trying to locate one of the implicit configuration files. @@ -29,7 +29,7 @@ --pdb Start the interactive Python debugger on errors. This also enables --wait to prevent pdb interruption. --spool Re-run after a delay (in milliseconds), allowing for - more file system events to queue up (default: 200 ms). + more file system events to queue up [default: 200]. -p --poll Use polling instead of OS events (useful in VMs). -v --verbose Increase verbosity of the output. -q --quiet Decrease verbosity of the output (precedence over -v). @@ -55,6 +55,8 @@ def main(argv=None): """ The entry point of the application. + + argv -- List of strings to parse. The default is taken from sys.argv[1:]. """ if argv is None: argv = sys.argv[1:] @@ -68,6 +70,8 @@ def main(argv=None): # Get paths and initial pytest arguments directories = args[''] pytest_args = list(directories) + + # Merge pytest arguments and directories if '--' in directories: index = directories.index('--') directories = directories[:index] @@ -76,6 +80,8 @@ def main(argv=None): # Adjust pytest and --collect-only args for ignore in args['--ignore']: pytest_args.extend(['--ignore', ignore]) + + # Set pytest config file if args['--config']: pytest_args.extend(['-c', args['--config']]) @@ -87,23 +93,27 @@ def main(argv=None): if args['--pdb']: pytest_args.append('--pdb') - # Parse extensions + # Parse extensions [default: .py] if args['--ext'] == '*': extensions = ALL_EXTENSIONS elif args['--ext']: extensions = [('.' if not e.startswith('.') else '') + e for e in args['--ext'].split(',')] - else: - extensions = None # Parse numeric arguments spool = args['--spool'] - if spool is not None: - try: - spool = int(spool) - except ValueError: - sys.stderr.write('Error: Spool must be an integer.\n') - return 2 + try: + spool = int(spool) + except ValueError: + sys.stderr.write('Error: Spool (--spool {}) must be an integer.\n' + .format(spool)) + return 2 + + if spool < 0: + sys.stderr.write('Error: Spool value(--spool {}) must be positive' + ' integer\n' + .format(spool)) + return 2 # Run pytest and watch for changes return watch(directories=directories, diff --git a/pytest_watch/config.py b/pytest_watch/config.py index 3312636..87bc247 100644 --- a/pytest_watch/config.py +++ b/pytest_watch/config.py @@ -71,8 +71,8 @@ def _collect_config(pytest_args, silent=True): except (Exception, SystemExit): pass # Print message and run again without silencing - print('Error: Could not run --collect-only to handle the pytest config ' - 'file. Trying again without silencing output...', + print('Error: Could not run --collect-only to handle the pytest ' + 'config file. Trying again without silencing output...', file=sys.stderr) return _run_pytest_collect(pytest_args) diff --git a/pytest_watch/helpers.py b/pytest_watch/helpers.py index 157565e..87d14e9 100644 --- a/pytest_watch/helpers.py +++ b/pytest_watch/helpers.py @@ -1,3 +1,4 @@ +import ctypes import os import signal import subprocess @@ -51,12 +52,15 @@ def dequeue_all(queue, spool=None): return items +def canonize_path(path): + return os.path.realpath(path) + + def samepath(left, right): """ - Determines whether two paths are the same. + Determines whether two paths are the same based on their absolute paths. """ - return (os.path.abspath(os.path.normcase(left)) == - os.path.abspath(os.path.normcase(right))) + return canonize_path(left) == canonize_path(right) def send_keyboard_interrupt(proc): @@ -70,7 +74,6 @@ def send_keyboard_interrupt(proc): os.kill(0, signal.CTRL_C_EVENT) except AttributeError: # Python 2.6 and below - import ctypes ctypes.windll.kernel32.GenerateConsoleCtrlEvent(0, 0) # Immediately throws KeyboardInterrupt from the simulated CTRL-C proc.wait() diff --git a/pytest_watch/summary.py b/pytest_watch/summary.py new file mode 100644 index 0000000..dee4d31 --- /dev/null +++ b/pytest_watch/summary.py @@ -0,0 +1,81 @@ +import time + +from colorama import Fore, Style + + +STYLE_BRIGHT = Fore.WHITE + Style.NORMAL + Style.BRIGHT +STYLE_HIGHLIGHT = Fore.CYAN + Style.NORMAL + Style.BRIGHT + + +def _reduce_events(events): + # FUTURE: Reduce ['a -> b', 'b -> c'] renames to ['a -> c'] + + creates = [] + moves = [] + for event, src, dest in events: + if event == FileCreatedEvent: + creates.append(dest) + if event == FileMovedEvent: + moves.append(dest) + + seen = [] + filtered = [] + for event, src, dest in events: + # Skip 'modified' event during 'created' + if src in creates and event != FileCreatedEvent: + continue + + # Skip 'modified' event during 'moved' + if src in moves: + continue + + # Skip duplicate events + if src in seen: + continue + seen.append(src) + + filtered.append((event, src, dest)) + return filtered + + +def _bright(arg): + return STYLE_BRIGHT + arg + Style.RESET_ALL + + +def _highlight(arg): + return STYLE_HIGHLIGHT + arg + Style.RESET_ALL + + +def show_summary(argv, events, verbose=False): + command = ' '.join(argv) + bright = _bright + highlight = _highlight + + time_stamp = time.strftime("%c", time.localtime(time.time())) + run_command_info = '[{}] Running: {}'.format(time_stamp, + highlight(command)) + if not events: + print(run_command_info) + return + + events = _reduce_events(events) + if verbose: + lines = ['Changes detected:'] + m = max(map(len, map(lambda e: VERBOSE_EVENT_NAMES[e[0]], events))) + for event, src, dest in events: + event = VERBOSE_EVENT_NAMES[event].ljust(m) + lines.append(' {} {}'.format( + event, + highlight(src + (' -> ' + dest if dest else '')))) + lines.append('') + lines.append(run_command_info) + else: + lines = [] + for event, src, dest in events: + lines.append('{} detected: {}'.format( + EVENT_NAMES[event], + bright(src + (' -> ' + dest if dest else '')))) + lines.append('') + lines.append(run_command_info) + + print('\n'.join(lines)) diff --git a/pytest_watch/tests/__init__.py b/pytest_watch/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pytest_watch/tests/conftest.py b/pytest_watch/tests/conftest.py new file mode 100644 index 0000000..e04ba45 --- /dev/null +++ b/pytest_watch/tests/conftest.py @@ -0,0 +1,20 @@ +import pytest + + +@pytest.fixture +def merge_config_callee(mocker): + m = mocker.patch("pytest_watch.command.merge_config", + side_effect=lambda *args, **kwargs: True) + return m + + +@pytest.fixture +def beep_mock(mocker): + return mocker.patch("pytest_watch.helpers.beep") + + +@pytest.fixture +def watch_callee(mocker): + watch_mock = mocker.patch("pytest_watch.command.watch") + watch_mock.return_value.side_effect = lambda *args, **kwargs: 0 + return watch_mock diff --git a/pytest_watch/tests/test_command.py b/pytest_watch/tests/test_command.py new file mode 100644 index 0000000..da697a0 --- /dev/null +++ b/pytest_watch/tests/test_command.py @@ -0,0 +1,469 @@ +import pytest +import sys +import shutil + +try: + from unittest.mock import patch +except ImportError: + from mock import patch + +from pytest_watch.command import main +from pytest_watch.constants import ALL_EXTENSIONS + + +if sys.version_info[0] < 3: + from io import BytesIO as io_mock +else: + from io import StringIO as io_mock + + +@pytest.mark.usefixtures("watch_callee") +class TestCLIArguments(): + + def _get_default_args(self): + return dict( + directories=[], + ignore=[], + extensions=[".py"], + beep_on_failure=True, + auto_clear=False, + wait=False, + beforerun=None, + afterrun=None, + onpass=None, + onfail=None, + onexit=None, + runner=None, + spool=200, + poll=False, + verbose=False, + quiet=False, + pytest_args=[] + ) + + def test_default_parameters(self, watch_callee): + main([]) + + assert 1 == watch_callee.call_count + watch_callee.assert_called_once_with(**self._get_default_args()) + + def test_empty_argv(self, watch_callee): + sys.argv[1:] = [] + + main() + + assert 1 == watch_callee.call_count + watch_callee.assert_called_once_with(**self._get_default_args()) + + +@pytest.mark.usefixtures("watch_callee") +class TestPdbArgument(): + + def test_default_pdb_argument(self, watch_callee): + sys.argv[1:] = [] + + main() + + assert not watch_callee.call_args[1]["wait"] + + assert "pdb" not in watch_callee.call_args[1] + + assert "--pdb" not in watch_callee.call_args[1]["pytest_args"] + + def test_pdb_argument(self, watch_callee): + main(["--pdb"]) + + assert watch_callee.call_args[1]["wait"] + + assert "pdb" not in watch_callee.call_args[1] + + assert "--pdb" in watch_callee.call_args[1]["pytest_args"] + + def test_pdb_and_wait_arguments(self, watch_callee): + main("--pdb --wait".split()) + + assert watch_callee.call_args[1]["wait"] + + assert "pdb" not in watch_callee.call_args[1] + + assert "--pdb" in watch_callee.call_args[1]["pytest_args"] + + def test_pdb_off_and_wait_on_arguments(self, watch_callee): + main("--wait".split()) + + assert watch_callee.call_args[1]["wait"] + + assert "pdb" not in watch_callee.call_args[1] + + assert "--pdb" not in watch_callee.call_args[1]["pytest_args"] + + +@pytest.mark.usefixtures("watch_callee", "merge_config_callee") +class TestConfigArgument(): + + def test_default_config(self, watch_callee, merge_config_callee): + sys.argv[1:] = [] + + main() + + assert "config" not in watch_callee.call_args[1] + assert "-c" not in watch_callee.call_args[1]["pytest_args"] + + def test_config_argument(self, watch_callee, merge_config_callee): + self._assert_config_file(watch_callee, "pytest.ini") + watch_callee.reset_mock() + self._assert_config_file(watch_callee, "custom_config_file.txt") + + def _assert_config_file(self, watch_callee, filename): + main(["--config", filename]) + + assert "config" not in watch_callee.call_args[1] + + pytest_args = watch_callee.call_args[1]["pytest_args"] + assert "-c" in pytest_args + assert filename == pytest_args[pytest_args.index("-c")+1] + + +@pytest.mark.usefixtures("watch_callee", "tmpdir_factory") +class TestIgnoreArgument(): + def test_default_ignore_argument(self, watch_callee): + sys.argv[1:] = [] + + main() + + assert [] == watch_callee.call_args[1]["ignore"] + + assert "--ignore" not in watch_callee.call_args[1]["pytest_args"] + + def test_ignore_argument(self, watch_callee): + main(["--ignore", "pytest_watch"]) + + assert ["pytest_watch"] == watch_callee.call_args[1]["ignore"] + + assert "--ignore" in watch_callee.call_args[1]["pytest_args"] + + def test_multiple_ignore_argument(self, tmpdir_factory, watch_callee): + directories = [] + argv = [] + + for _ in range(2): + new_dir = str(tmpdir_factory.mktemp("_")) + argv.append("--ignore") + argv.append(new_dir) + directories.append(new_dir) + + main(argv) + + assert directories == watch_callee.call_args[1]["ignore"] + pytest_args = watch_callee.call_args[1]["pytest_args"] + assert "--ignore" in pytest_args + + ignore_idx = pytest_args.index("--ignore") + assert argv == pytest_args + + def test_multiple_ignore_argument_conflict(self, tmpdir_factory, + watch_callee): + directories = [] + argv = [] + + for _ in range(2): + new_dir = str(tmpdir_factory.mktemp("_")) + argv.append("--ignore") + argv.append(new_dir) + directories.append(new_dir) + + argv.append("--") + argv.append("--ignore") + argv.append(str(tmpdir_factory.mktemp("_"))) + + main(argv) + + assert directories == watch_callee.call_args[1]["ignore"] + + pytest_args = watch_callee.call_args[1]["pytest_args"] + assert "--ignore" in pytest_args + assert 3 == pytest_args.count("--ignore") + + +@pytest.mark.usefixtures("watch_callee") +class TestSpoolArguments(): + + def test_zero_spool_value(self, watch_callee): + main("--spool 0".split()) + + assert "spool" in watch_callee.call_args[1] + assert 0 == watch_callee.call_args[1]["spool"] + + assert 1 == watch_callee.call_count + + def test_positive_spool_value(self, watch_callee): + main("--spool 2000".split()) + + assert "spool" in watch_callee.call_args[1] + assert 2000 == watch_callee.call_args[1]["spool"] + assert 1 == watch_callee.call_count + + watch_callee.reset_mock() + + main("--spool 20".split()) + + assert "spool" in watch_callee.call_args[1] + assert 20 == watch_callee.call_args[1]["spool"] + assert 1 == watch_callee.call_count + + def test_default_spool_value(self, watch_callee): + main([]) + + assert "spool" in watch_callee.call_args[1] + assert 200 == watch_callee.call_args[1]["spool"] + assert 1 == watch_callee.call_count + + def _assert_spool_error(self, watch_callee, value, err): + with patch("pytest_watch.command.sys.stderr", new=io_mock()) as out: + assert 2 == main(["--spool", value]) + assert err == out.getvalue(), ("Status code for invalid 'spool'" + " argument should be 2") + watch_callee.assert_not_called() + + def test_cause_error_for_negative_spool_values(self, watch_callee): + err = "Error: Spool value(--spool -1) must be positive integer\n" + self._assert_spool_error(watch_callee, value="-1", err=err) + + def test_cause_error_for_invalid_spool_values(self, watch_callee): + value = "abc" + self._assert_spool_error(watch_callee, value=value, + err=str("Error: Spool (--spool {}) must be" + " an integer.\n").format(value)) + + value = "@" + self._assert_spool_error(watch_callee, value=value, + err=str("Error: Spool (--spool {}) must be" + " an integer.\n").format(value)) + + value = "[]" + self._assert_spool_error(watch_callee, value=value, + err=str("Error: Spool (--spool {}) must be" + " an integer.\n").format(value)) + + +@pytest.mark.usefixtures("watch_callee") +class TestExtensionsArguments(): + + def test_default_extensions(self, watch_callee): + main([]) + + assert "extensions" in watch_callee.call_args[1] + + assert [".py"] == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + def test_all_extensions(self, watch_callee): + main("--ext *".split()) + + assert isinstance(watch_callee.call_args[1]["extensions"], object) + + assert None is not watch_callee.call_args[1]["extensions"] + + assert ALL_EXTENSIONS == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + def test_single_without_dot_extensions(self, watch_callee): + main("--ext py".split()) + + assert "extensions" in watch_callee.call_args[1] + + assert [".py"] == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + def test_single_with_dot_extensions(self, watch_callee): + main("--ext .py".split()) + + assert "extensions" in watch_callee.call_args[1] + + assert [".py"] == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + def test_multiple_extensions(self, watch_callee): + main("--ext .py,.html".split()) + + assert "extensions" in watch_callee.call_args[1] + + assert [".py", ".html"] == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + def test_multiple_with_and_without_dots_extensions(self, watch_callee): + main("--ext .py,html".split()) + + assert "extensions" in watch_callee.call_args[1] + + assert [".py", ".html"] == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + watch_callee.reset_mock() + + main("--ext py,.html".split()) + + assert "extensions" in watch_callee.call_args[1] + + assert [".py", ".html"] == watch_callee.call_args[1]["extensions"] + + assert 1 == watch_callee.call_count + + +@pytest.mark.usefixtures("watch_callee", "tmpdir", "merge_config_callee") +class TestDirectoriesAndPytestArgsArgumentsSplit(): + + def test_no_directory_empty_pytest_arg(self, watch_callee): + main(["--"]) + + assert "pytest_args" in watch_callee.call_args[1] + assert [] == watch_callee.call_args[1]["pytest_args"] + assert 1 == watch_callee.call_count + + def test_no_directory_single_pytest_arg(self, watch_callee): + main("-- --pdb".split()) + + assert "pytest_args" in watch_callee.call_args[1] + + assert ["--pdb"] == watch_callee.call_args[1]["pytest_args"] + + assert 1 == watch_callee.call_count + + def test_no_directory_multiple_pytest_args(self, watch_callee, merge_config_callee): + main("-- --pdb --cov=.".split()) + + assert 1 == merge_config_callee.call_count + + assert "pytest_args" in watch_callee.call_args[1] + + assert ["--pdb", "--cov=."] == watch_callee.call_args[1]["pytest_args"] + + assert 1 == watch_callee.call_count + + def test_multiple_directory_no_pytest_args(self, tmpdir_factory, + watch_callee): + directories = [str(tmpdir_factory.mktemp("_")) for _ in range(2)] + directories.append("--") + + main(directories) + + assert "pytest_args" in watch_callee.call_args[1] + assert "directories" in watch_callee.call_args[1] + + fetched_pytest_args = watch_callee.call_args[1]["pytest_args"] + fetched_directories = watch_callee.call_args[1]["directories"] + + assert directories[:-1] == fetched_directories + + assert len(fetched_pytest_args) > 1 + assert len(fetched_pytest_args) == len(fetched_directories) + assert fetched_directories == fetched_pytest_args + assert 1 == watch_callee.call_count + + def test_single_directory_no_pytest_args(self, watch_callee, tmpdir): + root_tmp = str(tmpdir) + + main([root_tmp, "--"]) + + assert "pytest_args" in watch_callee.call_args[1] + + pytest_args = watch_callee.call_args[1]["pytest_args"] + assert len(pytest_args) > 0 + + assert [root_tmp] == pytest_args + assert 1 == watch_callee.call_count + + fetched_directories = watch_callee.call_args[1]["directories"] + assert [root_tmp] == fetched_directories + + def test_single_directory_single_pytest_args(self, watch_callee, tmpdir): + root_tmp = str(tmpdir) + vargs = [root_tmp, "--", "--pdb"] + + main(vargs) + + assert "pytest_args" in watch_callee.call_args[1] + assert "directories" in watch_callee.call_args[1] + + fetched_pytest_args = watch_callee.call_args[1]["pytest_args"] + fetched_directories = watch_callee.call_args[1]["directories"] + + assert [vargs[0]] == fetched_directories + + pytest_args = watch_callee.call_args[1]["pytest_args"] + assert len(pytest_args) > 0 + assert [root_tmp, "--pdb"] == pytest_args + assert 1 == watch_callee.call_count + + assert [root_tmp] == fetched_directories + + def test_single_directory_multiple_pytest_args(self, watch_callee, tmpdir): + root_tmp = str(tmpdir) + vargs = [root_tmp, "--", "--pdb", "--cov=."] + + main(vargs) + + assert "pytest_args" in watch_callee.call_args[1] + assert "directories" in watch_callee.call_args[1] + + fetched_pytest_args = watch_callee.call_args[1]["pytest_args"] + fetched_directories = watch_callee.call_args[1]["directories"] + + assert [vargs[0]] == fetched_directories + + pytest_args = watch_callee.call_args[1]["pytest_args"] + assert len(pytest_args) > 0 + + assert [root_tmp, "--pdb", "--cov=."] == pytest_args + + assert 1 == watch_callee.call_count + + assert [root_tmp] == fetched_directories + + +@pytest.mark.usefixtures("watch_callee", "tmpdir_factory") +class TestDirectoriesArguments(): + + def test_default_directories(self, watch_callee): + directories = [] + + main(directories) + + assert "directories" in watch_callee.call_args[1] + + fetched_directories = watch_callee.call_args[1]["directories"] + assert directories == fetched_directories + assert 1 == watch_callee.call_count + + def test_single_directory(self, watch_callee, tmpdir): + root_tmp = str(tmpdir) + directories = [root_tmp] + self._assert_directories(directories, watch_callee) + + def test_two_directory_values(self, tmpdir_factory, watch_callee): + directories = [str(tmpdir_factory.mktemp("_")) for _ in range(2)] + self._assert_directories(directories, watch_callee) + + def test_ten_directory_values(self, tmpdir_factory, watch_callee): + directories = [str(tmpdir_factory.mktemp("_")) for _ in range(10)] + self._assert_directories(directories, watch_callee) + + def _assert_directories(self, directories, watch_callee): + assert len(directories) > 0, \ + "Multiple directories should be declared for this test case" + + main(directories) + + assert "directories" in watch_callee.call_args[1] + + fetched_directories = watch_callee.call_args[1]["directories"] + assert len(directories) == len(fetched_directories) + + assert directories == fetched_directories + assert 1 == watch_callee.call_count diff --git a/pytest_watch/tests/test_helpers.py b/pytest_watch/tests/test_helpers.py new file mode 100644 index 0000000..aeaee35 --- /dev/null +++ b/pytest_watch/tests/test_helpers.py @@ -0,0 +1,203 @@ +import os +import signal +import subprocess +import sys +from time import sleep + +try: + from queue import Queue +except ImportError: + from Queue import Queue + +import pytest +from pytest_watch import helpers + + +@pytest.fixture +def windows_ctrlc_mock(mocker): + k32_mock = mocker.patch("pytest_watch.helpers.ctypes") + ctrlc_mock = mocker.patch.object(k32_mock.windll.kernel32, + "GenerateConsoleCtrlEvent") + return ctrlc_mock + + +@pytest.fixture +def python_version_proc(): + return subprocess.Popen(sys.executable, + shell=helpers.is_windows) + + +def test_linux_clear_with_clear_command(mocker): + is_windows = mocker.patch.dict("pytest_watch.helpers.__dict__", + {"is_windows": False}) + + call_mock = mocker.patch("pytest_watch.helpers.subprocess.call") + + helpers.clear() + + assert 1 == call_mock.call_count + assert ("clear",) == call_mock.call_args[0] + assert dict(shell=True) == call_mock.call_args[1] + + +def test_windows_clear_with_cls_command(mocker): + is_windows = mocker.patch.dict("pytest_watch.helpers.__dict__", + {"is_windows": True}) + + call_mock = mocker.patch("pytest_watch.helpers.subprocess.call") + + helpers.clear() + + assert 1 == call_mock.call_count + assert ("cls",) == call_mock.call_args[0] + assert dict(shell=True) == call_mock.call_args[1] + + +def test_linux_process_kill_is_called(mocker, python_version_proc): + is_windows = mocker.patch.dict("pytest_watch.helpers.__dict__", + {"is_windows": False}) + + os_mock = mocker.patch("pytest_watch.helpers.os") + + kill_mock = mocker.patch.object(os_mock, "kill", + side_effect=lambda pid, s: pid) + + helpers.send_keyboard_interrupt(python_version_proc) + + assert 1 == kill_mock.call_count + assert (python_version_proc.pid, signal.SIGINT) == kill_mock.call_args[0] + + +def test_windows_process_kill_for_python26upper_is_called(mocker, + python_version_proc, + windows_ctrlc_mock): + ctrl_c_code = signal.SIGINT + + is_windows = mocker.patch.dict("pytest_watch.helpers.__dict__", + {"is_windows": True}) + ctrl_c_event = mocker.patch.dict("pytest_watch.helpers.signal.__dict__", + {"CTRL_C_EVENT": ctrl_c_code}) + + os_mock = mocker.patch("pytest_watch.helpers.os") + + kill_mock = mocker.patch.object(os_mock, "kill", + side_effect=KeyboardInterrupt) + + mocker.patch.object(python_version_proc, "wait") + helpers.send_keyboard_interrupt(python_version_proc) + + assert 0 == windows_ctrlc_mock.call_count + assert 1 == kill_mock.call_count + assert (0, ctrl_c_code) == kill_mock.call_args[0] + + +def test_windows_process_kill_for_python26_is_called(mocker, + windows_ctrlc_mock, + python_version_proc): + ctrl_c_code = signal.SIGINT + + is_windows = mocker.patch.dict("pytest_watch.helpers.__dict__", + {"is_windows": True}) + ctrl_c_event = mocker.patch.dict("pytest_watch.helpers.signal.__dict__", + {"CTRL_C_EVENT": ctrl_c_code}) + + os_mock = mocker.patch("pytest_watch.helpers.os") + + kill_mock = mocker.patch.object(os_mock, "kill", + side_effect=AttributeError) + + mocker.patch.object(python_version_proc, "wait") + helpers.send_keyboard_interrupt(python_version_proc) + + assert 1 == windows_ctrlc_mock.call_count + assert (0, 0) == windows_ctrlc_mock.call_args[0] + + +def test_dequeall_from_an_empty_queue_with_no_spool(): + q = Queue() + assert [] == helpers.dequeue_all(q, 0) + + +def test_dequeall_from_a_single_queue_with_no_spool(): + q = Queue() + q.put("element 1") + assert ["element 1"] == helpers.dequeue_all(q, 0) + + +def test_dequeall_from_multi_queue_with_no_spool(): + q = Queue() + q.put("element 1") + assert ["element 1"] == helpers.dequeue_all(q, 0) + q.put("element 2") + q.put("element 3") + assert ["element 2", "element 3"] == helpers.dequeue_all(q, 0) + + +def test_dequeall_from_multi_with_spool_200(mocker): + def _is_first_empty(): + empty = False + yield empty + + sleep_mock = mocker.patch("pytest_watch.helpers.sleep", wrap=sleep) + q = Queue() + mocker.patch.object(q, "empty", side_effect=_is_first_empty) + q.put("element 1") + q.put("element 2") + dequeued = helpers.dequeue_all(q) + sleep(.3) + q.put("element 3") + assert (.2,) == sleep_mock.call_args[0] + assert ["element 1", "element 2"] == dequeued + assert ["element 3"] == helpers.dequeue_all(q, 0) + + +def test_samepath_for_non_existent_file_without_errors(tmpdir): + samedir = tmpdir.mkdir("samepath") + file1 = samedir.join("file1.txt") + with open(file1.strpath, "w") as f: + f.write(".") + file2 = samedir.join("inexistent.txt") + + assert file1.exists() + assert not file2.exists() + assert not helpers.samepath(file1.strpath, file2.strpath) + + +@pytest.mark.skipif(sys.platform == 'win32', + reason="does not run on windows. System doesnt support symlinks") +def test_samepath_for_name_spaced_symbolic_link(tmpdir): + samedir = tmpdir.mkdir("samepath") + file1 = samedir.join("file1.txt") + with open(file1.strpath, "w") as f: + f.write(".") + symlink = samedir.join("Symbolic Link.txt") + symlink.mksymlinkto(file1) + + assert os.path.islink(symlink.strpath) + assert helpers.samepath(file1.strpath, symlink.strpath) + + +@pytest.mark.skipif(sys.platform == 'win32', + reason="does not run on windows. System doesnt support symlinks") +def test_samepath_for_symbolic_link(tmpdir): + samedir = tmpdir.mkdir("samepath") + file1 = samedir.join("file1.txt") + with open(file1.strpath, "w") as f: + f.write(".") + symlink = samedir.join("symlink1.txt") + symlink.mksymlinkto(file1) + + assert os.path.islink(symlink.strpath) + assert helpers.samepath(file1.strpath, symlink.strpath) + + +def test_samepath_for_same_file(tmpdir): + samedir = tmpdir.mkdir("samepath") + file1 = samedir.join("file1.txt") + assert helpers.samepath(file1.strpath, file1.strpath) + + +def test_samepath_fail_for_different_absolute_path(tmpdir): + samedir = tmpdir.mkdir("samepath") + assert not helpers.samepath(samedir.join("file1.txt").strpath, + samedir.join("file2.txt").strpath) diff --git a/pytest_watch/tests/test_main.py b/pytest_watch/tests/test_main.py new file mode 100644 index 0000000..5d616f3 --- /dev/null +++ b/pytest_watch/tests/test_main.py @@ -0,0 +1,16 @@ +import os +import sys + +try: + from unittest.mock import patch +except ImportError: + from mock import patch + +from pytest_watch import __main__ +from pytest_watch.__main__ import run_cli + + +@patch("pytest_watch.command.main", side_effect=lambda argv=None: 0) +def test_add_pytest_watch_folder_to_path(main): + run_cli() + assert os.path.dirname(__main__.__file__) in sys.path diff --git a/pytest_watch/tests/test_util.py b/pytest_watch/tests/test_util.py new file mode 100644 index 0000000..caa335d --- /dev/null +++ b/pytest_watch/tests/test_util.py @@ -0,0 +1,17 @@ +import pytest + +from pytest_watch.util import silence + + +def test_handle_exception_type(): + with pytest.raises(ValueError) as ex: + with silence(): + raise ValueError("Custom message error") + assert ex.errisinstance(ValueError) + + +def test_handle_exception_message(): + with pytest.raises(KeyError) as ex: + with silence(): + raise KeyError("Custom message error") + assert ex.match("Custom message error") diff --git a/pytest_watch/tests/test_watcher.py b/pytest_watch/tests/test_watcher.py new file mode 100644 index 0000000..2f6a2f5 --- /dev/null +++ b/pytest_watch/tests/test_watcher.py @@ -0,0 +1,176 @@ +import errno +import os +import shutil +import sys +import tempfile +from unittest import skip + +try: + from unittest import mock +except ImportError: + import mock + +import pytest + +from pytest_watch.watcher import _split_recursive, run_hook, watch,\ + _get_pytest_runner + + +class TestDirectoriesFiltering(): + + def setup_method(self): + self.root_dir = tempfile.mkdtemp() + + def teardown_method(self): + try: + shutil.rmtree(self.root_dir) + except: + pass + + def test_empty_split_recursive(self): + dirs = [] + ignore = [] + assert (dirs, ignore) == _split_recursive(dirs, ignore) + + def test_non_empty_directories_empty_ignore(self): + dirs = ["."] + ignore = [] + + assert (dirs, ignore) == _split_recursive(dirs, ignore) + + def test_invalid_directories(self): + dirs = [self.root_dir] + + fake_dir = os.path.join(self.root_dir, "atrocadocapacausti") + + with pytest.raises(OSError) as err: + watch(directories=[fake_dir]) + assert errno.ENOENT == err.value.errno + + def test_ignore_all_subdirs(self): + dirs = [self.root_dir] + + a_folder = tempfile.mkdtemp(dir=self.root_dir) + b_folder = tempfile.mkdtemp(dir=self.root_dir) + + ignore = [a_folder, b_folder] + + assert ([], [self.root_dir]) == _split_recursive(dirs, ignore) + + def test_ignore_subdirs_partially(self): + """ + This test runs over the following tree structure: + self.root_dir + |_included_folder + |_excluded_folder + + Ignoring , the following behavior is expected: + . should be loaded non-recursivelly; + . and its children will be excluded; + . only will be loaded recursivelly. + """ + dirs = [self.root_dir] + + included_folder = tempfile.mkdtemp(dir=self.root_dir) + excluded_folder = tempfile.mkdtemp(dir=self.root_dir) + + ignore = [excluded_folder] + + fail_err = str("Ignoring {1}, the following behavior is expected:\n" + ". {0} should be loaded non-recursivelly;\n" + ". {1} and its children will be excluded;\n" + ". only {2} will be loaded recursivelly.\n") + fail_msg = fail_err.format(self.root_dir, excluded_folder, + included_folder) + + result = _split_recursive(dirs, ignore) + assert ([included_folder], [self.root_dir]) == result, fail_msg + + @skip("Depends on pytest_watch.watcher._split_recursive support" + " for deep recursive navigation through directory tree") + def test_ignore_deep_subtree_multichild(self): + """ + This test runs over the following tree structure: + self.root_dir + |_tree_folder + ..|_subtree_folder_a + ....|_sub_subtree_folder + ..|_subtree_folder + ....|_sub_subtree_folder + + Ignoring , the following behavior is expected: + . should be loaded non-recursivelly; + . and its children will be excluded; + . only will be loaded recursivelly. + """ + dirs = [self.root_dir] + + tree_folder = tempfile.mkdtemp(dir=self.root_dir) + subtree_folder_a = tempfile.mkdtemp(dir=tree_folder) + subtree_folder = tempfile.mkdtemp(dir=tree_folder) + sub_subtree_folder = tempfile.mkdtemp(dir=tree_folder) + + ignore = [subtree_folder] + + assert ([self.root_dir], [tree_folder]) == _split_recursive(dirs, + ignore) + + @skip("Depends on pytest_watch.watcher._split_recursive support" + " for deep recursive navigation through directory tree") + def test_ignore_deep_subtree_single(self): + """ + This test runs over the following tree structure: + self.root_dir + |_tree_folder + ..|_subtree_folder + ....|_sub_subtree_folder + + Ignoring , the following behavior is expected: + . should be loaded non-recursivelly; + . and its children will be excluded; + . only will be loaded recursivelly. + """ + dirs = [self.root_dir] + + tree_folder = tempfile.mkdtemp(dir=self.root_dir) + subtree_folder = tempfile.mkdtemp(dir=tree_folder) + sub_subtree_folder = tempfile.mkdtemp(dir=tree_folder) + + ignore = [subtree_folder] + + assert ([self.root_dir], [tree_folder]) == _split_recursive(dirs, + ignore) + + +class TestPytestRunner(): + DEFAULT_EXECUTABLE = [sys.executable, "-m", "pytest"] + + def setup_method(self): + self.virtual_env = os.getenv("VIRTUAL_ENV") + if "VIRTUAL_ENV" in os.environ: + del os.environ['VIRTUAL_ENV'] + + def teardown_method(self): + if self.virtual_env: + os.putenv("VIRTUAL_ENV", self.virtual_env) + + def test_default_sys_executable(self): + assert TestPytestRunner.DEFAULT_EXECUTABLE == _get_pytest_runner() + + def test_empty_string_returns_sys_executable(self): + assert TestPytestRunner.DEFAULT_EXECUTABLE == _get_pytest_runner("") + assert TestPytestRunner.DEFAULT_EXECUTABLE == _get_pytest_runner(" ") + assert TestPytestRunner.DEFAULT_EXECUTABLE == _get_pytest_runner(" "*8) + + def test_custom_sys_executable(self): + assert ["mypytest"] == _get_pytest_runner("mypytest") + assert ["mpytest", "runtest"] == _get_pytest_runner("mpytest runtest") + + def test_virtualenv_executable(self): + os.environ["VIRTUAL_ENV"] = "/tmp/venv" + + assert ["py.test"] == _get_pytest_runner() + + del os.environ["VIRTUAL_ENV"] + + assert TestPytestRunner.DEFAULT_EXECUTABLE == _get_pytest_runner() diff --git a/pytest_watch/tests/test_watcher_events.py b/pytest_watch/tests/test_watcher_events.py new file mode 100644 index 0000000..c3a1471 --- /dev/null +++ b/pytest_watch/tests/test_watcher_events.py @@ -0,0 +1,94 @@ +import shutil +import tempfile + +try: + from unittest import mock +except: + import mock + +from watchdog.events import FileModifiedEvent, FileMovedEvent, \ + FileCreatedEvent, FileDeletedEvent, FileCreatedEvent, DirModifiedEvent, \ + FileSystemEvent + +from pytest_watch.constants import ALL_EXTENSIONS +from pytest_watch.watcher import EventListener +from pytest_watch.watcher import os as wos + + +def _assert_watched_filesystem_event(event, event_listener=None): + listener = event_listener if event_listener else EventListener() + + assert listener.event_queue.empty() + listener.on_any_event(event) + + assert not listener.event_queue.empty() + + +def _assert_unwatched_filesystem_event(event, event_listener=None): + listener = event_listener if event_listener else EventListener() + + assert listener.event_queue.empty() + listener.on_any_event(event) + + assert listener.event_queue.empty() + + +def test_unwatched_event(): + _assert_unwatched_filesystem_event(FileSystemEvent("/tmp/file.py")) + _assert_unwatched_filesystem_event(DirModifiedEvent("/tmp/")) + + +def test_file_modify_event(): + _assert_watched_filesystem_event(FileModifiedEvent("/tmp/file.py")) + + +def test_file_create_event(): + _assert_watched_filesystem_event(FileCreatedEvent("/tmp/file.py")) + + +def test_file_delete_event(): + _assert_watched_filesystem_event(FileDeletedEvent("/tmp/file.py")) + + +@mock.patch.object(wos.path, "relpath") +def test_file_move_event(relpath): + relpath.side_effect = lambda *args, **kwargs: args[0] + src_path = "/tmp/file.py" + dest_path = "/tmp/file-new.py" + + _assert_watched_filesystem_event(FileMovedEvent(src_path, dest_path)) + + assert 2 == relpath.call_count, str("os.path.relpath should be called " + "twice when file is moved src,dst") + + relpath.assert_any_call(src_path) + relpath.assert_any_call(dest_path) + + +class TestExtensionsMatch(): + def setup_method(self): + self.root_dir = tempfile.mkdtemp() + + def teardown_method(self): + try: + shutil.rmtree(self.root_dir) + except: + pass + + def test_event_over_all_extesions(self): + _, filename = tempfile.mkstemp(dir=self.root_dir, suffix=".py") + event = FileCreatedEvent(filename) + listener = EventListener(extensions=ALL_EXTENSIONS) + _assert_watched_filesystem_event(event, event_listener=listener) + + def test_event_over_observed_file(self): + _, filename = tempfile.mkstemp(dir=self.root_dir, suffix=".py") + event = FileCreatedEvent(filename) + listener = EventListener(extensions=[".py"]) + _assert_watched_filesystem_event(event, event_listener=listener) + + def test_event_over_not_observed_file(self): + _, filename = tempfile.mkstemp(dir=self.root_dir, suffix=".pyc") + event = FileCreatedEvent(filename) + listener = EventListener(extensions=[".py"]) + _assert_unwatched_filesystem_event(event, event_listener=listener) diff --git a/pytest_watch/tests/test_watcher_hooks.py b/pytest_watch/tests/test_watcher_hooks.py new file mode 100644 index 0000000..65eb0bf --- /dev/null +++ b/pytest_watch/tests/test_watcher_hooks.py @@ -0,0 +1,212 @@ +import os +import subprocess +from subprocess import call as _subcall +import sys + +try: + from unittest import mock +except ImportError: + import mock + +import pytest + +import pytest_watch +from pytest_watch.constants import EXIT_NOTESTSCOLLECTED, EXIT_OK +from pytest_watch.helpers import is_windows +from pytest_watch.watcher import watch, run_hook +from pytest_watch.watcher import subprocess as wsubprocess +from pytest_watch import watcher + + +def build_popen_mock(popen, config): + mockmock = mock.Mock() + mockmock.configure_mock(**config) + popen.return_value = mockmock + + +def raise_keyboard_interrupt(*args, **kwargs): + # force keyboard interruption + raise KeyboardInterrupt() + + +def assertion_wrapper(expected, callee, message=None): + """ + Adapter to support assertions as side_effect for patched objects. + + TODO: This implementation can be more generalized for assertions, i.e. !=, + using lambdas. For the moment, its satisfies run_hook tests. + """ + def _wrapped(*args, **kwargs): + if message: + assert expected == callee(*args, **kwargs), message + else: + assert expected == callee(*args, **kwargs) + return _wrapped + + +def get_sys_path(p): + #p = os.path.normpath(p) + if is_windows: + p = '"%s"'%p + return p + + +@pytest.fixture +def subp_call_mock(mocker): + return mocker.patch.object(pytest_watch.watcher.subprocess, "call") + + +def test_run_hook_systemexit_0(subp_call_mock): + subp_call_mock.side_effect = side_effect=assertion_wrapper(0, _subcall) + + python_exec = get_sys_path(sys.executable) + cmd_parts = [python_exec, "-c", "'exit(0)'"] + cmd = " ".join(cmd_parts) + run_hook(cmd) + + assert 1 == subp_call_mock.call_count + assert (cmd,) == subp_call_mock.call_args[0] + assert dict(shell=True) == subp_call_mock.call_args[1] + + +def test_run_hook_systemexit_not_0(subp_call_mock): + subp_call_mock.side_effect = side_effect=assertion_wrapper(1, _subcall) + + python_exec = get_sys_path(sys.executable) + cmd_parts = [python_exec, "-c", "'raise Exception(\'force error\')'"] + cmd = " ".join(cmd_parts) + run_hook(cmd) + + assert 1 == subp_call_mock.call_count + assert (cmd,) == subp_call_mock.call_args[0] + assert dict(shell=True) == subp_call_mock.call_args[1] + + +orig = watcher.run_hook + + +def get_hook_stop_iteration(command_str): + def hook(cmd, *args): + orig(cmd, *args) + if cmd == command_str: + raise StopIteration("Force this only for tests purpose") + return hook + + +@mock.patch.object(wsubprocess, "Popen") +@mock.patch("pytest_watch.watcher.subprocess.call", + side_effect=lambda *args, **kwargs: 0) +class TestRunHookCallbacks(): + + def test_with_beforerun(self, call_mock, popen_mock): + """ + Test if beforerun callback is called if it is passed as argument + """ + config = {"poll.side_effect": raise_keyboard_interrupt, + "wait.return_value": 0} + build_popen_mock(popen_mock, config) + + beforerun = "{} -c 'exit(0) #it is beforerun'".format(sys.executable) + + watch(beforerun=beforerun) + + call_mock.assert_called_once_with(beforerun, shell=True) + + @mock.patch("pytest_watch.helpers.send_keyboard_interrupt") + def test_afterrun_on_keyboard_interruption(self, keyb_int, call_mock, + popen_mock): + config = {"poll.side_effect": raise_keyboard_interrupt, + "wait.return_value": 10} + build_popen_mock(popen_mock, config) + + afterrun = "{} -m this".format(sys.executable) + + watch(afterrun=afterrun, wait=True) + + keyb_int.assert_not_called() + + assert 1 == call_mock.call_count + + expected_cmd = afterrun + " 10" # should run with p.wait() arg + + call_mock.assert_called_once_with(expected_cmd, shell=True) + + @mock.patch("pytest_watch.helpers.send_keyboard_interrupt") + def test_afterrun_without_keyboard_interruption(self, keyb_int, call_mock, + popen_mock): + config = {"poll.side_effect": lambda: 999} + build_popen_mock(popen_mock, config) + + afterrun = "{} -c 'exit(0) #it is afterrun'".format(sys.executable) + + watcher.run_hook = get_hook_stop_iteration(afterrun) + + watch(afterrun=afterrun, wait=True) + + keyb_int.assert_not_called() + + assert 1 == call_mock.call_count + + expected_cmd = afterrun + " 999" # should run with exit_code arg + + call_mock.assert_called_once_with(expected_cmd, shell=True) + + def test_onpass_on_exit(self, call_mock, popen_mock): + config = {"poll.side_effect": lambda: EXIT_OK} + build_popen_mock(popen_mock, config) + + onpass = "{} -c 'exit(0) #it is afterpass on exit'" \ + .format(sys.executable) + + watcher.run_hook = get_hook_stop_iteration(onpass) + watch(onpass=onpass, wait=True) + + call_mock.assert_called_once_with(onpass, shell=True) + + def test_onpass_on_not_tests_collected(self, call_mock, popen_mock): + config = {"poll.side_effect": lambda: EXIT_NOTESTSCOLLECTED} + build_popen_mock(popen_mock, config) + + onpass = "{} -c 'exit(0) #it is afterpass on not_tests_collected'" \ + .format(sys.executable) + watcher.run_hook = get_hook_stop_iteration(onpass) + watch(onpass=onpass, wait=True) + + call_mock.assert_called_once_with(onpass, shell=True) + + @mock.patch("pytest_watch.watcher.beep") + def test_onfail_beep_off(self, beep_mock, call_mock, popen_mock): + config = {"poll.side_effect": lambda: -1000} + build_popen_mock(popen_mock, config) + + onfail = "{} -c 'exit(1) # failure happens'".format(sys.executable) + watcher.run_hook = get_hook_stop_iteration(onfail) + watch(onfail=onfail, wait=True, beep_on_failure=False) + + call_mock.assert_called_once_with(onfail, shell=True) + beep_mock.assert_not_called() + + @mock.patch("pytest_watch.watcher.beep") + def test_onfail_beep_on(self, beep_mock, call_mock, popen_mock): + config = {"poll.side_effect": lambda: -1000} + build_popen_mock(popen_mock, config) + + onfail = "{} -c 'exit(1) # failure happens'".format(sys.executable) + watcher.run_hook = get_hook_stop_iteration(onfail) + watch(onfail=onfail, wait=True, beep_on_failure=True) + + call_mock.assert_called_once_with(onfail, shell=True) + assert 1 == beep_mock.call_count + + +@pytest.mark.skip("baby steps") +class TestRunHooksSkiped(): + + def test_run_hook_with_args(self): + assert False, "Not yet implemented" + + def test_run_hook_without_args(self): + assert False, "Not yet implemented" + + def test_onexit(self): + assert False, "Not yet implemented." diff --git a/pytest_watch/watcher.py b/pytest_watch/watcher.py index 3bc6d62..e180d5b 100644 --- a/pytest_watch/watcher.py +++ b/pytest_watch/watcher.py @@ -11,7 +11,6 @@ except ImportError: from Queue import Queue -from colorama import Fore, Style from watchdog.events import ( FileSystemEventHandler, FileModifiedEvent, FileCreatedEvent, FileMovedEvent, FileDeletedEvent) @@ -22,6 +21,7 @@ ALL_EXTENSIONS, EXIT_NOTESTSCOLLECTED, EXIT_OK, DEFAULT_EXTENSIONS) from .helpers import ( beep, clear, dequeue_all, is_windows, samepath, send_keyboard_interrupt) +from .summary import show_summary EVENT_NAMES = { @@ -37,15 +37,13 @@ FileDeletedEvent: 'Deleted:', } WATCHED_EVENTS = tuple(EVENT_NAMES) -STYLE_BRIGHT = Fore.WHITE + Style.NORMAL + Style.BRIGHT -STYLE_HIGHLIGHT = Fore.CYAN + Style.NORMAL + Style.BRIGHT class EventListener(FileSystemEventHandler): """ Listens for changes to files and re-runs tests after each change. """ - def __init__(self, extensions=[]): + def __init__(self, extensions=None): super(EventListener, self).__init__() self.event_queue = Queue() self.extensions = extensions or DEFAULT_EXTENSIONS @@ -78,98 +76,47 @@ def on_any_event(self, event): self.event_queue.put((type(event), src_path, dest_path)) -def _get_pytest_runner(custom): - if custom: +def _get_pytest_runner(custom=None): + if custom and custom.strip(): return custom.split(' ') + if os.getenv('VIRTUAL_ENV'): return ['py.test'] - return [sys.executable, '-m', 'pytest'] - -def _reduce_events(events): - # FUTURE: Reduce ['a -> b', 'b -> c'] renames to ['a -> c'] - - creates = [] - moves = [] - for event, src, dest in events: - if event == FileCreatedEvent: - creates.append(dest) - if event == FileMovedEvent: - moves.append(dest) - - seen = [] - filtered = [] - for event, src, dest in events: - # Skip 'modified' event during 'created' - if src in creates and event != FileCreatedEvent: - continue - - # Skip 'modified' event during 'moved' - if src in moves: - continue - - # Skip duplicate events - if src in seen: - continue - seen.append(src) - - filtered.append((event, src, dest)) - return filtered - - -def _show_summary(argv, events, verbose=False): - command = ' '.join(argv) - bright = lambda arg: STYLE_BRIGHT + arg + Style.RESET_ALL - highlight = lambda arg: STYLE_HIGHLIGHT + arg + Style.RESET_ALL - - time_stamp = time.strftime("%c", time.localtime(time.time())) - run_command_info = '[{}] Running: {}'.format(time_stamp, - highlight(command)) - if not events: - print(run_command_info) - return - - events = _reduce_events(events) - if verbose: - lines = ['Changes detected:'] - m = max(map(len, map(lambda e: VERBOSE_EVENT_NAMES[e[0]], events))) - for event, src, dest in events: - event = VERBOSE_EVENT_NAMES[event].ljust(m) - lines.append(' {} {}'.format( - event, - highlight(src + (' -> ' + dest if dest else '')))) - lines.append('') - lines.append(run_command_info) - else: - lines = [] - for event, src, dest in events: - lines.append('{} detected: {}'.format( - EVENT_NAMES[event], - bright(src + (' -> ' + dest if dest else '')))) - lines.append('') - lines.append(run_command_info) - - print('\n'.join(lines)) + return [sys.executable, '-m', 'pytest'] def _split_recursive(directories, ignore): + if not ignore: - return directories, [] + # If ignore list is empty, all directories should be included. + # Return all + ignore = ignore if type(ignore) is list else [] + return directories, ignore # TODO: Have this work recursively recursedirs, norecursedirs = [], [] + join = os.path.join for directory in directories: - subdirs = [os.path.join(directory, d) + # Build subdirectories paths list + subdirs = [join(directory, d) for d in os.listdir(directory) - if os.path.isdir(d)] + if os.path.isdir(join(directory, d))] + + # Filter not ignored subdirs in current folder filtered = [subdir for subdir in subdirs - if not any(samepath(os.path.join(directory, d), subdir) - for d in ignore)] + if not any(samepath(join(directory, ignore_name), subdir) + for ignore_name in ignore)] + if len(subdirs) == len(filtered): + # No subdirs were ignored recursedirs.append(directory) else: + # If any subdir is ignored, this folder will not be recursivelly + # observed norecursedirs.append(directory) + # But, non-ignored subdirs should be observed recursivelly recursedirs.extend(filtered) return sorted(set(recursedirs)), sorted(set(norecursedirs)) @@ -177,25 +124,38 @@ def _split_recursive(directories, ignore): def run_hook(cmd, *args): """ - Runs a command hook, if specified. + Runs a command hook as subprocess of current process. + + If cmd is not specified, nothing is executed. + + cmd -- executable file path + args -- list of command line arguments appended to executable call """ if cmd: command = ' '.join(map(str, (cmd,) + args)) subprocess.call(command, shell=True) -def watch(directories=[], ignore=[], extensions=[], beep_on_failure=True, +def watch(directories=None, ignore=None, extensions=None, beep_on_failure=True, auto_clear=False, wait=False, beforerun=None, afterrun=None, onpass=None, onfail=None, onexit=None, runner=None, spool=None, - poll=False, verbose=False, quiet=False, pytest_args=[]): + poll=False, verbose=False, quiet=False, pytest_args=None): + + directories = [] if directories is None else directories + ignore = [] if ignore is None else ignore + extensions = [] if extensions is None else extensions + pytest_args = [] if pytest_args is None else pytest_args + argv = _get_pytest_runner(runner) + (pytest_args or []) + # Prepare directories if not directories: directories = ['.'] directories = [os.path.abspath(directory) for directory in directories] for directory in directories: if not os.path.isdir(directory): - raise ValueError('Directory not found: ' + directory) + import errno + raise OSError(errno.ENOENT, 'Directory not found: ' + directory) # Setup event handler event_listener = EventListener(extensions) @@ -221,7 +181,7 @@ def watch(directories=[], ignore=[], extensions=[], beep_on_failure=True, # Show event summary if not quiet: - _show_summary(argv, events, verbose) + show_summary(argv, events, verbose) # Run custom command run_hook(beforerun) diff --git a/requirements-test.txt b/requirements-test.txt new file mode 100644 index 0000000..ab3f3e9 --- /dev/null +++ b/requirements-test.txt @@ -0,0 +1,2 @@ +pytest-cov>=2.5.1 +pytest-mock>=1.7.0 diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..b7e4789 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,2 @@ +[aliases] +test=pytest diff --git a/setup.py b/setup.py index 2424673..ed00ab5 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,5 @@ import os +import sys from setuptools import setup, find_packages @@ -7,6 +8,12 @@ def read(filename): return f.read() +DEPS_MAIN = ["colorama>=0.3.3", "docopt>=0.6.2", "pytest>=2.6.4", + "watchdog>=0.6.0"] +DEPS_TESTING = ["pytest-mock>=1.7.0"] +DEPS_QA = DEPS_TESTING + ["pytest-cov>=2.5.1", "codecov", "pytest-pep8"] + + setup( name='pytest-watch', version='4.1.0', @@ -18,11 +25,25 @@ def read(filename): license='MIT', platforms='any', packages=find_packages(), - install_requires=read('requirements.txt').splitlines(), + install_requires=DEPS_MAIN, + setup_requires=['pytest-runner',], + tests_require=DEPS_TESTING, entry_points={ 'console_scripts': [ 'pytest-watch = pytest_watch:main', 'ptw = pytest_watch:main', - ] + ], + #'pytest11': ["watch = pytest_watch:main"] + }, + extras_require={ + 'testing': DEPS_TESTING, + 'dev': DEPS_TESTING + DEPS_QA, + 'qa': DEPS_QA, + 'testing:python_version in "2.6, 2.7, 3.2"': ['mock'], + 'dev:python_version in "2.6, 2.7, 3.2"': ['mock'], + 'qa:python_version in "2.6, 2.7, 3.2"': ['mock'], }, + classifiers=[ + "Framework :: Pytest", + ] )