Testcases for subscribing a single file.

tests/test_single_file_watched.py

@@ -0,0 +1,180 @@
+# coding: utf-8
+
+""" Test cases when a single file is subscribed. """
+
+from os import mkdir
+import time
+
+from .utils import P, StartWatching, TestEventQueue
+
+SLEEP = 0.001
+
+
+def test_selftest(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """ Check if all events are catched in SLEEP time. """
+
+    emitter = start_watching(path=p())  # Pretty sure this should work
+
+    with open(p('file1.bak'), 'wt') as fp:
+        fp.write('test1')
+
+    with open(p('file2.bak'), 'wt') as fp:
+        fp.write('test2')
+
+    time.sleep(SLEEP)
+
+    found_files = set()
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:  # Not catched on Windows
+                assert event.src_path == p()
+            else:
+                found_files.add(event.src_path)
+    finally:
+        emitter.stop()
+
+    assert len(found_files) == 2, 'Number of expected files differ. Increase sleep time.'
+
+
+def test_file_access(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """ Check if file fires events. """
+
+    file1 = 'file1.bak'
+    tmpfile = p(file1)
+
+    with open(tmpfile, 'wt') as fp:
+        fp.write('init1')
+
+    emitter = start_watching(path=tmpfile)
+
+    # This is what we want to see
+    with open(tmpfile, 'wt') as fp:
+        fp.write('test1')
+
+    time.sleep(SLEEP)
+
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:
+                assert event.src_path == p()
+                continue
+            else:
+                assert event.src_path.endswith(file1)
+            break
+        else:
+            assert False, 'No event catched.'
+    finally:
+        emitter.stop()
+
+
+def test_file_access_multiple(
+    p: P, start_watching: StartWatching, event_queue: TestEventQueue
+) -> None:
+    """ Check if file fires events multiple times. """
+
+    file1 = 'file1.bak'
+    tmpfile = p(file1)
+
+    with open(tmpfile, 'wt') as fp:
+        fp.write('init1')
+
+    emitter = start_watching(path=tmpfile)
+
+    try:
+        for i in range(5):
+            # This is what we want to see multiple times
+            with open(tmpfile, 'wt') as fp:
+                fp.write('test1')
+
+            time.sleep(SLEEP)
+
+            while event_queue.qsize():
+                event, _ = event_queue.get()
+                if event.is_directory:
+                    assert event.src_path == p()
+                    continue
+                else:
+                    assert event.src_path.endswith(file1)
+                break
+            else:
+                assert False, 'No event catched.'
+
+    finally:
+        emitter.stop()
+
+
+def test_file_access_other_file(
+    p: P, start_watching: StartWatching, event_queue: TestEventQueue
+) -> None:
+    """ Check if other files doesn't fires events. """
+
+    file1 = 'file1.bak'
+    tmpfile = p(file1)
+
+    with open(tmpfile, 'wt') as fp:
+        fp.write('init1')
+
+    emitter = start_watching(path=tmpfile)
+
+    # Don't wanted
+    with open(p('file2.bak'), 'wt') as fp:
+        fp.write('test2')
+
+    # but this
+    with open(tmpfile, 'wt') as fp:
+        fp.write('test1')
+
+    time.sleep(SLEEP)
+
+    found_files = set()
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:
+                assert event.src_path == p()
+            else:
+                found_files.add(event.src_path)
+                assert event.src_path.endswith(file1)
+    finally:
+        emitter.stop()
+
+    assert len(found_files) == 1, 'Number of expected files differ. Wrong file catched.'
+
+
+def test_create_folder(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """ Check if creation of a directory and inside files doesn't fires events. """
+
+    file1 = 'file1.bak'
+    tmpfile = p(file1)
+
+    with open(tmpfile, 'wt') as fp:
+        fp.write('init1')
+
+    emitter = start_watching(path=tmpfile)
+
+    # Don't wanted
+    mkdir(p('myfolder'))
+    with open(p('myfolder/file2.bak'), 'wt') as fp:
+        fp.write('test2')
+
+    # but this
+    with open(tmpfile, 'wt') as fp:
+        fp.write('test1')
+
+    time.sleep(SLEEP)
+
+    found_files = set()
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:
+                assert event.src_path == p()
+            else:
+                found_files.add(event.src_path)
+                assert event.src_path.endswith(file1)
+    finally:
+        emitter.stop()
+
+    assert len(found_files) == 1, 'Number of expected files differ. Wrong file catched.'