← Back to branch summary

~ubuntu-branches/ubuntu/precise/ubuntuone-client/precise-201112142106

« back to all changes in this revision

Viewing changes to tests/platform/windows/test_filesystem_notifications.py

  • Committer: Bazaar Package Importer
  • Author(s): Rodney Dawes
  • Date: 2011-08-25 16:11:47 UTC
  • mfrom: (1.1.54 upstream)
  • Revision ID: james.westby@ubuntu.com-20110825161147-v6zedpznh2evnurj
Tags: 1.7.2-0ubuntu1
https://launchpad.net/bugs/829186
* New upstream release.
  - Work correctly with static and GI bindings of gobject (LP: #829186)

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/platform/windows/test_filesystem_notifications.py
32
32
    IN_OPEN,
33
33
    IN_CLOSE_WRITE
34
34
)
 
35
from ubuntuone.platform.windows import filesystem_notifications
35
36
from ubuntuone.platform.windows.filesystem_notifications import (
36
37
    FilesystemMonitor,
37
38
    NotifyProcessor,
47
48
)
48
49
 
49
50
 
 
51
class FakeException(Exception):
 
52
    """A fake Exception used in tests."""
 
53
 
 
54
 
50
55
class TestCaseHandler(ProcessEvent):
51
56
    """ProcessEvent used for test cases."""
52
57
 
613
618
        watch = Watch(1, path, None, True, None)
614
619
        self.assertEqual(watch.started, watch._watch_started_deferred)
615
620
 
 
621
    def test_stopped_property(self):
 
622
        """Test that the stopped property returns the stopped deferred."""
 
623
        path = u'\\\\?\\C:\\path'  # a valid windows path
 
624
        watch = Watch(1, path, None, True, None)
 
625
        self.assertEqual(watch.stopped, watch._watch_stopped_deferred)
 
626
 
 
627
    def random_error(self, *args):
 
628
        """Throw a fake exception."""
 
629
        raise FakeException()
 
630
 
 
631
    @defer.inlineCallbacks
 
632
    def test_start_watching_fails_early_in_thread(self):
 
633
        """An early failure inside the thread should errback the deferred."""
 
634
        test_path = self.mktemp("test_directory")
 
635
        self.patch(filesystem_notifications, "CreateFileW", self.random_error)
 
636
        watch = Watch(1, test_path, None, True, None)
 
637
        d = watch.start_watching()
 
638
        yield self.assertFailure(d, FakeException)
 
639
 
 
640
    @defer.inlineCallbacks
 
641
    def test_start_watching_fails_late_in_thread(self):
 
642
        """A late failure inside the thread should errback the deferred."""
 
643
        test_path = self.mktemp("test_directory")
 
644
        self.patch(filesystem_notifications, "ReadDirectoryChangesW",
 
645
                   self.random_error)
 
646
        watch = Watch(1, test_path, None, True, None)
 
647
        d = watch.start_watching()
 
648
        yield self.assertFailure(d, FakeException)
 
649
 
 
650
    @defer.inlineCallbacks
 
651
    def test_close_handle_is_called_on_error(self):
 
652
        """CloseHandle is called when there's an error in the watch thread."""
 
653
        test_path = self.mktemp("test_directory")
 
654
        close_called = []
 
655
        self.patch(filesystem_notifications, "CreateFileW", lambda *_: None)
 
656
        self.patch(filesystem_notifications, "CloseHandle",
 
657
                   close_called.append)
 
658
        self.patch(filesystem_notifications, "ReadDirectoryChangesW",
 
659
                   self.random_error)
 
660
        watch = Watch(1, test_path, self.mask, True, None)
 
661
        d = watch.start_watching()
 
662
        yield self.assertFailure(d, FakeException)
 
663
        self.assertEqual(len(close_called), 1)
 
664
        yield watch.stop_watching()
 
665
 
 
666
    @defer.inlineCallbacks
 
667
    def test_stop_watching_fired_when_watch_thread_finishes(self):
 
668
        """The deferred returned is fired when the watch thread finishes."""
 
669
        events = []
 
670
        test_path = self.mktemp("another_test_directory")
 
671
        fake_processor = events.append
 
672
        watch = Watch(1, test_path, self.mask, True, fake_processor)
 
673
        yield watch.start_watching()
 
674
        self.assertNotEqual(watch._watch_handle, None)
 
675
        yield watch.stop_watching()
 
676
        self.assertEqual(watch._watch_handle, None)
 
677
 
616
678
 
617
679
class TestWatchManager(BaseTwistedTestCase):
618
680
    """Test the watch manager."""
626
688
        self.manager = WatchManager(None)
627
689
        self.manager._wdm = {1: self.watch}
628
690
 
 
691
    @defer.inlineCallbacks
629
692
    def test_stop(self):
630
 
        """Test that the different watches are stoppe."""
 
693
        """Test that the different watches are stopped."""
631
694
        self.was_called = False
632
695
 
633
 
        def stop_watching():
 
696
        def fake_stop_watching(watch):
634
697
            """Fake stop watch."""
635
698
            self.was_called = True
636
 
 
637
 
        self.watch.stop_watching = stop_watching
638
 
        self.manager.stop()
639
 
        self.assertTrue(self.was_called, 'The watch stop was not called.')
 
699
            return defer.succeed(True)
 
700
 
 
701
        self.patch(Watch, "stop_watching", fake_stop_watching)
 
702
        yield self.manager.stop()
 
703
        self.assertTrue(self.was_called, 'The watch stop should be called.')
 
704
 
 
705
    @defer.inlineCallbacks
 
706
    def test_stop_multiple(self):
 
707
        """Test that stop is fired when *all* watches have stopped."""
 
708
 
 
709
        def fake_stop_watching(watch):
 
710
            """Another fake stop watch."""
 
711
            return watch.stopped
 
712
 
 
713
        self.patch(Watch, "stop_watching", fake_stop_watching)
 
714
        second_path = self.parent_path + u"second_path"
 
715
        second_watch = Watch(2, second_path, None, True, None)
 
716
        self.manager._wdm[2] = second_watch
 
717
        d = self.manager.stop()
 
718
        self.assertFalse(d.called, "Not fired before all watches end")
 
719
        self.watch.stopped.callback(None)
 
720
        self.assertFalse(d.called, "Not fired before all watches end")
 
721
        second_watch.stopped.callback(None)
 
722
        yield d
 
723
        self.assertTrue(d.called, "Fired after the watches ended")
640
724
 
641
725
    def test_get_present_watch(self):
642
726
        """Test that we can get a Watch using is wd."""
646
730
        """Test that we get an error when trying to get a missing wd."""
647
731
        self.assertRaises(KeyError, self.manager.get_watch, (1,))
648
732
 
 
733
    @defer.inlineCallbacks
649
734
    def test_delete_present_watch(self):
650
735
        """Test that we can remove a present watch."""
651
736
        self.was_called = False
653
738
        def stop_watching():
654
739
            """Fake stop watch."""
655
740
            self.was_called = True
 
741
            return defer.succeed(True)
656
742
 
657
743
        self.watch.stop_watching = stop_watching
658
 
        self.manager.del_watch(1)
 
744
        yield self.manager.del_watch(1)
659
745
        self.assertRaises(KeyError, self.manager.get_watch, (1,))
660
746
 
661
747
    def test_add_single_watch(self):
662
748
        """Test the addition of a new single watch."""
663
749
        self.was_called = False
664
750
 
665
 
        def start_watching(*args):
666
 
            """Fake stop watch."""
 
751
        def fake_start_watching(*args):
 
752
            """Fake start watch."""
667
753
            self.was_called = True
668
754
 
669
 
        Watch.start_watching = start_watching
 
755
        self.patch(Watch, "start_watching", fake_start_watching)
670
756
        self.manager._wdm = {}
671
757
 
672
758
        mask = 'bit_mask'
706
792
        """A watch on an unwatched path returns None."""
707
793
        self.assertEqual(None, self.manager.get_wd(self.parent_path))
708
794
 
 
795
    @defer.inlineCallbacks
709
796
    def test_rm_present_wd(self):
710
797
        """Test the removal of a present watch."""
711
 
        self.manager.rm_watch(1)
 
798
 
 
799
        def fake_stop_watching():
 
800
            """Fake stop watch."""
 
801
            return defer.succeed(True)
 
802
 
 
803
        self.patch(self.watch, "stop_watching", fake_stop_watching)
 
804
        yield self.manager.rm_watch(1)
712
805
        self.assertEqual(None, self.manager._wdm.get(1))
713
806
 
714
807
    def test_rm_root_path(self):
749
842
 
750
843
class TestWatchManagerAddWatches(BaseTwistedTestCase):
751
844
    """Test the watch manager."""
 
845
    timeout = 5
752
846
 
753
847
    def test_add_watch_twice(self):
754
848
        """Adding a watch twice succeeds when the watch is running."""
755
849
        self.patch(Watch, "start_watching", lambda self: self.started)
756
850
        manager = WatchManager(None)
757
 
        self.addCleanup(manager.stop)
 
851
        # no need to stop watching because start_watching is fake
758
852
 
759
853
        path = u'\\\\?\\C:\\test'  # a valid windows path
760
854
        mask = 'fake bit mask'
1201
1295
 
1202
1296
class FilesystemMonitorTestCase(BaseTwistedTestCase):
1203
1297
    """Tests for the FilesystemMonitor."""
 
1298
    timeout = 5
1204
1299
 
1205
1300
    def test_add_watch_twice(self):
1206
1301
        """Check the deferred returned by a second add_watch."""
1207
1302
        self.patch(Watch, "start_watching", lambda self: self.started)
1208
1303
        monitor = FilesystemMonitor(None, None)
1209
 
        self.addCleanup(monitor.shutdown)
 
1304
        # no need to stop watching because start_watching is fake
1210
1305
 
1211
1306
        parent_path = 'C:\\test'  # a valid windows path in utf-8 bytes
1212
1307
        child_path = parent_path + "\\child"