cosmetic refactor of tests and utilities

tests/shell.py

@@ -21,6 +21,7 @@
     :synopsis: Common shell operations for testing.
     :author: yesudeep@google.com (Yesudeep Mangalapilly)
 """
+
 from __future__ import with_statement
 
 import os.path
@@ -29,7 +30,7 @@
 import errno
 
 
-#def tree(path='.', show_files=False):
+# def tree(path='.', show_files=False):
 #    print(path)
 #    padding = ''
 #    for root, directories, filenames in os.walk(path):
@@ -40,64 +41,64 @@
 
 
 def cd(path):
-  os.chdir(path)
+    os.chdir(path)
 
 
 def pwd():
-  path = os.getcwd()
-  print(path)
-  return path
+    path = os.getcwd()
+    print(path)
+    return path
 
 
 def mkdir(path, parents=False):
-  """Creates a directory (optionally also creates all the parent directories
-in the path)."""
-  if parents:
-    try:
-      os.makedirs(path)
-    except OSError as e:
-      if not e.errno == errno.EEXIST:
-        raise
-  else:
-    os.mkdir(path)
+    """Creates a directory (optionally also creates all the parent directories
+  in the path)."""
+    if parents:
+        try:
+            os.makedirs(path)
+        except OSError as e:
+            if not e.errno == errno.EEXIST:
+                raise
+    else:
+        os.mkdir(path)
 
 
 def rm(path, recursive=False):
-  """Deletes files or directories."""
-  if os.path.isdir(path):
-    if recursive:
-      shutil.rmtree(path)
-    #else:
-    #    os.rmdir(path)
+    """Deletes files or directories."""
+    if os.path.isdir(path):
+        if recursive:
+            shutil.rmtree(path)
+        # else:
+        #    os.rmdir(path)
+        else:
+            raise OSError("rm: %s: is a directory." % path)
     else:
-      raise OSError("rm: %s: is a directory." % path)
-  else:
-    os.remove(path)
+        os.remove(path)
 
 
 def touch(path, times=None):
-  """Updates the modified timestamp of a file or directory."""
-  if os.path.isdir(path):
-    os.utime(path, times)
-  else:
-    with open(path, 'ab'):
-      os.utime(path, times)
+    """Updates the modified timestamp of a file or directory."""
+    if os.path.isdir(path):
+        os.utime(path, times)
+    else:
+        with open(path, 'ab'):
+            os.utime(path, times)
 
 
 def truncate(path):
-  """Truncates a file."""
-  with open(path, 'wb'):
-    os.utime(path, None)
+    """Truncates a file."""
+    with open(path, 'wb'):
+        os.utime(path, None)
 
 
 def mv(src_path, dest_path):
-  """Moves files or directories."""
-  shutil.move(src_path, dest_path)
+    """Moves files or directories."""
+    shutil.move(src_path, dest_path)
 
 
 def mkdtemp():
-  return tempfile.mkdtemp()
+    return tempfile.mkdtemp()
 
 
 def ls(path='.'):
-  return os.listdir(path)
+    return os.listdir(path)

tests/test_skip_repeats_queue.py

@@ -1,7 +1,27 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2011 Yesudeep Mangalapilly <yesudeep@gmail.com>
+# Copyright 2012 Google, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import unittest2
 from watchdog.utils.bricks import SkipRepeatsQueue
 
+
 class TestSkipRepeatsQueue(unittest2.TestCase):
+
     def test_basic_queue(self):
         q = SkipRepeatsQueue()
 
@@ -33,7 +53,6 @@ def test_allow_nonconsecutive(self):
         self.assertEqual(e1, q.get())
         self.assertTrue(q.empty())
 
-
     def test_prevent_consecutive(self):
         q = SkipRepeatsQueue()
 

tests/test_watch_observers_winapi.py

@@ -21,74 +21,76 @@
 import unittest2
 
 try:
-  import queue  # IGNORE:F0401
+    import queue  # IGNORE:F0401
 except ImportError:
-  import Queue as queue  # IGNORE:F0401
+    import Queue as queue  # IGNORE:F0401
 
 from time import sleep
-from tests.shell import\
-  mkdir,\
-  mkdtemp,\
-  touch,\
-  rm,\
-  mv
-
-from watchdog.events import DirModifiedEvent, DirCreatedEvent,\
-  FileCreatedEvent,\
-  FileMovedEvent, FileModifiedEvent, DirMovedEvent, FileDeletedEvent,\
-  DirDeletedEvent
+
+from tests.shell import (
+    mkdir,
+    mkdtemp,
+    mv
+)
+
+
+from watchdog.events import (
+    DirCreatedEvent,
+    DirMovedEvent,
+)
 
 from watchdog.observers.api import ObservedWatch
 from watchdog.utils import platform
 
 if platform.is_windows():
-  from watchdog.observers.read_directory_changes import WindowsApiEmitter as Emitter
-
-  temp_dir = mkdtemp()
-
-  def p(*args):
-    """
-    Convenience function to join the temporary directory path
-    with the provided arguments.
-    """
-    return os.path.join(temp_dir, *args)
-
-  class TestWindowsApiEmitter(unittest2.TestCase):
-    def setUp(self):
-      self.event_queue = queue.Queue()
-      self.watch = ObservedWatch(temp_dir, True)
-      self.emitter = Emitter(self.event_queue, self.watch, timeout=0.2)
-
-    def teardown(self):
-      pass
-
-    def test___init__(self):
-      SLEEP_TIME = 1
-      self.emitter.start()
-      sleep(SLEEP_TIME)
-      mkdir(p('fromdir'))
-      sleep(SLEEP_TIME)
-      mv(p('fromdir'), p('todir'))
-      sleep(SLEEP_TIME)
-      self.emitter.stop()
-
-      # What we need here for the tests to pass is a collection type
-      # that is:
-      #   * unordered
-      #   * non-unique
-      # A multiset! Python's collections.Counter class seems appropriate.
-      expected = set([
-        DirCreatedEvent(p('fromdir')),
-        DirMovedEvent(p('fromdir'),p('todir')),
-        ])
-      got = set()
-
-      while True:
-        try:
-          event, _ = self.event_queue.get_nowait()
-          got.add(event)
-        except queue.Empty:
-          break
-
-      print(got)
-      self.assertEqual(expected, got)
+    from watchdog.observers.read_directory_changes import WindowsApiEmitter as Emitter
+
+    temp_dir = mkdtemp()
+
+    def p(*args):
+        """
+        Convenience function to join the temporary directory path
+        with the provided arguments.
+        """
+        return os.path.join(temp_dir, *args)
+
+    class TestWindowsApiEmitter(unittest2.TestCase):
+
+        def setUp(self):
+            self.event_queue = queue.Queue()
+            self.watch = ObservedWatch(temp_dir, True)
+            self.emitter = Emitter(self.event_queue, self.watch, timeout=0.2)
+
+        def teardown(self):
+            pass
+
+        def test___init__(self):
+            SLEEP_TIME = 1
+            self.emitter.start()
+            sleep(SLEEP_TIME)
+            mkdir(p('fromdir'))
+            sleep(SLEEP_TIME)
+            mv(p('fromdir'), p('todir'))
+            sleep(SLEEP_TIME)
+            self.emitter.stop()
+
+            # What we need here for the tests to pass is a collection type
+            # that is:
+            #   * unordered
+            #   * non-unique
+            # A multiset! Python's collections.Counter class seems appropriate.
+            expected = set([
+                           DirCreatedEvent(p('fromdir')),
+                           DirMovedEvent(p('fromdir'), p('todir')),
+                           ])
+            got = set()
+
+            while True:
+                try:
+                    event, _ = self.event_queue.get_nowait()
+                    got.add(event)
+                except queue.Empty:
+                    break
+
+            print(got)
+            self.assertEqual(expected, got)