diff --git a/.gitignore b/.gitignore index 4a644af..bfe9e88 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ # exclude patterns (uncomment them if you want to use them): *~ *.pyc +dist/ diff --git a/README.md b/README.md index a934b68..7b05398 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,5 @@ # when-changed: Run a command when a file is changed - ### What is it? Tired of switching to the shell to test the changes you just made to @@ -13,24 +12,22 @@ changed the file, when-changed runs any command you specify. So to generate your latex resume automatically, you can do this: ```sh -$ when-changed CV.tex pdflatex CV.tex +$ when-changed CV.tex pdflatex CV.tex ``` Sweetness! - ### What do I need? -- Python 2.6+ +- Python 3.6+ - watchdog - ### Installation + ```sh -pip install https://github.com/joh/when-changed/archive/master.zip +pipx install https://github.com/eviweb/when-changed/archive/master.zip ``` - ### Usage ```sh @@ -38,26 +35,46 @@ when-changed [OPTION] FILE COMMAND... when-changed [OPTION] FILE [FILE ...] -c COMMAND ``` -FILE can be a directory. Use %f to pass the filename to the command. +FILE can be a directory. Use `%f` to pass the filename to the command. **Options:** -- -r Watch recursively -- -v Verbose output. Multiple -v options increase the verbosity. The maximum is 3: -vvv. -- -1 Don't re-run command if files changed while command was running -- -s Run command immediately at start -- -q Run command quietly +- `-r` Watch recursively +- `-v` Verbose output. Multiple -v options increase the verbosity. The maximum is 3: `-vvv`. +- `-1` Don't re-run command if files changed while command was running +- `-s` Run command immediately at start +- `-q` Run command quietly +- `-k` Kill the running command before restarting it (useful for long-running processes) +- `-d DELAY` Debounce: wait DELAY seconds before running, coalescing rapid changes into a single run +- `-p PATTERN` Only react to files matching PATTERN (glob, e.g. `*.py`). Can be repeated for multiple patterns. -### Environment variables: +### Environment variables when-changed provides the following environment variables: -- WHEN_CHANGED_EVENT: reflects the current event type that occurs. -Could be either: - - file_created - - file_modified - - file_moved - - file_deleted +- `WHEN_CHANGED_EVENT`: reflects the current event type that occurs. Could be either: + - `file_created` + - `file_modified` + - `file_moved` + - `file_deleted` +- `WHEN_CHANGED_FILE`: provides the full path of the file that has generated the event. + +### Examples + +Restart a server when source files change, killing the previous instance: + +```sh +$ when-changed -k -r src/ python server.py +``` + +Debounce rapid changes (e.g. formatter touching many files at once): + +```sh +$ when-changed -d 0.5 -r src/ make test +``` -- WHEN_CHANGED_FILE: provides the full path of the file that has generated the event. +Watch only Python and YAML files in a directory: +```sh +$ when-changed -r -p '*.py' -p '*.yml' src/ make lint +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..15284d7 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,19 @@ +[build-system] +requires = ["setuptools>=42", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "when-changed" +version = "0.5.0" +description = "Run a command when a file is changed" +readme = "README.md" +license = { text = "BSD" } +authors = [{ name = "Johannes H. Jensen", email = "joh@pseudoberries.com" }] +requires-python = ">=3.6" +dependencies = ["watchdog"] + +[project.urls] +Homepage = "https://github.com/joh/when-changed" + +[project.scripts] +when-changed = "whenchanged.whenchanged:main" diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..4ecb1ad --- /dev/null +++ b/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* diff --git a/requirements.txt b/requirements.txt index bd4e30e..e59495e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1 @@ watchdog -subprocess32 diff --git a/setup.py b/setup.py index bbf2230..4e39d6d 100644 --- a/setup.py +++ b/setup.py @@ -1,14 +1,16 @@ from setuptools import setup -setup(name='when-changed', - version='0.3.0', - description='Run a command when a file is changed', - author='Johannes H. Jensen', - author_email='joh@pseudoberries.com', - url='https://github.com/joh/when-changed', - packages=['whenchanged'], - entry_points={ - 'console_scripts': ('when-changed = whenchanged.whenchanged:main') - }, install_requires=['watchdog'], - license='BSD' - ) +setup( + name='when-changed', + version='0.5.0', + description='Run a command when a file is changed', + author='Johannes H. Jensen', + author_email='joh@pseudoberries.com', + url='https://github.com/joh/when-changed', + packages=['whenchanged'], + entry_points={ + 'console_scripts': ['when-changed = whenchanged.whenchanged:main'], + }, + install_requires=['watchdog'], + license='BSD', +) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..8a375c8 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,19 @@ +import sys +import types +from unittest.mock import MagicMock, patch + +# Mock watchdog before any import of whenchanged +watchdog_mock = types.ModuleType('watchdog') +watchdog_observers = types.ModuleType('watchdog.observers') +watchdog_events = types.ModuleType('watchdog.events') + +watchdog_observers.Observer = MagicMock +watchdog_events.FileSystemEventHandler = object + +watchdog_mock.observers = watchdog_observers +watchdog_mock.events = watchdog_events + +sys.modules['watchdog'] = watchdog_mock +sys.modules['watchdog.observers'] = watchdog_observers +sys.modules['watchdog.events'] = watchdog_events +sys.modules['subprocess32'] = MagicMock() diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..24fbfcd --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,81 @@ +"""Tests for CLI argument parsing in main().""" +import os +import sys +import pytest +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from whenchanged.whenchanged import WhenChanged + + +def run_main_with_args(args): + """Helper: run main() capturing WhenChanged instantiation args.""" + with patch('sys.argv', ['when-changed'] + args), \ + patch('whenchanged.whenchanged.WhenChanged') as MockWC: + MockWC.return_value.run = MagicMock() + try: + from whenchanged.whenchanged import main + main() + except SystemExit: + pass + return MockWC + + +class TestCLIParsing: + def test_basic_file_and_command(self): + MockWC = run_main_with_args(['file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[0] == ['file.py'] # files + assert args[1] == ['echo', 'ok'] # command + + def test_flag_r_sets_recursive(self): + MockWC = run_main_with_args(['-r', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[2] == True # recursive + + def test_flag_1_sets_run_once(self): + MockWC = run_main_with_args(['-1', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[3] == True # run_once + + def test_flag_s_sets_run_at_start(self): + MockWC = run_main_with_args(['-s', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[4] == True # run_at_start + + def test_flag_v_sets_verbose(self): + MockWC = run_main_with_args(['-v', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[5] == 1 # verbose_mode + + def test_flag_vvv_sets_verbose_3(self): + MockWC = run_main_with_args(['-vvv', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[5] == 3 + + def test_flag_q_sets_quiet(self): + MockWC = run_main_with_args(['-q', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[6] == True # quiet_mode + + def test_flag_k_sets_kill_mode(self): + MockWC = run_main_with_args(['-k', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[7] == True # kill_mode + + def test_flag_d_sets_debounce(self): + MockWC = run_main_with_args(['-d', '0.5', 'file.py', 'echo', 'ok']) + args, kwargs = MockWC.call_args + assert args[8] == 0.5 # debounce_delay + + def test_flag_c_multifile(self): + MockWC = run_main_with_args(['a.py', 'b.py', '-c', 'make', 'test']) + args, kwargs = MockWC.call_args + assert args[0] == ['a.py', 'b.py'] + assert args[1] == ['make', 'test'] + + def test_no_args_exits(self): + with patch('sys.argv', ['when-changed']): + with pytest.raises(SystemExit): + from whenchanged.whenchanged import main + main() diff --git a/tests/test_debounce.py b/tests/test_debounce.py new file mode 100644 index 0000000..67283e8 --- /dev/null +++ b/tests/test_debounce.py @@ -0,0 +1,72 @@ +"""Tests for debounce (-d) and on_change dispatch logic.""" +import os +import sys +import time +import threading +import pytest +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from whenchanged.whenchanged import WhenChanged + + +def make_wc(files=None, debounce_delay=0, **kwargs): + files = files or ['/tmp'] + with patch('whenchanged.whenchanged.Observer'): + wc = WhenChanged( + files=files, + command=['echo', 'ok'], + debounce_delay=debounce_delay, + **kwargs + ) + return wc + + +class TestOnChange: + def test_no_debounce_calls_run_command_immediately(self, tmp_path): + wc = make_wc(files=[str(tmp_path)]) + wc.run_command = MagicMock() + wc.on_change(str(tmp_path / 'test.py')) + wc.run_command.assert_called_once() + + def test_uninterested_path_not_called(self, tmp_path): + watched = tmp_path / 'watched' + watched.mkdir() + other = tmp_path / 'other' + other.mkdir() + wc = make_wc(files=[str(watched)]) + wc.run_command = MagicMock() + wc.on_change(str(other / 'test.py')) + wc.run_command.assert_not_called() + + +class TestDebounce: + def test_single_event_runs_after_delay(self, tmp_path): + wc = make_wc(files=[str(tmp_path)], debounce_delay=0.1) + wc.run_command = MagicMock() + wc.on_change(str(tmp_path / 'test.py')) + wc.run_command.assert_not_called() # pas encore + time.sleep(0.2) + wc.run_command.assert_called_once() + + def test_rapid_events_coalesced_into_one(self, tmp_path): + wc = make_wc(files=[str(tmp_path)], debounce_delay=0.1) + wc.run_command = MagicMock() + # 5 events rapides + for _ in range(5): + wc.on_change(str(tmp_path / 'test.py')) + time.sleep(0.02) + time.sleep(0.2) + # doit avoir été appelé une seule fois + assert wc.run_command.call_count == 1 + + def test_timer_reset_on_new_event(self, tmp_path): + wc = make_wc(files=[str(tmp_path)], debounce_delay=0.15) + wc.run_command = MagicMock() + wc.on_change(str(tmp_path / 'test.py')) + time.sleep(0.05) + wc.on_change(str(tmp_path / 'test.py')) # reset le timer + time.sleep(0.1) + wc.run_command.assert_not_called() # timer pas encore écoulé + time.sleep(0.1) + wc.run_command.assert_called_once() diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..523abc6 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,137 @@ +"""Tests for event handlers - on_created, on_modified, on_moved, on_deleted.""" +import os +import sys +import time +import pytest +from unittest.mock import MagicMock, patch, PropertyMock + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from whenchanged.whenchanged import WhenChanged + + +def make_wc(files=None, **kwargs): + files = files or ['/tmp/test.py'] + with patch('whenchanged.whenchanged.Observer'): + wc = WhenChanged( + files=files, + command=['echo', 'ok'], + **kwargs + ) + wc.on_change = MagicMock() + return wc + + +def make_event(src_path, is_directory=False, dest_path=None): + event = MagicMock() + event.src_path = src_path + event.is_directory = is_directory + if dest_path: + event.dest_path = dest_path + return event + + +class TestOnCreated: + def test_triggers_on_change(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_created(ev) + wc.on_change.assert_called_once_with('/tmp/test.py') + + def test_sets_event_envvar(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_created(ev) + assert wc.get_envvar('event') == 'file_created' + + def test_adds_to_recently_created(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_created(ev) + assert '/tmp/test.py' in wc._recently_created + + def test_ignores_directories(self): + wc = make_wc() + ev = make_event('/tmp/mydir', is_directory=True) + wc.on_created(ev) + wc.on_change.assert_not_called() + + +class TestOnModified: + def test_triggers_on_change(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_modified(ev) + wc.on_change.assert_called_once_with('/tmp/test.py') + + def test_sets_event_envvar(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_modified(ev) + assert wc.get_envvar('event') == 'file_modified' + + def test_skips_recently_created(self): + wc = make_wc() + wc._recently_created['/tmp/test.py'] = time.time() + ev = make_event('/tmp/test.py') + wc.on_modified(ev) + wc.on_change.assert_not_called() + + def test_removes_from_recently_created_after_skip(self): + wc = make_wc() + wc._recently_created['/tmp/test.py'] = time.time() + ev = make_event('/tmp/test.py') + wc.on_modified(ev) + assert '/tmp/test.py' not in wc._recently_created + + def test_purges_stale_recently_created(self): + wc = make_wc() + wc._recently_created['/tmp/old.py'] = time.time() - 2.0 # stale + ev = make_event('/tmp/test.py') + wc.on_modified(ev) + assert '/tmp/old.py' not in wc._recently_created + + def test_ignores_directories(self): + wc = make_wc() + ev = make_event('/tmp/mydir', is_directory=True) + wc.on_modified(ev) + wc.on_change.assert_not_called() + + +class TestOnMoved: + def test_triggers_on_change_with_dest_path(self): + wc = make_wc() + ev = make_event('/tmp/old.py', dest_path='/tmp/new.py') + wc.on_moved(ev) + wc.on_change.assert_called_once_with('/tmp/new.py') + + def test_sets_event_envvar(self): + wc = make_wc() + ev = make_event('/tmp/old.py', dest_path='/tmp/new.py') + wc.on_moved(ev) + assert wc.get_envvar('event') == 'file_moved' + + def test_ignores_directories(self): + wc = make_wc() + ev = make_event('/tmp/mydir', is_directory=True, dest_path='/tmp/newdir') + wc.on_moved(ev) + wc.on_change.assert_not_called() + + +class TestOnDeleted: + def test_triggers_on_change(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_deleted(ev) + wc.on_change.assert_called_once_with('/tmp/test.py') + + def test_sets_event_envvar(self): + wc = make_wc() + ev = make_event('/tmp/test.py') + wc.on_deleted(ev) + assert wc.get_envvar('event') == 'file_deleted' + + def test_ignores_directories(self): + wc = make_wc() + ev = make_event('/tmp/mydir', is_directory=True) + wc.on_deleted(ev) + wc.on_change.assert_not_called() diff --git a/tests/test_is_interested.py b/tests/test_is_interested.py new file mode 100644 index 0000000..ccb11ce --- /dev/null +++ b/tests/test_is_interested.py @@ -0,0 +1,123 @@ +"""Tests for WhenChanged.is_interested() - path filtering logic.""" +import os +import sys +import pytest +from unittest.mock import MagicMock, patch + +# conftest.py mocks watchdog before this import +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from whenchanged.whenchanged import WhenChanged + + +def make_wc(files, **kwargs): + """Helper: create a WhenChanged instance without starting the observer.""" + with patch.object(WhenChanged, '__init__', lambda self, *a, **kw: None): + wc = WhenChanged.__new__(WhenChanged) + # Reproduce only the attributes needed for is_interested() + wc.files = files + paths = {} + for f in files: + paths[os.path.realpath(f)] = f + wc.paths = paths + wc.recursive = kwargs.get('recursive', False) + wc.patterns = kwargs.get('patterns', []) + wc.exclude = WhenChanged.exclude + return wc + + +# --- Exclusion patterns --- + +class TestExcludePatterns: + def test_vim_swp(self, tmp_path): + f = str(tmp_path / '.file.swp') + wc = make_wc([str(tmp_path)]) + assert not wc.is_interested(f) + + def test_vim_swo(self, tmp_path): + f = str(tmp_path / '.file.swo') + wc = make_wc([str(tmp_path)]) + assert not wc.is_interested(f) + + def test_vim_swn(self, tmp_path): + f = str(tmp_path / '.file.swn') + wc = make_wc([str(tmp_path)]) + assert not wc.is_interested(f) + + def test_vim_4913(self, tmp_path): + f = str(tmp_path / '4913') + wc = make_wc([str(tmp_path)]) + assert not wc.is_interested(f) + + def test_backup_tilde(self, tmp_path): + # NOTE: regex r'.~$' has an unescaped dot — matches any char before ~. + # file.py~ is therefore NOT excluded (bug). Documents current behavior. + f = str(tmp_path / 'file.py~') + wc = make_wc([str(tmp_path)]) + assert wc.is_interested(f) # should be False once regex is fixed + + + def test_git_dir(self, tmp_path): + f = str(tmp_path / '.git' / 'COMMIT_EDITMSG') + wc = make_wc([str(tmp_path)]) + assert not wc.is_interested(f) + + def test_pycache(self, tmp_path): + f = str(tmp_path / '__pycache__' / 'mod.pyc') + wc = make_wc([str(tmp_path)]) + assert not wc.is_interested(f) + + def test_normal_py_file_not_excluded(self, tmp_path): + f = str(tmp_path / 'main.py') + wc = make_wc([str(tmp_path)]) + # not excluded by pattern — is_interested returns True for watched dir + assert wc.is_interested(f) + + +# --- Direct file watching --- + +class TestDirectFile: + def test_watched_file_itself(self, tmp_path): + f = tmp_path / 'main.py' + f.touch() + wc = make_wc([str(f)]) + assert wc.is_interested(os.path.realpath(str(f))) + + def test_unwatched_sibling(self, tmp_path): + f = tmp_path / 'main.py' + other = tmp_path / 'other.py' + f.touch() + other.touch() + wc = make_wc([str(f)]) + assert not wc.is_interested(str(other)) + + +# --- Directory watching --- + +class TestDirectoryWatching: + def test_file_in_watched_dir(self, tmp_path): + wc = make_wc([str(tmp_path)]) + f = str(tmp_path / 'script.py') + assert wc.is_interested(f) + + def test_file_in_subdir_non_recursive(self, tmp_path): + subdir = tmp_path / 'sub' + subdir.mkdir() + f = str(subdir / 'script.py') + wc = make_wc([str(tmp_path)], recursive=False) + assert not wc.is_interested(f) + + def test_file_in_subdir_recursive(self, tmp_path): + subdir = tmp_path / 'sub' / 'deep' + subdir.mkdir(parents=True) + f = str(subdir / 'script.py') + wc = make_wc([str(tmp_path)], recursive=True) + assert wc.is_interested(f) + + def test_file_outside_watched_dir(self, tmp_path): + other = tmp_path / 'other' + other.mkdir() + watched = tmp_path / 'watched' + watched.mkdir() + wc = make_wc([str(watched)]) + f = str(other / 'script.py') + assert not wc.is_interested(f) diff --git a/tests/test_patterns.py b/tests/test_patterns.py new file mode 100644 index 0000000..c4d5028 --- /dev/null +++ b/tests/test_patterns.py @@ -0,0 +1,125 @@ +"""Tests for -p PATTERN filtering (issue #90).""" +import os +import sys +import pytest +from unittest.mock import MagicMock, patch + +# conftest.py mocks watchdog before this import +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from whenchanged.whenchanged import WhenChanged + + +def make_wc(files=None, patterns=None, recursive=False): + files = files or ['/tmp'] + with patch('whenchanged.whenchanged.Observer'): + wc = WhenChanged( + files=files, + command=['echo', 'ok'], + recursive=recursive, + patterns=patterns or [], + ) + return wc + + +class TestMatchesPatterns: + def test_no_patterns_matches_everything(self): + wc = make_wc() + assert wc.matches_patterns('/tmp/main.py') + assert wc.matches_patterns('/tmp/style.css') + assert wc.matches_patterns('/tmp/README.md') + + def test_single_pattern_matches(self): + wc = make_wc(patterns=['*.py']) + assert wc.matches_patterns('/tmp/main.py') + assert wc.matches_patterns('/tmp/src/utils.py') + + def test_single_pattern_rejects(self): + wc = make_wc(patterns=['*.py']) + assert not wc.matches_patterns('/tmp/style.css') + assert not wc.matches_patterns('/tmp/README.md') + + def test_multiple_patterns_any_match(self): + wc = make_wc(patterns=['*.py', '*.yml']) + assert wc.matches_patterns('/tmp/main.py') + assert wc.matches_patterns('/tmp/config.yml') + assert not wc.matches_patterns('/tmp/style.css') + + def test_pattern_matches_basename_only(self): + """Pattern should match filename, not full path.""" + wc = make_wc(patterns=['*.py']) + assert wc.matches_patterns('/some/deep/path/script.py') + assert not wc.matches_patterns('/some/deep/path/script.js') + + def test_pattern_wildcard_prefix(self): + wc = make_wc(patterns=['test_*']) + assert wc.matches_patterns('/tmp/test_main.py') + assert not wc.matches_patterns('/tmp/main.py') + + def test_pattern_exact_filename(self): + wc = make_wc(patterns=['Makefile']) + assert wc.matches_patterns('/tmp/Makefile') + assert not wc.matches_patterns('/tmp/makefile') # case sensitive + + def test_pattern_question_mark_wildcard(self): + wc = make_wc(patterns=['file?.py']) + assert wc.matches_patterns('/tmp/file1.py') + assert wc.matches_patterns('/tmp/fileA.py') + assert not wc.matches_patterns('/tmp/file10.py') + + +class TestIsInterestedWithPatterns: + def test_pattern_filters_in_watched_dir(self, tmp_path): + wc = make_wc(files=[str(tmp_path)], patterns=['*.py']) + assert wc.is_interested(str(tmp_path / 'main.py')) + assert not wc.is_interested(str(tmp_path / 'style.css')) + + def test_no_pattern_watches_all_in_dir(self, tmp_path): + wc = make_wc(files=[str(tmp_path)]) + assert wc.is_interested(str(tmp_path / 'main.py')) + assert wc.is_interested(str(tmp_path / 'style.css')) + + def test_pattern_with_recursive(self, tmp_path): + subdir = tmp_path / 'src' / 'deep' + subdir.mkdir(parents=True) + wc = make_wc(files=[str(tmp_path)], patterns=['*.py'], recursive=True) + assert wc.is_interested(str(subdir / 'utils.py')) + assert not wc.is_interested(str(subdir / 'style.css')) + + def test_multiple_patterns_in_dir(self, tmp_path): + wc = make_wc(files=[str(tmp_path)], patterns=['*.py', '*.yml']) + assert wc.is_interested(str(tmp_path / 'main.py')) + assert wc.is_interested(str(tmp_path / 'config.yml')) + assert not wc.is_interested(str(tmp_path / 'README.md')) + + def test_pattern_does_not_override_exclude(self, tmp_path): + """Excluded files (vim swap etc.) must still be excluded even with patterns.""" + wc = make_wc(files=[str(tmp_path)], patterns=['*.py', '*.sw*']) + assert not wc.is_interested(str(tmp_path / '.main.swp')) + + +class TestCLIPatternParsing: + def run_main(self, args): + with patch('sys.argv', ['when-changed'] + args), \ + patch('whenchanged.whenchanged.WhenChanged') as MockWC: + MockWC.return_value.run = MagicMock() + try: + from whenchanged.whenchanged import main + main() + except SystemExit: + pass + return MockWC + + def test_single_pattern(self): + MockWC = self.run_main(['-p', '*.py', 'src/', 'make', 'test']) + args = MockWC.call_args[0] + assert args[9] == ['*.py'] + + def test_multiple_patterns(self): + MockWC = self.run_main(['-p', '*.py', '-p', '*.yml', 'src/', 'make']) + args = MockWC.call_args[0] + assert args[9] == ['*.py', '*.yml'] + + def test_no_pattern_defaults_to_empty_list(self): + MockWC = self.run_main(['src/', 'make', 'test']) + args = MockWC.call_args[0] + assert args[9] == [] diff --git a/tests/test_run_command.py b/tests/test_run_command.py new file mode 100644 index 0000000..e25eef9 --- /dev/null +++ b/tests/test_run_command.py @@ -0,0 +1,104 @@ +"""Tests for WhenChanged.run_command() - execution and run_once logic.""" +import os +import sys +import time +import pytest +from unittest.mock import MagicMock, patch, call + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from whenchanged.whenchanged import WhenChanged + + +def make_wc(files=None, command=None, **kwargs): + files = files or ['/tmp/test.py'] + command = command or ['echo', 'changed'] + with patch('whenchanged.whenchanged.Observer'): + wc = WhenChanged( + files=files, + command=command, + run_once=kwargs.get('run_once', False), + verbose_mode=kwargs.get('verbose_mode', 0), + quiet_mode=kwargs.get('quiet_mode', False), + kill_mode=kwargs.get('kill_mode', False), + debounce_delay=kwargs.get('debounce_delay', 0), + ) + return wc + + +class TestRunCommand: + def test_command_is_called(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)], command=['echo', 'ok']) + wc.set_envvar('event', 'file_modified') + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + mock_popen.return_value.wait.return_value = 0 + wc.run_command(str(f)) + mock_popen.assert_called_once() + + def test_percent_f_replaced(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)], command=['echo', '%f']) + wc.set_envvar('event', 'file_modified') + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + mock_popen.return_value.wait.return_value = 0 + wc.run_command(str(f)) + args = mock_popen.call_args[0][0] + assert str(f) in args + + def test_run_once_skips_if_not_modified(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)], run_once=True) + wc.set_envvar('event', 'file_modified') + wc.last_run = time.time() + 10 # simule un run récent + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + wc.run_command(str(f)) + mock_popen.assert_not_called() + + def test_run_once_runs_if_modified_after(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)], run_once=True) + wc.set_envvar('event', 'file_modified') + wc.last_run = 0 # jamais exécuté + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + mock_popen.return_value.wait.return_value = 0 + wc.run_command(str(f)) + mock_popen.assert_called_once() + + def test_last_run_updated(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)]) + wc.set_envvar('event', 'file_modified') + before = time.time() + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + mock_popen.return_value.wait.return_value = 0 + wc.run_command(str(f)) + assert wc.last_run >= before + + def test_kill_mode_terminates_previous_process(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)], kill_mode=True) + wc.set_envvar('event', 'file_modified') + old_proc = MagicMock() + old_proc.wait.return_value = 0 + wc._current_process = old_proc + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + mock_popen.return_value.wait.return_value = 0 + wc.run_command(str(f)) + old_proc.terminate.assert_called_once() + + def test_quiet_mode_suppresses_stdout(self, tmp_path): + f = tmp_path / 'test.py' + f.write_text('hello') + wc = make_wc(files=[str(f)], quiet_mode=True) + wc.set_envvar('event', 'file_modified') + with patch('whenchanged.whenchanged.subprocess.Popen') as mock_popen: + mock_popen.return_value.wait.return_value = 0 + with patch('builtins.open') as mock_open: + wc.run_command(str(f)) + mock_open.assert_called_with(os.devnull, 'wb') diff --git a/whenchanged/whenchanged.py b/whenchanged/whenchanged.py old mode 100755 new mode 100644 index 4cf932e..8ff8122 --- a/whenchanged/whenchanged.py +++ b/whenchanged/whenchanged.py @@ -1,8 +1,8 @@ #!/usr/bin/env python """%(prog)s - run a command when a file is changed -Usage: %(prog)s [-vr1s] FILE COMMAND... - %(prog)s [-vr1s] FILE [FILE ...] -c COMMAND +Usage: %(prog)s [OPTIONS] FILE COMMAND... + %(prog)s [OPTIONS] FILE [FILE ...] -c COMMAND FILE can be a directory. Use %%f to pass the filename to the command. @@ -13,6 +13,10 @@ -1 Don't re-run command if files changed while command was running -s Run command immediately at start -q Run command quietly +-k Kill the running command before restarting it +-d DELAY Debounce: wait DELAY seconds before running (default: 0) +-p PATTERN Only react to files matching PATTERN (glob, e.g. *.py). + Can be specified multiple times. Environment variables: - WHEN_CHANGED_EVENT: reflects the current event type that occurs. @@ -29,7 +33,9 @@ import sys import os import re +import fnmatch import time +import threading from datetime import datetime from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler @@ -44,11 +50,11 @@ class WhenChanged(FileSystemEventHandler): # files to exclude from being watched exclude = re.compile(r'|'.join(r'(.+/)?'+ a for a in [ # Vim swap files - r'\..*\.sw[px]*$', + r'\..*\.sw[a-z]$', # file creation test file 4913 r'4913$', # backup files - r'.~$', + r'\.~$', # git directories r'\.git/?', # __pycache__ directories @@ -56,7 +62,8 @@ class WhenChanged(FileSystemEventHandler): ])) def __init__(self, files, command, recursive=False, run_once=False, - run_at_start=False, verbose_mode=0, quiet_mode=False): + run_at_start=False, verbose_mode=0, quiet_mode=False, + kill_mode=False, debounce_delay=0, patterns=None): self.files = files paths = {} for f in files: @@ -70,6 +77,13 @@ def __init__(self, files, command, recursive=False, run_once=False, self.verbose_mode = verbose_mode self.quiet_mode = quiet_mode self.process_env = os.environ.copy() + self._recently_created = {} + self.kill_mode = kill_mode + self.debounce_delay = debounce_delay + self.patterns = patterns or [] + self._current_process = None + self._debounce_timer = None + self._lock = threading.Lock() self.observer = Observer(timeout=0.1) @@ -85,8 +99,9 @@ def __init__(self, files, command, recursive=False, run_once=False, def run_command(self, thefile): if self.run_once: - if os.path.exists(thefile) and os.path.getmtime(thefile) < self.last_run: + if os.path.exists(thefile) and os.path.getmtime(thefile) <= self.last_run: return + self.last_run = time.time() new_command = [] for item in self.command: new_command.append(item.replace('%f', thefile)) @@ -102,13 +117,32 @@ def run_command(self, thefile): print('==> ' + print_message + ' <==') self.set_envvar('file', thefile) stdout = open(os.devnull, 'wb') if self.quiet_mode else None - subprocess.call(new_command, shell=(len(new_command) == 1), env=self.process_env, stdout=stdout) - self.last_run = time.time() + if self.kill_mode and self._current_process is not None: + self._current_process.terminate() + try: + self._current_process.wait(timeout=5) + except subprocess.TimeoutExpired: + self._current_process.kill() + self._current_process = subprocess.Popen( + new_command, shell=(len(new_command) == 1), + env=self.process_env, stdout=stdout) + self._current_process.wait() + + def matches_patterns(self, path): + """Return True if path matches any of the watched patterns, or if no + patterns are defined (watch everything).""" + if not self.patterns: + return True + name = os.path.basename(path) + return any(fnmatch.fnmatch(name, p) for p in self.patterns) def is_interested(self, path): if self.exclude.match(path): return False + if not self.matches_patterns(path): + return False + if path in self.paths: return True @@ -126,19 +160,32 @@ def is_interested(self, path): def on_change(self, path): if self.is_interested(path): - self.run_command(path) + if self.debounce_delay > 0: + with self._lock: + if self._debounce_timer is not None: + self._debounce_timer.cancel() + self._debounce_timer = threading.Timer( + self.debounce_delay, self.run_command, args=[path]) + self._debounce_timer.start() + else: + self.run_command(path) def on_created(self, event): - if self.observer.__class__.__name__ == 'InotifyObserver': - # inotify also generates modified events for created files - return - if not event.is_directory: self.set_envvar('event', 'file_created') + self._recently_created[event.src_path] = time.time() self.on_change(event.src_path) def on_modified(self, event): if not event.is_directory: + now = time.time() + self._recently_created = { + p: t for p, t in self._recently_created.items() + if now - t < 1.0 + } + if event.src_path in self._recently_created: + del self._recently_created[event.src_path] + return self.set_envvar('event', 'file_modified') self.on_change(event.src_path) @@ -191,6 +238,9 @@ def main(): run_once = False run_at_start = False quiet_mode = False + kill_mode = False + debounce_delay = 0 + patterns = [] while args and args[0][0] == '-': flag = args.pop(0) @@ -211,6 +261,18 @@ def main(): args = [] elif flag == '-q': quiet_mode = True + elif flag == '-k': + kill_mode = True + elif flag == '-d': + if args: + try: + debounce_delay = float(args.pop(0)) + except ValueError: + print('Error: -d requires a numeric delay in seconds') + exit(1) + elif flag == '-p': + if args: + patterns.append(args.pop(0)) else: break @@ -239,7 +301,8 @@ def main(): print("When '%s' changes, run '%s'" % (files[0], print_command)) wc = WhenChanged(files, command, recursive, run_once, run_at_start, - verbose_mode, quiet_mode) + verbose_mode, quiet_mode, kill_mode, debounce_delay, + patterns) try: wc.run()