Lib/xdrlib.py

   1 """Implements (a subset of) Sun XDR -- eXternal Data Representation.
   2
   3 See: RFC 1014
   4
   5 """
   6
   7 import struct
   8
   9 # exceptions
  10 class Error:
  11     """Exception class for this module. Use:
  12
  13     except xdrlib.Error, var:
  14         # var has the Error instance for the exception
  15
  16     Public ivars:
  17         msg -- contains the message
  18
  19     """
  20     def __init__(self, msg):
  21         self.msg = msg
  22     def __repr__(self):
  23         return repr(self.msg)
  24     def __str__(self):
  25         return str(self.msg)
  26
  27
  28 class ConversionError(Error):
  29     pass
  30
  31
  32 \f
  33 class Packer:
  34     """Pack various data representations into a buffer."""
  35
  36     def __init__(self):
  37         self.reset()
  38
  39     def reset(self):
  40         self.__buf = ''
  41
  42     def get_buffer(self):
  43         return self.__buf
  44     # backwards compatibility
  45     get_buf = get_buffer
  46
  47     def pack_uint(self, x):
  48         self.__buf = self.__buf + struct.pack('>L', x)
  49
  50     pack_int = pack_uint
  51     pack_enum = pack_int
  52
  53     def pack_bool(self, x):
  54         if x: self.__buf = self.__buf + '\0\0\0\1'
  55         else: self.__buf = self.__buf + '\0\0\0\0'
  56
  57     def pack_uhyper(self, x):
  58         self.pack_uint(x>>32 & 0xffffffffL)
  59         self.pack_uint(x & 0xffffffffL)
  60
  61     pack_hyper = pack_uhyper
  62
  63     def pack_float(self, x):
  64         try: self.__buf = self.__buf + struct.pack('>f', x)
  65         except struct.error, msg:
  66             raise ConversionError, msg
  67
  68     def pack_double(self, x):
  69         try: self.__buf = self.__buf + struct.pack('>d', x)
  70         except struct.error, msg:
  71             raise ConversionError, msg
  72
  73     def pack_fstring(self, n, s):
  74         if n < 0:
  75             raise ValueError, 'fstring size must be nonnegative'
  76         n = ((n+3)/4)*4
  77         data = s[:n]
  78         data = data + (n - len(data)) * '\0'
  79         self.__buf = self.__buf + data
  80
  81     pack_fopaque = pack_fstring
  82
  83     def pack_string(self, s):
  84         n = len(s)
  85         self.pack_uint(n)
  86         self.pack_fstring(n, s)
  87
  88     pack_opaque = pack_string
  89     pack_bytes = pack_string
  90
  91     def pack_list(self, list, pack_item):
  92         for item in list:
  93             self.pack_uint(1)
  94             pack_item(item)
  95         self.pack_uint(0)
  96
  97     def pack_farray(self, n, list, pack_item):
  98         if len(list) <> n:
  99             raise ValueError, 'wrong array size'
 100         for item in list:
 101             pack_item(item)
 102
 103     def pack_array(self, list, pack_item):
 104         n = len(list)
 105         self.pack_uint(n)
 106         self.pack_farray(n, list, pack_item)
 107
 108
 109 \f
 110 class Unpacker:
 111     """Unpacks various data representations from the given buffer."""
 112
 113     def __init__(self, data):
 114         self.reset(data)
 115
 116     def reset(self, data):
 117         self.__buf = data
 118         self.__pos = 0
 119
 120     def get_position(self):
 121         return self.__pos
 122
 123     def set_position(self, position):
 124         self.__pos = position
 125
 126     def get_buffer(self):
 127         return self.__buf
 128
 129     def done(self):
 130         if self.__pos < len(self.__buf):
 131             raise Error('unextracted data remains')
 132
 133     def unpack_uint(self):
 134         i = self.__pos
 135         self.__pos = j = i+4
 136         data = self.__buf[i:j]
 137         if len(data) < 4:
 138             raise EOFError
 139         x = struct.unpack('>L', data)[0]
 140         try:
 141             return int(x)
 142         except OverflowError:
 143             return x
 144
 145     def unpack_int(self):
 146         i = self.__pos
 147         self.__pos = j = i+4
 148         data = self.__buf[i:j]
 149         if len(data) < 4:
 150             raise EOFError
 151         return struct.unpack('>l', data)[0]
 152
 153     unpack_enum = unpack_int
 154     unpack_bool = unpack_int
 155
 156     def unpack_uhyper(self):
 157         hi = self.unpack_uint()
 158         lo = self.unpack_uint()
 159         return long(hi)<<32 | lo
 160
 161     def unpack_hyper(self):
 162         x = self.unpack_uhyper()
 163         if x >= 0x8000000000000000L:
 164             x = x - 0x10000000000000000L
 165         return x
 166
 167     def unpack_float(self):
 168         i = self.__pos
 169         self.__pos = j = i+4
 170         data = self.__buf[i:j]
 171         if len(data) < 4:
 172             raise EOFError
 173         return struct.unpack('>f', data)[0]
 174
 175     def unpack_double(self):
 176         i = self.__pos
 177         self.__pos = j = i+8
 178         data = self.__buf[i:j]
 179         if len(data) < 8:
 180             raise EOFError
 181         return struct.unpack('>d', data)[0]
 182
 183     def unpack_fstring(self, n):
 184         if n < 0:
 185             raise ValueError, 'fstring size must be nonnegative'
 186         i = self.__pos
 187         j = i + (n+3)/4*4
 188         if j > len(self.__buf):
 189             raise EOFError
 190         self.__pos = j
 191         return self.__buf[i:i+n]
 192
 193     unpack_fopaque = unpack_fstring
 194
 195     def unpack_string(self):
 196         n = self.unpack_uint()
 197         return self.unpack_fstring(n)
 198
 199     unpack_opaque = unpack_string
 200     unpack_bytes = unpack_string
 201
 202     def unpack_list(self, unpack_item):
 203         list = []
 204         while 1:
 205             x = self.unpack_uint()
 206             if x == 0: break
 207             if x <> 1:
 208                 raise ConversionError, '0 or 1 expected, got ' + `x`
 209             item = unpack_item()
 210             list.append(item)
 211         return list
 212
 213     def unpack_farray(self, n, unpack_item):
 214         list = []
 215         for i in range(n):
 216             list.append(unpack_item())
 217         return list
 218
 219     def unpack_array(self, unpack_item):
 220         n = self.unpack_uint()
 221         return self.unpack_farray(n, unpack_item)
 222
 223 \f
 224 # test suite
 225 def _test():
 226     p = Packer()
 227     packtest = [
 228         (p.pack_uint,    (9,)),
 229         (p.pack_bool,    (None,)),
 230         (p.pack_bool,    ('hello',)),
 231         (p.pack_uhyper,  (45L,)),
 232         (p.pack_float,   (1.9,)),
 233         (p.pack_double,  (1.9,)),
 234         (p.pack_string,  ('hello world',)),
 235         (p.pack_list,    (range(5), p.pack_uint)),
 236         (p.pack_array,   (['what', 'is', 'hapnin', 'doctor'], p.pack_string)),
 237         ]
 238     succeedlist = [1] * len(packtest)
 239     count = 0
 240     for method, args in packtest:
 241         print 'pack test', count,
 242         try:
 243             apply(method, args)
 244             print 'succeeded'
 245         except ConversionError, var:
 246             print 'ConversionError:', var.msg
 247             succeedlist[count] = 0
 248         count = count + 1
 249     data = p.get_buffer()
 250     # now verify
 251     up = Unpacker(data)
 252     unpacktest = [
 253         (up.unpack_uint,   (), lambda x: x == 9),
 254         (up.unpack_bool,   (), lambda x: not x),
 255         (up.unpack_bool,   (), lambda x: x),
 256         (up.unpack_uhyper, (), lambda x: x == 45L),
 257         (up.unpack_float,  (), lambda x: 1.89 < x < 1.91),
 258         (up.unpack_double, (), lambda x: 1.89 < x < 1.91),
 259         (up.unpack_string, (), lambda x: x == 'hello world'),
 260         (up.unpack_list,   (up.unpack_uint,), lambda x: x == range(5)),
 261         (up.unpack_array,  (up.unpack_string,),
 262          lambda x: x == ['what', 'is', 'hapnin', 'doctor']),
 263         ]
 264     count = 0
 265     for method, args, pred in unpacktest:
 266         print 'unpack test', count,
 267         try:
 268             if succeedlist[count]:
 269                 x = apply(method, args)
 270                 print pred(x) and 'succeeded' or 'failed', ':', x
 271             else:
 272                 print 'skipping'
 273         except ConversionError, var:
 274             print 'ConversionError:', var.msg
 275         count = count + 1
 276
 277 \f
 278 if __name__ == '__main__':
 279     _test()