1. JanKanis
  2. python-inotify

Commits

JanKanis  committed 853ed90 Draft

uncomment and fix all simpler tests that previously applied to Watcher for PathWatcher

  • Participants
  • Parent commits 2dc3e01
  • Branches pathwatcher

Comments (0)

Files changed (2)

File inotify/pathwatcher.py

View file
  • Ignore whitespace
             while do2 or self._pending_watch_removes > 0:
                 do2 = False
                 for e in self._read_events(bufsize):
-                    if e == lastevent:
+                    # only filter duplicate path_changed events to
+                    # prevent hiding of bugs that may cause duplicate
+                    # other events to be created.
+                    if e.path_changed and e == lastevent:
                         continue
                     self.events.append(e)
+                    lastevent = e
             self._do_reconnect()
         events, self.events = self.events, []
         return events
 
     def remove(self, path):
         '''Remove watch on the given path.'''
+        path = PosixPath(path)
         self._paths[path].remove()
         del self._paths[path]
 
     def watches(self):
         '''return an iterator of all active watches'''
-        return self._path.keys()
+        return [str(p) for p in self._paths.keys()]
 
     def getmask(self, path):
         '''returns the mask for the watch on the given path'''
 
     def close(self):
         'close this watcher instance'
+        if self.fd is None:
+            return
         os.close(self.fd)
         self._watchdescriptors.clear()
         self._paths.clear()
 
     def remove(self):
         self._poplinks_from(0)
-        # in principle we can still be reconnected, but we don't
-        # expect that to actually happen
-        self._register_reconnect()
 
     def __repr__(self):
         return '<_PathWatch for {}>'.format(str(self.path))
             self.active = False
         # The list of callbacks can be modified from the handlers, so make a
         # copy.
-        for l in list(self.callbacks.get(event.name, ())):
+        callbacks = list(self.callbacks.get(event.name, ()))
+        if event.name != None:
+            callbacks.extend(self.callbacks.get(None, ()))
+        for l in callbacks:
             if not event.mask & (l.mask | IN_IGNORED):
                 continue
             for e in l.handle_event(event):
             self.watcher._removewatch(self)
         
     def __repr__(self):
-        names = ', '.join(set(c.__self__.printname() for lst in self.callbacks.values() for m, c in lst))
+        names = ', '.join(set(l.printname() for lst in self.callbacks.values() for l in lst))
         return '<_Descriptor for wd {}: {}>'.format(self.wd, names)
 
 

File test/testpath.py

View file
  • Ignore whitespace
 
 
 @pytest.fixture
-def w():
-  return inotify.PathWatcher()
+def w(request):
+  w = inotify.PathWatcher()
+  request.addfinalizer(lambda w=w: w.close())
+  return w
 
 
 def test_constants():
   assert not c_events & path_events
   assert path_events > c_events
 
+def test_repr(w):
+  w.add('.', IN_ALL_EVENTS)
+  repr(w)
+  repr(w._paths)
+  repr(w._watchdescriptors)
+  repr(next(iter(w._paths.values())).links)
 
 def test_open(w):
   mask = IN_OPEN | IN_CLOSE
   assert w._watchdescriptors[watchdesc] == wd
   
   open('testfile').close()
-  ev1, ev2 = w.read(block=False)
+  evts = w.read(block=False)
+  ev1, ev2 = evts
   assert ev1.open
   assert ev2.close
   assert ev2.close_nowrite
   open('testfile').close()
   evts.extend(w.read())
   assert len(evts) == 4
-  print(evts)
   
 
-# def test_move(w):
-#   w.add('.', inotify.IN_MOVE)
-#   assert w.read(0) == []
-#   os.rename('testfile', 'targetfile')
-#   ev = w.read(0)
-#   for e in ev:
-#     if e.name == 'testfile':
-#       assert e.moved_from
-#     if e.name == 'targetfile':
-#       assert e.moved_to
-#   assert ev[0].cookie and ev[0].cookie == ev[1].cookie
+def test_move(w):
+  w.add('.', IN_MOVE)
+  assert w.read(0) == []
+  os.rename('testfile', 'targetfile')
+  ev = w.read()
+  for e in ev:
+    if e.name == 'testfile':
+      assert e.moved_from
+    if e.name == 'targetfile':
+      assert e.moved_to
+  assert ev[0].cookie and ev[0].cookie == ev[1].cookie
 
 
-# def test_alias(w):
-#   '''The inotify system maps watch requests to aliases (e.g. symlinks) to the
-#   same watch descriptor, so we need to be sure that a watch is only really
-#   removed if all paths it is watching are dismissed.'''
+def test_alias(w):
+  '''The inotify system maps watch requests to aliases (e.g. symlinks) to the
+  same watch descriptor, so we need to be sure that a watch is only really
+  removed if all paths it is watching are dismissed.'''
 
-#   os.symlink('testfile', 'testlink')
-#   w1 = w.add('testfile', inotify.IN_OPEN)
-#   w2 = w.add('testlink', inotify.IN_OPEN)
-#   assert w1 == w2
-#   assert set(w.paths()) == {'testfile', 'testlink'}
-#   assert w.get_watch('testfile') == w.get_watch('testlink')
-#   assert len(w.watches()) == 1
-#   open('testlink').close()
-#   ev = w.read(0)
-#   assert len(ev) == 1
-#   w.remove_path('testfile')
-#   open('testlink').close()
-#   ev = w.read(0)
-#   assert len(ev) == 1
+  os.symlink('testfile', 'testlink')
+  w1 = w.add('testfile', inotify.IN_OPEN)
+  w2 = w.add('testlink', inotify.IN_OPEN)
+  assert set(w.watches()) == {'testfile', 'testlink'}
+  open('testlink').close()
+  ev = w.read(0)
+  assert len(ev) == 2
+  w.remove('testfile')
+  open('testlink').close()
+  ev = w.read(0)
+  assert len(ev) == 1
 
 
-# def test_delete(w):
-#     w.add('testfile', inotify.IN_DELETE_SELF)
-#     os.remove('testfile')
-#     ev1, ev2 = w.read(0)
-#     assert ev1.delete_self
-#     assert ev2.ignored
-#     assert w.num_watches() == 0
+def test_delete(w):
+    w.add('testfile', IN_DELETE_SELF | IN_IGNORED)
+    os.remove('testfile')
+    evts = w.read(0)
+    ev1, ev2, ev3 = evts
+    assert ev1.delete_self
+    assert ev2.ignored
+    assert ev3.path_delete
+    assert len(w.watches()) == 1
 
-# def test_wrongpath(w):
-#     with pytest.raises(OSError) as excinfo:
-#         w.add('nonexistant', inotify.IN_OPEN)
-#     assert excinfo.value.errno == os.errno.ENOENT
-#     with pytest.raises(OSError) as excinfo:
-#         w.add_all('nonexistant', inotify.IN_OPEN)
-#     assert excinfo.value.errno == os.errno.ENOENT
+def test_wrongpath(w):
+    w.add('nonexistant', IN_OPEN)
+    assert w.read(block=False) == []
+    open('nonexistant', 'w').close()
+    evts = w.read()
+    ev1 = evts[0]
+    assert ev1.path_create
+
+