[asterisk-commits] mjordan: branch mjordan/test_conditions r2189 - in /asterisk/team/mjordan/tes...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Mon Sep 12 12:47:50 CDT 2011
Author: mjordan
Date: Mon Sep 12 12:47:46 2011
New Revision: 2189
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=2189
Log:
Branch
Added:
asterisk/team/mjordan/test_conditions/trunk/ (props changed)
- copied from r2188, asterisk/trunk/
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py (with props)
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py (with props)
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/ThreadTestCondition.py (with props)
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/buildoptions.py (with props)
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/sippversion.py (with props)
Removed:
asterisk/team/mjordan/test_conditions/trunk/lib/python/sipp/
Modified:
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestCase.py
asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestState.py
asterisk/team/mjordan/test_conditions/trunk/runtests.py
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/authenticate_nominal/run-test
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/authenticate_nominal/test-config.yaml
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/check_voicemail_new_user/configs/ast1/extensions.conf
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/check_voicemail_new_user/run-test
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/check_voicemail_new_user/test-config.yaml
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/check_voicemail_nominal/configs/ast1/extensions.conf
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/check_voicemail_nominal/test-config.yaml
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/func_vmcount/configs/ast1/extensions.conf
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/func_vmcount/test-config.yaml
asterisk/team/mjordan/test_conditions/trunk/tests/apps/voicemail/leave_voicemail_nominal/test-config.yaml
Propchange: asterisk/team/mjordan/test_conditions/trunk/
------------------------------------------------------------------------------
reviewboard:url = https://reviewboard.asterisk.org
Propchange: asterisk/team/mjordan/test_conditions/trunk/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Mon Sep 12 12:47:46 2011
@@ -1,0 +1,2 @@
+asterisk-test-suite-report.xml
+*.pyc
Propchange: asterisk/team/mjordan/test_conditions/trunk/
------------------------------------------------------------------------------
svn:mergeinfo = /asterisk/trunk:1112
Modified: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestCase.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestCase.py?view=diff&rev=2189&r1=2188&r2=2189
==============================================================================
--- asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestCase.py (original)
+++ asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestCase.py Mon Sep 12 12:47:46 2011
@@ -12,26 +12,41 @@
import logging.config
import os
import datetime
+import time
from twisted.internet import reactor
from starpy import manager, fastagi
from asterisk import Asterisk
+from TestConfig import TestConfig
+from TestConditions import TestConditionController, TestCondition
+from ThreadTestCondition import ThreadPreTestCondition, ThreadPostTestCondition
logger = logging.getLogger(__name__)
class TestCase(object):
- ast = []
- ami = []
- fastagi = []
- reactor_timeout = 30
- passed = False
- defaultLogLevel = "WARN"
- defaultLogFileName = "logger.conf"
+ """
+ The base class object for python tests. This class provides common functionality to all
+ tests, including management of Asterisk instances, AMI, Twisted Reactor, and various
+ other utilities.
+ """
def __init__(self):
+ """
+ Create a new instance of a TestCase. Must be called by inheriting
+ classes.
+ """
+
self.test_name = os.path.dirname(sys.argv[0])
self.base = self.test_name.lstrip("tests/")
+ self.ast = []
+ self.ami = []
+ self.fastagi = []
+ self.reactor_timeout = 30
+ self.passed = False
+ self.defaultLogLevel = "WARN"
+ self.defaultLogFileName = "logger.conf"
self.timeoutId = None
+ self.test_config = TestConfig(self.test_name)
self.testStateController = None
""" Set up logging """
@@ -42,14 +57,43 @@
print "WARNING: no logging.conf file found; using default configuration"
logging.basicConfig(level=self.defaultLogLevel)
+ self.testConditionController = TestConditionController(self.test_config, self.ast, self.stop_reactor)
+ self.__setup_conditions()
+
logger.info("Executing " + self.test_name)
reactor.callWhenRunning(self.run)
+ def __setup_conditions(self):
+ """
+ Register pre and post-test conditions. Note that we have to first register condition checks
+ without related conditions, so that those that have dependencies can find them
+ """
+ self.conditions = self.test_config.get_conditions()
+ for c in self.conditions:
+ """ c is a 3-tuple of object, pre-post type, and related name """
+ if (c[2] == ""):
+ if (c[1] == "PRE"):
+ self.testConditionController.register_pre_test_condition(c[0])
+ elif (c[1] == "POST"):
+ self.testConditionController.register_post_test_condition(c[0])
+ else:
+ logger.warning("Unknown condition type [%s]" % c[1])
+ for c in self.conditions:
+ if (c[2] != ""):
+ if (c[1] == "POST"):
+ self.testConditionController.register_post_test_condition(c[0], c[2])
+ else:
+ logger.warning("Unsupported type [%s] with related condition %s" % (c[1], c[2]))
+ self.testConditionController.register_observer(self.handle_condition_failure, 'Failed')
+
def create_asterisk(self, count=1):
"""
-
- Keywork arguments:
- count --
+ Create n instances of Asterisk
+
+ Keyword arguments:
+ count -- the number of Asterisk instances to create. Each Asterisk instance
+ will be hosted on 127.0.0.x, where x is the 1-based index of the instance
+ created
"""
for c in range(count):
@@ -58,20 +102,25 @@
self.ast.append(Asterisk(base=self.base, host=host))
# Copy shared config files
self.ast[c].install_configs("%s/configs" %
- (self.test_name))
+ (self.test_name))
# Copy test specific config files
self.ast[c].install_configs("%s/configs/ast%d" %
- (self.test_name, c + 1))
+ (self.test_name, c + 1))
def create_ami_factory(self, count=1, username="user", secret="mysecret", port=5038):
"""
-
- Keywork arguments:
- count --
- username --
- secret --
- port --
- """
+ Create n instances of AMI. Each AMI instance will attempt to connect to
+ a previously created instance of Asterisk. When a connection is complete,
+ the ami_connect method will be called.
+
+ Keyword arguments:
+ count -- The number of instances of AMI to create
+ username -- The username to login with
+ secret -- The password to login with
+ port -- The port to connect over
+
+ """
+
for c in range(count):
host = "127.0.0.%d" % (c + 1)
self.ami.append(None)
@@ -82,19 +131,28 @@
def create_fastagi_factory(self, count=1):
"""
-
- """
+ Create n instances of AGI. Each AGI instance will attempt to connect to
+ a previously created instance of Asterisk. When a connection is complete,
+ the fastagi_connect method will be called.
+
+ Keyword arguments:
+ count -- The number of instances of AGI to create
+
+ """
+
for c in range(count):
host = "127.0.0.%d" % (c + 1)
self.fastagi.append(None)
logger.info("Creating FastAGI Factory %d" % (c + 1))
self.fastagi_factory = fastagi.FastAGIFactory(self.fastagi_connect)
reactor.listenTCP(4573, self.fastagi_factory,
- self.reactor_timeout, host)
+ self.reactor_timeout, host)
def start_asterisk(self):
"""
-
+ Start the instances of Asterisk that were previously created. See
+ create_asterisk. Note that this should be called before the reactor is
+ told to run.
"""
for index, item in enumerate(self.ast):
logger.info("Starting Asterisk instance %d" % (index + 1))
@@ -102,36 +160,59 @@
def stop_asterisk(self):
"""
-
- """
+ Stop the instances of Asterisk that were previously started. See
+ start_asterisk. Note that this should be called after the reactor has
+ returned from its run.
+ """
+ for amiInstance in self.ami:
+ try:
+ logger.debug("Logging off of AMI instance %d" % amiInstance.id)
+ amiInstance.logoff()
+ except:
+ logger.warning("Exception occurred while logging off of AMI instance %d" % amiInstance.id)
+
+ """ Pause for one second to allow dialplan to catch up with AMI logoffs """
+ time.sleep(1)
+ self.testConditionController.evaluate_post_checks()
for index, item in enumerate(self.ast):
logger.info("Stopping Asterisk instance %d" % (index + 1))
self.ast[index].stop()
def stop_reactor(self):
"""
-
+ Stop the reactor and cancel the test.
"""
logger.info("Stopping Reactor")
if reactor.running:
reactor.stop()
def __reactor_timeout(self):
- """
+ '''
A wrapper function for stop_reactor(), so we know when a reactor timeout
has occurred.
- """
+ '''
logger.warning("Reactor timeout: '%s' seconds" % self.reactor_timeout)
self.stop_reactor()
def run(self):
"""
-
- """
+ Base implementation of the test execution method, run. Derived classes
+ should override this and start their Asterisk dependent logic from this method.
+ Derived classes must call this implementation, as this method provides a fail
+ out mechanism in case the test hangs.
+ """
+ self.testConditionController.evaluate_pre_checks()
if (self.reactor_timeout > 0):
self.timeoutId = reactor.callLater(self.reactor_timeout, self.__reactor_timeout)
def ami_login_error(self, ami):
+ """
+ Handler for login errors into AMI. This will stop the test.
+
+ Keyword arguments:
+ ami -- The instance of AMI that raised the login error
+
+ """
logger.error("Error logging into AMI")
self.stop_reactor()
@@ -148,7 +229,14 @@
self.ami_connect(ami)
def handleOriginateFailure(self, reason):
- """ Convenience callback handler for twisted deferred errors for an AMI originate call """
+ """
+ Convenience callback handler for twisted deferred errors for an AMI originate call. Derived
+ classes can choose to add this handler to originate calls in order to handle them safely when
+ they fail. This will stop the test if called.
+
+ Keyword arguments:
+ reason -- The reason the originate failed
+ """
logger.error("Error sending originate:")
logger.error(reason.getTraceback())
self.stop_reactor()
@@ -163,3 +251,13 @@
self.timeoutId.reset(self.reactor_timeout)
newTime = datetime.datetime.fromtimestamp(self.timeoutId.getTime())
logger.info("Reactor timeout originally scheduled for %s, rescheduled for %s" % (str(originalTime), str(newTime)))
+
+ def handle_condition_failure(self, test_condition):
+ """
+ Callback handler for condition failures
+ """
+ if test_condition.pass_expected:
+ logger.error("Test Condition %s failed; setting test passed status to False" % test_condition.getName())
+ self.passed = False
+ else:
+ logger.info("Test Condition %s failed but expected failure was set; test status not modified" % test_condition.getName())
Added: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py?view=auto&rev=2189
==============================================================================
--- asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py (added)
+++ asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py Mon Sep 12 12:47:46 2011
@@ -1,0 +1,264 @@
+#!/usr/bin/env python
+'''
+Copyright (C) 2011, Digium, Inc.
+Matt Jordan <mjordan at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+'''
+
+import logging
+import logging.config
+
+from buildoptions import AsteriskBuildOptions
+from TestConfig import TestConfig
+
+logger = logging.getLogger(__name__)
+
+def enum(**enums):
+ return type('Enum', (), enums)
+
+class TestConditionController(object):
+ """
+ This class manages the pre and post-test condition checking. Pre and post-test
+ conditions can be added to this controller and will be automatically called
+ before and after a test runs. Based on the pass / fail of the test, the test
+ will be automatically stopped, halted, or otherwise affected.
+ """
+
+ def __init__(self, test_config, asterisk_instances = [], stop_test_callback = None):
+ """
+ Initialize the TestConditionController
+
+ Keyword arguments:
+ test_config -- the TestConfig object containing the configuration for this test
+ asterisk_instances -- Predefined instances of Asterisk that already exist
+ stop_test_callback -- The method to call when a test should be halted
+ """
+ self.__prechecks = []
+ self.__postchecks = []
+ self.__ast = asterisk_instances
+ self.__stop_test_callback = stop_test_callback
+ self.__observers = []
+ self.register_observer(self.__handle_condition_failure, "Failed")
+ self.register_observer(self.__handle_condition_failure, "Inconclusive")
+ self.test_config = test_config
+
+ def register_asterisk_instance(self, ast):
+ """
+ Registers an Asterisk instance with the controller. All instances of
+ Asterisk registered with the controller will be passed to the pre and post
+ test condition objects for condition checking.
+
+ Keyword arguments:
+ ast -- The asterisk instance to register for pre/post test checks
+ """
+ self.__ast.append(ast)
+
+ def register_pre_test_condition(self, pre_test_condition):
+ """
+ Register a pre-test conditional check. These test conditions will be evaluated
+ prior to execution of the main test body.
+
+ Keyword arguments:
+ pre_test_condition -- The pre-test condition to check
+ """
+ t = pre_test_condition, None
+ self.__prechecks.append(t)
+
+ def register_post_test_condition(self, post_test_condition, matching_pre_condition_name = ""):
+ """
+ Register a post-test conditional check. These test conditions are evaluated after
+ execution of the main test body.
+
+ Keyword arguments:
+ post_test_condition -- The post-test condition to check
+ matching_pre_condition_name -- A matching pre-test condition name to be passed to the post-test condition
+ during its evaluation
+
+ Note: will throw ValueError if matching_pre_condition_name is specified and is not
+ registered with the controller
+ """
+ matching_pre_condition = None
+ if (matching_pre_condition_name != ""):
+ for pre in self.__prechecks:
+ if pre[0].getName() == matching_pre_condition_name:
+ matching_pre_condition = pre[0]
+ break;
+ if (matching_pre_condition == None):
+ errMsg = "No pre condition found matching %s" % matching_pre_condition_name
+ raise ValueError(errMsg)
+ t = post_test_condition, matching_pre_condition
+ self.__postchecks.append(t)
+
+ def register_observer(self, callback_method, filter=""):
+ """
+ Register an observer for the pre- and post-test condition checks. An observer
+ will be called if a pre or post-test condition check matches the passed in filter.
+ If a filter is not provided, the observer will always be called.
+ """
+ t = callback_method, filter
+ logger.debug("Registering: " + str(callback_method))
+ self.__observers.append(t)
+
+ def evaluate_pre_checks(self):
+ """
+ Evaluate the pre-test conditions
+ """
+ logger.debug("Evaluating pre checks")
+ self.__evaluate_checks(self.__prechecks)
+
+ def evaluate_post_checks(self):
+ """
+ Evaluate the post-test conditions
+ """
+ logger.debug("Evaluating post checks")
+ self.__evaluate_checks(self.__postchecks)
+
+ def __evaluate_checks(self, check_list):
+ """ Register the instances of Asterisk and evaluate """
+ for check in check_list:
+ if (check[0].check_build_options()):
+ for ast in self.__ast:
+ check[0].register_asterisk_instance(ast)
+
+ logger.debug("Evaluating %s" % check[0].getName())
+ if (check[1] != None):
+ check[0].evaluate(check[1])
+ else:
+ check[0].evaluate()
+ self.__check_observers(check[0])
+
+ def __check_observers(self, test_condition):
+ for observerTuple in self.__observers:
+ if (observerTuple[1] == "" or (observerTuple[1] != "" and observerTuple[1] == test_condition.getStatus())):
+ observerTuple[0](test_condition)
+
+ if test_condition.getStatus() == 'Failed' and self.__stop_test_callback != None:
+ self.__stop_test_callback()
+
+ def __handle_condition_failure(self, test_condition):
+ if (test_condition.getStatus() == 'Inconclusive'):
+ logger.warning(test_condition)
+ elif (test_condition.getStatus() == 'Failed'):
+ if test_condition.pass_expected:
+ logger.error(test_condition)
+ else:
+ logger.warning(test_condition)
+
+TestStatuses = enum(Inconclusive='Inconclusive', Failed='Failed', Passed='Passed')
+
+class TestCondition(object):
+ """
+ The test condition base class. This object provides the evaluate method signature
+ which must be overriden, and various helper methods to print out and save the status
+ of the test condition check
+ """
+
+ __buildOptions = AsteriskBuildOptions()
+
+ def __init__(self, name):
+ """
+ Initialize a new test condition
+
+ Keyword arguments:
+ name -- The name of the condition that is being checked
+ """
+ self.failureReasons = []
+ self.__name = name
+ self.__testStatus = TestStatuses.Inconclusive
+ self.ast = []
+ self.build_options = []
+ self.pass_expected = True
+
+ def __str__(self):
+ """
+ Convert the object to a string representation
+ """
+ status = "Test Condition [%s]: [%s]" % (self.__name, self.__testStatus)
+ if (self.__testStatus == TestStatuses.Failed):
+ for reason in self.failureReasons:
+ status += "\n\tReason: %s" % reason
+ return status
+
+ def check_build_options(self):
+ """
+ Checks the required Asterisk Build Options for this TestCondition.
+ Returns true if all conditions are met, false if a condition was not met.
+ """
+ retVal = True
+ for option in self.build_options:
+ if not TestCondition.__buildOptions.checkOption(option[0], option[1]):
+ logger.debug("Build option %s not set to %s; test condition [%s] will not be checked" % (option[0], option[1], self.__name))
+ retVal = False
+ return retVal
+
+ def add_build_option(self, optionName, expectedValue="1"):
+ """
+ Adds a build option in Asterisk to check for before executing this condition.
+
+ Keyword arguments:
+ optionName -- The name of the Asterisk build option to check for
+ expectedValue -- The value the build option must have in order for this condition to execute
+ """
+ o = optionName, expectedValue
+ self.build_options.append(o)
+
+ def register_asterisk_instance(self, ast):
+ """
+ Registers an instance of asterisk to be tested by this condition. Note that
+ you are guaranteed to have all instances of Asterisk that should be checked
+ by the condition by the time the evaluate method is called.
+
+ Keyword arguments:
+ ast -- an instance of Asterisk to check
+ """
+ self.ast.append(ast)
+
+ def evaluate(self, related_test_condition = None):
+ """
+ Evaluate the test condition
+
+ Derived classes must implement this method and check their test condition
+ here. If the test condition passes they should call passCheck, otherwise they
+ should call failCheck.
+
+ Keyword arguments:
+ related_test_condition -- A test condition object that is related to this one. Provided if specified
+ when this test condition is registered to the test condition controller.
+ """
+ pass
+
+ def getName(self):
+ """
+ Return the name of the test condition
+ """
+ return self.__name
+
+ def getStatus(self):
+ """
+ Return the current status of the test
+ """
+ return str(self.__testStatus)
+
+ def passCheck(self):
+ """
+ Mark that the test condition has passed. Note that this will not overwrite
+ a previous indication that the test condition failed.
+ """
+ if (self.__testStatus == TestStatuses.Inconclusive):
+ self.__testStatus = TestStatuses.Passed
+
+ def failCheck(self, reason = ""):
+ """
+ Mark that the test condition has failed. Note that the test condition cannot
+ be changed once placed in this state.
+
+ Keyword arguments:
+ reason -- The reason the test failed
+ """
+ self.__testStatus = TestStatuses.Failed
+ if (reason != ""):
+ self.failureReasons.append(reason)
+
+
Propchange: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConditions.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py?view=auto&rev=2189
==============================================================================
--- asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py (added)
+++ asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py Mon Sep 12 12:47:46 2011
@@ -1,0 +1,363 @@
+#!/usr/bin/env python
+'''Asterisk external test suite driver.
+
+Copyright (C) 2011, Digium, Inc.
+Russell Bryant <russell at digium.com>
+Matt Jordan <mjordan at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+'''
+
+import sys
+import os
+import subprocess
+import optparse
+import time
+import yaml
+import socket
+
+sys.path.append("lib/python")
+
+import utils
+from version import AsteriskVersion
+from asterisk import Asterisk
+from buildoptions import AsteriskBuildOptions
+from sippversion import SIPpVersion
+
+class TestConditionConfig:
+ """
+ This class creates a test condition config and will build up an
+ object that derives from TestCondition based on that configuration
+ """
+
+ def __init__(self, config):
+ """
+ Construct a new test condition from a config sequence constructed
+ from the test-config.yaml file
+ """
+ self.classTypeName = ""
+ self.passExpected = True
+ self.type = ""
+ self.relatedCondition = ""
+ if 'name' in config:
+ self.classTypeName = config['name']
+ if 'expectedResult' in config:
+ try:
+ self.passExpected = not (config["expectedResult"].upper().strip() == "FAIL")
+ except:
+ self.passExpected = False
+ print "ERROR: '%s' is not a valid value for expectedResult" % config["expectedResult"]
+ if 'type' in config:
+ self.type = config['type'].upper().strip()
+ if 'relatedCondition' in config:
+ self.relatedCondition = config['relatedCondition'].strip()
+
+ def get_type(self):
+ """
+ The type of TestCondition (either PRE or POST)
+ """
+ return self.type
+
+ def get_related_condition(self):
+ """
+ The type name of the related condition that this condition will use
+ """
+ return self.relatedCondition
+
+ def make_condition(self):
+ """
+ Build up and return the condition object defined by this configuration
+ """
+ parts = self.classTypeName.split('.')
+ module = ".".join(parts[:-1])
+ if module != "":
+ m = __import__(module)
+ for comp in parts[1:]:
+ m = getattr(m, comp)
+ obj = m()
+ if not self.passExpected:
+ obj.pass_expected = False
+ return obj
+ return None
+
+
+class Dependency:
+ """
+ This class checks and stores the dependencies for a particular Test.
+ """
+
+ __asteriskBuildOptions = AsteriskBuildOptions()
+
+ def __init__(self, dep):
+ """
+ Construct a new dependency
+
+ Keyword arguments:
+ dep -- A tuple containing the dependency type name and its subinformation.
+ """
+
+ self.name = ""
+ self.version = ""
+ self.met = False
+ if "app" in dep:
+ self.name = dep["app"]
+ self.met = utils.which(self.name) is not None
+ elif "python" in dep:
+ self.name = dep["python"]
+ try:
+ __import__(self.name)
+ self.met = True
+ except ImportError:
+ pass
+ elif "sipp" in dep:
+ self.name = "SIPp"
+ version = None
+ feature = None
+ if 'version' in dep['sipp']:
+ version = dep['sipp']['version']
+ if 'feature' in dep['sipp']:
+ feature = dep['sipp']['feature']
+ self.sipp_version = SIPpVersion()
+ self.version = SIPpVersion(version, feature)
+ if self.sipp_version >= self.version:
+ self.met = True
+ if self.version.tls and not self.sipp_version.tls:
+ self.met = False
+ if self.version.pcap and not self.sipp_version.pcap:
+ self.met = False
+ elif "custom" in dep:
+ self.name = dep["custom"]
+ method = "depend_%s" % self.name
+ found = False
+ for m in dir(self):
+ if m == method:
+ self.met = getattr(self, m)()
+ found = True
+ if not found:
+ print "Unknown custom dependency - '%s'" % self.name
+ elif "asterisk" in dep:
+ self.name = dep["asterisk"]
+ self.met = self.__find_asterisk_module(self.name)
+ elif "buildflag" in dep:
+ self.name = dep["buildflag"]
+ self.met = self.__find_build_flag(self.name)
+ else:
+ print "Unknown dependency type specified:"
+ for key in dep.keys():
+ print key
+
+ def depend_soundcard(self):
+ """
+ Check the soundcard custom dependency
+ """
+ try:
+ f = open("/dev/dsp", "r")
+ f.close()
+ return True
+ except:
+ return False
+
+ def depend_ipv6(self):
+ """
+ Check the ipv6 custom dependency
+ """
+ try:
+ s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+ s.close()
+ return True
+ except:
+ return False
+
+ def depend_pjsuav6(self):
+ '''
+ This tests if pjsua was compiled with IPv6 support. To do this,
+ we run pjsua --help and parse the output to determine if --ipv6
+ is a valid option
+ '''
+ if utils.which('pjsua') is None:
+ return False
+
+ help_output = subprocess.Popen(['pjsua', '--help'],
+ stdout=subprocess.PIPE).communicate()[0]
+ if help_output.find('--ipv6') == -1:
+ return False
+ return True
+
+ def depend_fax(self):
+ """
+ Checks if Asterisk contains the necessary modules for fax tests
+ """
+ fax_providers = [
+ "app_fax",
+ "res_fax_spandsp",
+ "res_fax_digium",
+ ]
+
+ for f in fax_providers:
+ if self.__find_asterisk_module(f):
+ return True
+
+ return False
+
+ def __find_build_flag(self, name):
+ return (self.__asteriskBuildOptions.checkOption(name))
+
+ def __find_asterisk_module(self, name):
+ ast = Asterisk()
+ if "astmoddir" not in ast.directories:
+ return False
+
+ module = "%s/%s.so" % (ast.directories["astmoddir"], name)
+ if os.path.exists(module):
+ return True
+
+ return False
+
+
+class TestConfig:
+ """
+ Class that contains the configuration for a specific test, as parsed
+ by that tests test.yaml file.
+ """
+
+ def __init__(self, test_name):
+ """
+ Create a new TestConfig
+
+ Keyword arguments:
+ test_name -- The path to the directory containing the test-config.yaml file to load
+ """
+ self.can_run = True
+ self.test_name = test_name
+ self.skip = None
+ self.config = None
+ self.summary = None
+ self.maxversion = None
+ self.maxversion_check = False
+ self.minversion = None
+ self.minversion_check = False
+ self.deps = []
+ self.expectPass = True
+
+ self.__parse_config()
+
+ def __process_testinfo(self):
+ self.summary = "(none)"
+ self.description = "(none)"
+ if "testinfo" not in self.config:
+ return
+ testinfo = self.config["testinfo"]
+ if "skip" in testinfo:
+ self.skip = testinfo['skip']
+ self.can_run = False
+ if "summary" in testinfo:
+ self.summary = testinfo["summary"]
+ if "description" in testinfo:
+ self.description = testinfo["description"]
+
+ def __process_properties(self):
+ self.minversion = AsteriskVersion("1.4")
+ if "properties" not in self.config:
+ return
+ properties = self.config["properties"]
+ if "minversion" in properties:
+ try:
+ self.minversion = AsteriskVersion(properties["minversion"])
+ except:
+ self.can_run = False
+ print "ERROR: '%s' is not a valid minversion" % \
+ properties["minversion"]
+ if "maxversion" in properties:
+ try:
+ self.maxversion = AsteriskVersion(properties["maxversion"])
+ except:
+ self.can_run = False
+ print "ERROR: '%s' is not a valid maxversion" % \
+ properties["maxversion"]
+ if "expectedResult" in properties:
+ try:
+ self.expectPass = not (properties["expectedResult"].upper().strip() == "FAIL")
+ except:
+ self.can_run = False
+ print "ERROR: '%s' is not a valid value for expectedResult" %\
+ properties["expectedResult"]
+
+ def __parse_config(self):
+ test_config = "%s/test-config.yaml" % self.test_name
+ try:
+ f = open(test_config, "r")
+ except IOError:
+ print "Failed to open %s" % test_config
+ return
+ except:
+ print "Unexpected error: %s" % sys.exc_info()[0]
+ return
+
+ self.config = yaml.load(f)
+ f.close()
+
+ if not self.config:
+ print "ERROR: Failed to load configuration for test '%s'" % \
+ self.test_name
+ return
+
+ self.__process_testinfo()
+ self.__process_properties()
+
+ def get_conditions(self):
+ """
+ Gets the pre- and post-test conditions associated with this test
+
+ Returns a list of 3-tuples containing the following:
+ 0: The actual condition object
+ 1: The condition type (either PRE or POST)
+ 2: The name of the related condition that this one depends on
+ """
+ conditions = []
+ if not self.config:
+ return conditions
+
+ conditionsTemp = [TestConditionConfig(c) for c in self.config["properties"].get("testconditions") or [] ]
+
+ for cond in conditionsTemp:
+ c = cond.make_condition(), cond.get_type(), cond.get_related_condition()
+ conditions.append(c)
+
+ return conditions
+
+ def check_deps(self, ast_version):
+ """
+ Check whether or not a test should execute based on its configured dependencies
+
+ Keyword arguments:
+ ast_version -- The AsteriskVersion object containing the version of Asterisk that
+ will be executed
+ returns can_run - True if the test can execute, False otherwise
+ """
+
+ if not self.config:
+ return False
+
+ self.deps = [
+ Dependency(d)
+ for d in self.config["properties"].get("dependencies") or []
+ ]
+
+ self.minversion_check = True
+ if ast_version < self.minversion:
+ self.can_run = False
+ self.minversion_check = False
+ return self.can_run
+
+ self.maxversion_check = True
+ if self.maxversion is not None and ast_version > self.maxversion:
+ self.can_run = False
+ self.maxversion_check = False
+ return self.can_run
+
+ for d in self.deps:
+ if d.met is False:
+ self.can_run = False
+ break
+ return self.can_run
Propchange: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestConfig.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestState.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestState.py?view=diff&rev=2189&r1=2188&r2=2189
==============================================================================
--- asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestState.py (original)
+++ asterisk/team/mjordan/test_conditions/trunk/lib/python/asterisk/TestState.py Mon Sep 12 12:47:46 2011
@@ -8,24 +8,20 @@
'''
import sys
-import datetime
import logging
-
-from datetime import datetime
logger = logging.getLogger(__name__)
-"""
-The controller for the TestEvent state machine
-"""
class TestStateController(object):
+ """
+ The controller for the TestEvent state machine
+ """
-
- """
- testCase The TestCase derived class that owns this controller
- amiReceiver The AMI instance that will send the controller TestEvent notifications
- """
def __init__(self, testCase, amiReceiver):
+ """
+ testCase The TestCase derived class that owns this controller
+ amiReceiver The AMI instance that will send the controller TestEvent notifications
+ """
self.__testCase = testCase
self.__currentState = None
self.__assertHandler = None
@@ -33,26 +29,23 @@
""" Register for TestEvent updates """
amiReceiver.registerEvent('TestEvent', self.handleTestEvent)
+ def printTestEvent(self, event):
+ """
+ Print out a test event
- """
- Print out a test event
-
- event The TestEvent
- """
- def printTestEvent(self, event):
+ event The TestEvent
+ """
logger.debug(" Test Event received:")
for k, v in event.items():
logger.debug("\t" + k + "\t=\t" + v)
+ def handleTestEvent(self, ami, event):
+ """
+ Handler for a TestEvent
- """
- Handler for a TestEvent
-
- ami The AMI instance that sent us the TestEvent
- event The TestEvent
- """
- def handleTestEvent(self, ami, event):
-
+ ami The AMI instance that sent us the TestEvent
+ event The TestEvent
+ """
if (logger.getEffectiveLevel() == logging.DEBUG):
self.printTestEvent(event)
@@ -69,80 +62,75 @@
logger.warn("ASSERT received but no handler defined; test will now fail")
self.failTest()
+ def changeState(self, testState):
+ """
+ Change the current state machine state to a new state
- """
- Change the current state machine state to a new state
-
- testState The TestState to change to
- """
- def changeState(self, testState):
+ testState The TestState to change to
+ """
self.__currentState = testState
-
- """
- Fail and stop the test
- """
def failTest(self):
+ """
[... 1266 lines stripped ...]
More information about the asterisk-commits
mailing list