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 
21 import test_objects
22 import importlib
23 importlib.reload(test_objects)
24 from test_objects import *
25 
26 import pdb
27 #pdb.set_trace()
28 
29 from atlas.codecs import get_codec
30 
31 from atlas.transport.negotiation import NegotiationClient
32 from atlas.transport.file import read_file
33 
34 for 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