Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_observer__ne__():
watch1 = ObservedWatch('/foobar', True)
watch2 = ObservedWatch('/foobar', True)
watch_ne1 = ObservedWatch('/foo', True)
watch_ne2 = ObservedWatch('/foobar', False)
assert not watch1.__ne__(watch2)
assert watch1.__ne__(watch_ne1)
assert watch1.__ne__(watch_ne2)
def test_observer__ne__():
watch1 = ObservedWatch('/foobar', True)
watch2 = ObservedWatch('/foobar', True)
watch_ne1 = ObservedWatch('/foo', True)
watch_ne2 = ObservedWatch('/foobar', False)
assert not watch1.__ne__(watch2)
assert watch1.__ne__(watch_ne1)
assert watch1.__ne__(watch_ne2)
def emitter(event_queue):
watch = ObservedWatch(temp_dir, True)
em = Emitter(event_queue, watch, timeout=0.2)
em.start()
yield em
em.stop()
em.join(5)
def test_move(p):
mkdir(p('dir1'))
mkdir(p('dir2'))
touch(p('dir1', 'a'))
event_queue = Queue()
emitter = InotifyEmitter(event_queue, ObservedWatch(p(''), recursive=True))
emitter.start()
mv(p('dir1', 'a'), p('dir2', 'b'))
event = event_queue.get(timeout=5)[0]
assert event.event_type == 'moved'
assert not event.is_directory
assert event.src_path == p('dir1', 'a').encode()
assert event.dest_path == p('dir2', 'b').encode()
event = event_queue.get(timeout=5)[0]
assert event.event_type == 'modified'
assert event.is_directory
assert event.src_path == p('dir1').encode()
event = event_queue.get(timeout=5)[0]
assert event.event_type == 'modified'
def test_observer__eq__():
watch1 = ObservedWatch('/foobar', True)
watch2 = ObservedWatch('/foobar', True)
watch_ne1 = ObservedWatch('/foo', True)
watch_ne2 = ObservedWatch('/foobar', False)
assert watch1 == watch2
assert watch1.__eq__(watch2)
assert not watch1.__eq__(watch_ne1)
assert not watch1.__eq__(watch_ne2)
def test_event_emitter():
event_queue = EventQueue()
watch = ObservedWatch('/foobar', True)
event_emitter = EventEmitter(event_queue, watch, timeout=1)
event_emitter.queue_event(FileModifiedEvent('/foobar/blah'))
def test_event_dispatcher():
event = FileModifiedEvent('/foobar')
watch = ObservedWatch('/path', True)
class TestableEventDispatcher(EventDispatcher):
def dispatch_event(self, event, watch):
assert True
event_dispatcher = TestableEventDispatcher()
event_dispatcher.event_queue.put((event, watch))
event_dispatcher.start()
time.sleep(1)
event_dispatcher.stop()
def start_watching(path=None, use_full_emitter=False, recursive=True):
path = p('') if path is None else path
global emitter
if platform.is_linux() and use_full_emitter:
emitter = InotifyFullEmitter(event_queue, ObservedWatch(path, recursive=recursive))
else:
emitter = Emitter(event_queue, ObservedWatch(path, recursive=recursive))
if platform.is_darwin():
# FSEvents will report old events (like create for mkdtemp in test
# setup. Waiting for a considerable time seems to 'flush' the events.
time.sleep(10)
emitter.start()
def _set_root(self, root):
"""Set up watching `root` or closest existing parent."""
# schedule new root watch
while True:
try:
watch = self.root_observer.schedule(
event_handler=self, path=root, recursive=False
)
except OSError as sched_err:
# root doesn't exist, move up one directory and try again
try:
# clean up from failed scheduling
self.root_observer.unschedule(ObservedWatch(root, False))
except KeyError:
pass
newroot = os.path.dirname(root)
if newroot == root:
# we've gotten to system root and still failed
raise sched_err
else:
root = newroot
else:
# watch was set, break out of while loop
break
# add regex for root and next subdirectory
regexes = []
regexes.append("^" + re.escape(root) + "$")
nextdir = self._get_next_dir_in_path(root)
"""
instr_folder = Directory(os.path.join(MODULE_PATH, 'profiles'))
instrs = Dict(Str, File)
instrs_name = List(Str)
selected_instr_name = Str
selected_instr = Instance(InstrumentForm)
add_instr = Button('Add')
edit_instr = Button('Edit')
delete_instr = Button('Delete')
observer = Instance(Observer,())
event_handler = Instance(FileListUpdater)
watch = Instance(ObservedWatch)
instrs_view = View(
HGroup(
UItem('instrs_name',
editor = ListStrEditor(
selected = 'selected_instr_name'),
),
UItem('selected_instr', style = 'custom'),
VGroup(
UItem('add_instr'),
UItem('edit_instr'),
UItem('delete_instr'),
),
),
handler = InstrumentManagerHandler(),
title = 'Instrument Manager',