Atlas 0.7.0
Networking protocol for the Worldforge system.
test_codecs.py
1#test various codecs
2
3#Copyright 2000, 2001 by Aloril
4#Copyright 2002 by AIR-IX SUUNNITTELU/Ahiplan Oy
5
6#This library is free software; you can redistribute it and/or
7#modify it under the terms of the GNU Lesser General Public
8#License as published by the Free Software Foundation; either
9#version 2.1 of the License, or (at your option) any later version.
10
11#This library is distributed in the hope that it will be useful,
12#but WITHOUT ANY WARRANTY; without even the implied warranty of
13#MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14#Lesser General Public License for more details.
15
16#You should have received a copy of the GNU Lesser General Public
17#License along with this library; if not, write to the Free Software
18#Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19
20
21import test_objects
22import importlib
23importlib.reload(test_objects)
24from test_objects import *
25
26import pdb
27#pdb.set_trace()
28
29from atlas.codecs import get_codec
30
31from atlas.transport.negotiation import NegotiationClient
32from atlas.transport.file import read_file
33
34for codec_id, file_extension in (("Bach_beta2", "bach"),
35 ("XML", "xml"),
36 ("Packed", "packed"),
37 ):
38 try:
39
41 co = get_codec(codec_id)
42 def test_encode_decode(obj, e = co.encode, d = co.decode):
43 #print repr(obj)
44 #s = e(obj)
45 #pdb.set_trace()
46 #print d("<map/>")
47 #print d("<map/>")
48 #print d(s)
49 #assert( e(obj) == e(d(e(obj))) )
50 assert(co.decoder.eos())
51
55 assert( d(e(obj)) == d(e(d(e(obj)))) )
56 assert( d(e(d(e(obj)))) == d(e(d(e(d(e(obj)))))) )
57 s = e(obj)
58 obj2 = None
59 for ch in s:
60 obj2 = d(ch)
61 if obj2!=None: break
62 assert(obj2!=None)
63 assert(d(e(obj2))==d(s))
64
65 for i in range(0,len(s),5):
66 obj2 = d(s[i:i+5])
67 if obj2!=None: break
68 assert(obj2!=None)
69 assert(d(e(obj2))==d(s))
70
71 test_encode_decode([])
72 test_encode_decode({})
73 test_encode_decode([{}])
74 test_encode_decode({"a":[]})
75 for obj in msg:
76 test_encode_decode(obj)
77 test_encode_decode(msg)
78 test_encode_decode(lst_simple)
79
80
82 co.set_stream_mode()
83
84 s1 = co.encode(Object(i=8, a=[5, [], {}], j=Object(o=3.4))) + co.encode(Object(say="Hello!")) + co.encoder.close()
85 s2 = co.encode(Object(say="Hello!")) + co.encoder.close()
86 s3 = co.encoder.close()
87 res1 = '[\012\011{\012\011\011a: [\012\011\011\0115,\012\011\011\011[],\012\011\011\011{}\012\011\011],\012\011\011i: 8,\012\011\011j: {\012\011\011\011o: 3.4\012\011\011}\012\011},\012\011{\012\011\011say: "Hello!"\012\011}\012]\012'
88 res2 = '[\012\011{\012\011\011say: "Hello!"\012\011}\012]\012'
89 res3 = '[]\012'
90 #print s1
91 #print repr(co.decode(s1))
92 #print str(co.decode(s1))
93 #print repr(str(co.decode(s1)))
94 def test_stream_decode(s, res):
95 assert(co.decoder.eos())
96 assert(str(co.decode(s)) == res)
97 lst = Messages()
98 for i in range(0,len(s),5):
99 res_lst = co.decode(s[i:i+5])
100 if res_lst:
101 obj = res_lst.pop(0)
102 assert(not res_lst) #if 2 test messages are some day codec inside 5 bytes: will fail falsely
103 lst.append(obj)
104 assert( str(lst)==res )
105
106 test_stream_decode(s1, res1)
107 test_stream_decode(s2, res2)
108 test_stream_decode(s3, res3)
109
110
111
113 co.set_stream_mode()
114 file_name = "test." + file_extension
115 file_content = open(file_name).read()
116 neg1 = NegotiationClient([codec_id])
117 assert(neg1(file_content)=="found")
118 str1 = co.encode(msg)+co.encoder.close()
119 if neg1.str != str1:
120 print(codec_id + " encoding not same!")
121
122 msg2 = co.decode(str1)
123 file_content = open(file_name).read()
124 neg2 = NegotiationClient([codec_id])
125 assert(neg2(file_content)=="found")
126 str2 = co.encode(msg2)+co.encoder.close()
127 if neg2.str != str2:
128 print("%(codec_id)s decoding and %(codec_id)s encoding not same!" % locals())
129
130
131 assert(co.decoder.eos())
132
133
135 objects = read_file(file_name)
136 assert(str(objects)==str(msg))
137 except:
138 print("Exception testing codec", codec_id)
139 raise