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
101
102
103
104
105
|
#!/usr/bin/env python
# sugar-lint: disable
import copy
from os.path import exists
from cStringIO import StringIO
from __init__ import tests
from sugar_network import toolkit
from sugar_network.toolkit import Seqno
class ToolkitTest(tests.Test):
def test_Seqno_commit(self):
seqno = Seqno(tests.tmpdir + '/seqno')
seqno.next()
seqno.commit()
seqno.next()
seqno = Seqno(tests.tmpdir + '/seqno')
self.assertEqual(1, seqno.value)
def test_readline(self):
def readlines(string):
result = []
stream = StringIO(string)
while True:
line = toolkit.readline(stream)
if not line:
break
result.append(line)
return result
self.assertEqual([], readlines(''))
self.assertEqual([' '], readlines(' '))
self.assertEqual([' a '], readlines(' a '))
self.assertEqual(['\n'], readlines('\n'))
self.assertEqual(['\n', 'b'], readlines('\nb'))
self.assertEqual([' \n', ' b \n'], readlines(' \n b \n'))
def test_Pool(self):
stack = toolkit.Pool()
stack.add('a')
stack.add('b')
stack.add('c')
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('a'))
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('b'))
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('c'))
self.assertEqual(
[('c', toolkit.Pool.ACTIVE), ('b', toolkit.Pool.ACTIVE), ('a', toolkit.Pool.ACTIVE)],
[(i, stack.get_state(i)) for i in stack])
self.assertEqual(
[],
[i for i in stack])
self.assertEqual(toolkit.Pool.PASSED, stack.get_state('a'))
self.assertEqual(toolkit.Pool.PASSED, stack.get_state('b'))
self.assertEqual(toolkit.Pool.PASSED, stack.get_state('c'))
stack.rewind()
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('a'))
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('b'))
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('c'))
self.assertEqual(
['c', 'b', 'a'],
[i for i in stack])
stack.add('c')
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('c'))
self.assertEqual(
[('c', toolkit.Pool.ACTIVE)],
[(i, stack.get_state(i)) for i in stack])
self.assertEqual(toolkit.Pool.PASSED, stack.get_state('c'))
stack.add('b')
stack.add('a')
self.assertEqual(
['a', 'b'],
[i for i in stack])
stack.rewind()
self.assertEqual(
['a', 'b', 'c'],
[i for i in stack])
stack.add('d')
self.assertEqual(toolkit.Pool.QUEUED, stack.get_state('d'))
self.assertEqual(
[('d', toolkit.Pool.ACTIVE)],
[(i, stack.get_state(i)) for i in stack])
self.assertEqual(toolkit.Pool.PASSED, stack.get_state('d'))
stack.rewind()
self.assertEqual(
['d', 'a', 'b', 'c'],
[i for i in stack])
if __name__ == '__main__':
tests.main()
|
