[Asterisk-code-review] Begin work on python3 compatability. (testsuite[15])
Corey Farrell
asteriskteam at digium.com
Fri Jun 29 08:45:14 CDT 2018
Corey Farrell has uploaded this change for review. ( https://gerrit.asterisk.org/9314
Change subject: Begin work on python3 compatability.
......................................................................
Begin work on python3 compatability.
* Use explicit relative import syntax.
* Correct syntax for print and except.
* Fix incorrect usage of IOError in astcdr.py - IOError is not a tuple.
* Use items, values instead of iteritems, itervalues.
* Use OrderedDict from collections if available (python 2.7+).
* Do not unpack tuples from argument list, save tuple to variable then
unpack within code.
* Use exception handler to import Python3 version of urlencode if the
Python2 version fails to import.
* Remove unused import unittest from opensslversion.
* Move executable tests to separate files to resolve issues with
relative imports:
- buildoptions
- cdr
- cel
- channel_test_condition
- config
- lock_test_condition
- sip_dialog_test_condition (disabled)
- sippversion
* Modify utils_socket test to use new harness_shared module.
* Remove unused executable flag and shebang from phones.py.
* Update self_test script to run tests from new location.
* Remove ability to use config.py to display a config file.
* Allow PYTHON environmental variable to tell run-local to use a
specific python binary.
* Execute test_runner.py using python -m syntax. This avoids conflicts
between needing to use relative imports for modules but not being
allowed to use them in executable scripts.
* Encode self.realbase for use with md5 function.
* Remove executable flag and shebang from test_runner.py. Remove
redundant appending of sys.path.
* Modify output for running test and status to print the name of the
test instead of the command used to run the test.
* Use key= sorter argument to compare sizes of tmp filesystems. This
may cause us to use /var/tmp instead of /tmp even if /var/tmp has only
a few kilobytes more space but python3 does not support cmp= argument
to sorter.
* Decode binary to utf-8
- CLI output
- runtest.py subprocess output
- XML printed to stdout
- SyncAMI data
- SIPp runtime and version output
* Update test_runner.load_and_parse_module to check for the module in
lib/python/asterisk, make required adjustments if found.
* Update __strip_illegal_xml_chars to work with unicode version of
str.translate which does not support second argument.
* Update import statements from all *.py modules within tests/ to use
absolute names for asterisk modules. This is required for python2 to
continue functioning with the new way that test_runner must be executed.
No attempt has been made to fix python3 compatibility issues in these
modules.
Some tests may pass using python3. Doing so requires using an updated
starpy and no python sources within tests/ have been updated. Unit
tests from lib/python can be run against python2 and python3 using
./self_test.
The goal of this is for all tests to continue functioning using python2.
Some tests may work under python3 but individual tests will need to be
addressed separately. For tests which do not contain a run-test script
the version of python used to execute ./runtests.py will be used:
./runtests.py -t tests/manager/originate
python2 ./runtests.py -t tests/manager/originate
python3 ./runtests.py -t tests/manager/originate
These commands will be python, python2 and python3 respectively. The
first example where no interpreter is specified uses the shebang from
./runtests.py. The interpreter used by ./run-local can be controlled
by setting PYTHON environmental variable:
PYTHON=python3 ./run-local run -t tests/manager/originate
Change-Id: If76c2d3e11e4ab4552d0df7841287c8bb2de7918
---
M lib/python/asterisk/ami.py
M lib/python/asterisk/apptest.py
M lib/python/asterisk/ari.py
M lib/python/asterisk/astconfigparser.py
M lib/python/asterisk/astcsv.py
M lib/python/asterisk/astdicts.py
M lib/python/asterisk/asterisk.py
M lib/python/asterisk/bridge_test_case.py
M lib/python/asterisk/buildoptions.py
M lib/python/asterisk/cdr.py
M lib/python/asterisk/cel.py
M lib/python/asterisk/channel_test_condition.py
M lib/python/asterisk/confbridge.py
M lib/python/asterisk/config.py
M lib/python/asterisk/fd_test_condition.py
M lib/python/asterisk/lock_test_condition.py
M lib/python/asterisk/matcher.py
M lib/python/asterisk/matcher_listener.py
M lib/python/asterisk/opensslversion.py
M lib/python/asterisk/originate.py
M lib/python/asterisk/pcap.py
M lib/python/asterisk/phones.py
M lib/python/asterisk/pjsua_mod.py
M lib/python/asterisk/pluggable_modules.py
M lib/python/asterisk/pluggable_registry.py
M lib/python/asterisk/realtime_converter.py
M lib/python/asterisk/realtime_odbc_module.py
M lib/python/asterisk/realtime_test_module.py
A lib/python/asterisk/self_test/harness_shared.py
A lib/python/asterisk/self_test/locks-backtrace.txt
A lib/python/asterisk/self_test/locks-fail.txt
A lib/python/asterisk/self_test/locks-large-multiple-object.txt
A lib/python/asterisk/self_test/locks-multiple-objects-no-backtrace.txt
A lib/python/asterisk/self_test/locks-pass.txt
A lib/python/asterisk/self_test/locks-single-object-no-held-info.txt
A lib/python/asterisk/self_test/locks-single-object.txt
A lib/python/asterisk/self_test/test_buildoptions.py
A lib/python/asterisk/self_test/test_cdr.py
A lib/python/asterisk/self_test/test_cel.py
A lib/python/asterisk/self_test/test_channel_test_condition.py
A lib/python/asterisk/self_test/test_config.py
A lib/python/asterisk/self_test/test_lock_test_condition.py
R lib/python/asterisk/self_test/test_matcher.py
A lib/python/asterisk/self_test/test_sip_dialog_test_condition.py.txt
A lib/python/asterisk/self_test/test_sippversion.py
M lib/python/asterisk/self_test/test_utils_socket.py
M lib/python/asterisk/sip_channel_test_condition.py
M lib/python/asterisk/sip_dialog_test_condition.py
M lib/python/asterisk/sipp.py
M lib/python/asterisk/sippversion.py
M lib/python/asterisk/syncami.py
M lib/python/asterisk/test_case.py
M lib/python/asterisk/test_conditions.py
M lib/python/asterisk/test_config.py
M lib/python/asterisk/test_runner.py
M lib/python/asterisk/test_suite_utils.py
M lib/python/asterisk/thread_test_condition.py
M lib/python/asterisk/voicemail.py
M lib/python/rlmi.py
M lib/python/sip_message.py
M run-local
M runtests.py
M self_test
M tests/cdr/cdr-tests.py
M tests/cdr/sqlite3/cdr_sqlite3.py
M tests/channels/SIP/tcpauthlimit/sipp_scenario.py
M tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv4/scenario_generator.py
M tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv6/scenario_generator.py
M tests/channels/pjsip/subscriptions/presence/verify_bodies/presence.py
M tests/channels/pjsip/subscriptions/rls/rls_test.py
M tests/channels/pjsip/transfers/attended_transfer/nominal/packet_sniffer.py
M tests/codecs/audio_analyzer.py
M tests/codecs/rtp_analyzer.py
M tests/hep/hep_capture_node.py
M tests/manager/config/ManagerConfigTest.py
M tests/manager/device_state_changed/ami_device_state.py
M tests/manager/device_state_list/ami_device_state_list.py
M tests/manager/exten_state_list/ami_exten_state_list.py
M tests/manager/presence_state_changed/ami_presence_state.py
M tests/manager/presence_state_list/ami_presence_state_list.py
M tests/pbx/manager_extensions/ami_extension_control.py
M tests/rest_api/external_interaction/attended_transfer/attended_transfer.py
M tests/rest_api/external_interaction/blind_transfer/call_transfer.py
M tests/rest_api/message/message_modules.py
M usage.py
85 files changed, 1,910 insertions(+), 1,841 deletions(-)
git pull ssh://gerrit.asterisk.org:29418/testsuite refs/changes/14/9314/1
diff --git a/lib/python/asterisk/ami.py b/lib/python/asterisk/ami.py
index fb52d1c..2e0fe4e 100644
--- a/lib/python/asterisk/ami.py
+++ b/lib/python/asterisk/ami.py
@@ -13,7 +13,8 @@
import logging
import re
import json
-from pluggable_registry import PLUGGABLE_EVENT_REGISTRY,\
+from .test_runner import load_and_parse_module
+from .pluggable_registry import PLUGGABLE_EVENT_REGISTRY,\
PLUGGABLE_ACTION_REGISTRY, var_replace
LOGGER = logging.getLogger(__name__)
@@ -361,7 +362,7 @@
lower_key = key.lower()
if lower_key == 'extra':
value = dict((key.lower(), value)
- for key, value in value.iteritems())
+ for key, value in value.items())
self.requirements[lower_key] = value
self.orderings = requirements.get('partialorder') or []
self.named_id = requirements.get('id')
@@ -529,8 +530,7 @@
def event_callback(self, ami, event):
"""Callback called when an event is received from AMI"""
- callback_module = __import__(self.callback_module)
- method = getattr(callback_module, self.callback_method)
+ method = load_and_parse_module(self.callback_module + '.' + self.callback_method)
self.passed = method(ami, event)
if self.passed is None:
LOGGER.error("Callback %s.%s returned None instead of a boolean",
@@ -828,7 +828,7 @@
def replace_ami_vars(mydict, values):
outdict = {}
- for key, value in mydict.iteritems():
+ for key, value in mydict.items():
outdict[key] = var_replace(value, values)
return outdict
diff --git a/lib/python/asterisk/apptest.py b/lib/python/asterisk/apptest.py
index 1fc8496..2c33ad9 100644
--- a/lib/python/asterisk/apptest.py
+++ b/lib/python/asterisk/apptest.py
@@ -18,8 +18,8 @@
from twisted.internet import reactor, defer
sys.path.append("lib/python")
-from test_case import TestCase
-from ami import AMIEventInstance
+from .test_case import TestCase
+from .ami import AMIEventInstance
LOGGER = logging.getLogger(__name__)
@@ -165,7 +165,7 @@
A Scenario is considered done if all results have been met and
all expected actions have been executed.
"""
- return (all(self._expected_results.itervalues()) and
+ return (all(self._expected_results.values()) and
all(i.ran_actions or i.unexpected
for i in self._event_instances))
diff --git a/lib/python/asterisk/ari.py b/lib/python/asterisk/ari.py
index da19cee..23672dd 100644
--- a/lib/python/asterisk/ari.py
+++ b/lib/python/asterisk/ari.py
@@ -12,12 +12,16 @@
import re
import requests
import traceback
-import urllib
+try:
+ from urllib.parse import urlencode
+except:
+ from urllib import urlencode
-from test_case import TestCase
-from pluggable_registry import PLUGGABLE_EVENT_REGISTRY,\
+from .test_case import TestCase
+from .test_runner import load_and_parse_module
+from .pluggable_registry import PLUGGABLE_EVENT_REGISTRY,\
PLUGGABLE_ACTION_REGISTRY, var_replace
-from test_suite_utils import all_match
+from .test_suite_utils import all_match
from twisted.internet import reactor
try:
from autobahn.websocket import WebSocketClientFactory, \
@@ -332,7 +336,7 @@
"""
url = "ws://%s:%d/ari/events?%s" % \
(host, port,
- urllib.urlencode({'app': apps, 'api_key': '%s:%s' % userpass}))
+ urlencode({'app': apps, 'api_key': '%s:%s' % userpass}))
if subscribe_all:
url += '&subscribeAll=true'
LOGGER.info("WebSocketClientFactory(url=%s)", url)
@@ -560,7 +564,7 @@
url = self.ari.build_url(uri)
requests_method = getattr(requests, self.method)
params = dict((key, var_replace(val, values))
- for key, val in self.params.iteritems())
+ for key, val in self.params.items())
response = requests_method(
url,
@@ -580,7 +584,7 @@
response.status_code, response.text)
return False
else:
- if response.status_code / 100 != 2:
+ if response.status_code // 100 != 2:
LOGGER.error('sent %s %s %s response %d %s',
self.method, self.uri, self.params,
response.status_code, response.text)
@@ -606,8 +610,7 @@
callback = self.instance_config.get('callback')
if callback:
- module = __import__(callback['module'])
- self.callback = getattr(module, callback['method'])
+ self.callback = load_and_parse_module(callback['module'] + '.' + callback['method'])
else:
# No callback; just use a no-op
self.callback = lambda *args, **kwargs: True
diff --git a/lib/python/asterisk/astconfigparser.py b/lib/python/asterisk/astconfigparser.py
index dc79e80..e8da255 100644
--- a/lib/python/asterisk/astconfigparser.py
+++ b/lib/python/asterisk/astconfigparser.py
@@ -8,8 +8,7 @@
import re
import itertools
-from astdicts import OrderedDict
-from astdicts import MultiOrderedDict
+from .astdicts import OrderedDict, MultiOrderedDict
def merge_values(left, right, key):
@@ -270,11 +269,11 @@
def write_dicts(config_file, mdicts):
"""Write the contents of the mdicts to the specified config file"""
- for section, sect_list in mdicts.iteritems():
+ for section, sect_list in mdicts.items():
# every section contains a list of dictionaries
for sect in sect_list:
config_file.write("[%s]\n" % section)
- for key, val_list in sect.iteritems():
+ for key, val_list in sect.items():
# every value is also a list
for v in val_list:
key_val = key
@@ -355,7 +354,7 @@
if self._includes:
res.extend(list(itertools.chain(*[
incl.get_sections(key, attr, searched)
- for incl in self._includes.itervalues()])))
+ for incl in self._includes.values()])))
if self._parent:
res += self._parent.get_sections(key, attr, searched)
return res
@@ -445,7 +444,7 @@
with open(filename, 'rt') as config_file:
self._read(config_file, sect)
except IOError:
- print "Could not open file ", filename, " for reading"
+ print("Could not open file ", filename, " for reading")
def _read(self, config_file, sect):
"""Parse configuration information from the config_file"""
@@ -478,7 +477,7 @@
def write(self, config_file):
"""Write configuration information out to a file"""
try:
- for key, val in self._includes.iteritems():
+ for key, val in self._includes.items():
val.write(key)
config_file.write('#include "%s"\n' % key)
@@ -490,4 +489,4 @@
with open(config_file, 'wt') as fp:
self.write(fp)
except IOError:
- print "Could not open file ", config_file, " for writing"
+ print("Could not open file ", config_file, " for writing")
diff --git a/lib/python/asterisk/astcsv.py b/lib/python/asterisk/astcsv.py
index 67d25d3..7619bf5 100644
--- a/lib/python/asterisk/astcsv.py
+++ b/lib/python/asterisk/astcsv.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Asterisk CSV-based testing
This module implements the basic CSV testing backend for things like
@@ -11,7 +10,6 @@
the GNU General Public License Version 2.
"""
-import unittest
import sys
import csv
import re
@@ -57,10 +55,10 @@
else:
cmp_fn = (lambda x, y: str(x).lower() == str(y).lower())
- for key, value in self.iteritems():
+ for key, value in self.items():
if None not in (value, other.get(key)) and not cmp_fn(value, other.get(key)):
if not silent:
- LOGGER.warn("CSV MATCH FAILED, Expected: %s: '%s' "
+ LOGGER.warning("CSV MATCH FAILED, Expected: %s: '%s' "
"Got: %s: '%s'" % (key, value, key,
other.get(key)))
return False
@@ -70,9 +68,9 @@
"""Retrieve a value from the specified column key"""
return self.__columns.get(key)
- def iteritems(self):
+ def items(self):
"""Iterate over the values in the columns"""
- return self.__columns.iteritems()
+ return self.__columns.items()
def __str__(self):
return ",".join(["\"%s\"" % (self.__dict__[x]) for x in self.__fields])
@@ -94,12 +92,14 @@
self.__records = []
csvreader = None
+ csvfile = None
try:
- csvreader = csv.DictReader(open(self.filename, "r"), fields, ",")
- except IOError as (errno, strerror):
+ csvfile = open(self.filename, "r")
+ csvreader = csv.DictReader(csvfile, fields, ",")
+ except IOError as e:
LOGGER.error("IOError %d[%s] while opening file '%s'" %
- (errno, strerror, self.filename))
+ (e.errno, e.strerror, self.filename))
except:
LOGGER.error("Unexpected error: %s" % (sys.exc_info()[0]))
@@ -110,6 +110,8 @@
for row in csvreader:
record = self.row_factory(**row)
self.__records.append(record)
+
+ csvfile.close()
def __len__(self):
return len(self.__records)
@@ -125,7 +127,7 @@
each record"""
if not partial and (len(self) != len(other)):
- LOGGER.warn("CSV MATCH FAILED, different number of records, "
+ LOGGER.warning("CSV MATCH FAILED, different number of records, "
"self=%d and other=%d" % (len(self), len(other)))
return False
@@ -147,7 +149,7 @@
size = len(list_a)
# attempt two orderings: forward and reversed
- guess_orders = (range(size), list(reversed(range(size))))
+ guess_orders = (list(range(size)), list(reversed(range(size))))
found_orders = []
for guess_order in guess_orders:
@@ -185,7 +187,7 @@
# have it complain immediately.
for i, item in enumerate(self):
if not item.match(other[i], exact=exactness):
- LOGGER.warn("Failed to match entry %d" % (i,))
+ LOGGER.warning("Failed to match entry %d" % (i,))
return False
assert False
@@ -193,7 +195,7 @@
pass # joy!
elif len(matches) > 1:
- LOGGER.warn("More than one CSV permutation results in success")
+ LOGGER.warning("More than one CSV permutation results in success")
return True
@@ -205,9 +207,6 @@
try:
open(self.filename, "w").close()
except:
- LOGGER.warn("Unable to empty CSV file %s" % (self.filename))
-
-if __name__ == '__main__':
- unittest.main()
+ LOGGER.warning("Unable to empty CSV file %s" % (self.filename))
# vim:sw=4:ts=4:expandtab:textwidth=79
diff --git a/lib/python/asterisk/astdicts.py b/lib/python/asterisk/astdicts.py
index ae63075..4e70eec 100644
--- a/lib/python/asterisk/astdicts.py
+++ b/lib/python/asterisk/astdicts.py
@@ -3,260 +3,264 @@
# copied from http://code.activestate.com/recipes/576693/
try:
- from thread import get_ident as _get_ident
+ # Use builtin OrderedDict() from Python2.7.
+ from collections import OrderedDict
except ImportError:
- from dummy_thread import get_ident as _get_ident
+ try:
+ from thread import get_ident as _get_ident
+ except ImportError:
+ from dummy_thread import get_ident as _get_ident
-try:
- from _abcoll import KeysView, ValuesView, ItemsView
-except ImportError:
- pass
+ try:
+ from _abcoll import KeysView, ValuesView, ItemsView
+ except ImportError:
+ pass
-class OrderedDict(dict):
- 'Dictionary that remembers insertion order'
- # An inherited dict maps keys to values.
- # The inherited dict provides __getitem__, __len__, __contains__, and get.
- # The remaining methods are order-aware.
- # Big-O running times for all methods are the same as for regular dictionaries.
+ class OrderedDict(dict):
+ 'Dictionary that remembers insertion order'
+ # An inherited dict maps keys to values.
+ # The inherited dict provides __getitem__, __len__, __contains__, and get.
+ # The remaining methods are order-aware.
+ # Big-O running times for all methods are the same as for regular dictionaries.
- # The internal self.__map dictionary maps keys to links in a doubly linked list.
- # The circular doubly linked list starts and ends with a sentinel element.
- # The sentinel element never gets deleted (this simplifies the algorithm).
- # Each link is stored as a list of length three: [PREV, NEXT, KEY].
+ # The internal self.__map dictionary maps keys to links in a doubly linked list.
+ # The circular doubly linked list starts and ends with a sentinel element.
+ # The sentinel element never gets deleted (this simplifies the algorithm).
+ # Each link is stored as a list of length three: [PREV, NEXT, KEY].
- def __init__(self, *args, **kwds):
- '''Initialize an ordered dictionary. Signature is the same as for
- regular dictionaries, but keyword arguments are not recommended
- because their insertion order is arbitrary.
+ def __init__(self, *args, **kwds):
+ '''Initialize an ordered dictionary. Signature is the same as for
+ regular dictionaries, but keyword arguments are not recommended
+ because their insertion order is arbitrary.
- '''
- if len(args) > 1:
- raise TypeError('expected at most 1 arguments, got %d' % len(args))
- try:
- self.__root
- except AttributeError:
- self.__root = root = [] # sentinel node
- root[:] = [root, root, None]
- self.__map = {}
- self.__update(*args, **kwds)
+ '''
+ if len(args) > 1:
+ raise TypeError('expected at most 1 arguments, got %d' % len(args))
+ try:
+ self.__root
+ except AttributeError:
+ self.__root = root = [] # sentinel node
+ root[:] = [root, root, None]
+ self.__map = {}
+ self.__update(*args, **kwds)
- def __setitem__(self, key, value, dict_setitem=dict.__setitem__):
- 'od.__setitem__(i, y) <==> od[i]=y'
- # Setting a new item creates a new link which goes at the end of the linked
- # list, and the inherited dictionary is updated with the new key/value pair.
- if key not in self:
+ def __setitem__(self, key, value, dict_setitem=dict.__setitem__):
+ 'od.__setitem__(i, y) <==> od[i]=y'
+ # Setting a new item creates a new link which goes at the end of the linked
+ # list, and the inherited dictionary is updated with the new key/value pair.
+ if key not in self:
+ root = self.__root
+ last = root[0]
+ last[1] = root[0] = self.__map[key] = [last, root, key]
+ dict_setitem(self, key, value)
+
+ def __delitem__(self, key, dict_delitem=dict.__delitem__):
+ 'od.__delitem__(y) <==> del od[y]'
+ # Deleting an existing item uses self.__map to find the link which is
+ # then removed by updating the links in the predecessor and successor nodes.
+ dict_delitem(self, key)
+ link_prev, link_next, key = self.__map.pop(key)
+ link_prev[1] = link_next
+ link_next[0] = link_prev
+
+ def __iter__(self):
+ 'od.__iter__() <==> iter(od)'
root = self.__root
- last = root[0]
- last[1] = root[0] = self.__map[key] = [last, root, key]
- dict_setitem(self, key, value)
+ curr = root[1]
+ while curr is not root:
+ yield curr[2]
+ curr = curr[1]
- def __delitem__(self, key, dict_delitem=dict.__delitem__):
- 'od.__delitem__(y) <==> del od[y]'
- # Deleting an existing item uses self.__map to find the link which is
- # then removed by updating the links in the predecessor and successor nodes.
- dict_delitem(self, key)
- link_prev, link_next, key = self.__map.pop(key)
- link_prev[1] = link_next
- link_next[0] = link_prev
-
- def __iter__(self):
- 'od.__iter__() <==> iter(od)'
- root = self.__root
- curr = root[1]
- while curr is not root:
- yield curr[2]
- curr = curr[1]
-
- def __reversed__(self):
- 'od.__reversed__() <==> reversed(od)'
- root = self.__root
- curr = root[0]
- while curr is not root:
- yield curr[2]
- curr = curr[0]
-
- def clear(self):
- 'od.clear() -> None. Remove all items from od.'
- try:
- for node in self.__map.itervalues():
- del node[:]
+ def __reversed__(self):
+ 'od.__reversed__() <==> reversed(od)'
root = self.__root
- root[:] = [root, root, None]
- self.__map.clear()
- except AttributeError:
- pass
- dict.clear(self)
+ curr = root[0]
+ while curr is not root:
+ yield curr[2]
+ curr = curr[0]
- def popitem(self, last=True):
- '''od.popitem() -> (k, v), return and remove a (key, value) pair.
- Pairs are returned in LIFO order if last is true or FIFO order if false.
+ def clear(self):
+ 'od.clear() -> None. Remove all items from od.'
+ try:
+ for node in self.__map.itervalues():
+ del node[:]
+ root = self.__root
+ root[:] = [root, root, None]
+ self.__map.clear()
+ except AttributeError:
+ pass
+ dict.clear(self)
- '''
- if not self:
- raise KeyError('dictionary is empty')
- root = self.__root
- if last:
- link = root[0]
- link_prev = link[0]
- link_prev[1] = root
- root[0] = link_prev
- else:
- link = root[1]
- link_next = link[1]
- root[1] = link_next
- link_next[0] = root
- key = link[2]
- del self.__map[key]
- value = dict.pop(self, key)
- return key, value
+ def popitem(self, last=True):
+ '''od.popitem() -> (k, v), return and remove a (key, value) pair.
+ Pairs are returned in LIFO order if last is true or FIFO order if false.
- # -- the following methods do not depend on the internal structure --
-
- def keys(self):
- 'od.keys() -> list of keys in od'
- return list(self)
-
- def values(self):
- 'od.values() -> list of values in od'
- return [self[key] for key in self]
-
- def items(self):
- 'od.items() -> list of (key, value) pairs in od'
- return [(key, self[key]) for key in self]
-
- def iterkeys(self):
- 'od.iterkeys() -> an iterator over the keys in od'
- return iter(self)
-
- def itervalues(self):
- 'od.itervalues -> an iterator over the values in od'
- for k in self:
- yield self[k]
-
- def iteritems(self):
- 'od.iteritems -> an iterator over the (key, value) items in od'
- for k in self:
- yield (k, self[k])
-
- def update(*args, **kwds):
- '''od.update(E, **F) -> None. Update od from dict/iterable E and F.
-
- If E is a dict instance, does: for k in E: od[k] = E[k]
- If E has a .keys() method, does: for k in E.keys(): od[k] = E[k]
- Or if E is an iterable of items, does: for k, v in E: od[k] = v
- In either case, this is followed by: for k, v in F.items(): od[k] = v
-
- '''
- if len(args) > 2:
- raise TypeError('update() takes at most 2 positional '
- 'arguments (%d given)' % (len(args),))
- elif not args:
- raise TypeError('update() takes at least 1 argument (0 given)')
- self = args[0]
- # Make progressively weaker assumptions about "other"
- other = ()
- if len(args) == 2:
- other = args[1]
- if isinstance(other, dict):
- for key in other:
- self[key] = other[key]
- elif hasattr(other, 'keys'):
- for key in other.keys():
- self[key] = other[key]
- else:
- for key, value in other:
- self[key] = value
- for key, value in kwds.items():
- self[key] = value
-
- __update = update # let subclasses override update without breaking __init__
-
- __marker = object()
-
- def pop(self, key, default=__marker):
- '''od.pop(k[,d]) -> v, remove specified key and return the corresponding value.
- If key is not found, d is returned if given, otherwise KeyError is raised.
-
- '''
- if key in self:
- result = self[key]
- del self[key]
- return result
- if default is self.__marker:
- raise KeyError(key)
- return default
-
- def setdefault(self, key, default=None):
- 'od.setdefault(k[,d]) -> od.get(k,d), also set od[k]=d if k not in od'
- if key in self:
- return self[key]
- self[key] = default
- return default
-
- def __repr__(self, _repr_running={}):
- 'od.__repr__() <==> repr(od)'
- call_key = id(self), _get_ident()
- if call_key in _repr_running:
- return '...'
- _repr_running[call_key] = 1
- try:
+ '''
if not self:
- return '%s()' % (self.__class__.__name__,)
- return '%s(%r)' % (self.__class__.__name__, self.items())
- finally:
- del _repr_running[call_key]
+ raise KeyError('dictionary is empty')
+ root = self.__root
+ if last:
+ link = root[0]
+ link_prev = link[0]
+ link_prev[1] = root
+ root[0] = link_prev
+ else:
+ link = root[1]
+ link_next = link[1]
+ root[1] = link_next
+ link_next[0] = root
+ key = link[2]
+ del self.__map[key]
+ value = dict.pop(self, key)
+ return key, value
- def __reduce__(self):
- 'Return state information for pickling'
- items = [[k, self[k]] for k in self]
- inst_dict = vars(self).copy()
- for k in vars(OrderedDict()):
- inst_dict.pop(k, None)
- if inst_dict:
- return (self.__class__, (items,), inst_dict)
- return self.__class__, (items,)
+ # -- the following methods do not depend on the internal structure --
- def copy(self):
- 'od.copy() -> a shallow copy of od'
- return self.__class__(self)
+ def keys(self):
+ 'od.keys() -> list of keys in od'
+ return list(self)
- @classmethod
- def fromkeys(cls, iterable, value=None):
- '''OD.fromkeys(S[, v]) -> New ordered dictionary with keys from S
- and values equal to v (which defaults to None).
+ def values(self):
+ 'od.values() -> list of values in od'
+ return [self[key] for key in self]
- '''
- d = cls()
- for key in iterable:
- d[key] = value
- return d
+ def items(self):
+ 'od.items() -> list of (key, value) pairs in od'
+ return [(key, self[key]) for key in self]
- def __eq__(self, other):
- '''od.__eq__(y) <==> od==y. Comparison to another OD is order-sensitive
- while comparison to a regular mapping is order-insensitive.
+ def iterkeys(self):
+ 'od.iterkeys() -> an iterator over the keys in od'
+ return iter(self)
- '''
- if isinstance(other, OrderedDict):
- return len(self)==len(other) and self.items() == other.items()
- return dict.__eq__(self, other)
+ def itervalues(self):
+ 'od.itervalues -> an iterator over the values in od'
+ for k in self:
+ yield self[k]
- def __ne__(self, other):
- return not self == other
+ def iteritems(self):
+ 'od.iteritems -> an iterator over the (key, value) items in od'
+ for k in self:
+ yield (k, self[k])
- # -- the following methods are only used in Python 2.7 --
+ def update(*args, **kwds):
+ '''od.update(E, **F) -> None. Update od from dict/iterable E and F.
- def viewkeys(self):
- "od.viewkeys() -> a set-like object providing a view on od's keys"
- return KeysView(self)
+ If E is a dict instance, does: for k in E: od[k] = E[k]
+ If E has a .keys() method, does: for k in E.keys(): od[k] = E[k]
+ Or if E is an iterable of items, does: for k, v in E: od[k] = v
+ In either case, this is followed by: for k, v in F.items(): od[k] = v
- def viewvalues(self):
- "od.viewvalues() -> an object providing a view on od's values"
- return ValuesView(self)
+ '''
+ if len(args) > 2:
+ raise TypeError('update() takes at most 2 positional '
+ 'arguments (%d given)' % (len(args),))
+ elif not args:
+ raise TypeError('update() takes at least 1 argument (0 given)')
+ self = args[0]
+ # Make progressively weaker assumptions about "other"
+ other = ()
+ if len(args) == 2:
+ other = args[1]
+ if isinstance(other, dict):
+ for key in other:
+ self[key] = other[key]
+ elif hasattr(other, 'keys'):
+ for key in other.keys():
+ self[key] = other[key]
+ else:
+ for key, value in other:
+ self[key] = value
+ for key, value in kwds.items():
+ self[key] = value
- def viewitems(self):
- "od.viewitems() -> a set-like object providing a view on od's items"
- return ItemsView(self)
+ __update = update # let subclasses override update without breaking __init__
+
+ __marker = object()
+
+ def pop(self, key, default=__marker):
+ '''od.pop(k[,d]) -> v, remove specified key and return the corresponding value.
+ If key is not found, d is returned if given, otherwise KeyError is raised.
+
+ '''
+ if key in self:
+ result = self[key]
+ del self[key]
+ return result
+ if default is self.__marker:
+ raise KeyError(key)
+ return default
+
+ def setdefault(self, key, default=None):
+ 'od.setdefault(k[,d]) -> od.get(k,d), also set od[k]=d if k not in od'
+ if key in self:
+ return self[key]
+ self[key] = default
+ return default
+
+ def __repr__(self, _repr_running={}):
+ 'od.__repr__() <==> repr(od)'
+ call_key = id(self), _get_ident()
+ if call_key in _repr_running:
+ return '...'
+ _repr_running[call_key] = 1
+ try:
+ if not self:
+ return '%s()' % (self.__class__.__name__,)
+ return '%s(%r)' % (self.__class__.__name__, self.items())
+ finally:
+ del _repr_running[call_key]
+
+ def __reduce__(self):
+ 'Return state information for pickling'
+ items = [[k, self[k]] for k in self]
+ inst_dict = vars(self).copy()
+ for k in vars(OrderedDict()):
+ inst_dict.pop(k, None)
+ if inst_dict:
+ return (self.__class__, (items,), inst_dict)
+ return self.__class__, (items,)
+
+ def copy(self):
+ 'od.copy() -> a shallow copy of od'
+ return self.__class__(self)
+
+ @classmethod
+ def fromkeys(cls, iterable, value=None):
+ '''OD.fromkeys(S[, v]) -> New ordered dictionary with keys from S
+ and values equal to v (which defaults to None).
+
+ '''
+ d = cls()
+ for key in iterable:
+ d[key] = value
+ return d
+
+ def __eq__(self, other):
+ '''od.__eq__(y) <==> od==y. Comparison to another OD is order-sensitive
+ while comparison to a regular mapping is order-insensitive.
+
+ '''
+ if isinstance(other, OrderedDict):
+ return len(self)==len(other) and self.items() == other.items()
+ return dict.__eq__(self, other)
+
+ def __ne__(self, other):
+ return not self == other
+
+ # -- the following methods are only used in Python 2.7 --
+
+ def viewkeys(self):
+ "od.viewkeys() -> a set-like object providing a view on od's keys"
+ return KeysView(self)
+
+ def viewvalues(self):
+ "od.viewvalues() -> an object providing a view on od's values"
+ return ValuesView(self)
+
+ def viewitems(self):
+ "od.viewitems() -> a set-like object providing a view on od's items"
+ return ItemsView(self)
###############################################################################
### MultiOrderedDict
@@ -292,7 +296,7 @@
# TODO - find out why for some reason copies
# the [] as an [[]], so do manually
c = MultiOrderedDict() #self.__class__(self)
- for key, val in self.iteritems():
+ for key, val in self.items():
for v in val:
c[key] = v
return c
diff --git a/lib/python/asterisk/asterisk.py b/lib/python/asterisk/asterisk.py
old mode 100755
new mode 100644
index 985afea..c3996f6
--- a/lib/python/asterisk/asterisk.py
+++ b/lib/python/asterisk/asterisk.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Asterisk Instances in Python.
This module provides an interface for creating instances of Asterisk
@@ -18,9 +17,9 @@
import logging
import fileinput
-import test_suite_utils
+from . import test_suite_utils
-from config import ConfigFile
+from .config import ConfigFile
from twisted.internet import reactor, protocol, defer, utils, error
from twisted.python.failure import Failure
@@ -54,7 +53,7 @@
def dataReceived(self, data):
"""Called when we receive data back"""
LOGGER.debug(data)
- self.output += data
+ self.output += data.decode('utf-8', 'ignore')
def connectionLost(self, reason):
"""Called when the connect is lost"""
@@ -206,7 +205,8 @@
def _set_properties(self, result):
"""Set the properties based on the result of the
getProcessOutputAndValue call"""
- self.output, self.err, self.exitcode = result
+ bintxt, self.err, self.exitcode = result
+ self.output = bintxt.decode('utf-8', 'ignore')
class AsteriskProtocol(protocol.ProcessProtocol):
@@ -232,7 +232,7 @@
def outReceived(self, data):
"""Override of ProcessProtocol.outReceived"""
- self.output += data
+ self.output += data.decode('utf-8', 'ignore')
def connectionMade(self):
"""Override of ProcessProtocol.connectionMade"""
@@ -311,20 +311,6 @@
we're not running into the asterisk.ctl AF_UNIX limit.
"""
- def compare_free_space(x, y):
- if os.stat(y).st_dev == os.stat(x).st_dev:
- return 0
- # statvfs can return a long; comparison functions must return an
- # int. Where both are same filesystem, bavail might change from
- # one call to the next, but hopefully it is not more than 1000.
- difference = os.statvfs(y).f_bavail - os.statvfs(x).f_bavail
- if (difference > 1000):
- return 1
- elif (difference < 1000):
- return -1
- else:
- return 0
-
localtest_root = os.getenv("AST_TEST_ROOT")
if localtest_root:
# Base location of the temporary files created by the testsuite
@@ -333,7 +319,7 @@
default_etc_directory = os.path.join(localtest_root, "etc/asterisk")
else:
# select tmp path with most available space
- best_tmp = sorted(['/tmp', '/var/tmp'], cmp=compare_free_space)[0]
+ best_tmp = sorted(['/tmp', '/var/tmp'], key=lambda path: os.statvfs(path).f_bavail)[0]
# Base location of the temporary files created by the testsuite
test_suite_root = best_tmp + "/asterisk-testsuite"
# The default etc directory for Asterisk
@@ -955,7 +941,7 @@
ast_file.write("#include \"%s/asterisk.options.conf.inc\"\n" %
(self.astetcdir))
if ast_conf_options:
- for (var, val) in ast_conf_options.iteritems():
+ for (var, val) in ast_conf_options.items():
ast_file.write("%s = %s\n" % (var, val))
for (var, val) in cat.options:
if not ast_conf_options or var not in ast_conf_options:
diff --git a/lib/python/asterisk/bridge_test_case.py b/lib/python/asterisk/bridge_test_case.py
index ea1767e..155b3a1 100644
--- a/lib/python/asterisk/bridge_test_case.py
+++ b/lib/python/asterisk/bridge_test_case.py
@@ -13,7 +13,7 @@
from time import sleep
sys.path.append("lib/python")
-from test_case import TestCase
+from .test_case import TestCase
LOGGER = logging.getLogger(__name__)
diff --git a/lib/python/asterisk/buildoptions.py b/lib/python/asterisk/buildoptions.py
index 3b0b4b9..833053a 100644
--- a/lib/python/asterisk/buildoptions.py
+++ b/lib/python/asterisk/buildoptions.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Asterisk Build Options Handling
This module implements an Asterisk build options parser. It
@@ -13,7 +12,6 @@
"""
import sys
-import unittest
class AsteriskBuildOptions(object):
@@ -51,7 +49,7 @@
except IOError:
return ret_val
except:
- print "Unexpected error: %s" % sys.exc_info()[0]
+ print("Unexpected error: %s" % sys.exc_info()[0])
return ret_val
for line in file_lines:
if "#define" in line:
@@ -59,7 +57,7 @@
if (define_tuple[0] == "" or define_tuple[2] == ""):
msg = ("Unable to parse build option line [%s] into "
"compiler flag token and value" % line)
- print msg
+ print(msg)
else:
flag = define_tuple[0].strip()
allowed = define_tuple[2].strip()
@@ -89,22 +87,3 @@
elif expected_value == "0":
return True
return False
-
-
-class AsteriskBuildOptionsTests(unittest.TestCase):
- """Unit tests for AsteriskBuildOptions"""
-
- def test_1(self):
- """Test the defaults paths"""
- build_options = AsteriskBuildOptions()
- self.assertTrue(1)
-
-
-def main():
- """Main entry point for unit tests"""
-
- unittest.main()
-
-
-if __name__ == "__main__":
- main()
diff --git a/lib/python/asterisk/cdr.py b/lib/python/asterisk/cdr.py
index cc5226b..c5d62db 100644
--- a/lib/python/asterisk/cdr.py
+++ b/lib/python/asterisk/cdr.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Asterisk call detail record testing
This module implements an Asterisk CDR parser.
@@ -10,9 +9,8 @@
the GNU General Public License Version 2.
"""
-import unittest
import sys
-import astcsv
+from . import astcsv
import logging
LOGGER = logging.getLogger(__name__)
@@ -137,33 +135,5 @@
self, filename, records,
AsteriskCSVCDRLine.fields, AsteriskCSVCDRLine)
-
-class AsteriskCSVCDRTests(unittest.TestCase):
- """Unit tests for AsteriskCSVCDR"""
-
- def test_cdr(self):
- """Test the self_test/Master.csv record"""
-
- cdr = AsteriskCSVCDR("self_test/Master.csv")
- self.assertEqual(len(cdr), 2)
- self.assertTrue(
- AsteriskCSVCDRLine(duration=7, lastapp="hangup").match(
- cdr[0],
- exact=(True, True)))
- self.assertTrue(cdr[0].match(
- AsteriskCSVCDRLine(duration=7, lastapp="hangup"),
- exact=(True, True)))
-
- self.assertFalse(cdr[1].match(cdr[0]))
- self.assertFalse(cdr[0].match(cdr[1]))
- self.assertEqual(cdr[0].billsec, "7")
-
- self.assertTrue(cdr.match(cdr))
- cdr2 = AsteriskCSVCDR("self_test/Master2.csv")
- self.assertFalse(cdr.match(cdr2))
-
-
-if __name__ == '__main__':
- unittest.main()
# vim:sw=4:ts=4:expandtab:textwidth=79
diff --git a/lib/python/asterisk/cel.py b/lib/python/asterisk/cel.py
index d22d67f..9965270 100644
--- a/lib/python/asterisk/cel.py
+++ b/lib/python/asterisk/cel.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Asterisk call detail record testing
This module implements an Asterisk CEL parser.
@@ -11,8 +10,7 @@
"""
import yaml
-import unittest
-import astcsv
+from . import astcsv
import re
import logging
@@ -176,37 +174,5 @@
self, filename, records,
AsteriskCSVCELLine.fields, AsteriskCSVCELLine)
-
-class AsteriskCSVCELTests(unittest.TestCase):
- """Unit tests for AsteriskCSVCEL"""
-
- def test_cel(self):
- """Test CEL using self_test/CELMaster1.csv"""
-
- cel = AsteriskCSVCEL("self_test/CELMaster1.csv")
- self.assertEqual(len(cel), 16)
- self.assertTrue(AsteriskCSVCELLine(
- eventtype="LINKEDID_END",
- channel="TinCan/string").match(cel[-1],
- silent=True,
- exact=(True, True)))
- self.assertTrue(cel[-1].match(
- AsteriskCSVCELLine(eventtype="LINKEDID_END",
- channel="TinCan/string"),
- silent=True,
- exact=(True, True)))
-
- self.assertFalse(cel[1].match(cel[0], silent=True))
- self.assertFalse(cel[0].match(cel[1], silent=True))
- self.assertEqual(cel[-1].channel, "TinCan/string")
-
- self.assertTrue(cel.match(cel))
- cel2 = AsteriskCSVCEL("self_test/CELMaster2.csv")
- self.assertFalse(cel.match(cel2))
-
-
-if __name__ == '__main__':
- logging.basicConfig()
- unittest.main()
# vim:sw=4:ts=4:expandtab:textwidth=79
diff --git a/lib/python/asterisk/channel_test_condition.py b/lib/python/asterisk/channel_test_condition.py
index ec1d774..36421a4 100644
--- a/lib/python/asterisk/channel_test_condition.py
+++ b/lib/python/asterisk/channel_test_condition.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Test condition for channels
Copyright (C) 2011-2012, Digium, Inc.
@@ -9,9 +8,7 @@
"""
from twisted.internet import defer
-from test_conditions import TestCondition
-import logging
-import unittest
+from .test_conditions import TestCondition
import re
@@ -86,155 +83,3 @@
defer.DeferredList(exec_list).addCallback(_raise_finished,
finish_deferred)
return finish_deferred
-
-
-class AstMockOutput(object):
- """mock cli output base class"""
-
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.1"
-
- def MockDefer(self, output):
- """use real defer to mock deferred output"""
- self.output = output
- deferred = defer.Deferred()
- deferred.callback(self)
- return deferred
-
-
-class AstMockObjectInactive(AstMockOutput):
- """mock cli output showing no active channels"""
-
- def cli_exec(self, command):
- """presume command is core show channels and generate output"""
- output = ""
- output += "Channel Location State Application(Data)\n"
- output += "0 active channels\n"
- output += "0 active calls\n"
- output += "2 calls processed\n"
- output += "Asterisk ending (0).\n"
- return self.MockDefer(output)
-
-
-class AstMockObjectSingle(AstMockOutput):
- """mock cli output showing single active channel"""
-
- def cli_exec(self, command):
- """presume command is core show channels and generate output"""
- output = ""
- output += "Channel Location State Application(Data)\n"
- output += "Local/123 at default-00 (None) Down ()\n"
- output += "1 active channels\n"
- output += "0 active calls\n"
- output += "2 calls processed\n"
- output += "Asterisk ending (0).\n"
- return self.MockDefer(output)
-
-
-class AstMockObjectMultiple(AstMockOutput):
- """mock cli output showing multiple active channels"""
-
- def cli_exec(self, command):
- """presume command is core show channels and generate output"""
- output = ""
- output += "Channel Location State Application(Data)\n"
- output += "PJSIP/123 at default-00 (None) Down ()\n"
- output += "Local/123 at default-00 (None) Down ()\n"
- output += "SIP/alice at default-00 (None) Down ()\n"
- output += "3 active channels\n"
- output += "0 active calls\n"
- output += "2 calls processed\n"
- output += "Asterisk ending (0).\n"
- return self.MockDefer(output)
-
-
-class AstMockObjectLeaked(AstMockOutput):
- """mock cli output showing leaked channel"""
-
- def cli_exec(self, command):
- """presume command is core show channels and generate output"""
- output = ""
- output += "Channel Location State Application(Data)\n"
- output += "Local/123 at default-00 (None) Down ()\n"
- output += "0 active channels\n"
- output += "0 active calls\n"
- output += "2 calls processed\n"
- output += "Asterisk ending (0).\n"
- return self.MockDefer(output)
-
-
-class TestConfig(object):
- """Fake TestConfig object for unittest"""
-
- def __init__(self):
- self.class_type_name = "bogus"
- self.config = {}
- self.enabled = True
- self.pass_expected = True
-
-
-class ChannelTestConditionUnitTest(unittest.TestCase):
- """Unit Tests for ChannelTestCondition"""
-
- def test_evaluate_inactive(self):
- """test inactive channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.register_asterisk_instance(AstMockObjectInactive())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_evaluate_multiple_fail(self):
- """test multiple channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.register_asterisk_instance(AstMockObjectMultiple())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_evaluate_multiple_fail2(self):
- """test multiple channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.allowed_channels = 2
- obj.register_asterisk_instance(AstMockObjectMultiple())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_evaluate_multiple_pass(self):
- """test multiple channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.allowed_channels = 3
- obj.register_asterisk_instance(AstMockObjectMultiple())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_evaluate_single_fail(self):
- """test single channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.register_asterisk_instance(AstMockObjectSingle())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_evaluate_single_pass(self):
- """test single channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.allowed_channels = 1
- obj.register_asterisk_instance(AstMockObjectSingle())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_evaluate_leaked(self):
- """test leaked channel condition"""
- obj = ChannelTestCondition(TestConfig())
- obj.register_asterisk_instance(AstMockObjectLeaked())
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
-
-def main():
- """Run the unit tests"""
-
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
-
-if __name__ == "__main__":
- main()
diff --git a/lib/python/asterisk/confbridge.py b/lib/python/asterisk/confbridge.py
index acca1e7..0ca364d 100644
--- a/lib/python/asterisk/confbridge.py
+++ b/lib/python/asterisk/confbridge.py
@@ -14,7 +14,7 @@
import sys
import logging
-from test_state import TestState
+from .test_state import TestState
from twisted.internet import reactor
sys.path.append("lib/python")
diff --git a/lib/python/asterisk/config.py b/lib/python/asterisk/config.py
old mode 100755
new mode 100644
index e26b156..c5662d2
--- a/lib/python/asterisk/config.py
+++ b/lib/python/asterisk/config.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Asterisk Configuration File Handling.
This module implements interfaces for dealing with Asterisk configuration
@@ -18,7 +17,6 @@
import sys
import re
-import unittest
import logging
LOGGER = logging.getLogger(__name__)
@@ -54,7 +52,7 @@
match = self.varval_re.match(line)
if match is None:
if not is_blank_line(line):
- LOGGER.warn("Invalid line: '%s'" % line.strip())
+ LOGGER.warning("Invalid line: '%s'" % line.strip())
return
self.options.append((match.group("name"), match.group("value").strip()))
@@ -109,94 +107,6 @@
)
elif len(self.categories) == 0:
if not is_blank_line(line):
- LOGGER.warn("Invalid line: '%s'" % line.strip())
+ LOGGER.warning("Invalid line: '%s'" % line.strip())
else:
self.categories[-1].parse_line(line)
-
-
-class ConfigFileTests(unittest.TestCase):
- """Unit tests for ConfigFile"""
-
- def test_conf(self):
- """Test parsing a blob of config data"""
- test = \
- "; stuff\n" \
- "this line is invalid on purpose\n" \
- "[this is] also invalid]\n" \
- ";-- comment --;\n" \
- ";-- \n" \
- "[this is commented out]\n" \
- " --;\n" \
- "[foo]\n" \
- "a = b\n" \
- " b = a \n" \
- "this line is invalid on purpose\n" \
- ";moo\n" \
- "c = d;asdadf;adfads;adsfasdf\n" \
- " [bar] ;asdfasdf\n" \
- "a-b=c-d\n" \
- "xyz=x|y|z\n" \
- "1234 => 4242,Example Mailbox,root at localhost,,var=val\n" \
- "\n" \
- "[template](!)\n" \
- "foo=bar\n" \
- "exten => _NXX.,n,Wait(1)\n" \
- "astetcdir => /etc/asterisk\n"
-
- conf = ConfigFile(fn=None, config_str=test)
-
- self.assertEqual(len(conf.categories), 3)
-
- self.assertEqual(conf.categories[0].name, "foo")
- self.assertFalse(conf.categories[0].template)
- self.assertEqual(len(conf.categories[0].options), 3)
- self.assertEqual(conf.categories[0].options[0][0], "a")
- self.assertEqual(conf.categories[0].options[0][1], "b")
- self.assertEqual(conf.categories[0].options[1][0], "b")
- self.assertEqual(conf.categories[0].options[1][1], "a")
- self.assertEqual(conf.categories[0].options[2][0], "c")
- self.assertEqual(conf.categories[0].options[2][1], "d")
-
- self.assertEqual(conf.categories[1].name, "bar")
- self.assertFalse(conf.categories[1].template)
- self.assertEqual(len(conf.categories[1].options), 3)
- self.assertEqual(conf.categories[1].options[0][0], "a-b")
- self.assertEqual(conf.categories[1].options[0][1], "c-d")
- self.assertEqual(conf.categories[1].options[1][0], "xyz")
- self.assertEqual(conf.categories[1].options[1][1], "x|y|z")
- self.assertEqual(conf.categories[1].options[2][0], "1234")
- self.assertEqual(conf.categories[1].options[2][1],
- "4242,Example Mailbox,root at localhost,,var=val")
-
- self.assertEqual(conf.categories[2].name, "template")
- self.assertTrue(conf.categories[2].template)
- self.assertEqual(len(conf.categories[2].options), 3)
- self.assertEqual(conf.categories[2].options[0][0], "foo")
- self.assertEqual(conf.categories[2].options[0][1], "bar")
- self.assertEqual(conf.categories[2].options[1][0], "exten")
- self.assertEqual(conf.categories[2].options[1][1],
- "_NXX.,n,Wait(1)")
- self.assertEqual(conf.categories[2].options[2][0], "astetcdir")
- self.assertEqual(conf.categories[2].options[2][1], "/etc/asterisk")
-
-
-def main(argv=None):
- """Read in and show a config file, or run unit tests"""
-
- if argv is None:
- argv = sys.argv
-
- if len(argv) == 2:
- conf = ConfigFile(argv[1])
- for cat in conf.categories:
- LOGGER.debug("[%s]" % cat.name)
- for (var, val) in cat.options:
- LOGGER.debug("%s = %s" % (var, val))
- else:
- return unittest.main()
-
- return 0
-
-
-if __name__ == "__main__":
- sys.exit(main() or 0)
diff --git a/lib/python/asterisk/fd_test_condition.py b/lib/python/asterisk/fd_test_condition.py
index 4a8f869..234af28 100644
--- a/lib/python/asterisk/fd_test_condition.py
+++ b/lib/python/asterisk/fd_test_condition.py
@@ -11,7 +11,7 @@
import logging
from twisted.internet import defer
-from test_conditions import TestCondition
+from .test_conditions import TestCondition
LOGGER = logging.getLogger(__name__)
diff --git a/lib/python/asterisk/lock_test_condition.py b/lib/python/asterisk/lock_test_condition.py
index 0296693..805e2ff 100644
--- a/lib/python/asterisk/lock_test_condition.py
+++ b/lib/python/asterisk/lock_test_condition.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Held locks test condition
Copyright (C) 2011-2012, Digium, Inc.
@@ -10,10 +9,9 @@
import logging
import logging.config
-import unittest
from twisted.internet import defer
-from test_conditions import TestCondition
+from .test_conditions import TestCondition
LOGGER = logging.getLogger(__name__)
@@ -199,7 +197,7 @@
def evaluate(self, related_test_condition=None):
"""Evaluate the condition"""
- def __lock_info_obtained(finished_deferred):
+ def __lock_info_obtained(lst, finished_deferred):
"""Callback when lock information has been obtained"""
if (len(self.locks) > 0):
#Sometimes, a lock will be held at the end of a test run
@@ -229,314 +227,3 @@
finished_deferred)
return finished_deferred
-
-class AstMockObjectPassed(object):
- """A lock output that passed"""
-
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.2"
-
- def cli_exec(self, command, sync):
- """Fake out a CLI command execution"""
- lock_lines = "=======================================================================\n"
- lock_lines += "=== Currently Held Locks ==============================================\n"
- lock_lines += "=======================================================================\n"
- lock_lines += "===\n"
- lock_lines += "=== <pending> <lock#> (<file>): <lock type> <line num> <function> <lock name> <lock addr> (times locked)\n"
- lock_lines += "===\n"
- lock_lines += "===\n"
- lock_lines += "=======================================================================\n"
- return lock_lines
-
-
-class AstMockObjectFailure(object):
- """A lock object that failed"""
-
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.1"
-
- def cli_exec(self, command, sync):
- """Fake out a CLI command execution"""
- lock_lines = "=======================================================================\n"
- lock_lines += "=== Currently Held Locks ==============================================\n"
- lock_lines += "=======================================================================\n"
- lock_lines += "===\n"
- lock_lines += "=== <pending> <lock#> (<file>): <lock type> <line num> <function> <lock name> <lock addr> (times locked)\n"
- lock_lines += "===\n"
- lock_lines += "=== Thread ID: 0x402c6940 (do_monitor started at [25114] chan_sip.c restart_monitor())\n"
- lock_lines += "=== ---> Lock #0 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe423ee8]\n"
- lock_lines += "=== ---> Lock #1 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)\n"
- lock_lines += "=== --- ---> Locked Here: channel.c line 4304 (ast_indicate_data)\n"
- lock_lines += "=== -------------------------------------------------------------------\n"
- lock_lines += "===\n"
- lock_lines += "=== Thread ID: 0x449ec940 (netconsole started at [ 1351] asterisk.c listener())\n"
- lock_lines += "=== ---> Waiting for Lock #0 (astobj2.c): MUTEX 842 internal_ao2_iterator_next a->c 0x2aaaac491f50 (1)\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x446cec]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_iterator_next+0x29) [0x447134]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_iterator_next+0x19) [0x46cf7d]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x489e43]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_full+0x222) [0x48eec4]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_multiple_full+0x92) [0x48f035]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x43d129]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]\n"
- lock_lines += "/lib64/libpthread.so.0 [0x3d1d80673d]\n"
- lock_lines += "/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]\n"
- lock_lines += "=== --- ---> Locked Here: astobj2.c line 657 (internal_ao2_callback)\n"
- lock_lines += "=== -------------------------------------------------------------------\n"
- lock_lines += "===\n"
- lock_lines += "=== Thread ID: 0x44a68940 (netconsole started at [ 1351] asterisk.c listener())\n"
- lock_lines += "=== ---> Waiting for Lock #0 (astobj2.c): MUTEX 842 internal_ao2_iterator_next a->c 0x2aaaac491f50 (1)\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x446cec]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_iterator_next+0x29) [0x447134]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_iterator_next+0x19) [0x46cf7d]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x489e43]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_full+0x222) [0x48eec4]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_multiple_full+0x92) [0x48f035]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x43d129]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]\n"
- lock_lines += "/lib64/libpthread.so.0 [0x3d1d80673d]\n"
- lock_lines += "/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]\n"
- lock_lines += "=== --- ---> Locked Here: astobj2.c line 657 (internal_ao2_callback)\n"
- lock_lines += "=== -------------------------------------------------------------------\n"
- lock_lines += "===\n"
- lock_lines += "=======================================================================\n"
- return lock_lines
-
-
-class TestConfig(object):
- """Fake TestConfig object"""
-
- def __init__(self):
- """ Values here don't matter much - we just need to have something """
- self.class_type_name = "asterisk.LockTestCondition.LockTestCondition"
- self.pass_expected = True
- self.type = "Post"
- self.related_condition = ""
- self.config = {}
-
-
-class LockTestConditionUnitTest(unittest.TestCase):
- """Unit tests for LockTestCondition"""
-
- def test_evaluate_failed(self):
- """Test a failed locking condition"""
- ast = AstMockObjectFailure()
- obj = LockTestCondition(TestConfig())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_evaluate_pass(self):
- """Test a passed locking condition"""
- ast = AstMockObjectPassed()
- obj = LockTestCondition(TestConfig())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_evaluate_multiple(self):
- """Test multiple results"""
- ast1 = AstMockObjectPassed()
- ast2 = AstMockObjectFailure()
- obj = LockTestCondition(TestConfig())
- obj.register_asterisk_instance(ast1)
- obj.register_asterisk_instance(ast2)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
-
-class LockSequenceUnitTest(unittest.TestCase):
- """Tests for parsing a lock sequence"""
-
- def test_single_object_no_held_info(self):
- """Test a lock sequence with no waiting lock"""
-
- lock_lines = "=== Thread ID: 0x7f668c142700 (do_monitor started at [25915] chan_sip.c restart_monitor())\n"
- lock_lines += "=== ---> Lock #0 (chan_sip.c): MUTEX 25390 handle_request_do &netlock 0x7f6652193900 (1)\n"
- lock_lines += "main/logger.c:1302 ast_bt_get_addresses() (0x505e53+1D)\n"
- lock_lines += "main/lock.c:193 __ast_pthread_mutex_lock() (0x4fe55c+D9)\n"
- lock_lines += "channels/chan_sip.c:25393 handle_request_do()\n"
- lock_lines += "channels/chan_sip.c:25352 sipsock_read()\n"
- lock_lines += "main/io.c:288 ast_io_wait() (0x4f8228+19C)\n"
- lock_lines += "channels/chan_sip.c:25882 do_monitor()\n"
- lock_lines += "main/utils.c:1010 dummy_start()\n"
- lock_lines += "libpthread.so.0 <unknown>()\n"
- lock_lines += "libc.so.6 clone() (0x31be0e0bc0+6D)\n"
-
- obj = LockSequence()
- obj.parse_lock_sequence(lock_lines)
-
- def test_large_multiple_object(self):
- """Test with a waiting lock"""
-
- lock_lines = "=== Thread ID: 0x449ec940 (netconsole started at [ 1351] asterisk.c listener())\n"
- lock_lines += "=== ---> Waiting for Lock #0 (astobj2.c): MUTEX 842 internal_ao2_iterator_next a->c 0x2aaaac491f50 (1)\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x446cec]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_iterator_next+0x29) [0x447134]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_iterator_next+0x19) [0x46cf7d]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x489e43]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_full+0x222) [0x48eec4]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_multiple_full+0x92) [0x48f035]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x43d129]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]\n"
- lock_lines += "/lib64/libpthread.so.0 [0x3d1d80673d]\n"
- lock_lines += "/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]\n"
- lock_lines += "=== --- ---> Locked Here: astobj2.c line 657 (internal_ao2_callback)\n"
-
- obj = LockSequence()
- obj.parse_lock_sequence(lock_lines)
- self.assertEqual(obj.thread_id, "0x449ec940")
- self.assertEqual(obj.thread_name, "netconsole")
- self.assertEqual(obj.thread_line, 1351)
- self.assertEqual(obj.thread_file, "asterisk.c")
- self.assertEqual(obj.thread_func, "listener")
- self.assertTrue(len(obj.locks) == 1)
- self.assertEqual(obj.locks[0].locked_file, "astobj2.c")
- self.assertEqual(obj.locks[0].locked_line, 657)
- self.assertEqual(obj.locks[0].locked_func, "internal_ao2_callback")
-
- def test_single_object(self):
- """Test a lock held somewhere else"""
-
- lock_lines = "=== Thread ID: 0x402c6940 (do_monitor started at [25114] chan_sip.c restart_monitor())\n"
- lock_lines += "=== ---> Lock #0 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe423ee8]\n"
- lock_lines += "=== --- ---> Locked Here: channel.c line 4304 (ast_indicate_data)\n"
-
- obj = LockSequence()
- obj.parse_lock_sequence(lock_lines)
- self.assertEqual(obj.thread_id, "0x402c6940")
- self.assertEqual(obj.thread_name, "do_monitor")
- self.assertEqual(obj.thread_line, 25114)
- self.assertEqual(obj.thread_file, "chan_sip.c")
- self.assertEqual(obj.thread_func, "restart_monitor")
- self.assertTrue(len(obj.locks) == 1)
- self.assertEqual(obj.locks[0].locked_file, "channel.c")
- self.assertEqual(obj.locks[0].locked_line, 4304)
- self.assertEqual(obj.locks[0].locked_func, "ast_indicate_data")
- self.assertEqual(obj.locks[0].id, 0)
- self.assertEqual(obj.locks[0].type, "MUTEX")
- self.assertEqual(obj.locks[0].file, "chan_sip.c")
- self.assertEqual(obj.locks[0].line_number, 24629)
- self.assertEqual(obj.locks[0].func, "handle_request_do")
- self.assertEqual(obj.locks[0].name, "&netlock")
- self.assertEqual(obj.locks[0].addr, "0x2aaabe671a40")
- self.assertEqual(obj.locks[0].lock_count, 1)
- self.assertTrue(obj.locks[0].held)
- self.assertTrue(len(obj.locks[0].backtrace) == 3)
-
- def test_multiple_objects_no_backtrace(self):
- """Test multiple locks with no backtrace"""
-
- lock_lines = "=== Thread ID: 0x402c6940 (do_monitor started at [25114] chan_sip.c restart_monitor())\n"
- lock_lines += "=== ---> Lock #0 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)\n"
- lock_lines += "=== ---> Lock #1 (astobj2.c): MUTEX 657 internal_ao2_callback c 0x2aaaac491f50 (1)\n"
- lock_lines += "=== ---> Waiting for Lock #2 (channel.c): MUTEX 1691 ast_channel_cmp_cb chan 0x2aaaacd3a4e0 (1)\n"
- lock_lines += "=== --- ---> Locked Here: channel.c line 4304 (ast_indicate_data)\n"
-
- obj = LockSequence()
- obj.parse_lock_sequence(lock_lines)
- self.assertEqual(obj.thread_id, "0x402c6940")
- self.assertEqual(obj.thread_name, "do_monitor")
- self.assertEqual(obj.thread_line, 25114)
- self.assertEqual(obj.thread_file, "chan_sip.c")
- self.assertEqual(obj.thread_func, "restart_monitor")
- self.assertTrue(len(obj.locks) == 3)
- self.assertEqual(obj.locks[2].locked_file, "channel.c")
- self.assertEqual(obj.locks[2].locked_line, 4304)
- self.assertEqual(obj.locks[2].locked_func, "ast_indicate_data")
- self.assertEqual(obj.locks[0].id, 0)
- self.assertEqual(obj.locks[0].type, "MUTEX")
- self.assertEqual(obj.locks[0].file, "chan_sip.c")
- self.assertEqual(obj.locks[0].line_number, 24629)
- self.assertEqual(obj.locks[0].func, "handle_request_do")
- self.assertEqual(obj.locks[0].name, "&netlock")
- self.assertEqual(obj.locks[0].addr, "0x2aaabe671a40")
- self.assertEqual(obj.locks[0].lock_count, 1)
- self.assertTrue(obj.locks[0].held)
- self.assertTrue(len(obj.locks[0].backtrace) == 0)
-
-
-class LockObjectUnitTest(unittest.TestCase):
- """Unit tests for LockObject"""
-
- def test_no_backtrace(self):
- """Test creating a lock object with no thread backtrace"""
-
- lock_line = "=== ---> Waiting for Lock #0 (sig_ss7.c): MUTEX 636 ss7_linkset &linkset->lock 0x2aaab8a6b588 (1)"
- obj = LockObject()
- obj.parse_lock_information(lock_line)
- self.assertEqual(obj.id, 0)
- self.assertEqual(obj.type, "MUTEX")
- self.assertEqual(obj.file, "sig_ss7.c")
- self.assertEqual(obj.line_number, 636)
- self.assertEqual(obj.func, "ss7_linkset")
- self.assertEqual(obj.name, "&linkset->lock")
- self.assertEqual(obj.addr, "0x2aaab8a6b588")
- self.assertEqual(obj.lock_count, 1)
- self.assertFalse(obj.held)
- self.assertTrue(len(obj.backtrace) == 0)
-
- def test_with_backtrace(self):
- """Test creating a lock object with a thread backtrace"""
-
- lock_lines = "=== ---> Lock #1 (astobj2.c): MUTEX 657 internal_ao2_callback c 0x2aaaac491f50 (1)\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x4464be]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_callback+0x59) [0x446a4e]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_find+0x2b) [0x446ba7]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x46d3a7]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_get_by_name+0x24) [0x46d3e3]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/func_channel.so [0x2aaabfba2468]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_func_write+0x16a) [0x50aacd]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(pbx_builtin_setvar_helper+0x10e) [0x51fff4]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe422d09]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe4240a0]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe423cf1]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_io_wait+0x1ba) [0x4dc2e4]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe425722]\n"
- lock_lines += "/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]\n"
- lock_lines += "/lib64/libpthread.so.0 [0x3d1d80673d]\n"
- lock_lines += "/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]\n"
-
- obj = LockObject()
- obj.parse_lock_information(lock_lines)
- self.assertEqual(obj.id, 1)
- self.assertEqual(obj.type, "MUTEX")
- self.assertEqual(obj.file, "astobj2.c")
- self.assertEqual(obj.line_number, 657)
- self.assertEqual(obj.func, "internal_ao2_callback")
- self.assertEqual(obj.name, "c")
- self.assertEqual(obj.addr, "0x2aaaac491f50")
- self.assertEqual(obj.lock_count, 1)
- self.assertTrue(obj.held)
- self.assertTrue(len(obj.backtrace) == 19)
-
-
-def main():
- """Run the unit tests"""
-
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
-
-
-if __name__ == "__main__":
- main()
diff --git a/lib/python/asterisk/matcher.py b/lib/python/asterisk/matcher.py
index 3ab9c24..8ef575e 100644
--- a/lib/python/asterisk/matcher.py
+++ b/lib/python/asterisk/matcher.py
@@ -9,13 +9,17 @@
"""
import logging
+import sys
-from test_suite_utils import all_match
-from pluggable_registry import PLUGGABLE_EVENT_REGISTRY,\
- PLUGGABLE_ACTION_REGISTRY
+from .test_runner import load_and_parse_module
+from .test_suite_utils import all_match
+from .pluggable_registry import PLUGGABLE_EVENT_REGISTRY
LOGGER = logging.getLogger(__name__)
+
+if sys.version_info[0] == 3:
+ unicode = str
class ConditionError(Exception):
@@ -217,7 +221,6 @@
matched.append(c)
-
if not matched:
return False
@@ -315,13 +318,9 @@
self.triggered_callback = triggered_callback
- module_name, _, obj_type = config['type'].partition('.')
-
- module = __import__(module_name, fromlist=[obj_type])
- if not module:
- raise Exception("Unable to import module '{0}'.".format(module_name))
-
- obj = getattr(module, obj_type)
+ obj = load_and_parse_module(config['type'])
+ if not obj:
+ raise Exception("Unable to import module '{0}'.".format(config['type']))
self.conditions = obj(config, test_object, self.__handle_match)
diff --git a/lib/python/asterisk/matcher_listener.py b/lib/python/asterisk/matcher_listener.py
index d6b621d..cd2ca5a 100644
--- a/lib/python/asterisk/matcher_listener.py
+++ b/lib/python/asterisk/matcher_listener.py
@@ -13,7 +13,7 @@
from twisted.internet.protocol import DatagramProtocol
from twisted.internet import reactor
-from matcher import PluggableConditions
+from .matcher import PluggableConditions
LOGGER = logging.getLogger(__name__)
diff --git a/lib/python/asterisk/opensslversion.py b/lib/python/asterisk/opensslversion.py
index a7377ea..3fa5cd5 100644
--- a/lib/python/asterisk/opensslversion.py
+++ b/lib/python/asterisk/opensslversion.py
@@ -9,11 +9,10 @@
"""
import sys
-import unittest
import re
sys.path.append("lib/python")
-import test_suite_utils
+from . import test_suite_utils
class OpenSSLVersion:
"""An OpenSSL Version.
diff --git a/lib/python/asterisk/originate.py b/lib/python/asterisk/originate.py
index e6befb5..6c87a57 100644
--- a/lib/python/asterisk/originate.py
+++ b/lib/python/asterisk/originate.py
@@ -16,7 +16,7 @@
import json
import requests
-import ari
+from . import ari
LOGGER = logging.getLogger(__name__)
diff --git a/lib/python/asterisk/pcap.py b/lib/python/asterisk/pcap.py
index d39574a..cb5fa5c 100644
--- a/lib/python/asterisk/pcap.py
+++ b/lib/python/asterisk/pcap.py
@@ -739,13 +739,14 @@
self.callbacks = {}
self.traces = {}
- def process_packet(self, packet, (host, port)):
+ def process_packet(self, packet, addr):
"""Store a known packet in our traces and call our callbacks
Keyword Arguments:
packet A raw packet received from ... something.
- (host, port) Tuple of received host and port
+ addr Tuple of received host and port
"""
+ (host, port) = addr
packet = self.packet_factory.interpret_packet(packet)
if packet is None:
return
@@ -812,13 +813,14 @@
self.rules = rules
self.cb = cb
- def datagramReceived(self, data, (host, port)):
+ def datagramReceived(self, data, addr):
"""Callback for when a datagram is received
Keyword Arguments:
data The actual packet
- (host, port) Tuple of source host and port
+ addr Tuple of received host and port
"""
+ (host, port) = addr
LOGGER.debug('Proxy received from {0}:{1}\n{2}'.format(
host, port, data))
diff --git a/lib/python/asterisk/phones.py b/lib/python/asterisk/phones.py
old mode 100755
new mode 100644
index 6a96b38..4987da5
--- a/lib/python/asterisk/phones.py
+++ b/lib/python/asterisk/phones.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Pluggable modules and classes to simulate phones.
Copyright (C) 2015, Digium, Inc.
@@ -85,7 +84,7 @@
def __setup_pjsua_acc_cb(self):
"""Setup PJSUA account callbacks"""
- for name, phone_obj in self.__pjsua_phones.iteritems():
+ for name, phone_obj in self.__pjsua_phones.items():
acc_cb = AccCallback()
phone_obj.account.set_callback(acc_cb)
LOGGER.info("%s is ready to receive calls." % name)
@@ -106,7 +105,7 @@
if name:
return self.__pjsua_phones.get(name)
if account:
- for name, phone_obj in self.__pjsua_phones.iteritems():
+ for name, phone_obj in self.__pjsua_phones.items():
if account is phone_obj.account:
return phone_obj
diff --git a/lib/python/asterisk/pjsua_mod.py b/lib/python/asterisk/pjsua_mod.py
index 63add6a..cdf3d18 100644
--- a/lib/python/asterisk/pjsua_mod.py
+++ b/lib/python/asterisk/pjsua_mod.py
@@ -17,6 +17,7 @@
sys.path.append("lib/python")
from twisted.internet import reactor
+from .test_runner import load_and_parse_module
LOGGER = logging.getLogger(__name__)
@@ -147,7 +148,7 @@
self.lib.set_null_snd_dev()
self.__create_accounts()
self.lib.start()
- except pj.Error, exception:
+ except pj.Error as exception:
LOGGER.error("Exception: " + str(exception))
self.lib.destroy()
self.lib = None
@@ -306,6 +307,5 @@
def do_callback(self):
"""Call the configured callback module/method"""
- callback_module = __import__(self.callback_module)
- callback_method = getattr(callback_module, self.callback_method)
+ callback_method = load_and_parse_module(self.callback_module + '.' + self.callback_method)
callback_method(self.test_object, self.pj_accounts)
diff --git a/lib/python/asterisk/pluggable_modules.py b/lib/python/asterisk/pluggable_modules.py
old mode 100755
new mode 100644
index 344f908..dddb036
--- a/lib/python/asterisk/pluggable_modules.py
+++ b/lib/python/asterisk/pluggable_modules.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Generic pluggable modules
Copyright (C) 2012, Digium, Inc.
@@ -14,15 +13,15 @@
import re
sys.path.append("lib/python")
-from ami import AMIEventInstance
+from .ami import AMIEventInstance
from twisted.internet import reactor
from starpy import fastagi
-from test_runner import load_and_parse_module
-from pluggable_registry import PLUGGABLE_ACTION_REGISTRY,\
+from .test_runner import load_and_parse_module
+from .pluggable_registry import PLUGGABLE_ACTION_REGISTRY,\
PLUGGABLE_EVENT_REGISTRY,\
PluggableRegistry
-import matcher
+from . import matcher
LOGGER = logging.getLogger(__name__)
@@ -576,7 +575,7 @@
return
current_trigger = config['trigger']['match']
- for key, value in current_trigger.iteritems():
+ for key, value in current_trigger.items():
if key.lower() not in event:
LOGGER.debug("Condition %s not in event, returning", key)
return
@@ -701,8 +700,7 @@
if self.commands:
return self.execute_command(agi, 0)
else:
- callback_module = __import__(self.callback_module)
- method = getattr(callback_module, self.callback_method)
+ method = load_and_parse_module(self.callback_module + '.' + self.callback_method)
method(self.test_object, agi)
def on_command_failure(self, reason, agi, idx):
@@ -784,7 +782,7 @@
def register_modules(config, registry):
"""Register pluggable modules into the registry"""
- for key, local_class_path in config.iteritems():
+ for key, local_class_path in config.items():
local_class = load_and_parse_module(local_class_path)
if not local_class:
raise Exception("Unable to load %s for module key %s"
@@ -804,7 +802,7 @@
for e_a_set in config["mapping"]:
plug_set = {"events": [], "actions": []}
- for plug_name, plug_config in e_a_set.iteritems():
+ for plug_name, plug_config in e_a_set.items():
self.parse_module_config(plug_set, plug_name, plug_config)
if 0 == len(plug_set["events"]):
@@ -923,8 +921,7 @@
def run(self, triggered_by, source, extra):
"""Call the callback."""
- module = __import__(self.module)
- method = getattr(module, self.method)
+ method = load_and_parse_module(self.module + '.' + self.method)
self.test_object.set_passed(method(self.test_object, triggered_by,
source, extra))
PLUGGABLE_ACTION_REGISTRY.register("callback", CallbackActionModule)
@@ -960,12 +957,12 @@
def __init__(self, test_object, config):
"""Setup the test start observer"""
self.test_object = test_object
- self.module = __import__("phones")
+ self.module = "phones"
self.method = config["action"]
self.config = config
def run(self, triggered_by, source, extra):
"""Instruct phone to perform action"""
- method = getattr(self.module, self.method)
+ method = load_and_parse_module(self.module + "." + self.method)
method(self.test_object, triggered_by, source, extra, self.config)
PLUGGABLE_ACTION_REGISTRY.register("pjsua_phone", PjsuaPhoneActionModule)
diff --git a/lib/python/asterisk/pluggable_registry.py b/lib/python/asterisk/pluggable_registry.py
old mode 100755
new mode 100644
index 8169795..9a50a26
--- a/lib/python/asterisk/pluggable_registry.py
+++ b/lib/python/asterisk/pluggable_registry.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Pluggable module registries
Copyright (C) 2014, Digium, Inc.
diff --git a/lib/python/asterisk/realtime_converter.py b/lib/python/asterisk/realtime_converter.py
index 3f50b5f..95ae579 100644
--- a/lib/python/asterisk/realtime_converter.py
+++ b/lib/python/asterisk/realtime_converter.py
@@ -9,7 +9,7 @@
import os
from sqlalchemy import create_engine, MetaData, Table
-import astconfigparser
+from . import astconfigparser
import logging
LOGGER = logging.getLogger(__name__)
@@ -83,8 +83,8 @@
self.filename = filename
self.sections = sections
# All affected database tables in list form. Used for convenience
- self.tables = [table for section in sections.itervalues() for table in
- section.itervalues()]
+ self.tables = [table for section in sections.values() for table in
+ section.values()]
self.sorcery = None
self.extconfig = None
@@ -111,9 +111,9 @@
config_dir: The directory where Asterisk configuration can be found
"""
with open(self.sorcery.file, 'a') as sorcery:
- for section, items in self.sections.iteritems():
+ for section, items in self.sections.items():
sorcery.write('[{0}]\n'.format(section))
- for obj, table in items.iteritems():
+ for obj, table in items.items():
sorcery.write('{0} = realtime,{1}\n'.format(obj, table))
def write_extconfig_conf(self):
@@ -141,7 +141,7 @@
"""
conf = astconfigparser.MultiOrderedConfigParser()
conf.read(os.path.join(config_dir, self.filename))
- for title, sections in conf.sections().iteritems():
+ for title, sections in conf.sections().items():
LOGGER.info("Inspecting objects with title {0}".format(title))
for section in sections:
obj_type = section.get('type')[0]
@@ -165,7 +165,7 @@
Keyword Arguments:
obj_type: The object type to find the section for
"""
- for section, contents in self.sections.iteritems():
+ for section, contents in self.sections.items():
if obj_type in contents:
return section
diff --git a/lib/python/asterisk/realtime_odbc_module.py b/lib/python/asterisk/realtime_odbc_module.py
index 3653553..bbd8a8e 100644
--- a/lib/python/asterisk/realtime_odbc_module.py
+++ b/lib/python/asterisk/realtime_odbc_module.py
@@ -53,7 +53,7 @@
self.res_odbc = {}
# generate configuration for each dsn
- for dsn, config in module_config.iteritems():
+ for dsn, config in module_config.items():
self._configure(dsn, config)
# set the odbc and conf files
@@ -109,7 +109,7 @@
with open(filepath, 'w') as filehandle:
for section in contents:
filehandle.write('[' + section + ']\n')
- for name, value in contents[section].iteritems():
+ for name, value in contents[section].items():
filehandle.write(name + '=' + value + '\n')
def _read_ini_file(self, filepath):
diff --git a/lib/python/asterisk/realtime_test_module.py b/lib/python/asterisk/realtime_test_module.py
index fd38aa9..da97422 100644
--- a/lib/python/asterisk/realtime_test_module.py
+++ b/lib/python/asterisk/realtime_test_module.py
@@ -91,7 +91,7 @@
return [row for row in table
if all(key in row and re.match(value, row[key])
- for key, value in where.iteritems())]
+ for key, value in where.items())]
def retrieve_rows(self, table_name, where):
"""Retrieve multiple rows from a table.
@@ -239,7 +239,7 @@
since we could use a dict comprehension.
"""
filtered_args = {}
- for key, values in args.iteritems():
+ for key, values in args.items():
if " LIKE" in key:
# Strip away " LIKE" and % from values
filtered_args[key[:-5]] = [val.replace('%', '.*')
@@ -250,7 +250,7 @@
LOGGER.debug('filtered args is %s' % filtered_args)
return dict((key, values[0] if values else '.*') for key, values in
- filtered_args.iteritems())
+ filtered_args.items())
def encode_row(self, row):
"""Encode a retrieved row for an HTTP response.
@@ -261,7 +261,7 @@
Example output: 'foo=cat&bar=dog&baz=donkey'
"""
string = '&'.join(['{0}={1}'.format(cgi.escape(key), cgi.escape(val))
- for key, val in row.iteritems()])
+ for key, val in row.items()])
LOGGER.debug("Returning response %s" % string)
return string
@@ -481,7 +481,7 @@
if not data:
return
- for table_name, rows in data.iteritems():
+ for table_name, rows in data.items():
self.rt_data.add_rows(table_name, rows)
def setup_http(self):
diff --git a/lib/python/asterisk/self_test/harness_shared.py b/lib/python/asterisk/self_test/harness_shared.py
new file mode 100644
index 0000000..91449ac
--- /dev/null
+++ b/lib/python/asterisk/self_test/harness_shared.py
@@ -0,0 +1,54 @@
+"""Unit test harness
+
+This module provides the entry-point for tests
+
+Copyright (C) 2018, CFWare, LLC.
+Corey Farrell <git at cfware.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+import logging
+import os
+import sys
+from twisted.internet import defer
+import unittest
+
+# Add directory where the modules to test can be found
+sys.path.append('lib/python')
+
+
+def ReadTestFile(filename, basepath="lib/python/asterisk/self_test"):
+ fd = open(os.path.join(basepath, filename), "r")
+ output = fd.read()
+ fd.close()
+ return output
+
+
+class AstMockOutput(object):
+ """mock cli output base class"""
+
+ def __init__(self, host="127.0.0.1"):
+ """Constructor"""
+ self.host = host
+
+ def MockDeferFile(self, filename):
+ return self.MockDefer(ReadTestFile(filename))
+
+ def MockDefer(self, output):
+ """use real defer to mock deferred output"""
+ self.output = output
+ deferred = defer.Deferred()
+ deferred.callback(self)
+ return deferred
+
+
+def main():
+ """Run the unit tests"""
+
+ logging.basicConfig()
+ unittest.main()
+
+
+__all__ = ["main", "AstMockOutput", "ReadTestFile"]
diff --git a/lib/python/asterisk/self_test/locks-backtrace.txt b/lib/python/asterisk/self_test/locks-backtrace.txt
new file mode 100644
index 0000000..0de89a4
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-backtrace.txt
@@ -0,0 +1,20 @@
+=== ---> Lock #1 (astobj2.c): MUTEX 657 internal_ao2_callback c 0x2aaaac491f50 (1)
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x4464be]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_callback+0x59) [0x446a4e]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_find+0x2b) [0x446ba7]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x46d3a7]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_get_by_name+0x24) [0x46d3e3]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/func_channel.so [0x2aaabfba2468]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_func_write+0x16a) [0x50aacd]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(pbx_builtin_setvar_helper+0x10e) [0x51fff4]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe422d09]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe4240a0]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe423cf1]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_io_wait+0x1ba) [0x4dc2e4]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe425722]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]
+/lib64/libpthread.so.0 [0x3d1d80673d]
+/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]
diff --git a/lib/python/asterisk/self_test/locks-fail.txt b/lib/python/asterisk/self_test/locks-fail.txt
new file mode 100644
index 0000000..fb50ca8
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-fail.txt
@@ -0,0 +1,52 @@
+=======================================================================
+=== Currently Held Locks ==============================================
+=======================================================================
+===
+=== <pending> <lock#> (<file>): <lock type> <line num> <function> <lock name> <lock addr> (times locked)
+===
+=== Thread ID: 0x402c6940 (do_monitor started at [25114] chan_sip.c restart_monitor())
+=== ---> Lock #0 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe423ee8]
+=== ---> Lock #1 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)
+=== --- ---> Locked Here: channel.c line 4304 (ast_indicate_data)
+=== -------------------------------------------------------------------
+===
+=== Thread ID: 0x449ec940 (netconsole started at [ 1351] asterisk.c listener())
+=== ---> Waiting for Lock #0 (astobj2.c): MUTEX 842 internal_ao2_iterator_next a->c 0x2aaaac491f50 (1)
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x446cec]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_iterator_next+0x29) [0x447134]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_iterator_next+0x19) [0x46cf7d]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x489e43]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_full+0x222) [0x48eec4]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_multiple_full+0x92) [0x48f035]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x43d129]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]
+/lib64/libpthread.so.0 [0x3d1d80673d]
+/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]
+=== --- ---> Locked Here: astobj2.c line 657 (internal_ao2_callback)
+=== -------------------------------------------------------------------
+===
+=== Thread ID: 0x44a68940 (netconsole started at [ 1351] asterisk.c listener())
+=== ---> Waiting for Lock #0 (astobj2.c): MUTEX 842 internal_ao2_iterator_next a->c 0x2aaaac491f50 (1)
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x446cec]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_iterator_next+0x29) [0x447134]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_iterator_next+0x19) [0x46cf7d]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x489e43]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_full+0x222) [0x48eec4]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_multiple_full+0x92) [0x48f035]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x43d129]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]
+/lib64/libpthread.so.0 [0x3d1d80673d]
+/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]
+=== --- ---> Locked Here: astobj2.c line 657 (internal_ao2_callback)
+=== -------------------------------------------------------------------
+===
+=======================================================================
diff --git a/lib/python/asterisk/self_test/locks-large-multiple-object.txt b/lib/python/asterisk/self_test/locks-large-multiple-object.txt
new file mode 100644
index 0000000..ea657b6
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-large-multiple-object.txt
@@ -0,0 +1,16 @@
+=== Thread ID: 0x449ec940 (netconsole started at [ 1351] asterisk.c listener())
+=== ---> Waiting for Lock #0 (astobj2.c): MUTEX 842 internal_ao2_iterator_next a->c 0x2aaaac491f50 (1)
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_lock+0x53) [0x4456fc]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x446cec]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ao2_iterator_next+0x29) [0x447134]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_channel_iterator_next+0x19) [0x46cf7d]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x489e43]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_full+0x222) [0x48eec4]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_cli_command_multiple_full+0x92) [0x48f035]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x43d129]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk [0x5661c6]
+/lib64/libpthread.so.0 [0x3d1d80673d]
+/lib64/libc.so.6(clone+0x6d) [0x3d1ccd44bd]
+=== --- ---> Locked Here: astobj2.c line 657 (internal_ao2_callback)
diff --git a/lib/python/asterisk/self_test/locks-multiple-objects-no-backtrace.txt b/lib/python/asterisk/self_test/locks-multiple-objects-no-backtrace.txt
new file mode 100644
index 0000000..3ed0ca5
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-multiple-objects-no-backtrace.txt
@@ -0,0 +1,5 @@
+=== Thread ID: 0x402c6940 (do_monitor started at [25114] chan_sip.c restart_monitor())
+=== ---> Lock #0 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)
+=== ---> Lock #1 (astobj2.c): MUTEX 657 internal_ao2_callback c 0x2aaaac491f50 (1)
+=== ---> Waiting for Lock #2 (channel.c): MUTEX 1691 ast_channel_cmp_cb chan 0x2aaaacd3a4e0 (1)
+=== --- ---> Locked Here: channel.c line 4304 (ast_indicate_data)
diff --git a/lib/python/asterisk/self_test/locks-pass.txt b/lib/python/asterisk/self_test/locks-pass.txt
new file mode 100644
index 0000000..bbb3b92
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-pass.txt
@@ -0,0 +1,8 @@
+=======================================================================
+=== Currently Held Locks ==============================================
+=======================================================================
+===
+=== <pending> <lock#> (<file>): <lock type> <line num> <function> <lock name> <lock addr> (times locked)
+===
+===
+=======================================================================
diff --git a/lib/python/asterisk/self_test/locks-single-object-no-held-info.txt b/lib/python/asterisk/self_test/locks-single-object-no-held-info.txt
new file mode 100644
index 0000000..e1bf47a
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-single-object-no-held-info.txt
@@ -0,0 +1,11 @@
+=== Thread ID: 0x7f668c142700 (do_monitor started at [25915] chan_sip.c restart_monitor())
+=== ---> Lock #0 (chan_sip.c): MUTEX 25390 handle_request_do &netlock 0x7f6652193900 (1)
+main/logger.c:1302 ast_bt_get_addresses() (0x505e53+1D)
+main/lock.c:193 __ast_pthread_mutex_lock() (0x4fe55c+D9)
+channels/chan_sip.c:25393 handle_request_do()
+channels/chan_sip.c:25352 sipsock_read()
+main/io.c:288 ast_io_wait() (0x4f8228+19C)
+channels/chan_sip.c:25882 do_monitor()
+main/utils.c:1010 dummy_start()
+libpthread.so.0 <unknown>()
+libc.so.6 clone() (0x31be0e0bc0+6D)
diff --git a/lib/python/asterisk/self_test/locks-single-object.txt b/lib/python/asterisk/self_test/locks-single-object.txt
new file mode 100644
index 0000000..db356ed
--- /dev/null
+++ b/lib/python/asterisk/self_test/locks-single-object.txt
@@ -0,0 +1,6 @@
+=== Thread ID: 0x402c6940 (do_monitor started at [25114] chan_sip.c restart_monitor())
+=== ---> Lock #0 (chan_sip.c): MUTEX 24629 handle_request_do &netlock 0x2aaabe671a40 (1)
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(ast_bt_get_addresses+0x1a) [0x4e9679]
+/usr/local/asterisk-1.8.6.0/sbin/asterisk(__ast_pthread_mutex_lock+0xf6) [0x4e22d9]
+/usr/local/asterisk-1.8.6.0/lib/asterisk/modules/chan_sip.so [0x2aaabe423ee8]
+=== --- ---> Locked Here: channel.c line 4304 (ast_indicate_data)
diff --git a/lib/python/asterisk/self_test/test_buildoptions.py b/lib/python/asterisk/self_test/test_buildoptions.py
new file mode 100755
index 0000000..1f4ca77
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_buildoptions.py
@@ -0,0 +1,26 @@
+#!/usr/bin/env python
+"""Asterisk Build Options Handling Unit Test
+
+Copyright (C) 2011-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.
+"""
+
+from harness_shared import main
+import unittest
+from asterisk.buildoptions import AsteriskBuildOptions
+
+
+class AsteriskBuildOptionsTests(unittest.TestCase):
+ """Unit tests for AsteriskBuildOptions"""
+
+ def test_1(self):
+ """Test the defaults paths"""
+ build_options = AsteriskBuildOptions()
+ self.assertTrue(build_options)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/python/asterisk/self_test/test_cdr.py b/lib/python/asterisk/self_test/test_cdr.py
new file mode 100755
index 0000000..251530b
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_cdr.py
@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+"""Asterisk call detail record unit tests
+
+This module implements an Asterisk CDR parser.
+
+Copyright (C) 2010, Digium, Inc.
+Terry Wilson<twilson at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+from harness_shared import main
+import unittest
+from asterisk.cdr import AsteriskCSVCDR, AsteriskCSVCDRLine
+
+
+class AsteriskCSVCDRTests(unittest.TestCase):
+ """Unit tests for AsteriskCSVCDR"""
+
+ def test_cdr(self):
+ """Test the self_test/Master.csv record"""
+
+ cdr = AsteriskCSVCDR("lib/python/asterisk/self_test/Master.csv")
+ self.assertEqual(len(cdr), 2)
+ self.assertTrue(
+ AsteriskCSVCDRLine(duration=7, lastapp="hangup").match(
+ cdr[0],
+ exact=(True, True)))
+ self.assertTrue(cdr[0].match(
+ AsteriskCSVCDRLine(duration=7, lastapp="hangup"),
+ exact=(True, True)))
+
+ self.assertFalse(cdr[1].match(cdr[0], silent=True, exact=(True, True)))
+ self.assertFalse(cdr[0].match(cdr[1], silent=True, exact=(True, True)))
+ self.assertEqual(cdr[0].billsec, "7")
+
+ self.assertTrue(cdr.match(cdr))
+ cdr2 = AsteriskCSVCDR("lib/python/asterisk/self_test/Master2.csv")
+ self.assertFalse(cdr.match(cdr2))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/lib/python/asterisk/self_test/test_cel.py b/lib/python/asterisk/self_test/test_cel.py
new file mode 100755
index 0000000..3ea8aa9
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_cel.py
@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+"""Asterisk call detail record unit tests
+
+This module implements an Asterisk CDR parser.
+
+Copyright (C) 2010, Digium, Inc.
+Terry Wilson<twilson at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+from harness_shared import main
+import unittest
+from asterisk.cel import AsteriskCSVCEL, AsteriskCSVCELLine
+
+
+class AsteriskCSVCELTests(unittest.TestCase):
+ """Unit tests for AsteriskCSVCEL"""
+
+ def test_cel(self):
+ """Test CEL using self_test/CELMaster1.csv"""
+
+ cel = AsteriskCSVCEL("lib/python/asterisk/self_test/CELMaster1.csv")
+ self.assertEqual(len(cel), 16)
+ self.assertTrue(AsteriskCSVCELLine(
+ eventtype="LINKEDID_END",
+ channel="TinCan/string").match(cel[-1],
+ silent=True,
+ exact=(True, True)))
+ self.assertTrue(cel[-1].match(
+ AsteriskCSVCELLine(eventtype="LINKEDID_END",
+ channel="TinCan/string"),
+ silent=True,
+ exact=(True, True)))
+
+ self.assertFalse(cel[1].match(cel[0], silent=True, exact=(True, True)))
+ self.assertFalse(cel[0].match(cel[1], silent=True, exact=(True, True)))
+ self.assertEqual(cel[-1].channel, "TinCan/string")
+
+ self.assertTrue(cel.match(cel))
+ cel2 = AsteriskCSVCEL("lib/python/asterisk/self_test/CELMaster2.csv")
+ self.assertFalse(cel.match(cel2))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/lib/python/asterisk/self_test/test_channel_test_condition.py b/lib/python/asterisk/self_test/test_channel_test_condition.py
new file mode 100755
index 0000000..e32d883
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_channel_test_condition.py
@@ -0,0 +1,144 @@
+#!/usr/bin/env python
+"""Tests for test condition for channels
+
+Copyright (C) 2011-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.
+"""
+
+from harness_shared import AstMockOutput, main
+import unittest
+from asterisk.channel_test_condition import ChannelTestCondition
+
+
+class AstMockObjectInactive(AstMockOutput):
+ """mock cli output showing no active channels"""
+
+ def cli_exec(self, command):
+ """presume command is core show channels and generate output"""
+ output = ""
+ output += "Channel Location State Application(Data)\n"
+ output += "0 active channels\n"
+ output += "0 active calls\n"
+ output += "2 calls processed\n"
+ output += "Asterisk ending (0).\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectSingle(AstMockOutput):
+ """mock cli output showing single active channel"""
+
+ def cli_exec(self, command):
+ """presume command is core show channels and generate output"""
+ output = ""
+ output += "Channel Location State Application(Data)\n"
+ output += "Local/123 at default-00 (None) Down ()\n"
+ output += "1 active channels\n"
+ output += "0 active calls\n"
+ output += "2 calls processed\n"
+ output += "Asterisk ending (0).\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectMultiple(AstMockOutput):
+ """mock cli output showing multiple active channels"""
+
+ def cli_exec(self, command):
+ """presume command is core show channels and generate output"""
+ output = ""
+ output += "Channel Location State Application(Data)\n"
+ output += "PJSIP/123 at default-00 (None) Down ()\n"
+ output += "Local/123 at default-00 (None) Down ()\n"
+ output += "SIP/alice at default-00 (None) Down ()\n"
+ output += "3 active channels\n"
+ output += "0 active calls\n"
+ output += "2 calls processed\n"
+ output += "Asterisk ending (0).\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectLeaked(AstMockOutput):
+ """mock cli output showing leaked channel"""
+
+ def cli_exec(self, command):
+ """presume command is core show channels and generate output"""
+ output = ""
+ output += "Channel Location State Application(Data)\n"
+ output += "Local/123 at default-00 (None) Down ()\n"
+ output += "0 active channels\n"
+ output += "0 active calls\n"
+ output += "2 calls processed\n"
+ output += "Asterisk ending (0).\n"
+ return self.MockDefer(output)
+
+
+class TestConfig(object):
+ """Fake TestConfig object for unittest"""
+
+ def __init__(self):
+ self.class_type_name = "bogus"
+ self.config = {}
+ self.enabled = True
+ self.pass_expected = True
+
+
+class ChannelTestConditionUnitTest(unittest.TestCase):
+ """Unit Tests for ChannelTestCondition"""
+
+ def test_evaluate_inactive(self):
+ """test inactive channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.register_asterisk_instance(AstMockObjectInactive())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_evaluate_multiple_fail(self):
+ """test multiple channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.register_asterisk_instance(AstMockObjectMultiple())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_evaluate_multiple_fail2(self):
+ """test multiple channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.allowed_channels = 2
+ obj.register_asterisk_instance(AstMockObjectMultiple())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_evaluate_multiple_pass(self):
+ """test multiple channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.allowed_channels = 3
+ obj.register_asterisk_instance(AstMockObjectMultiple())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_evaluate_single_fail(self):
+ """test single channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.register_asterisk_instance(AstMockObjectSingle())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_evaluate_single_pass(self):
+ """test single channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.allowed_channels = 1
+ obj.register_asterisk_instance(AstMockObjectSingle())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_evaluate_leaked(self):
+ """test leaked channel condition"""
+ obj = ChannelTestCondition(TestConfig())
+ obj.register_asterisk_instance(AstMockObjectLeaked())
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/python/asterisk/self_test/test_config.py b/lib/python/asterisk/self_test/test_config.py
new file mode 100755
index 0000000..cc88636
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_config.py
@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+"""Asterisk Configuration File Handling Unit Tests.
+
+Copyright (C) 2010, Digium, Inc.
+Russell Bryant <russell at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+from harness_shared import main
+import unittest
+from asterisk.config import ConfigFile
+
+
+class ConfigFileTests(unittest.TestCase):
+ """Unit tests for ConfigFile"""
+
+ def test_conf(self):
+ """Test parsing a blob of config data"""
+ test = \
+ "; stuff\n" \
+ "this line is invalid on purpose\n" \
+ "[this is] also invalid]\n" \
+ ";-- comment --;\n" \
+ ";-- \n" \
+ "[this is commented out]\n" \
+ " --;\n" \
+ "[foo]\n" \
+ "a = b\n" \
+ " b = a \n" \
+ "this line is invalid on purpose\n" \
+ ";moo\n" \
+ "c = d;asdadf;adfads;adsfasdf\n" \
+ " [bar] ;asdfasdf\n" \
+ "a-b=c-d\n" \
+ "xyz=x|y|z\n" \
+ "1234 => 4242,Example Mailbox,root at localhost,,var=val\n" \
+ "\n" \
+ "[template](!)\n" \
+ "foo=bar\n" \
+ "exten => _NXX.,n,Wait(1)\n" \
+ "astetcdir => /etc/asterisk\n"
+
+ conf = ConfigFile(filename=None, config_str=test)
+
+ self.assertEqual(len(conf.categories), 3)
+
+ self.assertEqual(conf.categories[0].name, "foo")
+ self.assertFalse(conf.categories[0].template)
+ self.assertEqual(len(conf.categories[0].options), 3)
+ self.assertEqual(conf.categories[0].options[0][0], "a")
+ self.assertEqual(conf.categories[0].options[0][1], "b")
+ self.assertEqual(conf.categories[0].options[1][0], "b")
+ self.assertEqual(conf.categories[0].options[1][1], "a")
+ self.assertEqual(conf.categories[0].options[2][0], "c")
+ self.assertEqual(conf.categories[0].options[2][1], "d")
+
+ self.assertEqual(conf.categories[1].name, "bar")
+ self.assertFalse(conf.categories[1].template)
+ self.assertEqual(len(conf.categories[1].options), 3)
+ self.assertEqual(conf.categories[1].options[0][0], "a-b")
+ self.assertEqual(conf.categories[1].options[0][1], "c-d")
+ self.assertEqual(conf.categories[1].options[1][0], "xyz")
+ self.assertEqual(conf.categories[1].options[1][1], "x|y|z")
+ self.assertEqual(conf.categories[1].options[2][0], "1234")
+ self.assertEqual(conf.categories[1].options[2][1],
+ "4242,Example Mailbox,root at localhost,,var=val")
+
+ self.assertEqual(conf.categories[2].name, "template")
+ self.assertTrue(conf.categories[2].template)
+ self.assertEqual(len(conf.categories[2].options), 3)
+ self.assertEqual(conf.categories[2].options[0][0], "foo")
+ self.assertEqual(conf.categories[2].options[0][1], "bar")
+ self.assertEqual(conf.categories[2].options[1][0], "exten")
+ self.assertEqual(conf.categories[2].options[1][1],
+ "_NXX.,n,Wait(1)")
+ self.assertEqual(conf.categories[2].options[2][0], "astetcdir")
+ self.assertEqual(conf.categories[2].options[2][1], "/etc/asterisk")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/python/asterisk/self_test/test_lock_test_condition.py b/lib/python/asterisk/self_test/test_lock_test_condition.py
new file mode 100755
index 0000000..6cda06a
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_lock_test_condition.py
@@ -0,0 +1,193 @@
+#!/usr/bin/env python
+"""Held locks test condition unit tests
+
+Copyright (C) 2011-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.
+"""
+
+from harness_shared import AstMockOutput, ReadTestFile, main
+import unittest
+from asterisk.lock_test_condition import LockSequence, LockObject, LockTestCondition
+
+
+class AstMockObjectPassed(AstMockOutput):
+ """A lock output that passed"""
+
+ def __init__(self):
+ """Constructor"""
+ self.host = "127.0.0.2"
+
+ def cli_exec(self, command):
+ """Fake out a CLI command execution"""
+ return self.MockDeferFile("locks-pass.txt")
+
+
+class AstMockObjectFailure(AstMockOutput):
+ """A lock object that failed"""
+
+ def cli_exec(self, command):
+ """Fake out a CLI command execution"""
+ return self.MockDeferFile("locks-fail.txt")
+
+
+class TestConfig(object):
+ """Fake TestConfig object"""
+
+ def __init__(self):
+ """ Values here don't matter much - we just need to have something """
+ self.class_type_name = "asterisk.LockTestCondition.LockTestCondition"
+ self.pass_expected = True
+ self.type = "Post"
+ self.related_condition = ""
+ self.config = {}
+ self.enabled = True
+
+
+class LockTestConditionUnitTest(unittest.TestCase):
+ """Unit tests for LockTestCondition"""
+
+ def test_evaluate_failed(self):
+ """Test a failed locking condition"""
+ ast = AstMockObjectFailure()
+ obj = LockTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_evaluate_pass(self):
+ """Test a passed locking condition"""
+ ast = AstMockObjectPassed()
+ obj = LockTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_evaluate_multiple(self):
+ """Test multiple results"""
+ ast1 = AstMockObjectPassed()
+ ast2 = AstMockObjectFailure()
+ obj = LockTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast1)
+ obj.register_asterisk_instance(ast2)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+
+class LockSequenceUnitTest(unittest.TestCase):
+ """Tests for parsing a lock sequence"""
+
+ def test_single_object_no_held_info(self):
+ """Test a lock sequence with no waiting lock"""
+
+ obj = LockSequence()
+ obj.parse_lock_sequence(ReadTestFile("locks-single-object-no-held-info.txt"))
+
+ def test_large_multiple_object(self):
+ """Test with a waiting lock"""
+
+ obj = LockSequence()
+ obj.parse_lock_sequence(ReadTestFile("locks-large-multiple-object.txt"))
+ self.assertEqual(obj.thread_id, "0x449ec940")
+ self.assertEqual(obj.thread_name, "netconsole")
+ self.assertEqual(obj.thread_line, 1351)
+ self.assertEqual(obj.thread_file, "asterisk.c")
+ self.assertEqual(obj.thread_func, "listener")
+ self.assertTrue(len(obj.locks) == 1)
+ self.assertEqual(obj.locks[0].locked_file, "astobj2.c")
+ self.assertEqual(obj.locks[0].locked_line, 657)
+ self.assertEqual(obj.locks[0].locked_func, "internal_ao2_callback")
+
+ def test_single_object(self):
+ """Test a lock held somewhere else"""
+
+ obj = LockSequence()
+ obj.parse_lock_sequence(ReadTestFile("locks-single-object.txt"))
+ self.assertEqual(obj.thread_id, "0x402c6940")
+ self.assertEqual(obj.thread_name, "do_monitor")
+ self.assertEqual(obj.thread_line, 25114)
+ self.assertEqual(obj.thread_file, "chan_sip.c")
+ self.assertEqual(obj.thread_func, "restart_monitor")
+ self.assertTrue(len(obj.locks) == 1)
+ self.assertEqual(obj.locks[0].locked_file, "channel.c")
+ self.assertEqual(obj.locks[0].locked_line, 4304)
+ self.assertEqual(obj.locks[0].locked_func, "ast_indicate_data")
+ self.assertEqual(obj.locks[0].id, 0)
+ self.assertEqual(obj.locks[0].type, "MUTEX")
+ self.assertEqual(obj.locks[0].file, "chan_sip.c")
+ self.assertEqual(obj.locks[0].line_number, 24629)
+ self.assertEqual(obj.locks[0].func, "handle_request_do")
+ self.assertEqual(obj.locks[0].name, "&netlock")
+ self.assertEqual(obj.locks[0].addr, "0x2aaabe671a40")
+ self.assertEqual(obj.locks[0].lock_count, 1)
+ self.assertTrue(obj.locks[0].held)
+ self.assertTrue(len(obj.locks[0].backtrace) == 3)
+
+ def test_multiple_objects_no_backtrace(self):
+ """Test multiple locks with no backtrace"""
+
+ obj = LockSequence()
+ obj.parse_lock_sequence(ReadTestFile("locks-multiple-objects-no-backtrace.txt"))
+ self.assertEqual(obj.thread_id, "0x402c6940")
+ self.assertEqual(obj.thread_name, "do_monitor")
+ self.assertEqual(obj.thread_line, 25114)
+ self.assertEqual(obj.thread_file, "chan_sip.c")
+ self.assertEqual(obj.thread_func, "restart_monitor")
+ self.assertTrue(len(obj.locks) == 3)
+ self.assertEqual(obj.locks[2].locked_file, "channel.c")
+ self.assertEqual(obj.locks[2].locked_line, 4304)
+ self.assertEqual(obj.locks[2].locked_func, "ast_indicate_data")
+ self.assertEqual(obj.locks[0].id, 0)
+ self.assertEqual(obj.locks[0].type, "MUTEX")
+ self.assertEqual(obj.locks[0].file, "chan_sip.c")
+ self.assertEqual(obj.locks[0].line_number, 24629)
+ self.assertEqual(obj.locks[0].func, "handle_request_do")
+ self.assertEqual(obj.locks[0].name, "&netlock")
+ self.assertEqual(obj.locks[0].addr, "0x2aaabe671a40")
+ self.assertEqual(obj.locks[0].lock_count, 1)
+ self.assertTrue(obj.locks[0].held)
+ self.assertTrue(len(obj.locks[0].backtrace) == 0)
+
+
+class LockObjectUnitTest(unittest.TestCase):
+ """Unit tests for LockObject"""
+
+ def test_no_backtrace(self):
+ """Test creating a lock object with no thread backtrace"""
+
+ lock_line = "=== ---> Waiting for Lock #0 (sig_ss7.c): " + \
+ "MUTEX 636 ss7_linkset &linkset->lock 0x2aaab8a6b588 (1)"
+ obj = LockObject()
+ obj.parse_lock_information(lock_line)
+ self.assertEqual(obj.id, 0)
+ self.assertEqual(obj.type, "MUTEX")
+ self.assertEqual(obj.file, "sig_ss7.c")
+ self.assertEqual(obj.line_number, 636)
+ self.assertEqual(obj.func, "ss7_linkset")
+ self.assertEqual(obj.name, "&linkset->lock")
+ self.assertEqual(obj.addr, "0x2aaab8a6b588")
+ self.assertEqual(obj.lock_count, 1)
+ self.assertFalse(obj.held)
+ self.assertTrue(len(obj.backtrace) == 0)
+
+ def test_with_backtrace(self):
+ """Test creating a lock object with a thread backtrace"""
+
+ obj = LockObject()
+ obj.parse_lock_information(ReadTestFile("locks-backtrace.txt"))
+ self.assertEqual(obj.id, 1)
+ self.assertEqual(obj.type, "MUTEX")
+ self.assertEqual(obj.file, "astobj2.c")
+ self.assertEqual(obj.line_number, 657)
+ self.assertEqual(obj.func, "internal_ao2_callback")
+ self.assertEqual(obj.name, "c")
+ self.assertEqual(obj.addr, "0x2aaaac491f50")
+ self.assertEqual(obj.lock_count, 1)
+ self.assertTrue(obj.held)
+ self.assertTrue(len(obj.backtrace) == 19)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/python/asterisk/self_test/test2_matcher.py b/lib/python/asterisk/self_test/test_matcher.py
similarity index 100%
rename from lib/python/asterisk/self_test/test2_matcher.py
rename to lib/python/asterisk/self_test/test_matcher.py
diff --git a/lib/python/asterisk/self_test/test_sip_dialog_test_condition.py.txt b/lib/python/asterisk/self_test/test_sip_dialog_test_condition.py.txt
new file mode 100644
index 0000000..717e183
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_sip_dialog_test_condition.py.txt
@@ -0,0 +1,426 @@
+#!/usr/bin/env python
+"""Test condition for verifying SIP dialogs unit tests
+
+Copyright (C) 2011-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.
+"""
+
+from harness_shared import AstMockOutput, main
+import unittest
+from asterisk.sip_dialog_test_condition import SipDialogPreTestCondition, \
+ SipDialogPostTestCondition
+
+
+class TestConfig(object):
+ """Mock TestConfig object"""
+
+ def __init__(self):
+ """Constructor
+
+ Values here don't matter much - we just need to have something"""
+ self.type_name = ("asterisk.sip_dialog_test_condition." +
+ "SipDialogPostTestCondition")
+ self.pass_expected = True
+ self.type = "Post"
+ self.related_condition = ""
+ self.class_type_name = "sip_dialog_test_condition"
+ self.config = {}
+ self.enabled = True
+
+
+class TestConfigWithHistory(TestConfig):
+ """Mock TestConfig object with history requirements"""
+
+ def __init__(self):
+ """Constructor"""
+ super(TestConfigWithHistory, self).__init__()
+
+ self.config['history_requirements'] = []
+ self.config['history_requirements'].append('Hangup')
+ self.config['history_requirements'].append('NewChan')
+
+
+class AstMockObjectPostTestNoDestructionFail(AstMockOutput):
+ """Mock out CLI execution from Asterisk instance
+
+ This mock object makes it appear as if a dialog failed to be destroyed
+ """
+ def __init__(self):
+ """Constructor"""
+ self.host = "127.0.0.6"
+
+ def cli_exec(self, command):
+ """Mock CLI execution/response"""
+
+ output = ""
+ if command == "sip show objects":
+ output = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Peer objects by IP =-\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Registry objects: 0 =-\n\n"
+ output += "-= Dialog objects:\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. SchedDestroy 32000 ms\n"
+ output += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ output += "13. Hangup Cause Normal Clearing\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ output += "12. Hangup Cause Normal Clearing\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectPostTestNoHangupFail(AstMockOutput):
+ """Mock out CLI execution from Asterisk instance
+
+ This mock object makes it appear as if a channel failed to hangup
+ """
+
+ def __init__(self):
+ """Constructor"""
+ self.host = "127.0.0.5"
+
+ def cli_exec(self, command):
+ """Mock CLI execution/response"""
+
+ output = ""
+ if command == "sip show objects":
+ output = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Peer objects by IP =-\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Registry objects: 0 =-\n\n"
+ output += "-= Dialog objects:\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. SchedDestroy 32000 ms\n"
+ output += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. SchedDestroy 32000 ms\n"
+ output += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ output += "13. Hangup Cause Normal Clearing\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectPostTestNoDialogsPass(AstMockOutput):
+ """Mock out CLI execution from Asterisk instance
+
+ This mock object makes it appear as if there were no dialogs, which is okay
+ """
+ def __init__(self):
+ """Constructor"""
+ self.host = "127.0.0.4"
+
+ def cli_exec(self, command):
+ """Mock CLI execution/response"""
+
+ output = ""
+ if command == "sip show objects":
+ output = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Peer objects by IP =-\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Registry objects: 0 =-\n\n"
+ output += "-= Dialog objects:\n\n"
+
+ return self.MockDefer(output)
+
+
+class AstMockObjectPostTestPass(AstMockOutput):
+ """Mock out CLI execution from Asterisk instance
+
+ This mock object provides two dialogs with acceptable history
+ """
+ def __init__(self):
+ """Constructor"""
+ self.host = "127.0.0.3"
+
+ def cli_exec(self, command):
+ """Mock CLI execution/response"""
+
+ output = ""
+ if command == "sip show objects":
+ output = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Peer objects by IP =-\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Registry objects: 0 =-\n\n"
+ output += "-= Dialog objects:\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. SchedDestroy 32000 ms\n"
+ output += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ output += "13. Hangup Cause Normal Clearing\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. SchedDestroy 32000 ms\n"
+ output += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ output += "13. Hangup Cause Normal Clearing\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectPreTestFail(AstMockOutput):
+ """Mock out CLI execution from Asterisk instance
+
+ This mock object provides history during a pre-test call, which is wrong
+ """
+ def __init__(self):
+ """Constructor"""
+ self.host = "127.0.0.2"
+
+ def cli_exec(self, command):
+ """Mock CLI execution/response"""
+
+ output = ""
+ if command == "sip show objects":
+ output = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Peer objects by IP =-\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Registry objects: 0 =-\n\n"
+ output += "-= Dialog objects:\n\n"
+ output += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
+ output += "type: dialog\n"
+ output += "objflags: 0\n"
+ output += "refcount: 2\n\n"
+ elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
+ output = "* SIP Call\n"
+ output += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
+ output += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
+ output += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
+ output += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
+ output += "5. TxReq ACK / 102 ACK - ACK\n"
+ output += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
+ output += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
+ output += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
+ output += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
+ output += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
+ output += "11. SchedDestroy 32000 ms\n"
+ output += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
+ output += "13. Hangup Cause Normal Clearing\n"
+ return self.MockDefer(output)
+
+
+class AstMockObjectPreTestPass(AstMockOutput):
+ """Mock out CLI execution from Asterisk instance
+
+ This mock object provides no history during a pre-test call, which is the
+ expected state
+ """
+ def cli_exec(self, command):
+ """Mock CLI execution/response"""
+
+ output = ""
+ if command == "sip show objects":
+ output = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Peer objects by IP =-\n\n"
+ output += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
+ output += "-= Registry objects: 0 =-\n\n"
+ output += "-= Dialog objects:\n"
+ elif command == "sip show history":
+ output = "\n"
+ return self.MockDefer(output)
+
+
+ at unittest.skip("sip_dialog_test_condition.py is broken")
+class SipDialogTestConditionUnitTest(unittest.TestCase):
+ """Unit tests for SipDialogTestCondition objects"""
+
+ def test_pre_test_pass(self):
+ """Verify that acceptable pre-test output passes"""
+ ast = AstMockObjectPreTestPass()
+ obj = SipDialogPreTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_pre_test_fail(self):
+ """Verify that unacceptable pre-test output fails"""
+ ast = AstMockObjectPreTestFail()
+ obj = SipDialogPreTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_pre_test_fail_multi_asterisk(self):
+ """Verify that pre-test output from multiple sources fails when one
+ of those sources is bad"""
+ ast1 = AstMockObjectPreTestFail()
+ ast2 = AstMockObjectPreTestPass()
+ obj = SipDialogPreTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast1)
+ obj.register_asterisk_instance(ast2)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_post_test_pass(self):
+ """Verify nominal post-test output"""
+ ast = AstMockObjectPostTestPass()
+ obj = SipDialogPostTestCondition(TestConfigWithHistory())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_post_test_no_dialog_pass(self):
+ """Verify nominal post-test output with no dialogs passes"""
+ ast = AstMockObjectPostTestNoDialogsPass()
+ obj = SipDialogPostTestCondition(TestConfig())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Passed')
+
+ def test_post_test_no_hangup_fail(self):
+ """Verify no hangup detection is caught and results in a failure"""
+ ast = AstMockObjectPostTestNoHangupFail()
+ obj = SipDialogPostTestCondition(TestConfigWithHistory())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_post_test_no_destruction_fail(self):
+ """Verify no destruction is caught and results in a failure"""
+ ast = AstMockObjectPostTestNoDestructionFail()
+ obj = SipDialogPostTestCondition(TestConfigWithHistory())
+ obj.register_asterisk_instance(ast)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+ def test_post_test_multi_asterisk_fail(self):
+ """Test multiple instances of Asterisk where a single failure causes
+ a failure in the overall result"""
+ ast1 = AstMockObjectPostTestNoHangupFail()
+ ast2 = AstMockObjectPostTestNoDestructionFail()
+ ast3 = AstMockObjectPostTestNoDialogsPass()
+ ast4 = AstMockObjectPostTestPass()
+ obj = SipDialogPostTestCondition(TestConfigWithHistory())
+ obj.register_asterisk_instance(ast1)
+ obj.register_asterisk_instance(ast2)
+ obj.register_asterisk_instance(ast3)
+ obj.register_asterisk_instance(ast4)
+ obj.evaluate()
+ self.assertEqual(obj.get_status(), 'Failed')
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/python/asterisk/self_test/test_sippversion.py b/lib/python/asterisk/self_test/test_sippversion.py
new file mode 100755
index 0000000..8c3d933
--- /dev/null
+++ b/lib/python/asterisk/self_test/test_sippversion.py
@@ -0,0 +1,221 @@
+#!/usr/bin/env python
+"""SIPp Version String Handling Tests
+
+Copyright (C) 2010, Digium, Inc.
+Paul Belanger <pabelanger at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+from harness_shared import main
+import unittest
+from asterisk.sippversion import SIPpVersion
+
+
+class SIPpVersionTests(unittest.TestCase):
+ def test_version(self):
+ v = SIPpVersion("v3.2", None)
+ self.assertEqual(str(v), "v3.2")
+ self.assertEqual(v.concept, "v3")
+ self.assertEqual(v.major, "2")
+ self.assertEqual(v.minor, None)
+ self.assertFalse(v.tls)
+ self.assertFalse(v.pcap)
+
+ def test_version2(self):
+ v = SIPpVersion("v2.0.1", None)
+ self.assertEqual(str(v), "v2.0.1")
+ self.assertEqual(v.concept, "v2")
+ self.assertEqual(v.major, "0")
+ self.assertEqual(v.minor, "1")
+ self.assertFalse(v.tls)
+ self.assertFalse(v.pcap)
+
+ def test_version3(self):
+ v = SIPpVersion("v3.1", "TLS")
+ self.assertEqual(str(v), "v3.1-TLS")
+ self.assertEqual(v.concept, "v3")
+ self.assertEqual(v.major, "1")
+ self.assertEqual(v.minor, None)
+ self.assertTrue(v.tls)
+ self.assertFalse(v.pcap)
+
+ def test_version4(self):
+ v = SIPpVersion("v2.0.1", "TLS-PCAP")
+ self.assertEqual(str(v), "v2.0.1-TLS-PCAP")
+ self.assertEqual(v.concept, "v2")
+ self.assertEqual(v.major, "0")
+ self.assertEqual(v.minor, "1")
+ self.assertTrue(v.tls)
+ self.assertTrue(v.pcap)
+
+ def test_version5(self):
+ v = SIPpVersion("v3.2", "PCAP")
+ self.assertEqual(str(v), "v3.2-PCAP")
+ self.assertEqual(v.concept, "v3")
+ self.assertEqual(v.major, "2")
+ self.assertEqual(v.minor, None)
+ self.assertFalse(v.tls)
+ self.assertTrue(v.pcap)
+
+ def test_version6(self):
+ v = SIPpVersion(None, "PCAP")
+ self.assertEqual(str(v), "PCAP")
+ self.assertEqual(v.concept, None)
+ self.assertEqual(v.major, None)
+ self.assertEqual(v.minor, None)
+ self.assertFalse(v.tls)
+ self.assertTrue(v.pcap)
+
+ def test_version7(self):
+ v = SIPpVersion(None, "TLS")
+ self.assertEqual(str(v), "TLS")
+ self.assertEqual(v.concept, None)
+ self.assertEqual(v.major, None)
+ self.assertEqual(v.minor, None)
+ self.assertTrue(v.tls)
+ self.assertFalse(v.pcap)
+
+ def test_version8(self):
+ v = SIPpVersion(None, "TLS-PCAP")
+ self.assertEqual(str(v), "TLS-PCAP")
+ self.assertEqual(v.concept, None)
+ self.assertEqual(v.major, None)
+ self.assertEqual(v.minor, None)
+ self.assertTrue(v.tls)
+ self.assertTrue(v.pcap)
+
+ def test_cmp(self):
+ v1 = SIPpVersion("v3.2", None)
+ v2 = SIPpVersion("v3.1", None)
+ self.assertTrue(v1 > v2)
+
+ def test_cmp2(self):
+ v1 = SIPpVersion("v2.0.1", None)
+ v2 = SIPpVersion("v3.1", None)
+ self.assertTrue(v1 < v2)
+
+ def test_cmp3(self):
+ v1 = SIPpVersion("v3.1", None)
+ v2 = SIPpVersion("v3.1", None)
+ self.assertTrue(v1 == v2)
+
+ def test_cmp4(self):
+ v1 = SIPpVersion("v3.1", None)
+ v2 = SIPpVersion("v3.1", None)
+ self.assertFalse(v1 != v2)
+
+ def test_cmp5(self):
+ v1 = SIPpVersion("v3.1", "TLS")
+ v2 = SIPpVersion("v3.1", "TLS")
+ self.assertTrue(v1 == v2)
+
+ def test_cmp6(self):
+ v1 = SIPpVersion(None, "TLS")
+ v2 = SIPpVersion(None, "TLS")
+ self.assertTrue(v1 == v2)
+
+ def test_cmp7(self):
+ v1 = SIPpVersion("v2.0.1", None)
+ v2 = SIPpVersion("v2.0.1", None)
+ self.assertTrue(v1 == v2)
+
+ def test_cmp8(self):
+ v1 = SIPpVersion("v3.2", "TLS")
+ v2 = SIPpVersion("v3.2", "PCAP")
+ self.assertTrue(v1 != v2)
+
+ def test_cmp9(self):
+ v1 = SIPpVersion(None, "TLS")
+ v2 = SIPpVersion(None, "PCAP")
+ self.assertTrue(v1 != v2)
+
+ def test_cmp10(self):
+ v1 = SIPpVersion("v3.2", "TLS")
+ v2 = SIPpVersion("v3.2", "PCAP")
+ self.assertFalse(v1 == v2)
+
+ def test_cmp11(self):
+ v1 = SIPpVersion(None, "TLS")
+ v2 = SIPpVersion(None, "PCAP")
+ self.assertFalse(v1 == v2)
+
+ def test_cmp12(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion("v3.2", "TLS")
+ self.assertTrue(v1 >= v2)
+
+ def test_cmp13(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion(None, "TLS")
+ self.assertTrue(v1 >= v2)
+
+ def test_cmp14(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion("v3.2", "PCAP")
+ self.assertTrue(v1 >= v2)
+
+ def test_cmp15(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion(None, "PCAP")
+ self.assertTrue(v1 >= v2)
+
+ def test_cmp16(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion("v3.2", "TLS")
+ self.assertTrue(v1 != v2)
+
+ def test_cmp17(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion(None, "TLS")
+ self.assertTrue(v1 != v2)
+
+ def test_cmp18(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion("v3.2", "TLS")
+ self.assertFalse(v1 == v2)
+
+ def test_cmp19(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion(None, "TLS")
+ self.assertFalse(v1 == v2)
+
+ def test_cmp20(self):
+ v1 = SIPpVersion("v3.1", "PCAP")
+ v2 = SIPpVersion("v3.0", "TLS")
+ self.assertTrue(v1 > v2)
+
+ def test_cmp21(self):
+ v1 = SIPpVersion("v3.2", "TLS-PCAP")
+ v2 = SIPpVersion("v2.0.1", "PCAP")
+ self.assertTrue(v1 >= v2)
+
+ def test_cmp22(self):
+ v1 = SIPpVersion("v3.1", "TLS")
+ v2 = SIPpVersion("v3.1", "PCAP")
+ self.assertFalse(v1 > v2)
+
+ def test_cmp23(self):
+ v1 = SIPpVersion("v3.1", "TLS")
+ v2 = SIPpVersion("v3.1", "PCAP")
+ self.assertFalse(v1 < v2)
+
+ def test_cmp24(self):
+ v1 = SIPpVersion("v3.2", "TLS")
+ v2 = SIPpVersion("v3.1", "TLS-PCAP")
+ self.assertTrue(v1 >= v2)
+
+ def test_cmp25(self):
+ v1 = SIPpVersion("v3.1", "TLS-PCAP")
+ v2 = SIPpVersion("v3.2", "PCAP")
+ self.assertTrue(v1 <= v2)
+
+ def test_cmp26(self):
+ v1 = SIPpVersion("v2.0.1", "TLS-PCAP")
+ v2 = SIPpVersion("v3.0", "TLS")
+ self.assertFalse(v1 >= v2)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/python/asterisk/self_test/test_utils_socket.py b/lib/python/asterisk/self_test/test_utils_socket.py
index 3a0c33d..34672c6 100755
--- a/lib/python/asterisk/self_test/test_utils_socket.py
+++ b/lib/python/asterisk/self_test/test_utils_socket.py
@@ -10,17 +10,10 @@
the GNU General Public License Version 2.
"""
-import logging
-import sys
+from harness_shared import main
import unittest
-
from socket import SOCK_STREAM, SOCK_DGRAM, AF_INET, AF_INET6
-
-sys.path.append('lib/python') # noqa
from asterisk.utils_socket import Ports, PortError, get_available_port, MIN_PORT
-
-
-LOGGER = logging.getLogger(__name__)
class PortTests(unittest.TestCase):
@@ -169,7 +162,4 @@
if __name__ == "__main__":
"""Run the unit tests"""
-
- logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
- format="%(module)s:%(lineno)d - %(message)s")
- unittest.main()
+ main()
diff --git a/lib/python/asterisk/sip_channel_test_condition.py b/lib/python/asterisk/sip_channel_test_condition.py
index 150d90f..0f265f0 100644
--- a/lib/python/asterisk/sip_channel_test_condition.py
+++ b/lib/python/asterisk/sip_channel_test_condition.py
@@ -10,7 +10,7 @@
"""
from twisted.internet import defer
-from test_conditions import TestCondition
+from .test_conditions import TestCondition
class SipChannelTestCondition(TestCondition):
diff --git a/lib/python/asterisk/sip_dialog_test_condition.py b/lib/python/asterisk/sip_dialog_test_condition.py
index 4c75710..619a388 100644
--- a/lib/python/asterisk/sip_dialog_test_condition.py
+++ b/lib/python/asterisk/sip_dialog_test_condition.py
@@ -10,9 +10,8 @@
import logging
import logging.config
-import unittest
-from test_conditions import TestCondition
+from .test_conditions import TestCondition
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
@@ -227,422 +226,3 @@
self._counter = -1
__get_dialogs()
return self._finished_deferred
-
-
-class TestConfig(object):
- """Mock TestConfig object"""
-
- def __init__(self):
- """Constructor
-
- Values here don't matter much - we just need to have something"""
- self.type_name = ("asterisk.sip_dialog_test_condition." +
- "SipDialogPostTestCondition")
- self.pass_expected = True
- self.type = "Post"
- self.related_condition = ""
- self.config = {}
-
-
-class TestConfigWithHistory(TestConfig):
- """Mock TestConfig object with history requirements"""
-
- def __init__(self):
- """Constructor"""
- super(TestConfigWithHistory, self).__init__()
-
- self.config['history_requirements'] = []
- self.config['history_requirements'].append('Hangup')
- self.config['history_requirements'].append('NewChan')
-
-
-class AstMockObjectPostTestNoDestructionFail(object):
- """Mock out CLI execution from Asterisk instance
-
- This mock object makes it appear as if a dialog failed to be destroyed
- """
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.6"
-
- def cli_exec(self, command):
- """Mock CLI execution/response"""
-
- ret_string = ""
- if command == "sip show objects":
- ret_string = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Peer objects by IP =-\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Registry objects: 0 =-\n\n"
- ret_string += "-= Dialog objects:\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. SchedDestroy 32000 ms\n"
- ret_string += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- ret_string += "13. Hangup Cause Normal Clearing\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- ret_string += "12. Hangup Cause Normal Clearing\n"
- return ret_string
-
-
-class AstMockObjectPostTestNoHangupFail(object):
- """Mock out CLI execution from Asterisk instance
-
- This mock object makes it appear as if a channel failed to hangup
- """
-
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.5"
-
- def cli_exec(self, command):
- """Mock CLI execution/response"""
-
- ret_string = ""
- if command == "sip show objects":
- ret_string = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Peer objects by IP =-\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Registry objects: 0 =-\n\n"
- ret_string += "-= Dialog objects:\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. SchedDestroy 32000 ms\n"
- ret_string += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. SchedDestroy 32000 ms\n"
- ret_string += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- ret_string += "13. Hangup Cause Normal Clearing\n"
- return ret_string
-
-
-class AstMockObjectPostTestNoDialogsPass(object):
- """Mock out CLI execution from Asterisk instance
-
- This mock object makes it appear as if there were no dialogs, which is okay
- """
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.4"
-
- def cli_exec(self, command):
- """Mock CLI execution/response"""
-
- ret_string = ""
- if command == "sip show objects":
- ret_string = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Peer objects by IP =-\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Registry objects: 0 =-\n\n"
- ret_string += "-= Dialog objects:\n\n"
-
- return ret_string
-
-
-class AstMockObjectPostTestPass(object):
- """Mock out CLI execution from Asterisk instance
-
- This mock object provides two dialogs with acceptable history
- """
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.3"
-
- def cli_exec(self, command):
- """Mock CLI execution/response"""
-
- ret_string = ""
- if command == "sip show objects":
- ret_string = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Peer objects by IP =-\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Registry objects: 0 =-\n\n"
- ret_string += "-= Dialog objects:\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. SchedDestroy 32000 ms\n"
- ret_string += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- ret_string += "13. Hangup Cause Normal Clearing\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c5 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. SchedDestroy 32000 ms\n"
- ret_string += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- ret_string += "13. Hangup Cause Normal Clearing\n"
- return ret_string
-
-
-class AstMockObjectPreTestFail(object):
- """Mock out CLI execution from Asterisk instance
-
- This mock object provides history during a pre-test call, which is wrong
- """
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.2"
-
- def cli_exec(self, command):
- """Mock CLI execution/response"""
-
- ret_string = ""
- if command == "sip show objects":
- ret_string = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Peer objects by IP =-\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Registry objects: 0 =-\n\n"
- ret_string += "-= Dialog objects:\n\n"
- ret_string += "name: 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060\n"
- ret_string += "type: dialog\n"
- ret_string += "objflags: 0\n"
- ret_string += "refcount: 2\n\n"
- elif command == "sip show history 2ec048aa4ed1239664f6408f0c5044c4 at 127.0.0.2:5060":
- ret_string = "* SIP Call\n"
- ret_string += "1. NewChan Channel SIP/ast1-00000002 - from 2ec048aa4ed1239664f6408f0c5044\n"
- ret_string += "2. TxReqRel INVITE / 102 INVITE - INVITE\n"
- ret_string += "3. Rx SIP/2.0 / 102 INVITE / 100 Trying\n"
- ret_string += "4. Rx SIP/2.0 / 102 INVITE / 200 OK\n"
- ret_string += "5. TxReq ACK / 102 ACK - ACK\n"
- ret_string += "6. Rx BYE / 102 BYE / sip:ast2 at 127.0.0.2:5060\n"
- ret_string += "7. RTCPaudio Quality:ssrc=28249381;themssrc=485141946;lp=0;rxjitter=0.000029\n"
- ret_string += "8. RTCPaudioJitter Quality:minrxjitter=0.000000;maxrxjitter=0.000000;avgrxjitter=0\n"
- ret_string += "9. RTCPaudioLoss Quality:minrxlost=0.000000;maxrxlost=0.000000;avgrxlost=0.00000\n"
- ret_string += "10. RTCPaudioRTT Quality:minrtt=0.000000;maxrtt=0.000000;avgrtt=0.000000;stdevrt\n"
- ret_string += "11. SchedDestroy 32000 ms\n"
- ret_string += "12. TxResp SIP/2.0 / 102 BYE - 200 OK\n"
- ret_string += "13. Hangup Cause Normal Clearing\n"
- return ret_string
-
-
-class AstMockObjectPreTestPass(object):
- """Mock out CLI execution from Asterisk instance
-
- This mock object provides no history during a pre-test call, which is the
- expected state
- """
- def __init__(self):
- """Constructor"""
- self.host = "127.0.0.1"
-
- def cli_exec(self, command):
- """Mock CLI execution/response"""
-
- ret_string = ""
- if command == "sip show objects":
- ret_string = "-= Peer objects: 4 static, 0 realtime, 0 autocreate =-\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7001\ntype: peer\nobjflags: 0\nrefcount: 1\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Peer objects by IP =-\n\n"
- ret_string += "name: zoiper_01\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: audio\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "name: 7002\ntype: peer\nobjflags: 0\nrefcount: 3\n\n"
- ret_string += "-= Registry objects: 0 =-\n\n"
- ret_string += "-= Dialog objects:\n"
- elif command == "sip show history":
- return "\n"
- return ret_string
-
-
-class SipDialogTestConditionUnitTest(unittest.TestCase):
- """Unit tests for SipDialogTestCondition objects"""
-
- def test_pre_test_pass(self):
- """Verify that acceptable pre-test output passes"""
- ast = AstMockObjectPreTestPass()
- obj = SipDialogPreTestCondition(TestConfig())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_pre_test_fail(self):
- """Verify that unacceptable pre-test output fails"""
- ast = AstMockObjectPreTestFail()
- obj = SipDialogPreTestCondition(TestConfig())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_pre_test_fail_multi_asterisk(self):
- """Verify that pre-test output from multiple sources fails when one
- of those sources is bad"""
- ast1 = AstMockObjectPreTestFail()
- ast2 = AstMockObjectPreTestPass()
- obj = SipDialogPreTestCondition(TestConfig())
- obj.register_asterisk_instance(ast1)
- obj.register_asterisk_instance(ast2)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_post_test_pass(self):
- """Verify nominal post-test output"""
- ast = AstMockObjectPostTestPass()
- obj = SipDialogPostTestCondition(TestConfigWithHistory())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_post_test_no_dialog_pass(self):
- """Verify nominal post-test output with no dialogs passes"""
- ast = AstMockObjectPostTestNoDialogsPass()
- obj = SipDialogPostTestCondition(TestConfig())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Passed')
-
- def test_post_test_no_hangup_fail(self):
- """Verify no hangup detection is caught and results in a failure"""
- ast = AstMockObjectPostTestNoHangupFail()
- obj = SipDialogPostTestCondition(TestConfigWithHistory())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_post_test_no_destruction_fail(self):
- """Verify no destruction is caught and results in a failure"""
- ast = AstMockObjectPostTestNoDestructionFail()
- obj = SipDialogPostTestCondition(TestConfigWithHistory())
- obj.register_asterisk_instance(ast)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
- def test_post_test_multi_asterisk_fail(self):
- """Test multiple instances of Asterisk where a single failure causes
- a failure in the overall result"""
- ast1 = AstMockObjectPostTestNoHangupFail()
- ast2 = AstMockObjectPostTestNoDestructionFail()
- ast3 = AstMockObjectPostTestNoDialogsPass()
- ast4 = AstMockObjectPostTestPass()
- obj = SipDialogPostTestCondition(TestConfigWithHistory())
- obj.register_asterisk_instance(ast1)
- obj.register_asterisk_instance(ast2)
- obj.register_asterisk_instance(ast3)
- obj.register_asterisk_instance(ast4)
- obj.evaluate()
- self.assertEqual(obj.get_status(), 'Failed')
-
-
-def main():
- """Execute the unit tests"""
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
-
-
-if __name__ == "__main__":
- main()
diff --git a/lib/python/asterisk/sipp.py b/lib/python/asterisk/sipp.py
index b84965e..540a544 100644
--- a/lib/python/asterisk/sipp.py
+++ b/lib/python/asterisk/sipp.py
@@ -10,12 +10,12 @@
"""
import logging
-import test_suite_utils
+from . import test_suite_utils
from abc import ABCMeta, abstractmethod
from twisted.internet import reactor, defer, protocol, error
-from test_case import TestCase
-from utils_socket import get_available_port
+from .test_case import TestCase
+from .utils_socket import get_available_port
LOGGER = logging.getLogger(__name__)
@@ -495,7 +495,7 @@
def outReceived(self, data):
"""Override of ProcessProtocol.outReceived"""
LOGGER.debug("Received from SIPp scenario %s: %s" % (self._name, data))
- self.output += data
+ self.output += data.decode('utf-8', 'ignore')
def connectionMade(self):
"""Override of ProcessProtocol.connectionMade"""
diff --git a/lib/python/asterisk/sippversion.py b/lib/python/asterisk/sippversion.py
index b1110fc..f942e9c 100644
--- a/lib/python/asterisk/sippversion.py
+++ b/lib/python/asterisk/sippversion.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""SIPp Version String Handling
Copyright (C) 2010, Digium, Inc.
@@ -9,11 +8,7 @@
"""
import subprocess
-import sys
-import unittest
-sys.path.append("lib/python")
-
-import test_suite_utils
+from . import test_suite_utils
class SIPpVersion:
@@ -48,8 +43,9 @@
except OSError:
return
for line in sipp_process.stdout:
- if line.strip().startswith('SIPp '):
- sipp = line.strip()[5:]
+ line = line.decode('utf-8').strip()
+ if line.startswith('SIPp '):
+ sipp = line[5:]
sipp = sipp.split(',', 1)
sipp = sipp[0].split('-', 1)
version = sipp[0]
@@ -88,7 +84,7 @@
def __cmp__(self, other):
"""Compare two SIPpVersion instances against each other"""
- return cmp(int(self), int(other))
+ return (int(self) > int(other)) - (int(self) < int(other))
def __ne__(self, other):
"""Determine if this SIPpVersion instance is not equal to another"""
@@ -105,6 +101,14 @@
if ((res == 0) and (self.tls == other.tls and self.pcap == other.pcap)):
return True
return False
+
+ def __le__(self, other):
+ """Determine if this SIPpVersion instance is less than or equal to another"""
+ return int(self) <= int(other)
+
+ def __lt__(self, other):
+ """Determine if this SIPpVersion instance is less than another"""
+ return int(self) < int(other)
def __parse_version(self, version):
"""Parse the version string returned from SIPp"""
@@ -124,215 +128,3 @@
self.tls = True
if value.find("PCAP") > -1:
self.pcap = True
-
-
-class SIPpVersionTests(unittest.TestCase):
- def test_version(self):
- v = SIPpVersion("v3.2", None)
- self.assertEqual(str(v), "v3.2")
- self.assertEqual(v.concept, "v3")
- self.assertEqual(v.major, "2")
- self.assertEqual(v.minor, None)
- self.assertFalse(v.tls)
- self.assertFalse(v.pcap)
-
- def test_version2(self):
- v = SIPpVersion("v2.0.1", None)
- self.assertEqual(str(v), "v2.0.1")
- self.assertEqual(v.concept, "v2")
- self.assertEqual(v.major, "0")
- self.assertEqual(v.minor, "1")
- self.assertFalse(v.tls)
- self.assertFalse(v.pcap)
-
- def test_version3(self):
- v = SIPpVersion("v3.1", "TLS")
- self.assertEqual(str(v), "v3.1-TLS")
- self.assertEqual(v.concept, "v3")
- self.assertEqual(v.major, "1")
- self.assertEqual(v.minor, None)
- self.assertTrue(v.tls)
- self.assertFalse(v.pcap)
-
- def test_version4(self):
- v = SIPpVersion("v2.0.1", "TLS-PCAP")
- self.assertEqual(str(v), "v2.0.1-TLS-PCAP")
- self.assertEqual(v.concept, "v2")
- self.assertEqual(v.major, "0")
- self.assertEqual(v.minor, "1")
- self.assertTrue(v.tls)
- self.assertTrue(v.pcap)
-
- def test_version5(self):
- v = SIPpVersion("v3.2", "PCAP")
- self.assertEqual(str(v), "v3.2-PCAP")
- self.assertEqual(v.concept, "v3")
- self.assertEqual(v.major, "2")
- self.assertEqual(v.minor, None)
- self.assertFalse(v.tls)
- self.assertTrue(v.pcap)
-
- def test_version6(self):
- v = SIPpVersion(None, "PCAP")
- self.assertEqual(str(v), "PCAP")
- self.assertEqual(v.concept, None)
- self.assertEqual(v.major, None)
- self.assertEqual(v.minor, None)
- self.assertFalse(v.tls)
- self.assertTrue(v.pcap)
-
- def test_version7(self):
- v = SIPpVersion(None, "TLS")
- self.assertEqual(str(v), "TLS")
- self.assertEqual(v.concept, None)
- self.assertEqual(v.major, None)
- self.assertEqual(v.minor, None)
- self.assertTrue(v.tls)
- self.assertFalse(v.pcap)
-
- def test_version8(self):
- v = SIPpVersion(None, "TLS-PCAP")
- self.assertEqual(str(v), "TLS-PCAP")
- self.assertEqual(v.concept, None)
- self.assertEqual(v.major, None)
- self.assertEqual(v.minor, None)
- self.assertTrue(v.tls)
- self.assertTrue(v.pcap)
-
- def test_cmp(self):
- v1 = SIPpVersion("v3.2", None)
- v2 = SIPpVersion("v3.1", None)
- self.assertTrue(v1 > v2)
-
- def test_cmp2(self):
- v1 = SIPpVersion("v2.0.1", None)
- v2 = SIPpVersion("v3.1", None)
- self.assertTrue(v1 < v2)
-
- def test_cmp3(self):
- v1 = SIPpVersion("v3.1", None)
- v2 = SIPpVersion("v3.1", None)
- self.assertTrue(v1 == v2)
-
- def test_cmp4(self):
- v1 = SIPpVersion("v3.1", None)
- v2 = SIPpVersion("v3.1", None)
- self.assertFalse(v1 != v2)
-
- def test_cmp5(self):
- v1 = SIPpVersion("v3.1", "TLS")
- v2 = SIPpVersion("v3.1", "TLS")
- self.assertTrue(v1 == v2)
-
- def test_cmp6(self):
- v1 = SIPpVersion(None, "TLS")
- v2 = SIPpVersion(None, "TLS")
- self.assertTrue(v1 == v2)
-
- def test_cmp7(self):
- v1 = SIPpVersion("v2.0.1", None)
- v2 = SIPpVersion("v2.0.1", None)
- self.assertTrue(v1 == v2)
-
- def test_cmp8(self):
- v1 = SIPpVersion("v3.2", "TLS")
- v2 = SIPpVersion("v3.2", "PCAP")
- self.assertTrue(v1 != v2)
-
- def test_cmp9(self):
- v1 = SIPpVersion(None, "TLS")
- v2 = SIPpVersion(None, "PCAP")
- self.assertTrue(v1 != v2)
-
- def test_cmp10(self):
- v1 = SIPpVersion("v3.2", "TLS")
- v2 = SIPpVersion("v3.2", "PCAP")
- self.assertFalse(v1 == v2)
-
- def test_cmp11(self):
- v1 = SIPpVersion(None, "TLS")
- v2 = SIPpVersion(None, "PCAP")
- self.assertFalse(v1 == v2)
-
- def test_cmp12(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion("v3.2", "TLS")
- self.assertTrue(v1 >= v2)
-
- def test_cmp13(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion(None, "TLS")
- self.assertTrue(v1 >= v2)
-
- def test_cmp14(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion("v3.2", "PCAP")
- self.assertTrue(v1 >= v2)
-
- def test_cmp15(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion(None, "PCAP")
- self.assertTrue(v1 >= v2)
-
- def test_cmp16(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion("v3.2", "TLS")
- self.assertTrue(v1 != v2)
-
- def test_cmp17(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion(None, "TLS")
- self.assertTrue(v1 != v2)
-
- def test_cmp18(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion("v3.2", "TLS")
- self.assertFalse(v1 == v2)
-
- def test_cmp19(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion(None, "TLS")
- self.assertFalse(v1 == v2)
-
- def test_cmp20(self):
- v1 = SIPpVersion("v3.1", "PCAP")
- v2 = SIPpVersion("v3.0", "TLS")
- self.assertTrue(v1 > v2)
-
- def test_cmp21(self):
- v1 = SIPpVersion("v3.2", "TLS-PCAP")
- v2 = SIPpVersion("v2.0.1", "PCAP")
- self.assertTrue(v1 >= v2)
-
- def test_cmp22(self):
- v1 = SIPpVersion("v3.1", "TLS")
- v2 = SIPpVersion("v3.1", "PCAP")
- self.assertFalse(v1 > v2)
-
- def test_cmp23(self):
- v1 = SIPpVersion("v3.1", "TLS")
- v2 = SIPpVersion("v3.1", "PCAP")
- self.assertFalse(v1 < v2)
-
- def test_cmp24(self):
- v1 = SIPpVersion("v3.2", "TLS")
- v2 = SIPpVersion("v3.1", "TLS-PCAP")
- self.assertTrue(v1 >= v2)
-
- def test_cmp25(self):
- v1 = SIPpVersion("v3.1", "TLS-PCAP")
- v2 = SIPpVersion("v3.2", "PCAP")
- self.assertTrue(v1 <= v2)
-
- def test_cmp26(self):
- v1 = SIPpVersion("v2.0.1", "TLS-PCAP")
- v2 = SIPpVersion("v3.0", "TLS")
- self.assertFalse(v1 >= v2)
-
-
-def main():
- unittest.main()
-
-
-if __name__ == "__main__":
- main()
diff --git a/lib/python/asterisk/syncami.py b/lib/python/asterisk/syncami.py
index 031ae99..5482400 100644
--- a/lib/python/asterisk/syncami.py
+++ b/lib/python/asterisk/syncami.py
@@ -8,7 +8,13 @@
the GNU General Public License Version 2.
"""
-from urllib import urlencode
+try:
+ # python 2 import
+ from urllib import urlencode
+except:
+ # python 3 import
+ from urllib.parse import urlencode
+
from email.parser import HeaderParser
try:
from httplib import *
@@ -87,7 +93,7 @@
if res.status != 200:
raise InvalidAMIResponse(res)
self.cookie = res.getheader('set-cookie', None)
- data = res.read()
+ data = res.read().decode('utf-8')
parser = HeaderParser()
return parser.parsestr(data)
diff --git a/lib/python/asterisk/test_case.py b/lib/python/asterisk/test_case.py
index 80321f7..7bf6d16 100644
--- a/lib/python/asterisk/test_case.py
+++ b/lib/python/asterisk/test_case.py
@@ -19,9 +19,9 @@
from twisted.python import log
from starpy import manager, fastagi
-from asterisk import Asterisk
-from test_config import TestConfig
-from test_conditions import TestConditionController
+from .asterisk import Asterisk
+from .test_config import TestConfig
+from .test_conditions import TestConditionController
try:
@@ -45,12 +45,12 @@
except:
msg = ("WARNING: failed to preserve existing loggers - some "
"logging statements may be missing")
- print msg
+ print(msg)
logging.config.fileConfig(config_file)
else:
msg = ("WARNING: no logging.conf file found; using default "
"configuration")
- print msg
+ print(msg)
logging.basicConfig(level=logging.DEBUG)
root_logger = logging.getLogger()
@@ -102,7 +102,7 @@
# for the rasterisk CLI connection. As a quick fix, we hash the path
# using md5, to make it unique enough.
self.realbase = self.test_name.replace("tests/", "", 1)
- self.base = md5(self.realbase).hexdigest()
+ self.base = md5(self.realbase.encode()).hexdigest()
# We provide a symlink to it from a named path.
named_dir = os.path.join(Asterisk.test_suite_root, self.realbase)
try:
diff --git a/lib/python/asterisk/test_conditions.py b/lib/python/asterisk/test_conditions.py
index dbf36a7..7053465 100644
--- a/lib/python/asterisk/test_conditions.py
+++ b/lib/python/asterisk/test_conditions.py
@@ -19,7 +19,7 @@
import logging
import logging.config
-from buildoptions import AsteriskBuildOptions
+from .buildoptions import AsteriskBuildOptions
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/lib/python/asterisk/test_config.py b/lib/python/asterisk/test_config.py
index 8fdb884..6d2318d 100644
--- a/lib/python/asterisk/test_config.py
+++ b/lib/python/asterisk/test_config.py
@@ -18,12 +18,13 @@
sys.path.append("lib/python")
-import test_suite_utils
+from . import test_suite_utils
+from .test_runner import load_and_parse_module
-from asterisk import Asterisk
-from buildoptions import AsteriskBuildOptions
-from sippversion import SIPpVersion
-from opensslversion import OpenSSLVersion
+from .asterisk import Asterisk
+from .buildoptions import AsteriskBuildOptions
+from .sippversion import SIPpVersion
+from .opensslversion import OpenSSLVersion
class TestConditionConfig(object):
@@ -70,14 +71,9 @@
def make_condition(self):
"""Build and return the condition object defined by this config"""
- parts = self.class_type_name.split('.')
- module = '.'.join(parts[:-1])
- if module != '':
- mod = __import__(module)
- for comp in parts[1:]:
- mod = getattr(mod, comp)
- obj = mod(self)
- return obj
+ mod = load_and_parse_module(self.class_type_name)
+ if mod is not None:
+ return mod(self)
return None
@@ -147,7 +143,7 @@
self.met = getattr(self, dir_method)()
found = True
if not found:
- print "Unknown custom dependency - '%s'" % self.name
+ print("Unknown custom dependency - '%s'" % self.name)
elif "asterisk" in dep:
if self.ast:
self.name = dep["asterisk"]
@@ -167,9 +163,9 @@
from test_case import PCAP_AVAILABLE
self.met = PCAP_AVAILABLE
else:
- print "Unknown dependency type specified:"
+ print("Unknown dependency type specified:")
for key in dep.keys():
- print key
+ print(key)
def depend_remote(self):
"""Check to see if we run against a remote instance of Asterisk"""
@@ -256,13 +252,13 @@
if self.asterisk_build_options:
return (self.asterisk_build_options.check_option(name))
else:
- print "Unable to evaluate build options: no build options found"
+ print("Unable to evaluate build options: no build options found")
return False
def _find_asterisk_module(self, name):
"""Determine if an Asterisk module exists"""
if not Dependency.ast:
- print "Unable to evaluate Asterisk modules: Asterisk not found"
+ print("Unable to evaluate Asterisk modules: Asterisk not found")
return False
if Dependency.ast.original_astmoddir == "":
@@ -333,8 +329,8 @@
if self.config is not None and 'exclude-tests' in self.config:
self.excluded_tests = self.config['exclude-tests']
else:
- print ("WARNING - test configuration [%s] not found in "
- "config file" % self.test_configuration)
+ print("WARNING - test configuration [%s] not found in "
+ "config file" % self.test_configuration)
def _process_testinfo(self):
"""Process the test information block"""
@@ -381,8 +377,8 @@
self.config = yaml.load(config_file)
if not self.config:
- print "ERROR: Failed to load configuration for test '%s'" % \
- self.test_name
+ print("ERROR: Failed to load configuration for test '%s'" %
+ self.test_name)
return
self._process_global_settings()
@@ -408,8 +404,8 @@
matches = [cond_def for cond_def in self.condition_definitions
if cond_def['name'] == conf['name']]
if len(matches) != 1:
- print ("Unknown or too many matches for condition: " +
- conf['name'])
+ print("Unknown or too many matches for condition: " +
+ conf['name'])
else:
pre_cond = TestConditionConfig(conf, matches[0], "Pre")
post_cond = TestConditionConfig(conf, matches[0], "Post")
diff --git a/lib/python/asterisk/test_runner.py b/lib/python/asterisk/test_runner.py
old mode 100755
new mode 100644
index 644d117..d278e02
--- a/lib/python/asterisk/test_runner.py
+++ b/lib/python/asterisk/test_runner.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
"""Module that spawns and manages running a test
This module provides an entry point, loading, and teardown of test
@@ -22,8 +21,6 @@
LOGGER = logging.getLogger('test_runner')
logging.basicConfig()
-
-sys.path.append('lib/python')
class TestModuleFinder(object):
@@ -144,9 +141,18 @@
module_name = ".".join(parts[:-1])
if not len(module_name):
- LOGGER.error("No module specified: %s" % module_name)
+ LOGGER.error("No module specified: %s" % typename)
return None
+ if os.path.exists('lib/python/asterisk/%s.py' % module_name):
+ # This is convoluted but required. lib/python/asterisk packages
+ # must be loaded using absolute package names and 'asterisk' must
+ # be included in the list of parts. We cannot simply prepend
+ # type_name from the start because this blocks load of modules
+ # that are local to the test (add-test-to-search-path: 'True').
+ module_name = 'asterisk.' + module_name
+ parts = ['asterisk'] + parts
+
module = __import__(module_name)
for comp in parts[1:]:
module = getattr(module, comp)
diff --git a/lib/python/asterisk/test_suite_utils.py b/lib/python/asterisk/test_suite_utils.py
index d8ef541..84f54f5 100644
--- a/lib/python/asterisk/test_suite_utils.py
+++ b/lib/python/asterisk/test_suite_utils.py
@@ -1,4 +1,3 @@
-#! /usr/bin/env python
"""Asterisk testsuite utils
This module provides access to Asterisk testsuite utility
@@ -14,15 +13,19 @@
import os
import logging
import re
+import sys
from os import close
from os import remove
from shutil import move
from tempfile import mkstemp
-from config import ConfigFile
+from .config import ConfigFile
LOGGER = logging.getLogger(__name__)
+
+if sys.version_info[0] == 3:
+ unicode = str
def which(program):
"""Find the executable for a specified program
@@ -100,7 +103,7 @@
elif isinstance(pattern, dict):
# Dict should match for every field in the pattern.
# extra fields in the message are fine.
- for key, value in pattern.iteritems():
+ for key, value in pattern.items():
to_check = message.get(key)
if to_check is None or not all_match(value, to_check):
return False
diff --git a/lib/python/asterisk/thread_test_condition.py b/lib/python/asterisk/thread_test_condition.py
index 595e3ce..c12321f 100644
--- a/lib/python/asterisk/thread_test_condition.py
+++ b/lib/python/asterisk/thread_test_condition.py
@@ -9,7 +9,7 @@
"""
import logging
-from test_conditions import TestCondition
+from .test_conditions import TestCondition
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/lib/python/asterisk/voicemail.py b/lib/python/asterisk/voicemail.py
index 283650b..3c4c7d5 100644
--- a/lib/python/asterisk/voicemail.py
+++ b/lib/python/asterisk/voicemail.py
@@ -20,9 +20,9 @@
import time
import random
-from config import ConfigFile
-from test_case import TestCase
-from test_state import TestState, TestStateController, FailureTestState
+from .config import ConfigFile
+from .test_case import TestCase
+from .test_state import TestState, TestStateController, FailureTestState
sys.path.append("lib/python")
diff --git a/lib/python/rlmi.py b/lib/python/rlmi.py
index 5e662a0..2abf19e 100644
--- a/lib/python/rlmi.py
+++ b/lib/python/rlmi.py
@@ -343,7 +343,7 @@
return None
@classmethod
def gds_reverse_node_mapping(cls, mapping):
- return dict(((v, k) for k, v in mapping.iteritems()))
+ return dict(((v, k) for k, v in mapping.items()))
#
diff --git a/lib/python/sip_message.py b/lib/python/sip_message.py
index 85cd9fd..087c760 100644
--- a/lib/python/sip_message.py
+++ b/lib/python/sip_message.py
@@ -118,7 +118,7 @@
def main():
msg = """INVITE sip:123 at example.com SIP/2.0\r\nContact : \tTerry Wilson\r\n <terry at example.com>\r\nCall-ID:\r\n Whatever\r\nContact: New Contact\r\n\r\nData!!!!!"""
sipmsg = SIPMessage(msg)
- print sipmsg
+ print(sipmsg)
if sipmsg.get_header('CoNtact') is None:
return -1
if len(sipmsg.get_header_all('contact')) != 2:
diff --git a/run-local b/run-local
index 026a063..326ac50 100755
--- a/run-local
+++ b/run-local
@@ -75,7 +75,12 @@
LIBDIR=`dirname $HERE/astroot/usr/lib*/libasteriskssl.so`
export LD_LIBRARY_PATH="${LIBDIR}${LD_LIBRARY_PATH}"
set +e
- ./runtests.py "$@"
+ if test -n "$PYTHON"; then
+ # Use specific interpreter
+ $PYTHON ./runtests.py "$@"
+ else
+ ./runtests.py "$@"
+ fi
status=$?
rm "$AST_TEST_ROOT"
set -e
diff --git a/runtests.py b/runtests.py
index 1351a3c..eac88a1 100755
--- a/runtests.py
+++ b/runtests.py
@@ -111,7 +111,7 @@
def stdout_print(self, msg):
self.stdout += msg + "\n"
- print msg
+ print(msg)
def run(self):
self.passed = False
@@ -124,13 +124,14 @@
]
if not os.path.exists(cmd[0]):
- cmd = ["./lib/python/asterisk/test_runner.py",
+ cmd = [sys.executable,
+ "-m", "asterisk.test_runner",
"%s" % self.test_name]
if os.path.exists(cmd[0]) and os.access(cmd[0], os.X_OK):
if self.options.pcap:
os.environ['PCAP'] = "yes"
- self.stdout_print("Running %s ..." % cmd)
+ self.stdout_print("Running %s ..." % self.test_name)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
self.pid = p.pid
@@ -139,6 +140,7 @@
poll.register(p.stdout, select.POLLIN)
timedout = False
+ has_unicode_error = False
try:
while (not abandon_test):
try:
@@ -148,20 +150,24 @@
except select.error as v:
if v[0] != errno.EINTR:
raise
- l = p.stdout.readline()
+ l = p.stdout.readline().decode('ascii', 'ignore').strip()
if not l:
break
self.stdout_print(l)
+ except UnicodeEncodeError:
+ self.stdout_print('Unicode error reading output from test!')
+ has_unicode_error = True
+ pass
except IOError:
pass
p.wait()
# Sanitize p.returncode so it's always a boolean.
- did_pass = (p.returncode == 0 and not abandon_test)
+ did_pass = (p.returncode == 0 and not abandon_test and not has_unicode_error)
if did_pass and not self.test_config.expect_pass:
self.stdout_print("Test passed but was expected to fail.")
if not did_pass and not self.test_config.expect_pass:
- print "Test failed as expected."
+ print("Test failed as expected.")
self.passed = (did_pass == self.test_config.expect_pass)
if abandon_test:
@@ -190,8 +196,8 @@
shutil.rmtree(absolute_dir)
os.remove(symlink_dir)
except:
- print "Unable to clean up directory for" \
- "test %s (non-fatal)" % self.test_name
+ print("Unable to clean up directory for"
+ "test %s (non-fatal)" % self.test_name)
self.__parse_run_output(self.stdout)
if timedout:
@@ -202,13 +208,13 @@
status = 'passed'
else:
status = 'failed'
- pass_str = 'Test %s %s\n' % (cmd, status)
- print pass_str
+ pass_str = 'Test %s %s\n' % (self.test_name, status)
+ print(pass_str)
if self.options.syslog:
syslog.syslog(pass_str)
else:
- print "FAILED TO EXECUTE %s, it must exist and be executable" % cmd
+ print("FAILED TO EXECUTE %s, it must exist and be executable" % cmd)
self.time = time.time() - start_time
def _check_for_core(self):
@@ -237,7 +243,7 @@
debug_level = email_config.get('debug', 0)
if not sender or len(recipients) == 0:
- print "--email-on-crash requires sender and 1+ recipients"
+ print("--email-on-crash requires sender and 1+ recipients")
return
with open(dest_file_name, 'r') as bt_file:
@@ -255,12 +261,12 @@
send_email(smtp_server, sender, recipients, message,
debug=debug_level)
except Exception as exception:
- print "Failed to send email\nError: {0}".format(exception)
+ print("Failed to send email\nError: {0}".format(exception))
def _archive_core_dumps(self, core_dumps):
for core in core_dumps:
if not os.path.exists(core):
- print "Unable to find core dump file %s, skipping" % core
+ print("Unable to find core dump file %s, skipping" % core)
continue
random_num = random.randint(0, 16000)
dest_dir = "./logs/%s" % self.test_relpath
@@ -275,36 +281,36 @@
"-ex", "thread apply all bt",
"--batch",
"-c", core]
- print "Running %s" % (" ".join(gdb_cmd),)
+ print("Running %s" % (" ".join(gdb_cmd),))
try:
res = subprocess.call(gdb_cmd, stdout=dest_file, stderr=subprocess.STDOUT)
if res != 0:
- print "error analyzing core dump; gdb exited with %d" % res
+ print("error analyzing core dump; gdb exited with %d" % res)
# Copy the backtrace over to the logs
- print "Archived backtrace: {0}".format(dest_file_name)
+ print("Archived backtrace: {0}".format(dest_file_name))
if self.options.email_on_crash:
self._email_crash_report(dest_file_name)
- except OSError, ose:
- print "OSError ([%d]: %s) occurred while executing %r" % \
- (ose.errno, ose.strerror, gdb_cmd)
+ except OSError as ose:
+ print("OSError ([%d]: %s) occurred while executing %r" %
+ (ose.errno, ose.strerror, gdb_cmd))
except:
- print "Unknown exception occurred while executing %r" % (gdb_cmd,)
+ print("Unknown exception occurred while executing %r" % (gdb_cmd,))
finally:
dest_file.close()
if self.options.keep_core:
try:
dst_core = os.path.join(dest_dir, "core_{0}".format(random_num))
shutil.copy(core, dst_core)
- print "Archived core file: {0}".format(dst_core)
+ print("Archived core file: {0}".format(dst_core))
except Exception as e:
- print "Error occurred while copying core: {0}".format(e)
+ print("Error occurred while copying core: {0}".format(e))
try:
os.unlink(core)
- except OSError, e:
- print "Error removing core file: %s: " \
- "Beware of the stale core file in CWD!" % (e,)
+ except OSError as e:
+ print("Error removing core file: %s: "
+ "Beware of the stale core file in CWD!" % (e,))
def _find_run_dirs(self):
test_run_dir = os.path.join(Asterisk.test_suite_root,
@@ -387,7 +393,7 @@
res = subprocess.call(refcounter,
stdout=dest_file,
stderr=subprocess.STDOUT)
- except Exception, e:
+ except Exception as e:
self.stdout_print("Exception occurred while processing REF_DEBUG")
finally:
dest_file.close()
@@ -415,10 +421,10 @@
srcfile = os.path.join(src_dir, filename)
if os.path.exists(srcfile):
hardlink_or_copy(srcfile, os.path.join(dest_dir, filename))
- except Exception, e:
- print "Exception occurred while archiving file '%s' to %s: %s" % (
+ except Exception as e:
+ print("Exception occurred while archiving file '%s' to %s: %s" % (
srcfile, dest_dir, e
- )
+ ))
def _archive_logs(self):
(run_num, run_dir, archive_dir) = self._find_run_dirs()
@@ -519,35 +525,35 @@
tags.sort(key=str.lower)
maxwidth = max(len(t) for t in tags)
- print "Available test tags:"
+ print("Available test tags:")
tags = chunks(tags, 3)
for tag in tags:
- print "\t%-*s %-*s %-*s" % (
+ print("\t%-*s %-*s %-*s" % (
maxwidth, tag[0],
maxwidth, len(tag) > 1 and tag[1] or '',
- maxwidth, len(tag) > 2 and tag[2] or '')
+ maxwidth, len(tag) > 2 and tag[2] or ''))
def list_tests(self):
- print "Configured tests:"
+ print("Configured tests:")
i = 1
for t in self.tests:
- print "%.3d) %s" % (i, t.test_config.test_name)
- print " --> Summary: %s" % t.test_config.summary
+ print("%.3d) %s" % (i, t.test_config.test_name))
+ print(" --> Summary: %s" % t.test_config.summary)
if t.test_config.skip is not None:
- print " --> Skip: %s" % t.test_config.skip
+ print(" --> Skip: %s" % t.test_config.skip)
if t.test_config.features:
- print " --> Features:"
+ print(" --> Features:")
for feature_name in t.test_config.features:
- print " --> %s: -- Met: %s" % \
- (feature_name, str(t.test_config.feature_check[feature_name]))
+ print(" --> %s: -- Met: %s" %
+ (feature_name, str(t.test_config.feature_check[feature_name])))
if t.test_config.tags:
- print " --> Tags: %s" % str(t.test_config.tags)
+ print(" --> Tags: %s" % str(t.test_config.tags))
for d in t.test_config.deps:
if d.version:
- print " --> Dependency: %s" % (d.name)
- print " --> Version: %s -- Met: %s" % (d.version, str(d.met))
+ print(" --> Dependency: %s" % (d.name))
+ print(" --> Version: %s -- Met: %s" % (d.version, str(d.met)))
else:
- print " --> Dependency: %s -- Met: %s" % (d.name, str(d.met))
+ print(" --> Dependency: %s -- Met: %s" % (d.name, str(d.met)))
i += 1
@@ -571,7 +577,7 @@
else:
deps += ("%s" % d.name)
- print "%04d %s %s%s" % (i, flag, t.test_config.test_name, deps)
+ print("%04d %s %s%s" % (i, flag, t.test_config.test_name, deps))
i += 1
def run(self):
@@ -587,8 +593,8 @@
if excluded in t.test_name:
continue
i += 1
- print "Tests to run: %d * %d time(s) = %d Maximum test inactivity time: %d sec." % \
- (i, self.options.number, i * self.options.number, (self.options.timeout / 1000))
+ print("Tests to run: %d * %d time(s) = %d Maximum test inactivity time: %d sec." %
+ (i, self.options.number, i * self.options.number, (self.options.timeout / 1000)))
for t in self.tests:
if abandon_test_suite:
@@ -596,18 +602,18 @@
if t.can_run is False:
if t.test_config.skip is not None:
- print "--> %s ... skipped '%s'" % (t.test_name, t.test_config.skip)
+ print("--> %s ... skipped '%s'" % (t.test_name, t.test_config.skip))
t.skipped_reason = t.test_config.skip
self.total_skipped += 1
continue
- print "--> Cannot run test '%s'" % t.test_name
+ print("--> Cannot run test '%s'" % t.test_name)
for f in t.test_config.features:
- print "--- --> Version Feature: %s - %s" % (
- f, str(t.test_config.feature_check[f]))
- print "--- --> Tags: %s" % (t.test_config.tags)
+ print("--- --> Version Feature: %s - %s" % (
+ f, str(t.test_config.feature_check[f])))
+ print("--- --> Tags: %s" % (t.test_config.tags))
for d in t.test_config.deps:
- print "--- --> Dependency: %s - %s" % (d.name, str(d.met))
- print
+ print("--- --> Dependency: %s - %s" % (d.name, str(d.met)))
+ print("")
self.total_skipped += 1
t.skipped_reason = "Failed dependency"
continue
@@ -615,14 +621,14 @@
exclude = False
for excluded in self.global_config.excluded_tests:
if excluded in t.test_name:
- print "--- ---> Excluded test: %s" % excluded
+ print("--- ---> Excluded test: %s" % excluded)
exclude = True
if exclude:
self.total_skipped += 1
continue
running_str = "--> Running test '%s' ..." % t.test_name
- print running_str
+ print(running_str)
if self.options.syslog:
syslog.syslog(running_str)
@@ -630,16 +636,16 @@
t.passed = True
else:
# Establish Preconditions
- print "Making sure Asterisk isn't running ..."
+ print("Making sure Asterisk isn't running ...")
if os.system("if pidof asterisk >/dev/null; then "
"killall -9 asterisk >/dev/null 2>&1; "
"sleep 1; ! pidof asterisk >/dev/null; fi"):
- print "Could not kill asterisk."
- print "Making sure SIPp isn't running..."
+ print("Could not kill asterisk.")
+ print("Making sure SIPp isn't running...")
if os.system("if pidof sipp >/dev/null; then "
"killall -9 sipp >/dev/null 2>&1; "
"sleep 1; ! pidof sipp >/dev/null; fi"):
- print "Could not kill sipp."
+ print("Could not kill sipp.")
# XXX TODO Hard coded path, gross.
os.system("rm -f /var/run/asterisk/asterisk.ctl")
os.system("rm -f /var/run/asterisk/asterisk.pid")
@@ -668,12 +674,12 @@
(0x86, 0x9f),
]
- char_list = []
+ tbl = {}
for r in bad_chars:
# we do +1 here to include the last item
for i in range(r[0], r[1] + 1):
- char_list.append(chr(i))
- return data.translate(None, ''.join(char_list))
+ tbl[chr(i)] = None
+ return data.translate(tbl)
def write_results_xml(self, doc, root):
@@ -728,10 +734,10 @@
except IOError:
# Ignore errors for the optional tests/custom folder.
if path != "tests/custom/tests.yaml":
- print "Failed to open %s" % path
+ print("Failed to open %s" % path)
return None
except:
- print "Unexpected error: %s" % sys.exc_info()[0]
+ print("Unexpected error: %s" % sys.exc_info()[0])
return None
config = yaml.load(f)
@@ -747,7 +753,7 @@
"""
global abandon_test_suite
- print "SIGUSR1 received; stopping test suite after current test..."
+ print("SIGUSR1 received; stopping test suite after current test...")
abandon_test_suite = True
@@ -760,7 +766,7 @@
global abandon_test
global abandon_test_suite
- print "SIGTREM received; abandoning current test and stopping..."
+ print("SIGTREM received; abandoning current test and stopping...")
abandon_test = True
abandon_test_suite = True
@@ -877,8 +883,8 @@
if options.valgrind:
if not ET:
- print "python lxml module not loaded, text summaries " \
- "from valgrind will not be produced.\n"
+ print("python lxml module not loaded, text summaries "
+ "from valgrind will not be produced.\n")
os.environ["VALGRIND_ENABLE"] = "true"
dom = xml.dom.getDOMImplementation()
@@ -894,7 +900,7 @@
running_str = "Running tests for Asterisk (run {0} of {1})...\n".format(
iteration + 1, options.number)
- print running_str
+ print(running_str)
if options.syslog:
syslog.syslog(running_str)
@@ -903,22 +909,22 @@
# If exactly one test was requested, then skip the summary.
if len(test_suite.tests) != 1:
- print "\n=== TEST RESULTS ===\n"
- print "PATH: %s\n" % os.getenv("PATH")
+ print("\n=== TEST RESULTS ===\n")
+ print("PATH: %s\n" % os.getenv("PATH"))
for t in test_suite.tests:
sys.stdout.write("--> %s --- " % t.test_name)
if t.did_run is False:
- print "SKIPPED"
+ print("SKIPPED")
for d in t.test_config.deps:
- print " --> Dependency: %s -- Met: %s" % (d.name, str(d.met))
+ print(" --> Dependency: %s -- Met: %s" % (d.name, str(d.met)))
if options.tags:
for t in t.test_config.tags:
- print " --> Tag: %s -- Met: %s" % (t, str(t in options.tags))
+ print(" --> Tag: %s -- Met: %s" % (t, str(t in options.tags)))
continue
if t.passed is True:
- print "PASSED"
+ print("PASSED")
else:
- print "FAILED"
+ print("FAILED")
iteration += 1
@@ -929,11 +935,11 @@
with open(TEST_RESULTS, "w") as f:
doc.writexml(f, addindent=" ", newl="\n", encoding="utf-8")
except IOError:
- print "Failed to open test results output file: %s" % TEST_RESULTS
+ print("Failed to open test results output file: %s" % TEST_RESULTS)
except:
- print "Unexpected error: %s" % sys.exc_info()[0]
- print "\n"
- print doc.toprettyxml(" ", encoding="utf-8")
+ print("Unexpected error: %s" % sys.exc_info()[0])
+ print("\n")
+ print(doc.toprettyxml(" ", encoding="utf-8").decode('utf-8', 'ignore'))
if options.syslog:
syslog.syslog("All tests concluded")
@@ -954,7 +960,7 @@
try:
os.link(source, destination)
- except OSError, e:
+ except OSError as e:
# Different partitions can cause hard links to fail (error 18),
# if there's a different error, bail out immediately.
if e.args[0] != errno.EXDEV:
diff --git a/self_test b/self_test
index 1936014..045c090 100755
--- a/self_test
+++ b/self_test
@@ -28,17 +28,3 @@
run_test $i python2 $PYTHON2
[ "${i#test2}" = "${i}" ] && run_test $i python3 $PYTHON3
done
-
-# Temporary code for running unit tests that are not compatible with python3
-run_legacy_test() {
- # Arguments: test_name python_name python_bin
- if test -n "$3"; then
- echo " ==> Executing $1 ($2)"
- $3 lib/python/asterisk/${1}.py
- fi
-}
-
-for i in buildoptions channel_test_condition sippversion; do
- run_legacy_test $i python $PYTHON
- run_legacy_test $i python2 $PYTHON2
-done
diff --git a/tests/cdr/cdr-tests.py b/tests/cdr/cdr-tests.py
index a284673..9090045 100644
--- a/tests/cdr/cdr-tests.py
+++ b/tests/cdr/cdr-tests.py
@@ -13,8 +13,8 @@
import time
sys.path.append("lib/python")
-from cdr import CDRModule
-from cdr import AsteriskCSVCDR
+from asterisk.cdr import CDRModule
+from asterisk.cdr import AsteriskCSVCDR
LOGGER = logging.getLogger(__name__)
diff --git a/tests/cdr/sqlite3/cdr_sqlite3.py b/tests/cdr/sqlite3/cdr_sqlite3.py
index 1def05e..32ed786 100644
--- a/tests/cdr/sqlite3/cdr_sqlite3.py
+++ b/tests/cdr/sqlite3/cdr_sqlite3.py
@@ -14,7 +14,7 @@
import re
sys.path.append("lib/python")
-from config import ConfigFile
+from asterisk.config import ConfigFile
LOGGER = logging.getLogger(__name__)
diff --git a/tests/channels/SIP/tcpauthlimit/sipp_scenario.py b/tests/channels/SIP/tcpauthlimit/sipp_scenario.py
index 76a21aa..55a26f5 100644
--- a/tests/channels/SIP/tcpauthlimit/sipp_scenario.py
+++ b/tests/channels/SIP/tcpauthlimit/sipp_scenario.py
@@ -9,7 +9,7 @@
import logging
-from sipp import SIPpScenario
+from asterisk.sipp import SIPpScenario
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv4/scenario_generator.py b/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv4/scenario_generator.py
index 2e2357a..4fcb67a 100755
--- a/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv4/scenario_generator.py
+++ b/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv4/scenario_generator.py
@@ -12,8 +12,8 @@
sys.path.append("lib/python")
-import test_suite_utils
-from sipp import ScenarioGenerator
+from asterisk import test_suite_utils
+from asterisk.sipp import ScenarioGenerator
LOGGER = logging.getLogger(__name__)
diff --git a/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv6/scenario_generator.py b/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv6/scenario_generator.py
index 2a52c96..541ba62 100755
--- a/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv6/scenario_generator.py
+++ b/tests/channels/pjsip/registration/inbound/nominal/contact_acl/ipv6/scenario_generator.py
@@ -12,8 +12,8 @@
sys.path.append("lib/python")
-import test_suite_utils
-from sipp import ScenarioGenerator
+from asterisk import test_suite_utils
+from asterisk.sipp import ScenarioGenerator
LOGGER = logging.getLogger(__name__)
diff --git a/tests/channels/pjsip/subscriptions/presence/verify_bodies/presence.py b/tests/channels/pjsip/subscriptions/presence/verify_bodies/presence.py
index 143ea42..bbff8f8 100644
--- a/tests/channels/pjsip/subscriptions/presence/verify_bodies/presence.py
+++ b/tests/channels/pjsip/subscriptions/presence/verify_bodies/presence.py
@@ -15,7 +15,7 @@
sys.path.append('lib/python')
-from pcap import VOIPListener
+from asterisk.pcap import VOIPListener
from twisted.internet import reactor
LOGGER = logging.getLogger(__name__)
diff --git a/tests/channels/pjsip/subscriptions/rls/rls_test.py b/tests/channels/pjsip/subscriptions/rls/rls_test.py
index 9c4051c..08d0567 100644
--- a/tests/channels/pjsip/subscriptions/rls/rls_test.py
+++ b/tests/channels/pjsip/subscriptions/rls/rls_test.py
@@ -13,7 +13,7 @@
sys.path.append("lib/python")
sys.path.append("tests/channels/pjsip/subscriptions/rls")
-from pcap import VOIPProxy
+from asterisk.pcap import VOIPProxy
from rls_element import RLSPacket
from rls_validation import ValidationInfo
from twisted.internet import reactor
diff --git a/tests/channels/pjsip/transfers/attended_transfer/nominal/packet_sniffer.py b/tests/channels/pjsip/transfers/attended_transfer/nominal/packet_sniffer.py
index a96a661..b49f268 100755
--- a/tests/channels/pjsip/transfers/attended_transfer/nominal/packet_sniffer.py
+++ b/tests/channels/pjsip/transfers/attended_transfer/nominal/packet_sniffer.py
@@ -35,8 +35,8 @@
sys.path.append('lib/python')
-from phones import PjsuaPhoneController
-from pcap import VOIPListener
+from asterisk.phones import PjsuaPhoneController
+from asterisk.pcap import VOIPListener
LOGGER = logging.getLogger(__name__)
diff --git a/tests/codecs/audio_analyzer.py b/tests/codecs/audio_analyzer.py
index 9903c6d..e656777 100644
--- a/tests/codecs/audio_analyzer.py
+++ b/tests/codecs/audio_analyzer.py
@@ -13,8 +13,8 @@
import logging
from twisted.internet import reactor
-import ami
-import ari
+from asterisk import ami
+from asterisk import ari
import tonetest
sys.path.append("lib/python")
diff --git a/tests/codecs/rtp_analyzer.py b/tests/codecs/rtp_analyzer.py
index e20cde3..74f161a 100644
--- a/tests/codecs/rtp_analyzer.py
+++ b/tests/codecs/rtp_analyzer.py
@@ -13,7 +13,7 @@
sys.path.append("lib/python")
-import pcap
+from asterisk import pcap
LOGGER = logging.getLogger(__name__)
diff --git a/tests/hep/hep_capture_node.py b/tests/hep/hep_capture_node.py
index f663056..3e7feae 100644
--- a/tests/hep/hep_capture_node.py
+++ b/tests/hep/hep_capture_node.py
@@ -28,7 +28,7 @@
LOGGER = logging.getLogger(__name__)
-from test_suite_utils import all_match
+from asterisk.test_suite_utils import all_match
def enum(**enums):
"""Make an enumeration out of the passed in values"""
diff --git a/tests/manager/config/ManagerConfigTest.py b/tests/manager/config/ManagerConfigTest.py
index b7dbd8a..fe2e65f 100644
--- a/tests/manager/config/ManagerConfigTest.py
+++ b/tests/manager/config/ManagerConfigTest.py
@@ -13,8 +13,8 @@
sys.path.append("lib/python")
-from test_case import TestCase
-from syncami import SyncAMI
+from asterisk.test_case import TestCase
+from asterisk.syncami import SyncAMI
LOGGER = logging.getLogger(__name__)
diff --git a/tests/manager/device_state_changed/ami_device_state.py b/tests/manager/device_state_changed/ami_device_state.py
index a3ac5f3..96d8021 100644
--- a/tests/manager/device_state_changed/ami_device_state.py
+++ b/tests/manager/device_state_changed/ami_device_state.py
@@ -8,7 +8,7 @@
'''
import logging
-from test_case import TestCase
+from asterisk.test_case import TestCase
LOGGER = logging.getLogger(__name__)
diff --git a/tests/manager/device_state_list/ami_device_state_list.py b/tests/manager/device_state_list/ami_device_state_list.py
index 5abd104..105780e 100755
--- a/tests/manager/device_state_list/ami_device_state_list.py
+++ b/tests/manager/device_state_list/ami_device_state_list.py
@@ -9,7 +9,7 @@
"""
import logging
-from test_case import TestCase
+from asterisk.test_case import TestCase
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/tests/manager/exten_state_list/ami_exten_state_list.py b/tests/manager/exten_state_list/ami_exten_state_list.py
index c9449f4..90bdc4f 100755
--- a/tests/manager/exten_state_list/ami_exten_state_list.py
+++ b/tests/manager/exten_state_list/ami_exten_state_list.py
@@ -9,7 +9,7 @@
"""
import logging
-from test_case import TestCase
+from asterisk.test_case import TestCase
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/tests/manager/presence_state_changed/ami_presence_state.py b/tests/manager/presence_state_changed/ami_presence_state.py
index e6861fc..d8ac551 100644
--- a/tests/manager/presence_state_changed/ami_presence_state.py
+++ b/tests/manager/presence_state_changed/ami_presence_state.py
@@ -8,7 +8,7 @@
'''
import logging
-from test_case import TestCase
+from asterisk.test_case import TestCase
LOGGER = logging.getLogger(__name__)
diff --git a/tests/manager/presence_state_list/ami_presence_state_list.py b/tests/manager/presence_state_list/ami_presence_state_list.py
index 13c16c7..f67c386 100755
--- a/tests/manager/presence_state_list/ami_presence_state_list.py
+++ b/tests/manager/presence_state_list/ami_presence_state_list.py
@@ -9,7 +9,7 @@
"""
import logging
-from test_case import TestCase
+from asterisk.test_case import TestCase
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/tests/pbx/manager_extensions/ami_extension_control.py b/tests/pbx/manager_extensions/ami_extension_control.py
index d4ac26c..01e53af 100755
--- a/tests/pbx/manager_extensions/ami_extension_control.py
+++ b/tests/pbx/manager_extensions/ami_extension_control.py
@@ -8,7 +8,7 @@
'''
import logging
-from test_case import TestCase
+from asterisk.test_case import TestCase
from twisted.internet import defer
LOGGER = logging.getLogger(__name__)
diff --git a/tests/rest_api/external_interaction/attended_transfer/attended_transfer.py b/tests/rest_api/external_interaction/attended_transfer/attended_transfer.py
index d03fbfe..6b7790c 100644
--- a/tests/rest_api/external_interaction/attended_transfer/attended_transfer.py
+++ b/tests/rest_api/external_interaction/attended_transfer/attended_transfer.py
@@ -9,7 +9,7 @@
import logging
from sys import path
-from sipp import SIPpScenario
+from asterisk.sipp import SIPpScenario
LOGGER = logging.getLogger(__name__)
diff --git a/tests/rest_api/external_interaction/blind_transfer/call_transfer.py b/tests/rest_api/external_interaction/blind_transfer/call_transfer.py
index 23f1a7b..ddef074 100755
--- a/tests/rest_api/external_interaction/blind_transfer/call_transfer.py
+++ b/tests/rest_api/external_interaction/blind_transfer/call_transfer.py
@@ -16,7 +16,7 @@
import pjsua as pj
from twisted.internet import reactor
-import pjsua_mod
+from asterisk import pjsua_mod
LOGGER = logging.getLogger(__name__)
THIS_FILE = os.path.basename(__file__)
diff --git a/tests/rest_api/message/message_modules.py b/tests/rest_api/message/message_modules.py
index 2e4cdd4..c2d8f2e 100644
--- a/tests/rest_api/message/message_modules.py
+++ b/tests/rest_api/message/message_modules.py
@@ -14,7 +14,7 @@
from twisted.internet import defer
-from sipp import SIPpScenario
+from asterisk.sipp import SIPpScenario
LOGGER = logging.getLogger(__name__)
diff --git a/usage.py b/usage.py
index 16dea94..9e7c33f 100755
--- a/usage.py
+++ b/usage.py
@@ -70,7 +70,7 @@
def occurances(self, **kwargs):
match = []
for test in self.tests:
- for key, value in kwargs.iteritems():
+ for key, value in kwargs.items():
if value in getattr(test, key):
match.append(test)
continue
@@ -78,14 +78,14 @@
return len(match)
def results_for(self, key):
- print key.title() + ":"
+ print(key.title() + ":")
things = self.unique(key)
width = max(len(t) for t in things)
results = [(self.occurances(**{key: t}), t) for t in things]
results.sort(key=lambda tup: tup[0], reverse=True)
for (count, name) in results:
- print "\t%-*s %5d" % (width, name, count)
- print ""
+ print("\t%-*s %5d" % (width, name, count))
+ print("")
def load_yaml_config(path):
@@ -95,10 +95,10 @@
except IOError:
# Ignore errors for the optional tests/custom folder.
if path != "tests/custom/tests.yaml":
- print "Failed to open %s" % path
+ print("Failed to open %s" % path)
return None
except:
- print "Unexpected error: %s" % sys.exc_info()[0]
+ print("Unexpected error: %s" % sys.exc_info()[0])
return None
config = yaml.load(f)
@@ -108,11 +108,11 @@
def main(argv=None):
- print "Testsuite Module Usage and Coverage Report"
- print ""
+ print("Testsuite Module Usage and Coverage Report")
+ print("")
test_suite = TestSuite()
- print "Number of tests:", len(test_suite.tests)
- print ""
+ print("Number of tests:", len(test_suite.tests))
+ print("")
test_suite.results_for('tags')
test_suite.results_for('test_objects')
test_suite.results_for('test_modules')
--
To view, visit https://gerrit.asterisk.org/9314
To unsubscribe, visit https://gerrit.asterisk.org/settings
Gerrit-Project: testsuite
Gerrit-Branch: 15
Gerrit-MessageType: newchange
Gerrit-Change-Id: If76c2d3e11e4ab4552d0df7841287c8bb2de7918
Gerrit-Change-Number: 9314
Gerrit-PatchSet: 1
Gerrit-Owner: Corey Farrell <git at cfware.com>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.digium.com/pipermail/asterisk-code-review/attachments/20180629/212d867b/attachment-0001.html>
More information about the asterisk-code-review
mailing list