-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathSync_test.py
More file actions
100 lines (69 loc) · 2.61 KB
/
Sync_test.py
File metadata and controls
100 lines (69 loc) · 2.61 KB
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
"""This module is part of Swampy, a suite of programs available from
allendowney.com/swampy.
Copyright 2011 Allen B. Downey
Distributed under the GNU General Public License at gnu.org/licenses/gpl.html.
"""
import unittest
import Sync
class Tests(unittest.TestCase):
def test_sync_mutex(self):
sync = Sync.Sync(['mutex.py'])
threads = sync.get_threads()
threadA = threads[0]
column = threadA.column
source = threadA.step()
self.assertEqual(source, 'mutex.wait()')
threadB = Sync.Thread(column)
source = threadB.step()
self.assertEqual(source, 'mutex.wait()')
self.assertFalse(threadA.queued)
self.assertTrue(threadB.queued)
source = threadB.step()
self.assertEqual(source, None)
source = threadA.step()
source = threadA.step()
source = threadA.step()
self.assertEqual(source, 'mutex.signal()')
self.assertFalse(threadA.queued)
self.assertFalse(threadB.queued)
source = threadA.exec_line('pid = pid()', sync)
self.assertEqual(sync.locals['pid'], 'A')
def test_sync_conditional(self):
sync = Sync.Sync(['sync_code/conditional.py'])
threads = sync.get_threads()
threadA = threads[0]
source = threadA.step()
self.assertEqual(source, 'if counter == 0:')
source = threadA.step()
self.assertEqual(source, ' print True')
source = threadA.step()
self.assertEqual(source, 'if counter == 1:')
source = threadA.step()
self.assertEqual(source, 'pass')
source = threadA.step()
source = threadA.step()
self.assertEqual(source, 'else:')
source = threadA.step()
self.assertEqual(source, ' print False')
source = threadA.step()
source = threadA.step()
self.assertEqual(source, ' print True')
source = threadA.step()
source = threadA.step()
self.assertEqual(source, 'pass')
def test_sync_while(self):
sync = Sync.Sync(['sync_code/while.py'])
threads = sync.get_threads()
threadA = threads[0]
source = threadA.step()
self.assertEqual(source, 'while counter == 1:')
source = threadA.step()
self.assertEqual(source, 'while counter < 1:')
source = threadA.step()
self.assertEqual(source, ' counter += 1')
source = threadA.step()
self.assertEqual(source, 'while counter < 1:')
source = threadA.step()
self.assertEqual(source, 'pass')
if __name__ == '__main__':
unittest.main()