253
250
assertEqual = assertEquals = failUnlessEquals = failUnlessEqual
256
class IgnoreFileTests(unittest.TestCase):
257
"""Tests the ignore files behaviour."""
259
def test_filter_none(self):
260
"""Still works ok even if not receiving a regex to ignore."""
261
p = _GeneralINotifyProcessor(None)
262
self.assertFalse(p.is_ignored("froo.pyc"))
264
def test_filter_one(self):
265
"""Filters stuff that matches (or not) this one regex."""
266
p = _GeneralINotifyProcessor(None, ['\A.*\\.pyc\Z'])
267
self.assertTrue(p.is_ignored("froo.pyc"))
268
self.assertFalse(p.is_ignored("froo.pyc.real"))
269
self.assertFalse(p.is_ignored("otherstuff"))
271
def test_filter_two_simple(self):
272
"""Filters stuff that matches (or not) these simple regexes."""
273
p = _GeneralINotifyProcessor(None, ['\A.*foo\Z', '\A.*bar\Z'])
274
self.assertTrue(p.is_ignored("blah_foo"))
275
self.assertTrue(p.is_ignored("blah_bar"))
276
self.assertFalse(p.is_ignored("bar_xxx"))
277
self.assertFalse(p.is_ignored("--foo--"))
278
self.assertFalse(p.is_ignored("otherstuff"))
280
def test_filter_two_complex(self):
281
"""Filters stuff that matches (or not) these complex regexes."""
282
p = _GeneralINotifyProcessor(None,
283
['\A.*foo\Z|\Afoo.*\Z', '\A.*bar\Z'])
284
self.assertTrue(p.is_ignored("blah_foo"))
285
self.assertTrue(p.is_ignored("blah_bar"))
286
self.assertTrue(p.is_ignored("foo_xxx"))
287
self.assertFalse(p.is_ignored("--foo--"))
288
self.assertFalse(p.is_ignored("otherstuff"))
291
253
class WatchTests(BaseFSMonitorTestCase):
292
254
"""Test the EQ API to add and remove watchs."""
439
401
self.assertFalse(self.monitor.has_watch(path_ancestor))
442
class MutedSignalsTests(BaseTwisted):
443
"""Test that EQ filter some signals on demand."""
445
def check_filter(self, _=None):
446
self.assertFalse(self.monitor._processor._to_mute._cnt)
449
def test_mute_and_remove(self):
450
"""Test add and remove the mute."""
452
self.monitor.add_to_mute_filter('FS_FILE_OPEN', path='somepath')
453
self.assertEqual(self.monitor._processor._to_mute._cnt,
454
{'FS_FILE_OPEN': [{'path': 'somepath'}]})
455
self.monitor.add_to_mute_filter('FS_FILE_OPEN', path='somepath')
456
self.assertEqual(self.monitor._processor._to_mute._cnt,
457
{'FS_FILE_OPEN': [{'path': 'somepath'},
458
{'path': 'somepath'}]})
459
self.monitor.add_to_mute_filter('FS_FILE_OPEN', path='otherpath')
460
self.assertEqual(self.monitor._processor._to_mute._cnt,
461
{'FS_FILE_OPEN': [{'path': 'somepath'},
462
{'path': 'somepath'},
463
{'path': 'otherpath'}]})
466
self.monitor.rm_from_mute_filter('FS_FILE_OPEN', path='somepath')
467
self.assertEqual(self.monitor._processor._to_mute._cnt,
468
{'FS_FILE_OPEN': [{'path': 'somepath'},
469
{'path': 'otherpath'}]})
470
self.monitor.rm_from_mute_filter('FS_FILE_OPEN', path='otherpath')
471
self.assertEqual(self.monitor._processor._to_mute._cnt,
472
{'FS_FILE_OPEN': [{'path': 'somepath'}]})
473
self.monitor.rm_from_mute_filter('FS_FILE_OPEN', path='somepath')
474
self.assertEqual(self.monitor._processor._to_mute._cnt, {})
476
def _how_many_muted(self):
477
"""Return how many events are muted."""
478
mute_filter = self.monitor._processor._to_mute
479
return sum(len(x) for x in mute_filter._cnt.values())
481
def test_file_open(self):
482
"""Test receiving the open signal on files."""
483
testfile = os.path.join(self.root_dir, "foo")
484
open(testfile, "w").close()
485
self.monitor.add_to_mute_filter("FS_FILE_OPEN", path=testfile)
486
self.monitor.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", path=testfile)
487
self.assertEqual(self._how_many_muted(), 2)
488
self.monitor.add_watch(self.root_dir)
492
reactor.callLater(.1, self.check_filter)
493
return self._deferred
495
def test_file_close_nowrite(self):
496
"""Test receiving the close_nowrite signal on files."""
497
testfile = os.path.join(self.root_dir, "foo")
498
open(testfile, "w").close()
500
self.monitor.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", path=testfile)
501
self.assertEqual(self._how_many_muted(), 1)
502
self.monitor.add_watch(self.root_dir)
506
reactor.callLater(.1, self.check_filter)
507
return self._deferred
509
def test_file_create_close_write(self):
510
"""Test receiving the create and close_write signals on files."""
511
testfile = os.path.join(self.root_dir, "foo")
512
self.monitor.add_to_mute_filter("FS_FILE_CREATE", path=testfile)
513
self.monitor.add_to_mute_filter("FS_FILE_OPEN", path=testfile)
514
self.monitor.add_to_mute_filter("FS_FILE_CLOSE_WRITE", path=testfile)
515
self.assertEqual(self._how_many_muted(), 3)
516
self.monitor.add_watch(self.root_dir)
519
open(testfile, "w").close()
520
reactor.callLater(.1, self.check_filter)
521
return self._deferred
523
def test_dir_create(self):
524
"""Test receiving the create signal on dirs."""
525
testdir = os.path.join(self.root_dir, "foo")
526
self.monitor.add_to_mute_filter("FS_DIR_CREATE", path=testdir)
527
self.assertEqual(self._how_many_muted(), 1)
528
self.monitor.add_watch(self.root_dir)
532
reactor.callLater(.1, self.check_filter)
533
return self._deferred
535
def test_file_delete(self):
536
"""Test the delete signal on a file."""
537
testfile = os.path.join(self.root_dir, "foo")
538
open(testfile, "w").close()
539
self.monitor.add_to_mute_filter("FS_FILE_DELETE", path=testfile)
540
self.assertEqual(self._how_many_muted(), 1)
541
self.monitor.add_watch(self.root_dir)
545
reactor.callLater(.1, self.check_filter)
546
return self._deferred
548
def test_dir_delete(self):
549
"""Test the delete signal on a dir."""
550
testdir = os.path.join(self.root_dir, "foo")
552
self.monitor.add_to_mute_filter("FS_DIR_DELETE", path=testdir)
553
self.assertEqual(self._how_many_muted(), 1)
554
self.monitor.add_watch(self.root_dir)
558
reactor.callLater(.1, self.check_filter)
559
return self._deferred
561
def test_file_moved_inside(self):
562
"""Test the synthesis of the FILE_MOVE event."""
563
fromfile = os.path.join(self.root_dir, "foo")
564
self.fs.create(fromfile, "")
565
self.fs.set_node_id(fromfile, "from_node_id")
566
tofile = os.path.join(self.root_dir, "bar")
567
self.fs.create(tofile, "")
568
self.fs.set_node_id(tofile, "to_node_id")
569
open(fromfile, "w").close()
570
self.monitor.add_to_mute_filter("FS_FILE_MOVE",
571
path_from=fromfile, path_to=tofile)
572
self.assertEqual(self._how_many_muted(), 1)
573
self.monitor.add_watch(self.root_dir)
576
os.rename(fromfile, tofile)
577
reactor.callLater(.1, self.check_filter)
578
return self._deferred
580
def test_dir_moved_inside(self):
581
"""Test the synthesis of the DIR_MOVE event."""
582
fromdir = os.path.join(self.root_dir, "foo")
583
self.fs.create(fromdir, "")
584
self.fs.set_node_id(fromdir, "from_node_id")
585
todir = os.path.join(self.root_dir, "bar")
586
self.fs.create(todir, "")
587
self.fs.set_node_id(todir, "to_node_id")
589
self.monitor.add_to_mute_filter("FS_DIR_MOVE",
590
path_from=fromdir, path_to=todir)
591
self.assertEqual(self._how_many_muted(), 1)
592
self.monitor.add_watch(self.root_dir)
595
os.rename(fromdir, todir)
596
reactor.callLater(.1, self.check_filter)
597
return self._deferred
599
def test_file_moved_from_conflict(self):
600
"""Test the handling of the FILE_MOVE event when source is conflict."""
601
fromfile = os.path.join(self.root_dir, "foo.u1conflict")
602
self.fs.create(fromfile, "")
603
self.fs.set_node_id(fromfile, "from_node_id")
604
tofile = os.path.join(self.root_dir, "foo")
605
self.fs.create(tofile, "")
606
self.fs.set_node_id(tofile, "to_node_id")
607
open(fromfile, "w").close()
608
self.monitor.add_to_mute_filter("FS_FILE_MOVE",
609
path_from=fromfile, path_to=tofile)
610
self.assertEqual(self._how_many_muted(), 2)
611
self.monitor.add_watch(self.root_dir)
614
os.rename(fromfile, tofile)
615
reactor.callLater(.1, self.check_filter)
616
return self._deferred
618
def test_file_moved_from_partial(self):
619
"""Test the handling of the FILE_MOVE event when source is partial."""
620
fromfile = os.path.join(self.root_dir, "mdid.u1partial.foo")
621
root_dir = os.path.join(self.root_dir, "my_files")
622
tofile = os.path.join(root_dir, "foo")
624
open(fromfile, "w").close()
625
self.monitor.add_to_mute_filter("FS_FILE_CREATE", path=tofile)
626
self.monitor.add_to_mute_filter("FS_FILE_CLOSE_WRITE", path=tofile)
627
self.assertEqual(self._how_many_muted(), 2)
628
self.monitor.add_watch(root_dir)
631
os.rename(fromfile, tofile)
632
reactor.callLater(.1, self.check_filter)
633
return self._deferred
636
404
def test_suite():
637
405
"""Collect the tests."""
added
removed
tests/platform/linux/test_filesystem_notifications.py
