1#Licensed to the Apache Software Foundation (ASF) under one 2#or more contributor license agreements. See the NOTICE file 3#distributed with this work for additional information 4#regarding copyright ownership. The ASF licenses this file 5#to you under the Apache License, Version 2.0 (the 6#"License"); you may not use this file except in compliance 7#with the License. You may obtain a copy of the License at 8 9# http://www.apache.org/licenses/LICENSE-2.0 10 11#Unless required by applicable law or agreed to in writing, software 12#distributed under the License is distributed on an "AS IS" BASIS, 13#WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14#See the License for the specific language governing permissions and 15#limitations under the License. 16import unittest, os, sys, re, threading, time 17 18myDirectory = os.path.realpath(sys.argv[0]) 19rootDirectory = re.sub("/testing/.*", "", myDirectory) 20 21sys.path.append(rootDirectory) 22 23from hodlib.Common.xmlrpc import hodXRClient 24from hodlib.Common.socketServers import hodXMLRPCServer 25from hodlib.GridServices.service import ServiceUtil 26from hodlib.Common.util import hodInterrupt, HodInterruptException 27 28from testing.lib import BaseTestSuite 29 30excludes = [] 31 32global serverPort 33serverPort = None 34 35class test_HodXRClient(unittest.TestCase): 36 def setUp(self): 37 pass 38 39 # All testMethods have to have their names start with 'test' 40 def testSuccess(self): 41 global serverPort 42 client = hodXRClient('http://localhost:' + str(serverPort), retryRequests=False) 43 self.assertEqual(client.testing(), True) 44 pass 45 46 def testFailure(self): 47 """HOD should raise Exception when unregistered rpc is called""" 48 global serverPort 49 client = hodXRClient('http://localhost:' + str(serverPort), retryRequests=False) 50 self.assertRaises(Exception, client.noMethod) 51 pass 52 53 def testTimeout(self): 54 """HOD should raise Exception when rpc call times out""" 55 # Give client some random nonexistent url 56 serverPort = ServiceUtil.getUniqRandomPort(h='localhost',low=40000,high=50000) 57 client = hodXRClient('http://localhost:' + str(serverPort), retryRequests=False) 58 self.assertRaises(Exception, client.testing) 59 pass 60 61 def testInterrupt(self): 62 """ HOD should raise HodInterruptException when interrupted""" 63 64 def interrupt(testClass): 65 testClass.assertRaises(HodInterruptException, client.testing) 66 67 serverPort = ServiceUtil.getUniqRandomPort(h='localhost',low=40000,high=50000) 68 client = hodXRClient('http://localhost:' + str(serverPort)) 69 myThread = threading.Thread(name='testinterrupt', target=interrupt,args=(self,)) 70 # Set the global interrupt 71 hodInterrupt.setFlag() 72 myThread.start() 73 myThread.join() 74 pass 75 76 def tearDown(self): 77 pass 78 79class XmlrpcTestSuite(BaseTestSuite): 80 def __init__(self): 81 # suite setup 82 BaseTestSuite.__init__(self, __name__, excludes) 83 84 def rpcCall(): 85 return True 86 87 global serverPort 88 serverPort = ServiceUtil.getUniqRandomPort(h='localhost',low=40000,high=50000) 89 self.server = hodXMLRPCServer('localhost', [serverPort]) 90 self.server.register_function(rpcCall, 'testing') 91 self.thread = threading.Thread(name="server", 92 target=self.server._serve_forever) 93 self.thread.start() 94 time.sleep(1) # give some time to start server 95 96 def cleanUp(self): 97 # suite tearDown 98 self.server.stop() 99 self.thread.join() 100 101def RunXmlrpcTests(): 102 # modulename_suite 103 suite = XmlrpcTestSuite() 104 testResult = suite.runTests() 105 suite.cleanUp() 106 return testResult 107 108if __name__ == "__main__": 109 RunXmlrpcTests() 110