aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_netstring.py
blob: 3c56acd9aca8f130e802d86b148d3a2ba3589b3f (plain)
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
from libmproxy import netstring
from cStringIO import StringIO
import libpry


class uNetstring(libpry.AutoTree):
    def setUp(self):
        self.test_data = "Netstring module by Will McGugan"
        self.encoded_data = "9:Netstring,6:module,2:by,4:Will,7:McGugan,"

    def test_header(self):
        tests = [ ("netstring", "9:"),
                  ("Will McGugan", "12:"),
                  ("", "0:") ]
        for test, result in tests:
            assert netstring.header(test) == result 
            
    def test_file_encoder(self):
        file_out = StringIO()
        data = self.test_data.split()
        encoder = netstring.FileEncoder(file_out)
        for s in data:
            encoder.write(s)
        encoded_data = file_out.getvalue()            
        assert encoded_data == self.encoded_data
        
    def test_decode_file(self):
        data = self.test_data.split()
        for buffer_size in range(1, len(self.encoded_data)):
            file_in = StringIO(self.encoded_data[:])            
            decoded_data = list(netstring.decode_file(file_in, buffer_size = buffer_size))
            assert decoded_data == data
        
    def test_decoder(self):
        encoded_data = self.encoded_data
        for step in range(1, len(encoded_data)):
            i = 0
            chunks = []
            while i < len(encoded_data):
                chunks.append(encoded_data[i:i+step])
                i += step                
            decoder = netstring.Decoder()
            decoded_data = [] 
            for chunk in chunks:
                for s in decoder.feed(chunk):
                    decoded_data.append(s)
            assert decoded_data == self.test_data.split()

    def test_errors(self):
        d = netstring.Decoder()
        libpry.raises("Illegal digit", list, d.feed("1:foo"))
        d = netstring.Decoder()
        libpry.raises("Preceding zero", list, d.feed("01:f"))
        d = netstring.Decoder(5)
        libpry.raises("Maximum size", list, d.feed("500:f"))
        d = netstring.Decoder()
        libpry.raises("Illegal digit", list, d.feed(":f"))




tests = [
    uNetstring()
]