[svn-commits] mjordan: branch mjordan/iax2-driver r4376 - /asterisk/team/mjordan/iax2-drive...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Wed Nov 27 09:23:13 CST 2013
Author: mjordan
Date: Wed Nov 27 09:23:10 2013
New Revision: 4376
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=4376
Log:
Add some stuff
The biggest problem here was the approach that I took to packing the binary
data. This could get tweaked to use a better underlying Python library,
which would make it a lot easier. It's still not a bad starting point for
a sipp-esque thing for IAX2.
Added:
asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/
asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/__init__.py (with props)
asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py (with props)
asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py (with props)
asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py (with props)
Added: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/__init__.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/__init__.py?view=auto&rev=4376
==============================================================================
(empty)
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/__init__.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/__init__.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/__init__.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py?view=auto&rev=4376
==============================================================================
--- asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py (added)
+++ asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py Wed Nov 27 09:23:10 2013
@@ -1,0 +1,365 @@
+#!/usr/bin/env python
+''' Frame base classes
+
+Detailed description.
+
+Copyright (C) 2010-2012, Digium, Inc.
+Matt Jordan <mjordan at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+'''
+
+import unittest
+import struct
+import binascii
+
+from iax2_utils import enum
+
+Subclass = enum(INTEGER = 0, POWER_2 = 1)
+
+class ParameterOverflowException(Exception):
+ ''' An exception raised when a parameter is given a value that would
+ cause it to overflow the accepted bounds of its value when sent as an
+ IAX2 frame '''
+
+ def __init__(self, parameter, value):
+ ''' Constructor
+
+ Keyword Arguments:
+ parameter The parameter that whose bounds were exceeded
+ value The value passed in
+ '''
+
+ self.parameter = parameter
+ self.value = value
+
+
+ def __str__(self):
+ ''' Return the exception as a string '''
+ return ''.join([repr(self.parameter), ': ', repr(self.value)])
+
+
+class MiniFrameHeader(object):
+ ''' An IAX2 mini frame header '''
+
+ def __init__(self):
+ ''' Constructor '''
+ self._src_call_number = 0
+ self._timestamp = 0
+ self._data_packer = struct.Struct('!HH')
+
+
+ def get_timestamp(self):
+ ''' Return the timestamp '''
+ return self._timestamp
+
+
+ def set_timestamp(self, value):
+ ''' Set the timestamp '''
+ if (value > 65536):
+ raise ParameterOverflowException('timestamp', value)
+ self._timestamp = value
+
+
+ def get_src_call_number(self):
+ ''' Return the source call number '''
+ return self._src_call_number
+
+
+ def set_src_call_number(self, value):
+ ''' Set the source call number '''
+ if (value > 32768):
+ raise ParameterOverflowException('src_call_number', value)
+ self._src_call_number = value
+
+
+ def write(self):
+ ''' Return the frame as a byte array '''
+ msg = self._data_packer.pack(self._src_call_number, self._timestamp)
+ return msg
+
+
+ def read(self, buffer):
+ ''' Populate the frame from a byte array '''
+ (self._src_call_number,
+ self._timestamp) = self._data_packer.unpack_from(buffer)
+ self._src_call_number = self._src_call_number & 0x7FFF
+
+
+class FullFrameHeader(object):
+ ''' An IAX2 full frame header '''
+
+ def __init__(self):
+ ''' Constructor '''
+ self._src_call_number = 0
+ self._retransmitted = False
+ self._dest_call_number = 0
+ self._timestamp = 0
+ self._outbound_seq_num = 0
+ self._inbound_seq_num = 0
+ self._frame_type = 0
+ self._subclass_type = Subclass.INTEGER
+ self._subclass = 0
+ self._data_packer = struct.Struct('!HHIBBBB')
+
+
+ def write(self):
+ ''' Return the frame as a byte array '''
+ retrans = 0x8000 if self._retransmitted else 0x0000
+ subclass_type = 0x80 if self._subclass_type == Subclass.POWER_2 else 0x00
+ msg = self._data_packer.pack(
+ (self._src_call_number & 0x7FFF) | 0x8000,
+ (self._dest_call_number & 0x7FFF) | retrans,
+ self._timestamp,
+ self._outbound_seq_num,
+ self._inbound_seq_num,
+ self._frame_type,
+ (self._subclass & 0x7F) | subclass_type)
+ return msg
+
+
+ def read(self, buffer):
+ ''' Populate the frame from a byte array '''
+ if len(buffer) < 12:
+ raise ValueError('Buffer length cannot be less than 12')
+ (self._src_call_number,
+ self._dest_call_number,
+ self._timestamp,
+ self._outbound_seq_num,
+ self._inbound_seq_num,
+ self._frame_type,
+ self._subclass) = self._data_packer.unpack_from(buffer)
+ self._src_call_number = self._src_call_number & 0x7FFF
+ self._retransmitted = (self._dest_call_number & 0x8000 == 0x8000)
+ self._dest_call_number = self._dest_call_number & 0x7FFF
+ if self._subclass & 0x80 == 0x80:
+ self._subclass_type = Subclass.POWER_2
+ else:
+ self._subclass_type = Subclass.INTEGER
+ self._subclass = self._subclass & 0x7F
+
+
+ def set_frame_type(self, value):
+ ''' Set the frame type '''
+ if (value > 256):
+ raise ParameterOverflowException('frame_type', value)
+ self._frame_type = value
+
+
+ def get_frame_type(self):
+ ''' Return the frame type '''
+ return self._frame_type
+
+
+ def set_subclass(self, value):
+ ''' Set the subclass '''
+ if (value > 128):
+ raise ParameterOverflowException('subclass', value)
+ self._subclass = value
+
+
+ def get_subclass(self):
+ ''' Return the subclass '''
+ return self._subclass
+
+
+ def set_subclass_type(self, value):
+ ''' Set the subclass type '''
+ if ((value != Subclass.POWER_2) or (value != Subclass.INTEGER)):
+ raise ValueError('subclass_type must be of type Subclass')
+ self._subclass_type = value
+
+
+ def get_subclass_type(self):
+ ''' Return the subclass type '''
+ return self._subclass_type
+
+
+ def set_inbound_seq_num(self, value):
+ ''' Set the inbound sequence number '''
+ if (value > 256):
+ raise ParameterOverflowException('inbound_seq_num', value)
+ self._inbound_seq_num = value
+
+
+ def get_inbound_seq_num(self):
+ ''' Return the inbound sequence number '''
+ return self._inbound_seq_num
+
+
+ def set_outbound_seq_num(self, value):
+ ''' Set the outbound sequence number '''
+ if (value > 256):
+ raise ParameterOverflowException('outbound_seq_num', value)
+ self._outbound_seq_num = value
+
+
+ def get_outbound_seq_num(self):
+ ''' Return the outbound sequence number '''
+ return self._outbound_seq_num
+
+
+ def get_timestamp(self):
+ ''' Return the timestamp '''
+ return self._timestamp
+
+
+ def set_timestamp(self, value):
+ ''' Set the timestamp '''
+ if (value > 4294967296):
+ raise ParameterOverflowException('timestamp', value)
+ self._timestamp = value
+
+
+ def set_dest_call_number(self, value):
+ ''' Set the destination call number '''
+ if (value > 32768):
+ raise ParameterOverflowException('dest_call_number', value)
+ self._dest_call_number = value
+
+
+ def get_dest_call_number(self):
+ ''' Return the destination call number '''
+ return self._dest_call_number
+
+
+ def get_src_call_number(self):
+ ''' Return the source call number '''
+ return self._src_call_number
+
+
+ def set_src_call_number(self, value):
+ ''' Set the source call number '''
+ if (value > 32768):
+ raise ParameterOverflowException('src_call_number', value)
+ self._src_call_number = value
+
+
+ def get_retransmitted(self):
+ ''' Return whether or not this is a retransmitted frame '''
+ return self._retransmitted
+
+
+ def set_retransmitted(self, value):
+ ''' Set whether or not this is a retransmitted frame '''
+ self._retransmitted = value
+
+
+class MiniFrameHeaderTest(unittest.TestCase):
+ ''' MiniFrameHeader unit tests '''
+
+ def verify_frames(self, left, right):
+ self.assertEquals(left.get_timestamp(), right.get_timestamp())
+ self.assertEquals(left.get_src_call_number(),
+ right.get_src_call_number())
+
+ def test_constructor(self):
+ obj = MiniFrameHeader()
+ self.assertEquals(0, obj.get_timestamp())
+ self.assertEquals(0, obj.get_src_call_number())
+
+ def test_invalid_parameters(self):
+ obj = MiniFrameHeader()
+ self.assertRaises(ParameterOverflowException,
+ obj.set_timestamp,
+ 0x1FFFF)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_src_call_number,
+ 0xFFFF)
+
+ def test_mini_frame_read_write(self):
+ obj = MiniFrameHeader()
+ obj2 = MiniFrameHeader()
+ obj.set_timestamp(0xABAB)
+ obj.set_src_call_number(0x0A12)
+ buf = obj.write()
+ obj2.read(buf)
+ self.verify_frames(obj, obj2)
+
+
+class FullFrameHeaderTest(unittest.TestCase):
+ ''' FullFrameHeader unit tests '''
+
+ def verify_frames(self, left, right):
+ self.assertEquals(left.get_dest_call_number(),
+ right.get_dest_call_number())
+ self.assertEquals(left.get_frame_type(), right.get_frame_type())
+ self.assertEquals(left.get_inbound_seq_num(),
+ right.get_inbound_seq_num())
+ self.assertEquals(left.get_outbound_seq_num(),
+ right.get_outbound_seq_num())
+ self.assertEquals(left.get_retransmitted(), right.get_retransmitted())
+ self.assertEquals(left.get_src_call_number(),
+ right.get_src_call_number())
+ self.assertEquals(left.get_subclass(), right.get_subclass())
+ self.assertEquals(left.get_subclass_type(), right.get_subclass_type())
+ self.assertEquals(left.get_timestamp(), right.get_timestamp())
+
+ def test_constructor(self):
+ obj = FullFrameHeader()
+ self.assertEquals(Subclass.INTEGER, obj.get_subclass_type())
+ self.assertEquals(0, obj.get_inbound_seq_num())
+ self.assertEquals(0, obj.get_outbound_seq_num())
+ self.assertEquals(0, obj.get_timestamp())
+ self.assertEquals(0, obj.get_src_call_number())
+ self.assertEquals(0, obj.get_dest_call_number())
+ self.assertFalse(obj.get_retransmitted())
+ self.assertEquals(0, obj.get_frame_type())
+
+ def test_default_write(self):
+ obj = FullFrameHeader()
+ buf = obj.write()
+ self.assertEquals(12, len(buf))
+ self.assertEquals('800000000000000000000000', binascii.hexlify(buf))
+
+ def test_new_frame(self):
+ obj = FullFrameHeader()
+ obj2 = FullFrameHeader()
+ obj.set_src_call_number(9460)
+ obj.set_timestamp(18)
+ obj.set_frame_type(6)
+ obj.set_subclass(1)
+ buf = obj.write()
+ obj2.read(buf)
+ self.verify_frames(obj, obj2)
+
+ def test_answer_frame(self):
+ obj = FullFrameHeader()
+ obj2 = FullFrameHeader()
+ obj.set_src_call_number(6502)
+ obj.set_dest_call_number(9460)
+ obj.set_timestamp(22)
+ obj.set_outbound_seq_num(1)
+ obj.set_inbound_seq_num(1)
+ obj.set_frame_type(4)
+ obj.set_subclass(4)
+ buf = obj.write()
+ obj2.read(buf)
+ self.verify_frames(obj, obj2)
+
+ def test_invalid_parameters(self):
+ obj = FullFrameHeader()
+ self.assertRaises(ParameterOverflowException,
+ obj.set_src_call_number, 0xFFFF)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_dest_call_number, 0xFFFF)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_timestamp, 0xFFFFFFFFF)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_outbound_seq_num, 0xFFF)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_inbound_seq_num, 0xFFF)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_frame_type, 0xFFF)
+ self.assertRaises(ValueError,
+ obj.set_subclass_type, 3)
+ self.assertRaises(ParameterOverflowException,
+ obj.set_subclass, 0xFF)
+
+def main():
+ unittest.main()
+
+
+if __name__ == "__main__":
+ main()
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/frames.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py?view=auto&rev=4376
==============================================================================
--- asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py (added)
+++ asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py Wed Nov 27 09:23:10 2013
@@ -1,0 +1,14 @@
+#!/usr/bin/env python
+""" Utilities
+
+Detailed description.
+
+Copyright (C) 2010-2012, Digium, Inc.
+Matt Jordan <mjordan at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+def enum(**enums):
+ return type('Enum', (), enums)
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/iax2_utils.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py?view=auto&rev=4376
==============================================================================
--- asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py (added)
+++ asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py Wed Nov 27 09:23:10 2013
@@ -1,0 +1,383 @@
+#!/usr/bin/env python
+''' Frame base classes
+
+Detailed description.
+
+Copyright (C) 2010-2012, Digium, Inc.
+Matt Jordan <mjordan at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+'''
+
+import unittest
+import struct
+import binascii
+import array
+
+from iax2_utils import enum
+
+IEType = enum(
+ CALLED_NUMBER = 0x01,
+ CALLING_NUMBER = 0x02,
+ CALLING_ANI = 0x03,
+ CALLING_NAME = 0x04,
+ CALLED_CONTEXT = 0x05,
+ USERNAME = 0x06,
+ PASSWORD = 0x07,
+ CAPABILITY = 0x08,
+ FORMAT = 0x09,
+ LANGUAGE = 0x0a,
+ VERSION = 0x0b,
+ ADSICPE = 0x0c,
+ DNID = 0x0d,
+ AUTHMETHODS = 0x0e,
+ CHALLENGE = 0x0f,
+ MDF5_RESULT = 0x10,
+ RSA_RESULT = 0x11,
+ APPARENT_ADDR = 0x12,
+ REFRESH = 0x13,
+ DPSTATUS = 0x14,
+ CALLNO = 0x15,
+ CAUSE = 0x16,
+ IAX_UNKNOWN = 0x17,
+ MSGCOUNT = 0x18,
+ AUTOANSWER = 0x19,
+ MUSICONHOLD = 0x1a,
+ TRANSFERID = 0x1b,
+ RDNIS = 0x1c,
+ DATETIME = 0x1f,
+ CALLINGPRES = 0x26,
+ CALLINGTON = 0x27,
+ CALLINGTNS = 0x28,
+ SAMPLINGRATE = 0x29,
+ CAUSECODE = 0x2a,
+ ENCRYPTION = 0x2b,
+ ENCKEY = 0x2c,
+ CODEC_PREFS = 0x2d,
+ RR_JITTER = 0x2e,
+ RR_LOSS = 0x2f,
+ RR_PKTS = 0x30,
+ RR_DELAY = 0x31,
+ RR_DROPPED = 0x32,
+ RR_OOO = 0x33,
+ OSPTOKEN = 0x34)
+
+IAX2AuthMethod = enum(RESERVED = 0x0001, MD5 = 0x0002, RSA = 0x0004)
+
+class IE(object):
+ ''' A base class for an information element '''
+
+ _data_packer = struct.Struct('!BB')
+
+ def __init__(self, ie, data_len=0):
+ ''' Constructor '''
+ self._ie = ie
+ self._data_length = data_len
+
+
+ def get_ie(self):
+ ''' This IE's identifier '''
+ return self._ie
+
+
+ def get_data_length(self):
+ ''' The data length of the IE '''
+ return self._data_length
+
+
+ def write(self):
+ ''' Write the IE to a buffer '''
+ msg = self._data_packer.pack(self._ie, self._data_length)
+ msg = self.write_data(msg, 2)
+ return msg
+
+
+ def write_data(self, buffer, offset):
+ ''' Write the inherited class's data into the buffer. Derived classes
+ should write the value into the buffer at the passed in offset.
+
+ Keyword Arguments:
+ buffer The binary buffer to append the data into. This should already
+ be the length needed to store the entire user defined data, as
+ specified by this object's data_length
+ offset The offset at which data can be written into the buffer
+ '''
+ return buffer
+
+
+ def read(self, buffer):
+ ''' Read in the IE from a buffer '''
+ # Fudge the data here if we're only a length 2. Due to the fact that
+ # Struct will pad the first byte with an additional byte, we end up
+ # needing a total of three bytes in the buffer.
+# if (len(buffer) == 2 and self._data_packer.size != 2):
+# temp_buffer = binascii.hexlify(buffer)
+# temp_buffer += '00'
+# buffer = binascii.a2b_hex(temp_buffer)
+# (self._ie, self._data_length, discard) = self._data_packer.unpack_from(buffer)
+# else:
+ (self._ie, self._data_length) = self._data_packer.unpack_from(buffer)
+ self.read_data(buffer, 2)
+
+
+ def read_data(self, buffer, offset):
+ ''' Read the inherited class's data from a buffer. Derived classes
+ should read the value from the buffer at the passed in offset.
+
+ Keyword Arguments:
+ buffer The binary buffer to read the data from.
+ offset The offset at which to begin reading the data.
+ '''
+ pass
+
+
+class IEVersion(IE):
+ ''' An IE that describes the IAX version '''
+
+ def __init__(self):
+ super(IEVersion, self).__init__(IEType.VERSION, 0x02)
+ self.__data_packer = struct.Struct('!H')
+ self._version = 0x0002
+
+ def get_version (self):
+ return self._version
+
+ def set_version(self, value):
+ if (value > 256):
+ raise ValueError()
+ self._version = value
+
+ def write_data(self, buffer, offset):
+ return self.__data_packer.pack_into(buffer, offset, self._version)
+
+ def read_data(self, buffer, offset):
+ (self._version) = self.__data_packer.unpack_from(buffer, offset)
+
+
+class IEStringData(IE):
+ ''' An IE with some string data component '''
+
+ def __init__(self, ie, value):
+ ''' Constructor
+
+ ie The information element identifier
+ value The string value to packet into the IE. If populated, this value
+ will be overridden '''
+ super(IEStringData, self).__init__(ie, len(value))
+ self._value = value
+ self.__data_packer = struct.Struct('!%ds' % len(value))
+
+ def get_value(self):
+ ''' The string value '''
+ return self._value
+
+ def write_data(self, buffer, offset):
+ buf = array.array('c', self._value)
+ print buf
+ print self.__data_packer.format
+ print len(buffer)
+ return self.__data_packer.pack_into(buffer,
+ offset,
+ buf)
+
+ def read_data(self, buffer, offset):
+ (self._value) = self.__data_packer.unpack_from(buffer, offset)
+
+
+class IEADSICapability(IE):
+ ''' An IE for ADSI capability '''
+
+ def __init(self):
+ super(IEADSICapability, self).__init__(IEType.ADSICPE, 0x02)
+ self._adsi_capability = 0x0
+ self.__data_packer = struct.Struct('!H')
+
+ def get_value(self):
+ ''' The ADSI capability value '''
+ return self._adsi_capability
+
+ def write_data(self, buffer, offset):
+ return self.__data_packer.pack_into(buffer, offset, self._adsi_capability)
+
+ def read_data(self, buffer, offset):
+ (self._adsi_capability) = self.__data_packer.unpack_from(buffer, offset)
+
+
+class IEAuthMethods(IE):
+ ''' An IE for Authentication Methods '''
+
+ def __init__(self):
+ super(IEAuthMethods, self).__init__(IEType.AUTHMETHODS, 0x02)
+ self._auth_method = 0x0000
+ self.__data_packer = struct.Struct('!H')
+
+ def get_value(self):
+ ''' The authentication method '''
+ return self._auth_method
+
+ def write_data(self, buffer, offset):
+ return self._data_packer.pack_into(buffer, offset, self._auth_method)
+
+ def read_data(self, buffer, offset):
+ auth_method = self.__data_packer.unpack_from(buffer, offset)
+ if (auth_method == 0x0001):
+ self._auth_method = IAX2AuthMethod.RESERVED
+ elif (auth_method == 0x0002):
+ self._auth_method = IAX2AuthMethod.MD5
+ elif (auth_method == 0x0004):
+ self._auth_method = IAX2AuthMethod.RSA
+ else:
+ raise ValueError('Unknown authentication method %d' % auth_method)
+
+class IEIpAddr(IE):
+ def __init__(self, ie, len):
+ super(IEIpAddr, self).__init__(ie, len)
+ fmt = '!HHI'
+ if (len == 0x1c):
+ fmt += '32sI'
+ elif (len == 0x10):
+ fmt += '16s'
+ else:
+ raise ValueError('Data length %d does not correspond to a valid IPv4 or IPv6 address')
+ self.__data_packer = struct.Struct(fmt)
+ self._addr_family = 0
+ self._port = 0
+ self._addr = 0
+ self._flow_rate = 0
+
+ def get_port(self):
+ return self._port
+
+ def get_addr(self):
+ return self._addr
+
+ def get_flow_rate(self):
+ return self._flow_rate
+
+ def get_addr_family(self):
+ return self._addr_family
+
+# def write(self, buffer, offset):
+# if (len == 0x10):
+# self.__data_packer.pack_into(buffer, offset,
+# self._addr_family,
+# self._port,
+# self._addr,)
+# elif (len == 0x1c):
+# self.__data_packer.pack_into(buffer, offset,
+# self._addr_family,
+# self._port,
+# self._flow_rate,
+#
+
+class IEFactory(object):
+
+ __factory = {
+ IEType.CALLED_NUMBER: IEStringData,
+ IEType.CALLING_NUMBER: IEStringData,
+ IEType.CALLING_ANI: IEStringData,
+ IEType.CALLING_NAME: IEStringData,
+ IEType.CALLED_CONTEXT: IEStringData,
+ IEType.USERNAME: IEStringData,
+ IEType.CAPABILITY: None,
+ IEType.FORMAT: None,
+ IEType.LANGUAGE: IEStringData,
+ IEType.VERSION: IEVersion,
+ IEType.ADSICPE: IEADSICapability,
+ IEType.DNID: IEStringData,
+ IEType.AUTHMETHODS: IEAuthMethods,
+ IEType.CHALLENGE: IEStringData,
+ IEType.MDF5_RESULT: IEStringData,
+ IEType.RSA_RESULT: IEStringData,
+ IEType.APPARENT_ADDR: IEIpAddr,
+ IEType.REFRESH: None,
+ IEType.DPSTATUS: None,
+ IEType.CALLNO: None,
+ IEType.CAUSE: None,
+ IEType.IAX_UNKNOWN: None,
+ IEType.MSGCOUNT: None,
+ IEType.AUTOANSWER: None,
+ IEType.MUSICONHOLD: None,
+ IEType.TRANSFERID: None,
+ IEType.RDNIS: None,
+ IEType.DATETIME: None,
+ IEType.CALLINGPRES: None,
+ IEType.CALLINGTON: None,
+ IEType.CALLINGTNS: None,
+ IEType.SAMPLINGRATE: None,
+ IEType.CAUSECODE: None,
+ IEType.ENCRYPTION: None,
+ IEType.ENCKEY: None,
+ IEType.CODEC_PREFS: None,
+ IEType.RR_JITTER: None,
+ IEType.RR_LOSS: None,
+ IEType.RR_PKTS: None,
+ IEType.RR_DELAY: None,
+ IEType.RR_DROPPED: None,
+ IEType.RR_OOO: None,
+ IEType.OSPTOKEN: None}
+
+ @classmethod
+ def create(cls, ie):
+ if ie not in IEFactory.__factory:
+ raise ValueError('Unsupported IE %d' % ie)
+
+
+
+
+class IEUnitTest(unittest.TestCase):
+ ''' IE unit tests '''
+
+ def test_constructor(self):
+ obj = IE(0, 0)
+ self.assertEquals(0, obj.get_data_length())
+ self.assertEquals(0, obj.get_ie())
+
+ def test_constructor_data(self):
+ obj = IE(2, 1)
+ self.assertEquals(1, obj.get_data_length())
+ self.assertEquals(2, obj.get_ie())
+
+ def test_default_write(self):
+ obj = IE(0, 0)
+ buf = obj.write()
+ self.assertEquals(2, len(buf))
+ self.assertEquals('0000', binascii.hexlify(buf))
+
+ def test_default_read(self):
+ obj = IE(1, 1)
+ bin = binascii.a2b_hex('0FF0')
+ obj.read(bin)
+ self.assertEquals(15, obj.get_ie())
+ self.assertEquals(240, obj.get_data_length())
+
+
+class IEStringDataUnitTest(unittest.TestCase):
+ ''' IENoData unit tests '''
+
+ def test_constructor(self):
+ obj = IEStringData(0x05, "Called Name")
+ self.assertEquals(11, obj.get_data_length())
+ self.assertEquals(0x05, obj.get_ie())
+
+ def test_default_write(self):
+ obj = IEStringData(0x05, "Called Name")
+ buf = obj.write()
+ print binascii.hexlify(buf)
+ self.assertEquals(13, len(buf))
+ self.assertEquals('050b', binascii.hexlify(buf))
+
+ def test_default_read(self):
+ obj = IEStringData(0x05, "")
+ bin = binascii.a2b_hex('050B')
+ obj.read(bin)
+ self.assertEquals(0x05, obj.get_ie())
+ self.assertEquals(0, obj.get_data_length())
+
+
+def main():
+ unittest.main()
+
+if __name__ == "__main__":
+ main()
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/team/mjordan/iax2-driver/lib/python/iax2-test/ie.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
More information about the svn-commits
mailing list