Recap:
So, in the last two posts (Part 1 and Part 2), we've covered the motivations for a pluggable framework for the Asterisk Test Suite, and we've taken a test (cdr_userfield) and made it completely configuration driven. This same approach works well for most of the other CDR tests as well - such as cdr_accountcode, which was also mentioned in the first post. Both of these tests (and others) can be driven by the configuration we've defined for SimpleTestCase and the configuration we've defined for our CDRModule.
You'll note that I said "most of the other CDR tests".
We do need to inform TestRunner that it should load our test's module (ForkCdrModule) and instantiate the object (ForkCdrModuleEndTime) from some location other than the normal Asterisk Test Suite Python library location. To do that, we add a new YAML key, load-from-path, and specify the path to the module:
Now, in TestRunner, we have add a small portion to the usual pluggable module loading to look for this key and, if so, add the path to the TestModuleFinder class's search paths:You'll note that I said "most of the other CDR tests".
A Monkey in the Works
Enter the monkey wrench that are the ForkCDR tests. Lets take a look at one, cdr_fork_end_time. First, we have some expectations that are added in the constructor of the test:
class ForkEndTimingTest(CDRTestCase):
def __init__(self):
CDRTestCase.__init__(self)
self.add_expectation('cdrtest_local', AsteriskCSVCDRLine(
destination = "1",
lastapp = "ForkCDR",
dcontext = "default",
dchannel = "SIP/test-.*",
channel = "Local/1@default-.*",
disposition = "ANSWERED",
amaflags = "DOCUMENTATION"))
self.add_expectation('cdrtest_local', AsteriskCSVCDRLine(
destination = "1",
lastapp = "Hangup",
dcontext = "default",
dchannel = "SIP/test-.*",
channel = "Local/1@default-.*",
disposition = "NO ANSWER",
amaflags = "DOCUMENTATION"))
Well, that's not too bad - we can easily express that in the configuration for a CDRModule instance, by specifying two expected lines for file cdrtest_local. Unfortunately, there's more:
def match_cdrs(self):
CDRTestCase.match_cdrs(self)
if not self.passed:
return
cdr1 = AsteriskCSVCDR(fn = "%s/%s/cdr-csv/%s.csv" % (self.ast[0].base, self.ast[0].directories['astlogdir'], "cdrtest_local"))
#check for missing fields
for cdritem in cdr1:
if cdritem.duration is None or cdritem.start is None or cdritem.end is None:
logger.Error("EPIC FAILURE: CDR record %s is missing one or more key fields. This should never be able to happen." % cdritem)
self.passed = False
return
# The dialplan is set up so that these two CDRs should each last at least 4 seconds. Giving it wiggle room,
# we'll just say we want it to be greater than 1 second.
if ((int(cdr1[0].duration) <= 1) or (int(cdr1[1].duration) <= 1)):
logger.error("FAILURE: One or both CDRs only lasted a second or less (expected more)")
self.passed = False
return
end = time.strptime(cdr1[0].end, "%Y-%m-%d %H:%M:%S")
beg = time.strptime(cdr1[1].start, "%Y-%m-%d %H:%M:%S")
#check that the end of the first CDR occured within a 1 second split of the beginning of the second CDR
if (abs(time.mktime(end) - time.mktime(beg)) > 1):
logger.error("Time discrepency between end1 and start2 must be one second or less.\n")
logger.error("Actual times: end cdr1 = %s begin cdr2 = %s" % (cdr1[0].end, cdr1[1].start))
self.passed = False
return
Rut roh Shaggy. We now have custom logic in method match_cdrs that is pretty specific to this test. Here is what this method does that our CDRModule doesn't do:
- It ensures that certain field were left in each line
- Verifies that each duration is greater than 1 second
- Checks that the end of the first CDR entry is within one second of the beginning of the second CDR entry
The first two could be configuration driven in CDRModule; the third, on the other hand, is definitely specific to the usage of ForkCDR. It doesn't make sense for those behaviors to be exposed to other CDR tests.
So what do we do?
 |
| I look cheeky and fun, but I'm going to screw up your well planned architecture |
Overriding CDRModule
Ideally, ForkCDR tests would provide their own module that was specific only to their test. We wouldn't want to put them in the general purpose library because, well, the logic isn't general purpose - its specific to only a very few tests. Lets assume that we'll provide a concrete implementation of CDRModule, and extend it to provide the ForkCDR specific logic. You'll remember the CDRModule already provides a method, match_cdrs, that does the matching between the actual CDR CSV lines and what our expected results. We can put the logic of cdr_fork_end_time's match_cdrs into its own override of that method, call the base class's implementation, and be done. It would look something like this:
class ForkCdrModuleEndTime(CDRModule):
''' A class that adds some additional CDR checking of the end times on top
of CDRModule
In addition to checking the normal expectations, this class also checks
whether or not the end times of the CDRs are within some period of time
of each each other.
Note that this class assumes the CDRs are in cdrtest_local.
'''
def __init__(self, module_config, test_object):
super(ForkCdrModuleEndTime, self).__init__(module_config, test_object)
def match_cdrs(self):
super(ForkCdrModuleEndTime, self).match_cdrs()
if (not self.test_object.passed):
return
cdr1 = AsteriskCSVCDR(fn = "%s/%s/cdr-csv/%s.csv" %
(self.test_object.ast[0].base,
self.test_object.ast[0].directories['astlogdir'],
"cdrtest_local"))
#check for missing fields
for cdritem in cdr1:
if (cdritem.duration is None or
cdritem.start is None or
cdritem.end is None):
logger.Error("EPIC FAILURE: CDR record %s is missing one or " \
"more key fields. This should never be able to " \
"happen." % cdritem)
self.test_object.passed = False
return
# The dialplan is set up so that these two CDRs should each last at
# least 4 seconds. Giving it wiggle room, we'll just say we want it to
# be greater than 1 second.
if ((int(cdr1[0].duration) <= 1) or (int(cdr1[1].duration) <= 1)):
logger.error("FAILURE: One or both CDRs only lasted a second or " \
"less (expected more)")
self.test_object.passed = False
return
end = time.strptime(cdr1[0].end, "%Y-%m-%d %H:%M:%S")
beg = time.strptime(cdr1[1].start, "%Y-%m-%d %H:%M:%S")
#check that the end of the first CDR occured within a 1 second split of
# the beginning of the second CDR
if (abs(time.mktime(end) - time.mktime(beg)) > 1):
logger.error("Time discrepency between end1 and start2 must be " \
"one second or less.\n")
logger.error("Actual times: end cdr1 = %s begin cdr2 = %s" %
(cdr1[0].end, cdr1[1].start))
self.test_object.passed = False
return
But, you ask, what good is this? How do we get our TestRunner module to load this class, which - currently - is in the same directory as the test-config for this test (tests/cdr/cdr_fork_end_time)?
Dynamic Test Module Importing and Class Instantiation
Big words for a subheading. What do we mean?
Well, lets assume that ForkCdrModuleEndTime sits in module ForkCdrModule, which lives in folder tests/cdr/cdr_fork_end_time. Our normal python libraries, including TestRunner and CdrModule, live in lib/python/asterisk, and the Python search path is usually modified to include lib/python.
If we directly attempted to __import__ ForkCdrModule.ForkCdrModuleEndTime, by specifying its module type/class type in the test-config.yaml, we'd throw an Exception. That module doesn't live in the lib/python/asterisk package, nor does it exist in any package in the Python search path. How do we get around this?
- Move ForkCdrModule into the lib.python.asterisk package. Unfortunately, this means that test specific logic ends up in our general purpose libraries, which is what we want to avoid.
- Make tests/cdr/cdr_fork_end_time a package by adding an __init__.py file to the directory. We could then modify the python search path to include that directory before TestRunner starts an import. This feels less than ideal for two reasons:
- We end up having to make a lot of tests packages, which isn't really what we want. Packages with a single module/class is less than ideal, and would require a lot of empty __init__.py files.
- We'd end up having to modify TestRunner to either statically "know" of the various test modules and their locations, or we'll have to pass that information into TestRunner via some configuration so that TestRunner can dynamically modify the Python search path. The latter option isn't too bad, but it does feel a little less than optimal.
- Create a module importer using the Python imp package and import the module ourselves. This is more work, but should be the most flexible, and avoids having to turn the tests into Python packages.
Python module importing using the imp package can be broken down into two major stages:
- Implement an object that has a find_module method. The purpose of this is for your object to determine if it should be responsible for handling the import of the specified module. If it cannot handle the import, it raises an ImportError during construction. Otherwise, its find_module method should provide an object to do the actual loading.
- Implement an object to load the Python module into memory and return it for execution. Its up to this object to do all the heavy lifting, be that reading the module from some backing storage, to converting the contents into interpret-able Python code, setting module properties, etc. In our case, since we still expect the test's to write their modules in Python, this ends up being fairly straight forward.
TestModuleFinder
Code first, then discussion:
class TestModuleFinder(object):
''' Determines if a module is a test module that can be loaded '''
supported_paths = []
def __init__(self, path_entry):
if not path_entry in TestModuleFinder.supported_paths:
raise ImportError()
LOGGER.debug('TestModuleFinder supports path %s' % path_entry)
return
def find_module(self, fullname, suggested_path = None):
''' Attempts to find the specified module
Parameters:
fullname The full name of the module to load
suggested_path Optional path to find the module at
'''
search_paths = TestModuleFinder.supported_paths
if suggested_path:
search_paths.append(suggested_path)
for path in search_paths:
if os.path.exists('%s/%s.py' % (path, fullname)):
return TestModuleLoader(path)
LOGGER.warn("Unable to find module '%s'" % fullname)
return None
sys.path_hooks.append(TestModuleFinder)
So, what do we have here?
- An object that during construction checks to see if a class property, supported_paths, contains the path to some module that is passed to the constructor. If supported_paths contains the path we don't throw an ImportError - otherwise, we do.
- A method, find_module, that looks for some module along the supported_paths (plus an optional path provided to the method). If we find a file with the extension '.py' in that path, we return an instance of a new object, TestModuleLoader, and pass to it the full path to that file. Otherwise, we return None.
Not too hard. You'll notice that we add the type of the object to the system path_hooks. This lets Python know that when an import occurs, it should create an instance of this type to aid in the importing.
Onward!
TestModuleLoader
Again: code!
class TestModuleLoader(object):
''' Loads modules defined in the tests '''
def __init__(self, path_entry):
''' Constructor
Parameters:
path_entry The path the module is located at
'''
self._path_entry = path_entry
def _get_filename(self, fullname):
return '%s/%s.py' % (self._path_entry, fullname)
def load_module(self, fullname):
''' Load the module into memory
Parameters:
fullname The full name of the module to load
'''
if fullname in sys.modules:
mod = sys.modules[fullname]
else:
mod = sys.modules.setdefault(fullname,
imp.load_source(fullname, self._get_filename(fullname)))
return mod
So, since our file should still be valid Python code, the actual TestModuleLoader class ends up being very simple as well. We have a method, load_module, that will be called when the module should be loaded into memory. When this happens, we first check to see if the name of the module has already been added to the system modules - if so, we simply return the corresponding module.
If not, we use the imp package to load the module into memory from the source file specified (since we found the Python source in the TestModuleFinder). We associate that source to the name of the module and add it to the system modules - and again, return the corresponding module.
And... that's it!
Putting the Pieces Together
We do need to inform TestRunner that it should load our test's module (ForkCdrModule) and instantiate the object (ForkCdrModuleEndTime) from some location other than the normal Asterisk Test Suite Python library location. To do that, we add a new YAML key, load-from-path, and specify the path to the module:
test-modules:
test-object:
config-section: test-object-config
typename: 'SimpleTestCase.SimpleTestCase'
modules:
-
load-from-path: 'tests/cdr/cdr_fork_end_time'
config-section: 'cdr-config'
typename: 'ForkCdrModule.ForkCdrModuleEndTime'
if ('load-from-path' in module_spec):
TestModuleFinder.supported_paths.append(
module_spec['load-from-path'])
And... that's it!
Conclusions
At the end of this little series, we've taken a lot of common code throughout the Asterisk Test Suite and eliminated it by allowing it to be configured. Since we can't predict what every test will need, we've allowed a mechanism by which a test can write its own modules and have them loaded into the framework at run time - without having to turn the test into a Python package.
Going forward, I expect us to use these concepts a lot. We have a lot of plans to expand on this framework for CEL records, AMI events, SIPp tests, core Asterisk bridging tests, and a whole host of other things. It should be interesting to see what we'll need to expand on. Two things we will need at some point in time:
- Lots more entry points in the Test Objects for pluggable modules to insert themselves.
- A way for pluggable modules to refer to other pluggable modules. Currently, pluggable modules are only aware of the Test Object. We'll need a way for a module to determine if another module exists and - if it doesn't, create it; if it does exist, it may either want the existing object or it may want a new one. This starts to suspiciously sound a lot like an IOC container or dependency injection...
