A Recap
So, in the previous post, I introduced motivation of having a Pluggable Framework in the Asterisk Test Suite:- Reduce code duplication
- Reuse common logic across tests more efficiently then simply requiring test writers to add layers of inheritance
- Decrease time and complexity involved in writing tests
To recap, we have:
- A TestRunner module that has an entry point that:
- Reads a YAML file into memory
- Imports the specified package.modules and instantiates the specified objects, passing them their configuration from the YAML file
- Differentiates between the main 'Test Object' and 'Pluggable Modules' that are added to the Test Object
- Starts the twisted reactor, tells the test object to run, and reports the results
- Some Test Object, specified in the configuration (in our case, SimpleTestCase)
Not all of this was shown in the blogpost, as that's a lot of information. If you're curious at what the actual source code for this looks like, you can view it in the public Subversion repository here:
(Note: there's no guarantee that these links will always work - the Test Suite changes over time!)
So, what's next?
Well, the original tests that were used as a motivating example - cdr_accountfield and cdr_userfield - are supposed to verify CDRs. While the SimpleTestCase is able to originate a call into Asterisk (along with starting/stopping Asterisk and doing some other necessary plumbing), it doesn't do anything with CDRs. So what we need is a way to verify CDRs.
We don't want to simply tack this onto SimpleTestCase - not all tests need to verify CDRs, and we may have other Test Objects that do want to verify CDRs. Enter: the Pluggable Module.
 |
| The best thing I could find for "pluggable" via Google Images. Yay clipart! |
Introducing: CDRModule
So, we want to verify CDR records in a test. (FYI: CDR == Call Detail Records. What parties were involved in a call, how long they talked, etc. CDRs are widely used, but there are some problems with them - there's a lot of complex call scenarios that can't be expressed in CDRs, which is why we have CEL - but that's a topic for another day) What we had previously in the various tests already could do this - a set of Python libraries read in a CSV file (which happens to be the easiest CDR backend to test) and - using regular expressions - matches records to expected results. Lets take a look again at cdr_userfield:
Now we just need to modify the constructor of CDRModule to register itself with the concrete implementation of TestCase and specify what method we want called when the test stops:
And that's it! We now have a mechanism to fully configure a CDR test such that the run-test script is completely unnecessary. Lets look again at cdr_userfield:
Phew. That is a lot of code to look at in a blog post, so I don't blame anyone if they skipped through some of it. The important point to take away from this exercise is how the YAML method of defining the test describes the entire test. A number of details about the test were hidden from the test writer in the run-test script - for example, the fact that Local channels were used to originate the call into Asterisk was completely hidden. They had to know that's how the base class worked. While abstracting details is a feature of inheritance, when it happens with crucial information, you probably have poor class design - and getting inheritable class design "right" is incredibly hard. So why make a test writer have to do it?
Next time: the fly in the ointment that are the Fork CDR tests.
class CDRUserFieldTest(CDRTestCase):
def __init__(self):
CDRTestCase.__init__(self)
self.add_expectation('cdrtest_local',AsteriskCSVCDRLine(source="",
destination="1", dcontext="default", callerid="",
channel="Local/1@default-.*", dchannel="", lastapp="Hangup", lastarg="",
disposition="ANSWERED", amaflags="DOCUMENTATION", userfield="bazinga"
))
Under the hood in the CDRTestCase class, we have a dictionary of expected results, that the method add_expectation adds to. This maps a CSV CDR file (cdrtest_local, in this case), to a list of CDR CSV lines - managed by the class AsteriskCSVCDRLine. The AsteriskCSVCDRLine class contains fields that match the columns in the CDR record.
When a CDR record is verified, each line is read and verified against the matching entry in the list of AsteriskCSVCDRLine objects for that file. For each column in a record, if a field has a value, a regular expression is used to determine if the two fields match. If any field doesn't match - or if the file has fewer or more lines then what is expected - the CDRTestCase class will fail the test.
To replicate this functionality, the CDRModule can use the same principle: have a dictionary of file names that map to a list of AsteriskCSVCDRLine objects. Instead of hard coding in the values, we'll instead read them from a YAML file that is presented to us in our constructor.
As a first step, we can define our YAML layout. We'll need the following:
- The ability to support multiple files
- The ability to support multiple line definitions per file
- The ability to support any number of column names per line definition
- Each column name has a value that can be compiled as a regular expression
What we end up with is the following:
cdr-config: - file: 'cdrtest_local' lines: - source: '' destination: '1' dcontext: 'default' callerid: '' channel: 'Local/1@default-.*' dchannel: '' lastapp: 'Hangup' lastarg: '' disposition: 'ANSWERED' amaflags: 'DOCUMENTATION' userfield: 'bazinga'
Now we're ready for the CDRModule class itself.
class CDRModule(object):
''' A module that checks a test for expected CDR results '''
def __init__(self, module_config, test_object):
''' Constructor
Parameters:
module_config The yaml loaded configuration for the CDR Module
test_object A concrete implementation of TestClass
'''
self.test_object = test_object
# Build our expected CDR records
self.cdr_records = {}
for record in module_config:
file_name = record['file']
if file_name not in self.cdr_records:
self.cdr_records[file_name] = []
for csv_line in record['lines']:
# Set the record to the default fields, then update with what
# was passed in to us
dict_record = dict((k, None) for k in AsteriskCSVCDRLine.fields)
dict_record.update(csv_line)
self.cdr_records[file_name].append(AsteriskCSVCDRLine(
accountcode=dict_record['accountcode'], source=dict_record['source'],
destination=dict_record['destination'], dcontext=dict_record['dcontext'],
callerid=dict_record['callerid'], channel=dict_record['channel'],
dchannel=dict_record['dchannel'], lastapp=dict_record['lastapp'],
lastarg=dict_record['lastarg'], start=dict_record['start'],
answer=dict_record['answer'], end=dict_record['end'],
duration=dict_record['duration'], billsec=dict_record['billsec'],
disposition=dict_record['disposition'], amaflags=dict_record['amaflags'],
uniqueid=dict_record['uniqueid'], userfield=dict_record['userfield']))
This obviously looks very similar to our cdr_userfield test, except that now all of the expected results are being built from a YAML file. Note that the AsteriskCSVCDRLine class defines the allowed CDR fields and we use that to pre-populate each expected CDR line. That way a test writer does not have to provide default values for each possible CDR field for each expected CDR line.
Now that question is: how do we get our CDRModule to be called when the test ends, so that the CDR records can be verified?
By default, the Test Objects (concrete implementations of the TestCase class) do not expose places to hook modules onto them. What we need is for some method in CDRModule to be called when the concrete implementation of TestCase is finished running the test. To do this, we'll:
- Expose a method that an observer can use to register themselves for notifications when Asterisk has fully stopped
- Modify TestCase to call the observers when Asterisk and the twisted reactor have fully stopped, that is, when the test is in a state that it can no longer modify its data
The registration method:
def register_stop_observer(self, callback): ''' Register an observer that will be called when Asterisk is stopped
Parameters: callback The deferred callback function to be called when Asterisk is stopped
Note: This appends a callback to the deferred chain of callbacks executed when all instances of Asterisk are stopped. ''' self._stop_callbacks.append(callback)
In TestCase, the method stop_reactor is always used to stop the test. Here, we can modify the method such that the callbacks registered in register_stop_observer will be added to the deferred chain, such that they're each called after the test has finished running.
def stop_reactor(self): """ Stop the reactor and the test. """ def __stop_reactor(result): """ Called when the Asterisk instances are stopped """ logger.info("Stopping Reactor") if reactor.running: try: reactor.stop() except twisted.internet.error.ReactorNotRunning: # Something stopped it between our checks - at least we're stopped pass if not self._stopping: self._stopping = True df = self.__stop_asterisk() df.addCallback(__stop_reactor) for callback in self._stop_callbacks: df.addCallback(callback)
Now we just need to modify the constructor of CDRModule to register itself with the concrete implementation of TestCase and specify what method we want called when the test stops:
# Hook ourselves onto the test object
test_object.register_stop_observer(self._check_cdr_records)
Finally, we implement our CDR checking. Note that much of this was pulled out of the existing CDRTestCase classes - essentially, when we're told to check our CDRs, we take the expected results and match them against the actual CDR file that was created for the test. The AsteriskCSVCDR class does the heavy lifting for us in terms of actually matching records. Finally, we modify the pass/fail results of the Test Object based on the results of the AsteriskCSVCDR instances.
def _check_cdr_records(self, callback_param):
''' A deferred callback method that is called by the TestCase
derived object when all Asterisk instances have stopped
Parameters:
callback_param
'''
LOGGER.debug("Checking CDR records...")
self.match_cdrs()
return callback_param
def match_cdrs(self):
''' Called when all instances of Asterisk have exited. Derived
classes can override this to provide their own behavior for CDR
matching.
'''
expectations_met = True
for key in self.cdr_records:
cdr_expect = AsteriskCSVCDR(records=self.cdr_records[key])
cdr_file = AsteriskCSVCDR(fn="%s/%s/cdr-csv/%s.csv" %
(self.test_object.ast[0].base,
self.test_object.ast[0].directories['astlogdir'],
key))
if cdr_expect.match(cdr_file):
LOGGER.debug("%s.csv - CDR results met expectations" % key)
else:
LOGGER.error("%s.csv - CDR results did not meet expectations. Test Failed." % key)
expectations_met = False
self.test_object.passed = expectations_met
And that's it! We now have a mechanism to fully configure a CDR test such that the run-test script is completely unnecessary. Lets look again at cdr_userfield:
cdr_userfield as run-test script:
#!/usr/bin/env python
'''
Copyright (C) 2010, Digium, Inc.
Terry Wilson
This program is free software, distributed under the terms of
the GNU General Public License Version 2.
'''
import sys
sys.path.append("lib/python")
from asterisk.CDRTestCase import CDRTestCase
from asterisk.asterisk import Asterisk
from asterisk.cdr import AsteriskCSVCDR, AsteriskCSVCDRLine
from twisted.internet import reactor
import re
class CDRUserFieldTest(CDRTestCase):
def __init__(self):
CDRTestCase.__init__(self)
self.add_expectation('cdrtest_local',AsteriskCSVCDRLine(source="",
destination="1", dcontext="default", callerid="",
channel="Local/1@default-.*", dchannel="", lastapp="Hangup", lastarg="",
disposition="ANSWERED", amaflags="DOCUMENTATION", userfield="bazinga"
))
def main():
test = CDRUserFieldTest()
test.start_asterisk()
reactor.run()
test.stop_asterisk()
return test.results()
if __name__ == '__main__':
sys.exit(main())
cdr_userfield as YAML:
testinfo:
summary: 'Test that Set(CDR(userfield)=...) works'
description: |
'Test that setting the userfield field in the CDR works'
test-modules:
test-object:
config-section: test-object-config
typename: 'SimpleTestCase.SimpleTestCase'
modules:
-
config-section: 'cdr-config'
typename: 'cdr.CDRModule'
test-object-config:
spawn-after-hangup: True
test-iterations:
-
channel: 'Local/1@default'
application: 'Echo'
cdr-config:
-
file: 'cdrtest_local'
lines:
-
source: ''
destination: '1'
dcontext: 'default'
callerid: ''
channel: 'Local/1@default-.*'
dchannel: ''
lastapp: 'Hangup'
lastarg: ''
disposition: 'ANSWERED'
amaflags: 'DOCUMENTATION'
userfield: 'bazinga'
properties:
minversion: '1.8.0.0'
dependencies:
- python : 'twisted'
- python : 'starpy'
- asterisk : 'cdr_csv'
tags:
- CDR
- chan_local
Phew. That is a lot of code to look at in a blog post, so I don't blame anyone if they skipped through some of it. The important point to take away from this exercise is how the YAML method of defining the test describes the entire test. A number of details about the test were hidden from the test writer in the run-test script - for example, the fact that Local channels were used to originate the call into Asterisk was completely hidden. They had to know that's how the base class worked. While abstracting details is a feature of inheritance, when it happens with crucial information, you probably have poor class design - and getting inheritable class design "right" is incredibly hard. So why make a test writer have to do it?
Next time: the fly in the ointment that are the Fork CDR tests.
