Keep mounts monitor routines in separate module

2 files changed, 93 insertions, 0 deletions

diff --git a/tests/units/__main__.py b/tests/units/__main__.py
index 49fbd51..4624277 100644
--- a/tests/units/__main__.py
+++ b/tests/units/__main__.py
@@ -24,5 +24,6 @@ from datastore import *
 from dbus_datastore import *
 from files_sync import *
 from dbus_client import *
+from mounts_monitor import *
 
 tests.main()
diff --git a/tests/units/mounts_monitor.py b/tests/units/mounts_monitor.py
new file mode 100755
index 0000000..3ff5852
--- /dev/null
+++ b/tests/units/mounts_monitor.py
@@ -0,0 +1,92 @@
+#!/usr/bin/env python
+# sugar-lint: disable
+
+import os
+
+from __init__ import tests
+
+from sugar_network.toolkit import mounts_monitor
+from active_toolkit import coroutine
+
+
+class MountsMonitorTest(tests.Test):
+
+    def setUp(self):
+        tests.Test.setUp(self)
+        mounts_monitor._COMPLETE_MOUNT_TIMEOUT = 0
+
+    def test_Populate(self):
+        self.touch('mnt/foo')
+        self.touch('mnt/bar')
+        self.touch('mnt/fake')
+
+        mounts_monitor.start('mnt')
+
+        found = []
+        mounts_monitor.connect('foo', found.append, None)
+        mounts_monitor.connect('bar', found.append, None)
+        self.assertEqual(
+                ['mnt/foo', 'mnt/bar'],
+                found)
+
+    def test_Found(self):
+        os.makedirs('mnt')
+        mounts_monitor.start('mnt')
+
+        found = []
+        mounts_monitor.connect('foo', found.append, None)
+        mounts_monitor.connect('bar', found.append, None)
+
+        coroutine.dispatch()
+        self.touch('mnt/foo')
+        self.touch('mnt/bar')
+        self.touch('mnt/fake')
+        coroutine.sleep(.5)
+
+        self.assertEqual(
+                ['mnt/foo', 'mnt/bar'],
+                found)
+
+    def test_Lost(self):
+        os.makedirs('mnt')
+        mounts_monitor.start('mnt')
+
+        found = []
+        lost = []
+        mounts_monitor.connect('foo', found.append, lost.append)
+        mounts_monitor.connect('bar', found.append, lost.append)
+
+        coroutine.dispatch()
+        self.touch('mnt/foo')
+        self.touch('mnt/bar')
+        self.touch('mnt/fake')
+        os.unlink('mnt/foo')
+        os.unlink('mnt/bar')
+        os.unlink('mnt/fake')
+        coroutine.sleep(.5)
+
+        self.assertEqual(
+                ['mnt/foo', 'mnt/bar'],
+                found)
+        self.assertEqual(
+                ['mnt/foo', 'mnt/bar'],
+                lost)
+
+    def test_FoundTimeout(self):
+        mounts_monitor._COMPLETE_MOUNT_TIMEOUT = 2
+        os.makedirs('mnt')
+        mounts_monitor.start('mnt')
+
+        found = []
+        mounts_monitor.connect('probe', found.append, None)
+
+        coroutine.dispatch()
+        self.touch('mnt/probe')
+        coroutine.sleep(1)
+        self.assertEqual([], found)
+        coroutine.sleep(1.5)
+        self.assertEqual(['mnt/probe'], found)
+
+
+if __name__ == '__main__':
+    tests.main()