Skip to content

Commit

Permalink
Add test for issue gorakhargosh#797
Browse files Browse the repository at this point in the history
This works for me locally, let's see what the CI machine says.
  • Loading branch information
CCP-Aporia committed May 18, 2021
1 parent 50af6eb commit a7aa848
Showing 1 changed file with 48 additions and 2 deletions.
50 changes: 48 additions & 2 deletions tests/test_fsevents.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,10 @@ def teardown_function(function):
rm(p(""), recursive=True)


def start_watching(path=None, use_full_emitter=False):
def start_watching(path=None, recursive=True, use_full_emitter=False):
global emitter
path = p("") if path is None else path
emitter = FSEventsEmitter(event_queue, ObservedWatch(path, recursive=True), suppress_history=True)
emitter = FSEventsEmitter(event_queue, ObservedWatch(path, recursive=recursive), suppress_history=True)
emitter.start()


Expand Down Expand Up @@ -247,6 +247,52 @@ def test_converting_cfstring_to_pyunicode():
emitter.stop()


def test_issue_797():
from watchdog.events import (
PatternMatchingEventHandler,
DirCreatedEvent,
DirModifiedEvent,
FileCreatedEvent,
FileModifiedEvent
)

class TestEventHandler(PatternMatchingEventHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.expected_events = [
FileCreatedEvent(p('foo.json')),
FileModifiedEvent(p('foo.json'))
]
self.observed_events = set()

def on_any_event(self, event):
logger.info(event)
self.expected_events.remove(event)
self.observed_events.add(event)
# expected_event = self.expected_events.pop(0)
# assert expected_event == event

def done(self):
return not self.expected_events

event_handler = TestEventHandler(patterns=["*.json"], ignore_patterns=[], ignore_directories=True)
observer = Observer()
observer.schedule(event_handler, p())
observer.start()
time.sleep(0.1)

try:
touch(p('foo.json'))
timeout_at = time.time() + 5
while not event_handler.done() and time.time() < timeout_at:
time.sleep(0.1)

assert event_handler.done()
finally:
observer.stop()
observer.join()


def test_watchdog_recursive():
""" See https://github.com/gorakhargosh/watchdog/issues/706
"""
Expand Down

0 comments on commit a7aa848

Please sign in to comment.