Commits

Christopher Nilsson  committed d1890fd

Added copy & move unittests.

Made the unittest output a bit less spammy as well.

  • Participants
  • Parent commits c53117e

Comments (0)

Files changed (2)

File dirwatch/base.py

 import time
 
 LOG_NAME = "dirwatch"
-LOG_STANDARD_FMT = "%(asctime)s p=%(process)d t=%(thread)d [%(levelname)s] %(message)s"
+LOG_STANDARD_FMT = "%(asctime)s p=%(process)d t=%(thread)d [%(levelname)s] %(message)s\n"
 _LOG_HANDLER_ADDED = False
 
 class DirWatcher(object):
         self.monitor_events = Queue.Queue()
         self._exit_notification = object()
         self.monitoring = False      
-        self.init_logging() 
+        self.init_logging(verbosity=logging.WARNING) 
         self._worker_thread = None
         self.snapshot = {}
         

File dirwatch/test/test_dirwatch.py

             if path not in self.path_events:
                 self.path_events[path] = []
             self.path_events[path].append(event) 
-            print "%s [%s] %s" % (time.ctime(event.timestamp), event.event_type, event.path)           
+            #print "%s [%s] %s" % (time.ctime(event.timestamp), event.event_type, event.path)           
         
     def test_on_create(self):
-        """ Check we see some events on creating a test file. """                
+        """ Check we see some events on creating a test file.         
+        """                
         self.dw.add_watch(self.testdir)
         
         self.dw.start_monitor()
             self.dw.stop_monitor()
             
     def test_multiple_creates(self):
+        """ Check we see the events for ALL our files.         
+        """
         self.dw.add_watch(self.testdir)
         self.dw.start_monitor()
         try:
                     testfile.write("test")
                     testfile.close()
                     self.cleanup.append(self.testfilename + "_" + str(idx))
-                time.sleep(0.02)
+                time.sleep(0.02) # yield cpu in between file drops...
         finally:
             self.dw.stop_monitor()
         self.assertEquals(num_of_paths, len(self.path_events.keys()))
+        
+    def test_copyfile_to_watch_dir(self):
+        """ Check we notice when a file is copied into our watched directory.         
+        """
+        self.dw.add_watch(self.testdir)
+        self.dw.start_monitor()
+        outfilename = get_test_filename()
+        try:
+            with open(__file__, "r") as srcf, open(outfilename, "w") as outf:
+                for line in srcf:
+                    outf.write(line)
+            self.cleanup.append(outfilename)
+        finally:
+            self.dw.stop_monitor()
+        
+        self.assertIn(outfilename, self.path_events, "Test file '{0}' not seen".format(outfilename))
+        
+    def test_movefile_to_watch_dir(self):
+        """ Check we notice when a file is moved into our watched directory (usually no modify involved).         
+        """
+        self.dw.add_watch(self.testdir)
+        self.dw.start_monitor()
+        outfilename = get_test_filename()
+        srcfilename = os.path.join(os.getcwd(), "testsrcfile")
+        try:
+            with open(srcfilename, "w+") as srcfile:
+                srcfile.write("testing...\n")
+            try:
+                os.rename(srcfilename, outfilename)
+            except:
+                self.cleanup.append(srcfilename)
+            else:
+                self.cleanup.append(outfilename)
+        finally:
+            self.dw.stop_monitor()
+        
+        self.assertIn(outfilename, self.path_events, "Test file '{0}' not seen".format(outfilename))    
+        
             
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main(verbosity=2)