1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
#!/usr/bin/env python
# sugar-lint: disable
import os
import shutil
from __init__ import tests
from sugar_network.toolkit import coroutine, mountpoints
class MountpointsTest(tests.Test):
def setUp(self):
tests.Test.setUp(self)
mountpoints._COMPLETE_MOUNT_TIMEOUT = 0.01
def test_Populate(self):
self.touch('mnt/1/foo')
self.touch('mnt/2/foo')
self.touch('mnt/2/bar')
self.touch('mnt/3/fake')
os.makedirs('mnt/4')
found = []
mountpoints.connect('foo', found.append, None)
mountpoints.connect('bar', found.append, None)
mountpoints.populate('mnt')
self.assertEqual(
sorted(['mnt/1', 'mnt/2', 'mnt/2']),
sorted(found))
def test_Found(self):
os.makedirs('mnt')
coroutine.spawn(mountpoints.monitor, 'mnt')
found = []
mountpoints.connect('foo', found.append, None)
mountpoints.connect('bar', found.append, None)
coroutine.dispatch()
self.touch('mnt/1/foo')
self.touch('mnt/2/foo')
self.touch('mnt/2/bar')
self.touch('mnt/3/fake')
os.makedirs('mnt/4')
coroutine.sleep(.5)
self.assertEqual(
sorted(['mnt/1', 'mnt/2', 'mnt/2']),
sorted(found))
def test_Lost(self):
os.makedirs('mnt')
coroutine.spawn(mountpoints.monitor, 'mnt')
found = []
lost = []
mountpoints.connect('foo', found.append, lost.append)
mountpoints.connect('bar', found.append, lost.append)
coroutine.dispatch()
self.touch('mnt/1/foo')
self.touch('mnt/2/foo')
self.touch('mnt/2/bar')
self.touch('mnt/3/fake')
os.makedirs('mnt/4')
coroutine.sleep(.1)
shutil.rmtree('mnt/1')
shutil.rmtree('mnt/2')
shutil.rmtree('mnt/3')
shutil.rmtree('mnt/4')
coroutine.sleep(.1)
self.assertEqual(
sorted(['mnt/1', 'mnt/2', 'mnt/2']),
sorted(found))
self.assertEqual(
sorted(['mnt/1', 'mnt/2', 'mnt/2']),
sorted(lost))
def test_FoundTimeout(self):
mountpoints._COMPLETE_MOUNT_TIMEOUT = 2
os.makedirs('mnt')
coroutine.spawn(mountpoints.monitor, 'mnt')
found = []
mountpoints.connect('probe', found.append, None)
coroutine.dispatch()
self.touch('mnt/1/probe')
coroutine.sleep(1)
self.assertEqual([], found)
coroutine.sleep(1.5)
self.assertEqual(['mnt/1'], found)
if __name__ == '__main__':
tests.main()
|
