Bug fixes, new features and test suite

.gitignore

@@ -4,3 +4,4 @@
 # exclude patterns (uncomment them if you want to use them):
 *~
 *.pyc
+dist/

README.md

@@ -13,7 +13,7 @@ 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!
@@ -27,7 +27,7 @@ Sweetness!
 
 ### Installation
 ```sh
-pip install https://github.com/joh/when-changed/archive/master.zip
+pipx install https://github.com/joh/when-changed/archive/master.zip
 ```
 
 
@@ -47,6 +47,9 @@ FILE can be a directory. Use %f to pass the filename to the command.
 - -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:
 
@@ -61,3 +64,22 @@ Could be either:
 
 - 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
+```
+
+Watch only Python and YAML files in a directory:
+
+```sh
+$ when-changed -r -p '*.py' -p '*.yml' src/ make lint
+```

pyproject.toml

@@ -0,0 +1,19 @@
+[build-system]
+requires = ["setuptools>=42", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "when-changed"
+version = "0.3.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"

pytest.ini

@@ -0,0 +1,5 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*

requirements.txt

@@ -1,2 +1 @@
 watchdog
-subprocess32

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()

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()

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()

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()