1from __future__ import annotations 2 3import os 4from queue import Empty, Queue 5from time import sleep 6 7import pytest 8 9from watchdog.events import ( 10 DirCreatedEvent, 11 DirDeletedEvent, 12 DirModifiedEvent, 13 DirMovedEvent, 14 FileCreatedEvent, 15 FileDeletedEvent, 16 FileModifiedEvent, 17 FileMovedEvent, 18) 19from watchdog.observers.api import ObservedWatch 20from watchdog.observers.polling import PollingEmitter as Emitter 21 22from .shell import mkdir, mkdtemp, msize, mv, rm, touch 23 24SLEEP_TIME = 0.4 25TEMP_DIR = mkdtemp() 26 27 28def p(*args): 29 """ 30 Convenience function to join the temporary directory path 31 with the provided arguments. 32 """ 33 return os.path.join(TEMP_DIR, *args) 34 35 36@pytest.fixture 37def event_queue(): 38 return Queue() 39 40 41@pytest.fixture 42def emitter(event_queue): 43 watch = ObservedWatch(TEMP_DIR, recursive=True) 44 em = Emitter(event_queue, watch, timeout=0.2) 45 em.start() 46 yield em 47 em.stop() 48 em.join(5) 49 50 51def test___init__(event_queue, emitter): 52 sleep(SLEEP_TIME) 53 mkdir(p("project")) 54 55 sleep(SLEEP_TIME) 56 mkdir(p("project", "blah")) 57 58 sleep(SLEEP_TIME) 59 touch(p("afile")) 60 61 sleep(SLEEP_TIME) 62 touch(p("fromfile")) 63 64 sleep(SLEEP_TIME) 65 mv(p("fromfile"), p("project", "tofile")) 66 67 sleep(SLEEP_TIME) 68 touch(p("afile")) 69 70 sleep(SLEEP_TIME) 71 mv(p("project", "blah"), p("project", "boo")) 72 73 sleep(SLEEP_TIME) 74 rm(p("project"), recursive=True) 75 76 sleep(SLEEP_TIME) 77 rm(p("afile")) 78 79 sleep(SLEEP_TIME) 80 msize(p("bfile")) 81 82 sleep(SLEEP_TIME) 83 rm(p("bfile")) 84 85 sleep(SLEEP_TIME) 86 emitter.stop() 87 88 # What we need here for the tests to pass is a collection type 89 # that is: 90 # * unordered 91 # * non-unique 92 # A multiset! Python's collections.Counter class seems appropriate. 93 expected = { 94 DirModifiedEvent(p()), 95 DirCreatedEvent(p("project")), 96 DirModifiedEvent(p("project")), 97 DirCreatedEvent(p("project", "blah")), 98 FileCreatedEvent(p("afile")), 99 DirModifiedEvent(p()), 100 FileCreatedEvent(p("fromfile")), 101 DirModifiedEvent(p()), 102 DirModifiedEvent(p()), 103 FileModifiedEvent(p("afile")), 104 DirModifiedEvent(p("project")), 105 DirModifiedEvent(p()), 106 FileDeletedEvent(p("project", "tofile")), 107 DirDeletedEvent(p("project", "boo")), 108 DirDeletedEvent(p("project")), 109 DirModifiedEvent(p()), 110 FileDeletedEvent(p("afile")), 111 DirModifiedEvent(p()), 112 FileCreatedEvent(p("bfile")), 113 FileModifiedEvent(p("bfile")), 114 DirModifiedEvent(p()), 115 FileDeletedEvent(p("bfile")), 116 } 117 118 expected.add(FileMovedEvent(p("fromfile"), p("project", "tofile"))) 119 expected.add(DirMovedEvent(p("project", "blah"), p("project", "boo"))) 120 121 got = set() 122 123 while True: 124 try: 125 event, _ = event_queue.get_nowait() 126 got.add(event) 127 except Empty: 128 break 129 130 assert expected == got 131 132 133def test_delete_watched_dir(event_queue, emitter): 134 rm(p(""), recursive=True) 135 136 sleep(SLEEP_TIME) 137 emitter.stop() 138 139 # What we need here for the tests to pass is a collection type 140 # that is: 141 # * unordered 142 # * non-unique 143 # A multiset! Python's collections.Counter class seems appropriate. 144 expected = { 145 DirDeletedEvent(os.path.dirname(p(""))), 146 } 147 148 got = set() 149 150 while True: 151 try: 152 event, _ = event_queue.get_nowait() 153 got.add(event) 154 except Empty: 155 break 156 157 assert expected == got 158