old_contents: from sparts.tasks.file import DirectoryWatcherTask
  from ..base import SingleTaskTestCase
  from tempfile import mkdtemp
  from shutil import rmtree
  
  import os.path
  import time
  
  
  class MyTask(DirectoryWatcherTask):
      def __init__(self, *args, **kwargs):
          super(MyTask, self).__init__(*args, **kwargs) 
          self.onFileCreated = self.service.test.mock.Mock()
          self.onFileDeleted = self.service.test.mock.Mock()
          self.onFileChanged = self.service.test.mock.Mock()
  
  class TestMyTask(SingleTaskTestCase):
      TASK = MyTask
  
      def setUp(self):
          self.testpath = mkdtemp()
          MyTask.PATH = self.testpath
          MyTask.INTERVAL = 0.25
          super(TestMyTask, self).setUp()
          self.task.execute()
  
      def tearDown(self):
          rmtree(self.testpath)
          super(TestMyTask, self).tearDown()
  
      def test_file_create(self):
          fn = os.path.join(self.testpath, 'foo')
          with open(fn, mode='w'):
              pass
          os.utime(fn, None)
          self.task.execute()
          self.assertTrue(self.task.onFileCreated.called)
          self.assertEquals(self.task.onFileCreated.call_count, 1)
          self.assertEquals(self.task.onFileCreated.call_args[0][0], 'foo',
                            self.task.onFileCreated.call_args)
  
      def test_file_delete(self):
          self.test_file_create()
          os.remove(os.path.join(self.testpath, 'foo'))
          self.task.execute()
  
          self.assertTrue(self.task.onFileDeleted.called)
          self.assertEquals(self.task.onFileDeleted.call_count, 1)
          self.assertEquals(self.task.onFileDeleted.call_args[0][0], 'foo',
                            self.task.onFileDeleted.call_args)
  
      def test_file_update(self):
          self.test_file_create()
          # Sleep one second so that mtime/atime will change appropriately
          time.sleep(1.0)
          os.utime(os.path.join(self.testpath, 'foo'), None)
          self.task.execute()
          self.assertTrue(self.task.onFileChanged.called)
          self.assertEquals(self.task.onFileChanged.call_count, 1)
          self.assertEquals(self.task.onFileChanged.call_args[0][0], 'foo',
                            self.task.onFileChanged.call_args)
  
new_contents: from sparts.tasks.file import DirectoryWatcherTask
  from ..base import SingleTaskTestCase
  from tempfile import mkdtemp
  from shutil import rmtree
  
  import os.path
  import time
  
  
  class MyTask(DirectoryWatcherTask):
      def __init__(self, *args, **kwargs):
          super(MyTask, self).__init__(*args, **kwargs) 
          self.onFileCreated = self.service.test.mock.Mock()
          self.onFileDeleted = self.service.test.mock.Mock()
          self.onFileChanged = self.service.test.mock.Mock()
  
  class TestMyTask(SingleTaskTestCase):
      TASK = MyTask
  
      def setUp(self):
          self.testpath = mkdtemp()
          MyTask.PATH = self.testpath
          MyTask.INTERVAL = 0.25
          super(TestMyTask, self).setUp()
          self.task.execute()
  
      def tearDown(self):
          rmtree(self.testpath)
          super(TestMyTask, self).tearDown()
  
      def test_file_create(self):
          fn = os.path.join(self.testpath, 'foo')
          with open(fn, mode='w'):
              pass
          os.utime(fn, None)
          self.task.execute()
          self.assertTrue(self.task.onFileCreated.called)
          self.assertEquals(self.task.onFileCreated.call_count, 1)
          self.assertEquals(self.task.onFileCreated.call_args[0][0], 'foo',
                            self.task.onFileCreated.call_args)
  
      def test_file_delete(self):
          self.test_file_create()
          os.remove(os.path.join(self.testpath, 'foo'))
          self.task.execute()
  
          self.assertTrue(self.task.onFileDeleted.called)
          self.assertEquals(self.task.onFileDeleted.call_count, 1)
          self.assertEquals(self.task.onFileDeleted.call_args[0][0], 'foo',
                            self.task.onFileDeleted.call_args)
  
      def test_file_update(self):
          self.test_file_create()
  
          path = os.path.join(self.testpath, 'foo')
  
          # Forcibly update the atime/mtime of the file
          st = os.stat(path)
          os.utime(path, (st.st_atime - 10, st.st_mtime - 10))
  
          self.task.execute()
          self.assertTrue(self.task.onFileChanged.called)
          self.assertEquals(self.task.onFileChanged.call_count, 1)
          self.assertEquals(self.task.onFileChanged.call_args[0][0], 'foo',
                            self.task.onFileChanged.call_args)
  
subject: Make file test case pass more quickly