#!/usr/bin/python # # arch-tag: unit tests for SOAP server # Time-stamp: <2006-03-22 10:49:40 mike> # import unittest import random import urllib2 from SOAPpy import * import SOAPpy.Types from elementsoap.ElementSOAP import * import threading import time WSDL_URL = '' EXTERNAL_LOADS = 16 INTERNAL_LOADS = 16 BUSES = 6 SENSORS = 25 def geturl(url): if url.startswith('file://'): src = open(url[7:], 'r') else: src = urllib2.urlopen(urllib2.Request(url)) contents = src.read() src.close() return contents class PmacsService(SoapService): _typemap = {type('') : 'string', type(1) : 'int', type(1.0) : 'float'} def __init__(self, wsdl): root = ElementTree.fromstring(geturl(wsdl)) soap_ns = '{http://schemas.xmlsoap.org/wsdl/soap/}' wsdl_ns = '{http://schemas.xmlsoap.org/wsdl/}' node = root.find('%sservice/%sport/%saddress' % (wsdl_ns, wsdl_ns, soap_ns)) self.url = node.get('location') self.ops = {} for node in root.findall('%sbinding/%soperation' % (wsdl_ns, wsdl_ns)): name = node.get('name') e = node.find(soap_ns + 'operation') action = e.get('soapAction') e = node.find('%sinput/%sbody' % (wsdl_ns, soap_ns)) ns = '{' + e.get('namespace') + '}' self.ops[name] = (ns, action) SoapService.__init__(self) def _to_struct(self, root): s = {} for node in root: s[node.tag] = self._to_type(node) return s def _to_array(self, root): a = [] for node in root.findall('item'): pass return a def setValidator(self, url, ns): namespace,action = self.ops['setValidator'] header = SoapHeader() sec = SoapSecurity(header) ut = SoapUsernameToken(sec, 'admin', 'pmacs') request = SoapRequest(namespace + 'setValidator') SoapElement(request, namespace + 'url', 'string', url) SoapElement(request, namespace + 'ns', 'string', ns) response = self.call(action, request, header=header) return response.findtext('Result') def setLoad(self, id, newstate): namespace,action = self.ops['setLoad'] header = SoapHeader() sec = SoapSecurity(header) ut = SoapUsernameToken(sec, 'admin', 'pmacs') request = SoapRequest(namespace + 'setLoad') SoapElement(request, namespace + 'id', 'string', id) s = SoapElement(request, self.ns + 'newstate') for k,v in newstate.items(): SoapElement(s, self.ns + k, self._typemap[type(v)], str(v)) response = self.call(action, request, header=header) return response def getstate(power, deadface): """Return the state of a Load (0=off, 1=on, -1=isolated)""" if deadface == 0: return -1 return power class getLoadsThread(threading.Thread): def __init__(self, server, id): self.server = server self.id = id self.go = 0 threading.Thread.__init__(self) self.setDaemon(1) def run(self): self.go = 1 while self.go: self.server.getLoads(id=self.id) time.sleep(1) def stop(self): self.go = 0 class checkGFthread(threading.Thread): def __init__(self, server, busname): self.server = server self.busname = busname self.resp_time = 0 threading.Thread.__init__(self) def run(self): self.req_time = time.time() resp = self.server.checkGF(id=self.busname) self.resp_time = time.time() class ServerTests(unittest.TestCase): def setUp(self): self.proxy = WSDL.Proxy(WSDL_URL) state = self.proxy.getNPCState() self.failUnless(state == 1, 'NPC is not responding') def testGetLoads(self): """Check the output of getLoads""" keys = [u'current_limit', u'deadface', u'bus', u'current', u'switch', u'error', u'breaker', u'port', u'name'] keys.sort() result = self.proxy.getLoads(id='/1/external') self.assertEqual(len(result.data), EXTERNAL_LOADS, 'bad external load count (%d)' % len(result.data)) keylist = result.data[0]['_keyord'] keylist.sort() self.assertEqual(keylist, keys, 'bad load structure fields: %s' % str(keylist)) result = self.proxy.getLoads(id='/1/internal') self.assertEqual(len(result.data), INTERNAL_LOADS, 'bad internal load count (%d)' % len(result.data)) keylist = result.data[0]['_keyord'] keylist.sort() self.assertEqual(keylist, keys, 'bad load structure fields: %s' % str(keylist)) def testGetMV(self): """Check the output of getMV""" keys = [u'voltage', u'current', u'current_diff', u'aworking', u'bworking', u'aok', u'bok', u'v400a', u'v400b', u'i400a', u'i400b', u's2open', u'astages', u'bstages', u'tsecs', u'tusecs'] keys.sort() result = self.proxy.getMV(id='/mv') keylist = result['_keyord'] keylist.sort() self.assertEqual(keylist, keys, 'bad load structure fields: %s' % str(keylist)) def testGetLoad(self): """Check the results of getLoads for a single port""" result = self.proxy.getLoads(id='/1/external/4/v48') self.assertEqual(len(result.data), 1, 'bad 48v external load count (%d)' % len(result.data)) result = self.proxy.getLoads(id='/1/external/4/v400') self.assertEqual(len(result.data), 1, 'bad 400v external load count (%d)' % len(result.data)) result = self.proxy.getLoads(id='/1/internal/4') self.assertEqual(len(result.data), 1, 'bad internal load count (%d)' % len(result.data)) def testBadName(self): """Check results of invalid load IDs""" self.proxy.soapproxy.throw_faults = 1 self.assertRaises(SOAPpy.Types.faultType, self.proxy.getLoads, id='/1/external/12/v48') self.assertRaises(SOAPpy.Types.faultType, self.proxy.getLoads, id='/1/internal/20/v48') self.assertRaises(SOAPpy.Types.faultType, self.proxy.setLoad, id='/1/external/20/v48', newstate={'name' : 'load20'}) def notestChangeState(self): """Check that we can change the state of a load""" n = random.choice(range(1, 9)) id = '/1/external/%d/v48' % n result = self.proxy.getLoads(id=id) state0 = getstate(result.data[0]['switch'], result.data[0]['deadface']) if state0 == 0: order = [1, -1, 0] elif state0 == 1: order = [-1, 0, 1] else: order = [0, 1, -1] for state in order: self.proxy.setLoad(id=id, newstate={'state' : state}) time.sleep(1.5) result = self.proxy.getLoads(id=id) check = getstate(result.data[0]['switch'], result.data[0]['deadface']) self.assertEqual(state, check, 'cannot change state of %s to %d' % (id, state)) def testSampling(self): """Check that sampled data is being updated (~5 second run time)""" r0 = self.proxy.getLoads(id='/1/internal/4') time.sleep(5) r1 = self.proxy.getLoads(id='/1/internal/4') dt = r1.tsecs - r0.tsecs self.failUnless(dt >= 4, 'samples not being updated properly (dt = %d)' % dt) def testGetBus(self): """Check the results of getBuses""" keys = [u'current', u'voltage', u'name'] keys.sort() result = self.proxy.getBuses(id='/1/bus') self.assertEqual(len(result.data), BUSES, 'bad bus count (%d)' % len(result.data)) keylist = result.data[0]['_keyord'] keylist.sort() self.assertEqual(keylist, keys, 'bad bus structure fields: %s' % str(keylist)) def testGetSensors(self): """Check the results of getSensors""" keys = [u'units', u'name', u'value'] keys.sort() result = self.proxy.getSensors(id='/1/sensor') keylist = result.data[0]['_keyord'] keylist.sort() self.assertEqual(keylist, keys, 'bad sensor structure fields: %s' % str(keylist)) def testGFcheck(self): """Check that we can run checkGF for all buses (~10 sec run time)""" for name in 'v400 v400r v48 v48r test'.split(): self.proxy.checkGF(id=name) def testGFmulti(self): """Check that multiple threads calling checkGF() will block""" t1 = checkGFthread(self.proxy, 'v400') t2 = checkGFthread(self.proxy, 'v48') t1.start() t2.start() t1.join(timeout=5) t2.join(timeout=5) self.failUnless(abs(t1.resp_time - t2.resp_time) >= 1, "call did not block") def testGetSPS(self): """Check the output of getSPS""" state = self.proxy.getPSCState() self.failUnless(state == 1, 'PSC is not responding') keys = [u'current', u'voltage', u'state', u'tc'] keys.sort() result = self.proxy.getSPS(id='/psc') keylist = result['_keyord'] keylist.sort() self.assertEqual(keylist, keys, 'bad load structure fields: %s' % str(keylist)) class OperationalServerTests(ServerTests): def testCurrent(self): """Check that loads which are off or isolated have zero current""" result = self.proxy.getLoads(id='/1') for e in result.data: if e['switch'] == 0 or e['deadface'] == 0: self.failUnless(e['current'] == 0, 'bad current reading (%s, %d)' % (e['bus'], e['port'])) if __name__ == '__main__': import sys try: WSDL_URL = sys.argv[1] except IndexError: WSDL_URL = 'http://pmacs.apl.washington.edu/wsdl/pmacs.wsdl' suite = unittest.TestSuite() suite.addTest(unittest.makeSuite(ServerTests)) unittest.TextTestRunner(verbosity=2).run(suite)