Commit 905e77b8 authored by Michael Weinrich's avatar Michael Weinrich
Browse files

Checks for discovery, description and presentation ready

parent dffe5098
#! /usr/bin/env python
# Licensed under the MIT license
# Copyright 2007, Michael Weinrich <>
import os, sys
from twisted.internet import reactor
from twisted.python import usage
from upntest.base import UPnT
import louie
from configobj import ConfigObj
class Options(usage.Options):
optParameters = [['configfile', 'c', '', 'path to configfile'],
def main(options):
# get settings or options
def setConfigFile(filename):
def findConfigDir():
configDir = os.path.expanduser('~')
configDir = os.getcwd()
return configDir
if filename is '':
filename = os.path.join( findConfigDir(), '.upntest')
return filename
config = ConfigObj( setConfigFile( options['configfile']))
u = UPnT(config)
if __name__ == '__main__':
options = Options()
except usage.UsageError, errortext:
print '%s: %s' % (sys.argv[0], errortext)
print '%s: Try --help for usage details.' % (sys.argv[0])
reactor.callWhenRunning(main, options)
\ No newline at end of file
from setuptools import setup, find_packages
from upntest import __version__
description="""UPnT - Test suite for UPnP devices""",
long_description="""In preparation""",
author="Michael Weinrich",
license = "MIT",
scripts = ['bin/upntest', 'bin/upntest-gui'],
url = "",
#download_url = '' % __version__,
download_url = '',
classifiers = ['Development Status :: 4 - Beta',
'Environment :: Console',
'Environment :: Web Environment',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python',
package_data = {
'coherence': ['xmlfiles/device/*.xml',
......@@ -3,62 +3,80 @@
# Copyright 2007, Michael Weinrich <>
from coherence.base import Coherence
from twisted.internet import reactor
import louie
from coherence.extern.logger import Logger
log = Logger('UPnT')
from twisted.internet import reactor, defer
from coherence.base import Coherence
from coherence import log
#from coherence.extern.logger import Logger
#log = Logger('UPnT')
from upntest.testhost import TestHost
"""The main class of the whole test suite"""
class UPnT:
class UPnT(log.Loggable):
logCategory = 'UPnT'
known_hosts = {}
runningDeferreds = []
shutdown_signal = None
def __init__(self, *args, **kwargs):
self._coherence = Coherence(config);
def __init__(self, config={}):
self._config = config
self._coherence = Coherence(self._config);'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)
if not config.get('GUI', False):
louie.connect(self.showInfoMessage, 'UPnT.infoMessage', 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)'signals connected, waiting for signals...')'connecting signals...')
louie.connect(self.datagramReceived, 'UPnT.ssdp_datagram_received', louie.Any)
louie.connect(self.datagramReceived, 'UPnT.msearch_datagram_received', louie.Any)
louie.connect(self.hostRemoved, 'UPnT.host_removed', louie.Any)
reactor.addSystemEventTrigger('before', 'shutdown', self.waitForShutdown)
def datagramReceived(self, data, ip, port):
"""debug output for received datagrams"""
#print '\nDatagram received from %s:%s' % (host, port)
#print '\nDatagram received from %s:%s' % (ip, port)
#print 'Datagram data:\n%s' % data
if not self.known_hosts.has_key(ip):
self.known_hosts[ip] = TestHost(ip, config)'Host added', ip)
self.known_hosts[ip] = TestHost(ip, self._config)'Host added', ip)
self.debug('Calling dispatchDiscoveryPacketData')
except Exception:
# GUI update signal
louie.send('UPnT.host_discovery_end', None, ip)
def hostRemoved(self, ip):
"""Remove host from the known hosts list"""
if __name__ == '__main__':
def removeHosts(self):
self.known_hosts = None
config = {}
config['logmode'] = 'warning'
config['logfile'] = None#'/home/micxer/development/UPnT/upnptest.log'
config['serverport'] = 30020
config['plugins'] = {}
config['plugins']['FSStore'] = {}
config['plugins']['FSStore']['name'] = 'Coherence MediaServer'
config['plugins']['FSStore']['content'] = '~/mediadata/'
config['subsystem_log'] = {}
config['subsystem_log']['UPnT'] = 'debug'
config['subsystem_log']['UPnT.TestHost'] = 'debug'
config['subsystem_log']['UPnT.TestDevice'] = 'debug'
config['subsystem_log']['UPnT.TestService'] = 'debug'
config['subsystem_log']['UPnT.utils'] = 'debug'
config['schema_file_dir'] = '/home/micxer/development/UPnT/schemas/'
config['templates_dir'] = '/home/micxer/development/UPnT/xmlfiles/'
u = UPnT(config)
def showInfoMessage(self, message):
print message
def waitForShutdown(self):'Shutting UPnT down...')
if len(self.runningDeferreds) == 0:
return None
return defer.DeferredList(self.runningDeferreds)
def addDeferredToList(self, d):
def removeDeferredFromList(self, d):
......@@ -8,15 +8,20 @@ from twisted.internet import reactor, defer
from lxml import etree
from StringIO import StringIO
import louie
from testservice import *
from coherence.upnp.core import utils
from coherence.extern.logger import Logger
log = Logger('UPnT.TestDevice')
from coherence import log
#from coherence.extern.logger import Logger
#log = Logger('UPnT.TestDevice')
class TestDevice:
class TestDevice(log.Loggable):
"""Class for storing data about active devices"""
logCategory = 'UPnT_TestDevice'
def __init__(self, headers, config):
self._uuid = ''
......@@ -49,7 +54,7 @@ class TestDevice:
location = property(getLocation)
def getDeviceType(self):
return self._device_type
return self._device_type + ':' + self._device_version
deviceType = property(getDeviceType)
def getUuid(self):
......@@ -84,11 +89,11 @@ class TestDevice:
cache_control = headers['cache-control'].split('=')
if new:'Creating new device for %s' % usn_split[0][5:])'Creating new device for %s' % usn_split[0][5:])
self._uuid = usn_split[0][5:] # UUID is uuid:1234567890
self._expiration_notification_func = reactor.callLater(int(cache_control[1].strip()), self.deviceExpired)
else:'Updating device for %s' % usn_split[0][5:])'Updating device for %s' % usn_split[0][5:])
......@@ -103,23 +108,19 @@ class TestDevice:
self._root_device = True
def updateSubDevice(self, headers):
print 'updateDevice called'
def updateService(self, headers):
print 'updateService called'
def addSubdevice(self, subdevice):
"""Add a subdevice"""
subdevice.parent = self
self._sub_devices[subdevice.UUID] = subdevice
def checkDescriptions(self, device_list, service_list):
"""Check if description of device is valid and if all devices and
services were announced hence existent in the _sub_devices and _services"""
log.debug('Checking device description from %s' % self._desc_location)
self.debug('Checking device description from %s' % self._desc_location)
d = utils.getPage(self._desc_location)
louie.send('UPnT.running_deferred', None, d)
d.addErrback(self.validateDescriptionFailed, self._desc_location)
......@@ -127,8 +128,8 @@ class TestDevice:
d.addCallback(self.checkDeviceAnnouncement, device_list)
d.addErrback(self.checkDeviceAnnouncementFailed, self._desc_location)
d.addCallback(self.checkServiceAnnouncement, device_list, service_list)
d.addErrback(self.checkServiceAnnouncementFailed, self._desc_location)
d.addCallback(self.checkServiceAnnouncement, device_list, service_list, d=d)
d.addErrback(self.checkServiceAnnouncementFailed, self._desc_location, d=d)
def validateDescription(self, desc):
"""Description was sucessfully received and will be checked"""
......@@ -139,33 +140,36 @@ class TestDevice:
doc = etree.parse(StringIO(description))
if self._xmlschema is None:
print 'No XML Schema to compare description against!'
louie.send('UPnT.infoMessage', None, 'No XML Schema to compare description against!')
return None
elif self._xmlschema.validate(doc) == 1:
print 'Device-Description valid! (UUID %s)' % self._uuid
louie.send('UPnT.infoMessage', None, 'Device-Description valid! (UUID %s)' % self._uuid)
#TODO: some strings have to be <=64 characters in length
error = self._xmlschema.error_log.last_error
print 'Device-Description not valid! (UUID %s)' % self._uuid'Description:\n%s' % description)
louie.send('UPnT.infoMessage', None, 'Device-Description not valid! (UUID %s)' % self._uuid)'Description:\n%s' % description)
return None
return doc
def validateDescriptionFailed(self, failure, url):
"""Error while receiving description"""
log.warning("error requesting", url)
self.warning("error requesting", url)
return failure
def checkDeviceAnnouncement(self, doc, device_list, xpath_exp='/upnp:root/upnp:device'):
"""Take the description, extract devices and look for a TestDevice object
in the existing devices list. If found, add it to the tree. If not, the
device wasn't announced."""
# some error happened in previous callbacks
if doc == None:
return None
ns = {'upnp': 'urn:schemas-upnp-org:device-1-0'}
devices = doc.xpath(xpath_exp, ns)
......@@ -174,27 +178,32 @@ class TestDevice:
uuid = udn[0].text[5:]
deviceType = device.xpath('upnp:deviceType', ns)
if not device_list.has_key(uuid):
print 'Device with type %s and UUID %s wasn\'t announced during discovery phase' % (deviceType[0].text, uuid)
louie.send('UPnT.infoMessage', None, 'Device %s/uuid:%s::%s wasn\'t announced during discovery phase' % (, uuid, deviceType[0].text))
self._sub_devices[uuid] = device_list[uuid]
self._sub_devices[uuid].parent = self
print 'Device with type %s and UUID %s was announced during discovery phase' % (deviceType[0].text, uuid)
louie.send('UPnT.infoMessage', None, 'Device %s/uuid:%s::%s was announced during discovery phase' % (, uuid, deviceType[0].text))
self.checkDeviceAnnouncement(device, device_list, xpath_exp='upnp:deviceList/upnp:device')
return doc
def checkDeviceAnnouncementFailed(self, failure, url):
self.debug('checkDeviceAnnouncementFailed (%s)' % url)
return failure
def checkServiceAnnouncement(self, doc, device_list, service_list, xpath_exp='/upnp:root/upnp:device'):
def checkServiceAnnouncement(self, doc, device_list, service_list, xpath_exp='/upnp:root/upnp:device', d=None):
"""Take the description, extract services and look for a TestService
object in the existing services list. If found, add it to the tree. If
not, the service wasn't announced."""
# some error happened in previous callbacks
if doc == None:
return None
ns = {'upnp': 'urn:schemas-upnp-org:device-1-0'}
self.debug(etree.tostring(xpath_exp, pretty_print=True))
devices = doc.xpath(xpath_exp, ns)
for device in devices:
......@@ -207,12 +216,12 @@ class TestDevice:
serviceType = service.xpath('upnp:serviceType', ns)
usn = udn[0].text + '::' + serviceType[0].text
if not service_list.has_key(usn):
print 'Service %s/%s wasn\'t announced during discovery phase' % (, usn)
louie.send('UPnT.infoMessage', None, 'Service %s/%s wasn\'t announced during discovery phase' % (, usn))
print 'Service %s/%s was announced during discovery phase' % (, usn)
louie.send('UPnT.infoMessage', None, 'Service %s/%s was announced during discovery phase' % (, usn))
self._services[usn] = service_list[usn]
self._services[usn].parentDevice = self
log.debug('Assinging %s as parentDevice to %s' % (self, self._services[usn]))
self.debug('Assinging %s as parentDevice to %s' % (self, self._services[usn]))
scpdurl = service.xpath('upnp:SCPDURL', ns)
if len(scpdurl) > 0:
......@@ -233,12 +242,16 @@ class TestDevice:
self.checkServiceAnnouncement(device, device_list, service_list, xpath_exp='upnp:deviceList/upnp:device')
if d is not None:
louie.send('UPnT.ending_deferred', None, d)
return doc
def checkServiceAnnouncementFailed(self, failure, url):
def checkServiceAnnouncementFailed(self, failure, url, d):
self.debug('checkServiceAnnouncementFailed (%s)' % url)
louie.send('UPnT.ending_deferred', None, d)
def removeUnknownTags(self, xmldoc):
"""Removes unknown elements from tree before validating."""
......@@ -253,7 +266,7 @@ class TestDevice:
for event, elem in parser:
remove_elem = True
for knowntag in known_tags:
#log.debug('Comparing %s to %s' % (elem.tag, knowntag))
#self.debug('Comparing %s to %s' % (elem.tag, knowntag))
if elem.tag.endswith(knowntag):
remove_elem = False
......@@ -261,7 +274,7 @@ class TestDevice:
for elem in elements_to_remove:
log.debug('Removing subtag %s from tag %s' % (elem.tag, elem.getparent().tag))
self.debug('Removing subtag %s from tag %s' % (elem.tag, elem.getparent().tag))
return etree.tostring(parser.root)
......@@ -270,5 +283,5 @@ class TestDevice:
"""Inform about expiration of device"""
self._expiration_notification_func = None
print 'Device announcement of %s:%s has expired!' % (self._device_type, self._device_version)
louie.send('UPnT.infoMessage', None, 'Device announcement of %s:%s has expired!' % (self._device_type, self._device_version))
\ No newline at end of file
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment