# -*- coding: UTF-8 -*- """The site URL server package test""" # Copyright 2005, 2006 EIAO Consoritum # This program is distributed under the terms of the GNU General # Public License. # # This file is part of the European Internet Accessibility Observatory # (EIAO) # # EIAO is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # EIAO is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with EIAO; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, # MA 02110-1301 USA __owner__ = "Terje Gjosater, Morten Goodwin Olsen" __maintainer__ = "Nils Ulltveit-More" __version__ = 0.1 import SOAPpy import os import sys import unittest import urllib import urlparse import time from sc import SystemConfiguration sc=SystemConfiguration() ETLSERVER=urllib.unquote(sc.etlserver) URLSERVER=urllib.unquote(sc.siteurlserver) """Test suite for the siteURLserver module""" class TestSequenceFunctions(unittest.TestCase): def __init__(self, testcase): unittest.TestCase.__init__(self, testcase) processes=os.popen("ps ax").read() try: self.assert_("siteurlserver" in processes) except AssertionError: print "Please start the site URL server to run this test" print "Run e.g. /etc/init.d/siteurlserver start" sys.exit(1) try: self.assert_("etlserver" in processes) except AssertionError: print "Please start the ETL server with \"nowrite\" option to run this test" print "Run e.g. etlserver nowrite" sys.exit(1) def setUp(self): self.urlserver = SOAPpy.SOAPProxy(URLSERVER) self.etlserver = SOAPpy.SOAPProxy(ETLSERVER) def tearDown(self): self.urlserver.removeTestRun() def test_1_addurls(self): success = self.urlserver.addURLs("testrun1", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.addURLsFromTable("testrun3", ETLSERVER, "Site") self.assert_(success) success = self.urlserver.addURLsFromTable("testrun3", ETLSERVER, "Site") self.assert_(not success) def test_2_removetestruns(self): success = self.urlserver.addURLs("testrun4", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.addURLs("testrun5", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.addURLs("testrun6", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.addURLs("testrun7", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.removeTestRun("testrun42") self.assert_(not success) success = self.urlserver.removeTestRun("current") self.assert_(success) success = self.urlserver.removeTestRun("testrun6") self.assert_(success) success = self.urlserver.removeTestRun() self.assert_(success) success = self.urlserver.removeTestRun("testrun6") self.assert_(not success) def test_3_geturls(self): #get urls across testruns success = self.urlserver.removeTestRun() success = self.urlserver.addURLs("testrun8", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.addURLs("testrun9", ETLSERVER, "http://odin.dep.no") self.assert_(success) success = self.urlserver.addURLs("testrun10", ETLSERVER, "http://odin.dep.no") self.assert_(success) urltuple = self.urlserver.getSiteURL() #print urltuple success = self.urlserver.finishedSiteURL(urltuple[0],self.urlserver) self.assert_(success) self.assert_("http://odin.dep.no" in urltuple) urltuple = self.urlserver.getSiteURL() #print urltuple success = self.urlserver.finishedSiteURL(urltuple[0],self.urlserver) self.assert_(success) self.assert_("http://odin.dep.no" in urltuple) urltuple = self.urlserver.getSiteURL() #print urltuple success = self.urlserver.finishedSiteURL(urltuple[0],self.urlserver) self.assert_(success) self.assert_("http://odin.dep.no" in urltuple) #urltuple = self.urlserver.getSiteURL() #print urltuple #success = self.urlserver.finishedSiteURL("http://odin.dep.no",self.urlserver) #self.assert_(not success) #self.assert_(urltuple == None) #get several urls from one testrun self.urlserver.removeTestRun() success = self.urlserver.addURLs("testrun11", ETLSERVER, ["http://odin.dep.no", "http://odin2.dep.no"]) self.assert_(success) success = self.urlserver.finishedSiteURL("http://odin2.dep.no",self.urlserver) self.assert_(not success) urltuple = self.urlserver.getSiteURL() #print urltuple success = self.urlserver.finishedSiteURL(urltuple[0],self.urlserver) self.assert_(success) self.assert_("http://odin.dep.no" in urltuple or "http://odin2.dep.no" in urltuple) urltuple = self.urlserver.getSiteURL() #print urltuple success = self.urlserver.finishedSiteURL(urltuple[0],self.urlserver) self.assert_(success) self.assert_("http://odin.dep.no" in urltuple or "http://odin2.dep.no" in urltuple) #urltuple = self.urlserver.getSiteURL() #print urltuple success = self.urlserver.finishedSiteURL("http://odin2.dep.no",self.urlserver) # self.assert_(urltuple == None) self.assert_(not success) def test_4_rescheduling(self): success = self.urlserver.removeTestRun() success = self.urlserver.addURLs("testrun12", ETLSERVER, ["odin.dep.no","eiao2.eiao.net"]) self.assert_(success) # success = self.urlserver.addURLs("testrun13", ETLSERVER, "http://odin.dep.no") urltuple = self.urlserver.getSiteURL() # print urltuple self.assert_("odin.dep.no" in urltuple or "eiao2.eiao.net" in urltuple) self.urlserver.putSiteURL(urltuple[0],urltuple[1],urltuple[2],urltuple[3]) urltuple = self.urlserver.getSiteURL() self.assert_("odin.dep.no" in urltuple or "eiao2.eiao.net" in urltuple) # print urltuple self.urlserver.putSiteURL(urltuple[0],urltuple[1],urltuple[2],urltuple[3]) urltuple = self.urlserver.getSiteURL() self.assert_("odin.dep.no" in urltuple or "eiao2.eiao.net" in urltuple) # print urltuple self.urlserver.putSiteURL(urltuple[0],urltuple[1],urltuple[2],urltuple[3]) urltuple = self.urlserver.getSiteURL() self.assert_("odin.dep.no" in urltuple or "eiao2.eiao.net" in urltuple) # print urltuple self.urlserver.putSiteURL(urltuple[0],urltuple[1],urltuple[2],urltuple[3]) urltuple = self.urlserver.getSiteURL() self.assert_("odin.dep.no" in urltuple or "eiao2.eiao.net" in urltuple) # print urltuple self.urlserver.putSiteURL(urltuple[0],urltuple[1],urltuple[2],urltuple[3]) urltuple = self.urlserver.getSiteURL() self.assert_("odin.dep.no" in urltuple or "eiao2.eiao.net" in urltuple) # print urltuple #self.urlserver.finishedSiteURL(urltuple[0],self.urlserver) def test_5_keepalive(self): # Test that keepalive functionality works serviceEndpoint=urllib.unquote(sc.siteurlserver) host = urlparse.urlparse(urllib.unquote(serviceEndpoint))[1] if ':' in host: host = host[:host.find(':')] self.urlserver.keepAlive('www.foo.com',1337,host) self.urlserver.keepAlive('www.foo.com',1337,host) sc.maxnotimeout = '1' #print "Sleeping for %d seconds to get timeout..." % (int(sc.maxnotimeout)+1) time.sleep(int(sc.maxnotimeout)+1) # Expect crawler with PID=1337 to have timed out now self.assert_(1337 in self.urlserver.notAlive(host,duration=int(sc.maxnotimeout))) # Queue should be empty now self.assert_(len(self.urlserver.notAlive(host))==0) if __name__ == "__main__": unittest.main()