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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
#!/usr/bin/env python
#
# Test script for producers.py
#
__revision__ = "$Id: test_producers.py,v 1.2 2002/09/18 20:16:40 akuchling Exp $"
import StringIO, zlib
from sancho.unittest import TestScenario, parse_args, run_scenarios
tested_modules = ["medusa.producers"]
from medusa import producers
test_string = ''
for i in range(16385):
test_string += chr(48 + (i%10))
class ProducerTest (TestScenario):
def setup (self):
pass
def shutdown (self):
pass
def _check_all (self, p, expected_string):
# Check that a producer returns all of the string,
# and that it's the unchanged string.
count = 0
data = ""
while 1:
s = p.more()
if s == "":
break
count += len(s)
data += s
self.test_val('count', len(expected_string))
self.test_val('data', expected_string)
self.test_val('p.more()', '')
return data
def check_simple (self):
p = producers.simple_producer(test_string)
self.test_val('p.more()', test_string[:1024])
p = producers.simple_producer(test_string, buffer_size = 5)
self._check_all(p, test_string)
def check_scanning (self):
p = producers.scanning_producer(test_string)
self.test_val('p.more()', test_string[:1024])
p = producers.scanning_producer(test_string, buffer_size = 5)
self._check_all(p, test_string)
def check_lines (self):
p = producers.lines_producer(['a']* 65)
self._check_all(p, 'a\r\n'*65)
def check_buffer (self):
p = producers.buffer_list_producer(['a']* 1027)
self._check_all(p, 'a'*1027)
def check_file (self):
f = StringIO.StringIO(test_string)
p = producers.file_producer(f)
self._check_all(p, test_string)
def check_output (self):
p = producers.output_producer()
for i in range(0,66):
p.write('a')
for i in range(0,65):
p.write('b\n')
self._check_all(p, 'a'*66 + 'b\r\n'*65)
def check_composite (self):
p1 = producers.simple_producer('a'*66, buffer_size = 5)
p2 = producers.lines_producer(['b']*65)
p = producers.composite_producer([p1, p2])
self._check_all(p, 'a'*66 + 'b\r\n'*65)
def check_glob (self):
p1 = producers.simple_producer(test_string, buffer_size = 5)
p = producers.globbing_producer(p1, buffer_size = 1024)
self.test_true('1024 <= len(p.more())')
def check_hooked (self):
def f (num_bytes):
self.test_val('num_bytes', len(test_string))
p1 = producers.simple_producer(test_string, buffer_size = 5)
p = producers.hooked_producer(p1, f)
self._check_all(p, test_string)
def check_chunked (self):
p1 = producers.simple_producer('the quick brown fox', buffer_size = 5)
p = producers.chunked_producer(p1, footers=['FOOTER'])
self._check_all(p, """5\r
the q\r
5\r
uick \r
5\r
brown\r
4\r
fox\r
0\r
FOOTER\r
\r\n""")
def check_compressed (self):
p1 = producers.simple_producer(test_string, buffer_size = 5)
p = producers.compressed_producer(p1)
compr_data = self._check_all(p, zlib.compress(test_string, 5))
self.test_val('zlib.decompress(compr_data)', test_string)
def check_escaping (self):
p1 = producers.simple_producer('the quick brown fox', buffer_size = 5)
p = producers.escaping_producer(p1,
esc_from = ' ',
esc_to = '_')
self._check_all(p, 'the_quick_brown_fox')
# class ProducerTest
if __name__ == "__main__":
(scenarios, options) = parse_args()
run_scenarios(scenarios, options)
|
