Commit 01764a03 authored by Michael Weinrich's avatar Michael Weinrich
Browse files

Moved ClientDevice and ServerDevice into on module,, documented
Minor code changes
parent 97320457
__version_info__ = (0,5,0)
#: C{tuple} with version information
__version_info__ = (0,3,0)
#: A C{str} variable that shows the actual version of this software.
__version__ = '%d.%d.%d' % (__version_info__[0],__version_info__[1],__version_info__[2],)
# Licensed under the MIT license
Base class of the UPnP Test Tool.
# Copyright 2007, Michael Weinrich <>
@license: Licensed under the MIT license
@author: Michael Weinrich <>
@copyright: Copyright 2007, Michael Weinrich
# external includes
import louie
from twisted.internet import reactor, defer
# Coherence includes
from coherence.base import Coherence
#from coherence.upnp.core.devices.control_point import ControlPoint
from coherence import log
#from coherence.extern.logger import Logger
#log = Logger('UPnT')
# local includes
from import Host
class UPnT(log.Loggable):
The main class of the whole test suite
Main class of the test tool. All messages from UPnP devices are send to
this class that is distibuting the messages to the right object in the
logCategory = 'UPnT'
"""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
@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.knownHosts = {}
self.runningDeferreds = []
self.shutdownSignal = None
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
need it."""
self._coherence = Coherence(self._config);
#if config.get('test_control_point', False):
#self._control_point = ControlPoint(self._coherence)
"""The C{Coherence} instance that is created based on the given config
C{dict}."""'connecting signals...')
louie.connect(self.datagramReceived, 'UPnT.ssdp_datagram_received', louie.Any, weak=False)
......@@ -43,28 +62,37 @@ class UPnT(log.Loggable):
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.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)'signals connected, waiting for signals...')
reactor.addSystemEventTrigger('before', 'shutdown', self.waitForShutdown)
#reactor.addSystemEventTrigger('before', 'shutdown', self.waitForShutdown)
def datagramReceived(self, data, ip, port):
Check if host exists in host list. If not create it and call the
validation method.
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
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.
# devices on the local host are ignored
if ip == self._coherence.hostname:
if not self.knownHosts.has_key(ip):
self.knownHosts[ip] = Host(ip, self._config, self._coherence)
if not self._knownHosts.has_key(ip):
self._knownHosts[ip] = Host(ip, self._config, self._coherence)'Host added', ip)
self.debug('Calling dispatchDiscoveryPacketData')
#except Exception, error:
# self.warning(error)
......@@ -73,28 +101,67 @@ class UPnT(log.Loggable):
def hostRemoved(self, ip):
Remove host from the known hosts list
Remove host from the known hosts list.
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})
"""'hostRemoved(%r)' % ip)
def removeHosts(self):
self.knownHosts = None
def showInfoMessage(self, message):
Show info messages giving hints on better implementation.
Messages sent through this channel don't define the tested device as
@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})
print type + ': ' + message
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
at least one C{Deferred} is still running.
"""'Shutting UPnT down...')
if len(self.runningDeferreds) == 0:
if len(self._runningDeferreds) == 0:
return None
return defer.DeferredList(self.runningDeferreds)
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.
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.
# Licensed under the MIT license
# Copyright 2007, Michael Weinrich <>
import louie
from coherence import log
from upntest.eventing import ClientEventMessageChecks
from upntest.controlling import ClientControl
class ClientDevice(log.Loggable):
Class for storing data about client devices like ControlPoints.
logCategory = 'UPnT_ClientDevice'
eventchecker = None
controlchecker = None
def __init__(self, config):
templates_dir = config.get('templates_dir', None)
if not self.eventchecker:
self.eventchecker = ClientEventMessageChecks()
if not self.controlchecker:
self.controlchecker = ClientControl(templates_dir)
\ No newline at end of file
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