Commit 6f563007 authored by Michael Weinrich's avatar Michael Weinrich
Browse files

Updated the signal names to the new ones

Changed the way the default file path for schema files is created
parent 9975e8e4
"""
Base class for the GUI.
@license: Licensed under the MIT license
@license: Licensed under the MIT license
http://opensource.org/licenses/mit-license.php
@author: Michael Weinrich <testsuite@michael-weinrich.de>
@copyright: Copyright 2007, Michael Weinrich
......@@ -12,6 +12,7 @@ import sys
import os
from twisted.internet import reactor, task, defer
from twisted.python.filepath import FilePath
import gtk, gtk.glade, gobject
......@@ -24,7 +25,7 @@ class UPnTGUI:
UPnT = None
"""Class variable for a single instance of UPnT."""
def __init__(self, config):
"""
Create the window and setup the underlying UPnT instance.
......@@ -32,7 +33,11 @@ class UPnTGUI:
# Try to locate the glade file
gladefile = "upntest.glade"
tryfiles = [gladefile, os.path.join(os.path.dirname(sys.argv[0]), gladefile), config.get('glade_dir', '') + gladefile]
tryfiles = [gladefile,
os.path.join(os.path.dirname(sys.argv[0]), gladefile),
config.get('glade_dir', '') + gladefile,
FilePath(__file__).parent().child('upntest.glade').path]
print FilePath(__file__).parent().child('upntest.glade').path
for file in tryfiles:
if os.access(file, os.F_OK):
gladefile = file
......@@ -48,7 +53,7 @@ class UPnTGUI:
"onActivateDeviceRow": self.showDetails
}
self.wTree.signal_autoconnect(dic)
# TreView for the devices
gfx_dir = config.get('gfx_dir', '');
self.device_icon = gtk.gdk.pixbuf_new_from_file_at_size(gfx_dir + 'gnome-fs-client.svg', 24, 24)
......@@ -63,14 +68,14 @@ class UPnTGUI:
column.set_attributes(imgrenderer, pixbuf=0)
column.set_attributes(textrenderer, text=1)
self.getWidget("treeviewDeviceTree").append_column(column)
# TreeModel for device tree
self.deviceTreeModel = gtk.TreeStore(gtk.gdk.Pixbuf,
str,
object)
self.getWidget("treeviewDeviceTree").set_model(self.deviceTreeModel)
"""Device var that holds the device that is tested when the Execute
"""Device var that holds the device that is tested when the Execute
Tests button is hit"""
self.serviceToTest = None
......@@ -88,17 +93,17 @@ class UPnTGUI:
column.set_attributes(textrenderer, text=1)
column.set_title('Value')
self.getWidget("treeviewDetails").append_column(column)
# ListStore for the device and service details
model = gtk.ListStore(str, str)
self.getWidget("treeviewDetails").set_model(model)
# buffer of the textview at the bottom showing all messages
self.messageTextBuffer = self.getWidget('textviewMessages').get_buffer()
# we are in GUI mode
config['GUI'] = True
# get all the signals
louie.connect(self.updateMessageBox, 'UPnT.infoMessage', louie.Any, weak=False)
louie.connect(self.updateUpnpDevices, 'UPnT.host_discovery_end', louie.Any, weak=False)
......@@ -126,7 +131,7 @@ class UPnTGUI:
model = self.getWidget("treeviewDeviceTree").get_model()
iter = model.get_iter_first()
current_parent = None
while iter is not None:
cur_dev = model.get_value(iter, 2)
if cur_dev._host.ip == ip:
......@@ -146,7 +151,7 @@ class UPnTGUI:
device_iter = model.append(current_parent, [self.device_icon, device.deviceType + ':' + device.deviceVersion, device])
self.addSubDevicesToTree(model, device_iter, device)
self.addServicesToTree(model, device_iter, device)
def addSubDevicesToTree(self, model, device_iter, device):
"""
Add all subdevices and their services to the tree
......@@ -155,22 +160,22 @@ class UPnTGUI:
subdevice_iter = model.append(device_iter, [self.device_icon, subdevice.deviceType + ':' + subdevice.deviceVersion, subdevice])
self.addSubDevicesToTree(model, subdevice_iter, subdevice)
self.addServicesToTree(model, subdevice_iter, subdevice)
def addServicesToTree(self, model, device_iter, device):
"""
Add all services of a device to the tree
"""
for uuid, service in device._services.items():
model.append(device_iter, [self.service_icon, service.serviceType + ':' + service.serviceVersion, service])
def executeTests(self, widget, data=None):
"""
Run the control test on a device.
"""
if self.serviceToTest is not None:
self.serviceToTest.startControlTesting(None, defer.succeed(None))
def updateMessageBox(self, message):
"""
Add the received message to the text box.
......@@ -198,7 +203,7 @@ class UPnTGUI:
model.append(['scpdURL', service.scpdUrl])
model.append(['controlURL', service.controlUrl])
model.append(['eventSubURL', service.eventSubUrl])
def refreshUI(self):
"""
Refresh the UI to show changes of the controls to the user.
......@@ -206,4 +211,3 @@ class UPnTGUI:
while gtk.events_pending():
gtk.main_iteration()
#reactor.callLater(1, self.refreshUI)
\ No newline at end of file
"""
Base class of the UPnP Test Tool.
@license: Licensed under the MIT license
@license: Licensed under the MIT license
http://opensource.org/licenses/mit-license.php
@author: Michael Weinrich <testsuite@michael-weinrich.de>
@copyright: Copyright 2007, Michael Weinrich
......@@ -18,67 +18,67 @@ from upntest.host import Host
class UPnT(log.Loggable):
"""
Main class of the test tool. All messages from UPnP devices are send to
this class which is distibuting the messages to the right object in the
Main class of the test tool. All messages from UPnP devices are send to
this class which is distibuting the messages to the right object in the
hierarchy.
"""
logCategory = 'UPnT'
"""Set a log category different to the Coherence one. All log categories
"""Set a log category different to the Coherence one. All log categories
of the upntest package will also start with C{UPnT_}."""
def __init__(self, config={}):
"""
Initialise the main class.
During initialisation the main Coherence instance is created and all
signals are connected to local funtions for processing of received
During initialisation the main Coherence instance is created and all
signals are connected to local funtions for processing of received
packets.
@param config: A C{dict} that holds all information of the config file
@param config: A C{dict} that holds all information of the config file
in the same hierarchical manner that is used in the config file itself.
"""
self._knownHosts = {}
"""All known hosts that have sent any kind of discovery packet."""
self._runningDeferreds = []
"""A C{list} that stores all currently running C{Deferred}s."""
self._hostsToRemove = []
"""A C{list} of hosts that already sent goodbye messages (not used yet).
"""
self._config = config
"""Stores the given config options to pass it on to other objects who
"""Stores the given config options to pass it on to other objects who
need it."""
self._coherence = Coherence(self._config);
"""The C{Coherence} instance that is created based on the given config
"""The C{Coherence} instance that is created based on the given config
C{dict}."""
self.info('connecting signals...')
louie.connect(self.datagramReceived, 'UPnT.ssdp_datagram_received', louie.Any, weak=False)
louie.connect(self.datagramReceived, 'UPnT.msearch_datagram_received', louie.Any, weak=False)
louie.connect(self.hostRemoved, 'UPnT.host_removed', louie.Any, weak=False)
louie.connect(self.datagramReceived, 'UPnP.SSDP.datagram_received', louie.Any, weak=False)
louie.connect(self.datagramReceived, 'UPnP.msearch_datagram_received', louie.Any, weak=False)
#louie.connect(self.hostRemoved, 'UPnP.host_removed', louie.Any, weak=False)
if not config.get('GUI', False):
louie.connect(self.showInfoMessage, 'UPnT.infoMessage', louie.Any, weak=False)
louie.connect(self.showErrorMessage, 'UPnT.errorMessage', louie.Any, weak=False)
louie.connect(self.addDeferredToList, 'UPnT.running_deferred', louie.Any, weak=False)
louie.connect(self.removeDeferredFromList, 'UPnT.ending_deferred', louie.Any, weak=False)
self.info('signals connected, waiting for signals...')
#reactor.addSystemEventTrigger('before', 'shutdown', self.waitForShutdown)
def datagramReceived(self, data, ip, port):
"""
Process the received SSDP datagram.
First check if sending host exists in host list. If not create it. Then
pass on the received packet data to the
L{dispatchDiscoveryPacketData<Host.dispatchDiscoveryPacketData>} of this
First check if sending host exists in host list. If not create it. Then
pass on the received packet data to the
L{dispatchDiscoveryPacketData<Host.dispatchDiscoveryPacketData>} of this
L{Host} object.
@param data: A C{str} containing all the packet data.
@param ip: The IP of the sending host (C{str}).
@param port: The port to which this packet was sent to. Not used.
......@@ -87,7 +87,7 @@ class UPnT(log.Loggable):
# devices on the local host are ignored
if ip == self._coherence.hostname:
return
#if ip == '192.168.82.1':
# return
......@@ -99,37 +99,37 @@ class UPnT(log.Loggable):
self._knownHosts[ip].dispatchDiscoveryPacketData(data)
#except Exception, error:
# self.warning(error)
def hostRemoved(self, ip):
"""
Remove host from the known hosts list.
Host with IP sent goodbye messages so it will be removed from the known
Host with IP sent goodbye messages so it will be removed from the known
host list.
@param ip: IP address of the host that is to be removed. (C{str})
"""
self.info('hostRemoved(%r)' % ip)
self._hostsToRemove.append(ip)
def showInfoMessage(self, message):
"""
Show info messages giving hints on better implementation.
Messages sent through this channel don't define the tested device as
Messages sent through this channel don't define the tested device as
invalid.
@param message: Message that is to be displayed. (C{str})
"""
print message
def showErrorMessage(self, type, message):
"""
Show error messages genereated by test routines.
Messages sent through ths channel define the tested device as invalid.
@param type: Type of error that occured. (C{str})
@param message: Message explaining the error. (C{str})
"""
......@@ -138,10 +138,10 @@ class UPnT(log.Loggable):
def waitForShutdown(self):
"""
Wait for all Deferreds before reactor shutdown.
Used to wait for all Deferreds to end before the reactor shuts down.
@return: None if no C{Deferred}s are running or a C{DeferredList} if
@return: None if no C{Deferred}s are running or a C{DeferredList} if
at least one C{Deferred} is still running.
"""
self.info('Shutting UPnT down...')
......@@ -150,19 +150,19 @@ class UPnT(log.Loggable):
return None
else:
return defer.DeferredList(self._runningDeferreds)
def addDeferredToList(self, d):
"""
Add the given Deferred to the list of currently running Deferreds.
@param d: C{Deferred} that should be added to the list.
"""
self._runningDeferreds.append(d)
def removeDeferredFromList(self, d):
"""
Remove the given Deferred from the list of currently running Deferreds.
@param d: C{Deferred} that should be removed from the list.
"""
self._runningDeferreds.remove(d)
"""
Module that holds the classes for checking control messages and responses
Module that holds the classes for checking control messages and responses
sent by client or server devices.
@license: Licensed under the MIT license
@license: Licensed under the MIT license
http://opensource.org/licenses/mit-license.php
@author: Michael Weinrich <testsuite@michael-weinrich.de>
@copyright: Copyright 2007, Michael Weinrich
......@@ -14,6 +14,7 @@ from lxml import etree
import os
import time
from twisted.internet import reactor, defer
from twisted.python.filepath import FilePath
# Coherence includes
from coherence import log
# local includes
......@@ -21,53 +22,57 @@ from upntest.soap_proxy import SOAPProxy
class ServerControl(log.Loggable):
"""
This class contains methods to invoke all actions that a service offers and
to check the responses for validity.
The actions that are invoked are read from the service template that is
selected based on the service type. The actions listed in there can be
customized or removed using an additional file containing the according
This class contains methods to invoke all actions that a service offers and
to check the responses for validity.
The actions that are invoked are read from the service template that is
selected based on the service type. The actions listed in there can be
customized or removed using an additional file containing the according
instructions (see usage documentation).
"""
logCategory = 'UPnT_ServerControl'
"""Log category for the server control checker."""
def __init__(self, parent, config, service_type, service_version, uuid):
"""
Initailise the server control messages checker.
@param parent: "Service that is to be tested.
@param config: A C{dict} containing the configuration data from the
@param config: A C{dict} containing the configuration data from the
config file.
@param service_type: Type of service that is to be tested.
@param service_version: Version of service type.
@param uuid: UUID of the service to be tested (C{str}).
"""
self.parentService = parent
"""Service that is tested."""
self._testcasesXml = None
"""XML file containing the service description with all state variables
"""XML file containing the service description with all state variables
and actions."""
self._uuidXml = None
"""Device specific service description that is merged with the general
"""Device specific service description that is merged with the general
one in L{_testcasesXml}."""
self._ready = True
"""A C{bool} indicating that the device initialisation was successful."""
if service_type != "" and \
service_version != "" and \
uuid != "":
test_file_dir = config.get('test_file_dir', None)
test_file_dir = config.get('test_file_dir', FilePath(__file__).parent().parent().child('testcases').path)
if not test_file_dir.endswith('/'):
test_file_dir += '/'
if test_file_dir is not None:
try:
self._testcasesXml = \
etree.parse(test_file_dir + 'common/' +
etree.parse(test_file_dir + 'common/' +
service_type + service_version + '.SyntaxTests.xml')
try:
self._uuidXml = \
etree.parse(test_file_dir + 'uuid/' +
etree.parse(test_file_dir + 'uuid/' +
service_type + service_version + '.' + uuid + '.SyntaxTests.xml')
except IOError, error:
self.info('Specific file for %s:%s on uuid:%s not found' % (service_type, service_version, uuid))
......@@ -87,25 +92,25 @@ class ServerControl(log.Loggable):
self._ready = False
if not self._ready:
return
self.loop_tests = False
"""A C{bool} indicating whether the tests should be run repeatedly until
"""A C{bool} indicating whether the tests should be run repeatedly until
this variable is set to C{False} (which is the default)."""
self._testcases = {}
"""A C{dict} that is used to store all the data neccessary to invoke an
"""A C{dict} that is used to store all the data neccessary to invoke an
action on the remote device."""
self.parseTestcases()
def parseTestcases(self):
"""
Parse files with test cases and create list for testing.
Parses the file with common information about the testcases. Afterwards
Parses the file with common information about the testcases. Afterwards
filter that input using the file with UUID specific information.
"""
self.debug('parseTestcases')
# get all testcases from the common file
testcases = self._testcasesXml.xpath('/ServiceControlSyntaxTestCases/TestCaseList/TestCase')
for testcase in testcases:
......@@ -119,9 +124,9 @@ class ServerControl(log.Loggable):
for argument in in_args:
self._testcases[id]['InArgs'][argument.tag] = argument.text
self._testcases[id]['ExpectedReturnCode'] = testcase.findtext('ExpectedReturnCode')
#self.debug(self._testcases)
# read the device specific file and make according changes
#self.debug(etree.tostring(self._uuidXml, pretty_print=True))
specific_changes = self._uuidXml.xpath('/ServiceControlSyntaxTestCases/TestCaseList/TestCase')
......@@ -136,7 +141,7 @@ class ServerControl(log.Loggable):
action_name = change.findtext('ActionName')
in_args = change.findall('./InArgs/*')
expected_return_code = change.findtext('ExpectedReturnCode')
bool1 = self._testcases[id].has_key('InArgs') and len(in_args) == len(self._testcases[id]['InArgs'])
bool2 = not self._testcases[id].has_key('InArgs') and len(in_args) == 0
if category != "" and action_name != "" and expected_return_code != "" and \
......@@ -151,12 +156,12 @@ class ServerControl(log.Loggable):
def startTesting(self, loop_tests=False):
"""
Set all variables and start the test asynchronusly.
@param loop_tests: If C{True}, the test loop starts again at the beginning
@param loop_tests: If C{True}, the test loop starts again at the beginning
of the method list.
@return: A C{Deferred} which is activated when the test run is
finished. If L{_ready} is still C{False}, an already activated C{Deferred}
@return: A C{Deferred} which is activated when the test run is
finished. If L{_ready} is still C{False}, an already activated C{Deferred}
is returned.
"""
if not self._ready:
......@@ -166,12 +171,12 @@ class ServerControl(log.Loggable):
self.test_deferred = defer.Deferred()
reactor.callLater(1, self.doTesting)
return self.test_deferred
def doTesting(self):
"""
Execution of the actual test loop.
Go through all testcases and send a control message for each test case.
Go through all testcases and send a control message for each test case.
If L{loop_tests} is C{True}, repeat the tests until it is set to C{False}.
"""
#print 'doTesting...'
......@@ -195,67 +200,67 @@ class ServerControl(log.Loggable):
d = callClient.callRemote(testcase['ActionName'], in_args)
d.addCallback(self.callSucceeded, action, instance_id)
d.addErrback(self.callFailed, action, url, in_args, testcase['ActionName'])
is_looped = self.loop_tests
self.test_deferred.callback(test_result)
def stopTesting(self):
"""
Set L{loop_tests} to C{False} and stop the test loop after the current
Set L{loop_tests} to C{False} and stop the test loop after the current
cycle.
"""
self.loop_tests = False
def callFailed(self, failure, action, url, args, action_name):
"""
Callback for an unsuccessful action invocation.
@param failure: The failure that caused the error.
@param action: Name of the action that failed including service type
@param action: Name of the action that failed including service type
(C{str}).
@param url: URL of the event server (C{str}).
@param args: C{dict} containing all IN arguments of the action.
@param action_name: Name of the action (C{str}).
"""
self.warning("error: invoking %s on %s with %r" % (action,
self.warning("error: invoking %s on %s with %r" % (action,
url,
args))
self.info(failure)
action_list = self.parentService._actionList
if action_list.has_key(action_name):
if not action_list[action_name]['Optional']:
louie.send('UPnT.infoMessage', None,
"Required action %s couldn't be invoked on %s with %r" % (action,
url,
louie.send('UPnT.infoMessage', None,
"Required action %s couldn't be invoked on %s with %r" % (action,
url,
args))
#return failure
def callSucceeded(self, results, action, instance_id):
"""
Callback for a successful action invocation.
@param results: The results of the action invocation.
@param action: Name of the action including service type (C{str}).
@param instance_id: Instance ID that was used during the invocation
@param instance_id: Instance ID that was used during the invocation
(C{int}).
"""
louie.send('UPnT.infoMessage', None,
louie.send('UPnT.infoMessage', None,
"\nok: call %s (instance %d) returned" % (action, instance_id))
if len(results) > 0:
for out_arg, out_val in results.items():
louie.send('UPnT.infoMessage', None,
louie.send('UPnT.infoMessage', None,
"%s: %s" % (out_arg, out_val))
else:
louie.send('UPnT.infoMessage', None,
louie.send('UPnT.infoMessage', None,
'no return values' % (action, instance_id))
class ClientControl(log.Loggable):
"""
Device to check is a control point so we can only perform passive checks on
Device to check is a control point so we can only perform passive checks on
the messages sent by the device.
"""
......@@ -263,29 +268,33 @@ class ClientControl(log.Loggable):
"""Log category for the client control checker."""
serviceActions = {}
"""C{dict} conatining all the actions that are found in all the available
service templates. This is a class variable so that the templates are
"""C{dict} conatining all the actions that are found in all the available
service templates. This is a class variable so that the templates are
available to all instances once the first instance parsed all templates."""
def __init__(self, templates_dir):
"""
Read all avilable service templates and store all actions found in
Read all avilable service templates and store all actions found in
L{serviceActions}.
@param templates_dir: Path to the directory where the service and device
@param templates_dir: Path to the directory where the service and device
templates are located (C{str}).
"""
if not templates_dir.endswith('/'):
templates_dir += '/'
if len(self.serviceActions) == 0:
template_files = [template_file for template_file in os.listdir(templates_dir + 'service/') if not template_file.startswith('.')]
ns = {'upnp': 'urn:schemas-upnp-org:service-1-0'}
for template_file in template_files:
service_template = etree.parse(templates_dir + 'service/' + template_file)