""" Testing the REST api, not the units. unittest the right way feels so unpythonic :( !! BIG FAT WARNING !!: this test deletes active torrents . !! BIG FAT WARNING 2!!: this test hammers the tracker that is tested against. """ import unittest import cookielib, urllib2 , urllib import WebUi.webserver_common as ws import operator ws.init_05() print 'test-env=',ws.ENV #CONFIG: BASE_URL = 'http://localhost:8112' PWD = 'deluge' def get_status(id): return ws.proxy.get_torrent_status(id,ws.TORRENT_KEYS) #BASE: #303 = see other #404 = not found #500 = server error #200 = OK, page exists. class TestWebUiBase(unittest.TestCase): def setUp(self): #cookie aware-opener that DOES NOT use redirects. opener = urllib2.OpenerDirector() self.cj = cookielib.CookieJar() for handler in [urllib2.HTTPHandler(),urllib2.HTTPDefaultErrorHandler(), urllib2.FileHandler(),urllib2.HTTPErrorProcessor(), urllib2.HTTPCookieProcessor(self.cj)]: opener.add_handler(handler) #/opener self.opener = opener def open_url(self, page, post=None): url = BASE_URL + page if post == 1: post = {'Force_a_post' : 'spam'} if post: post = urllib.urlencode(post) r = self.opener.open(url , data = post) #BUG: error-page does not return status 500, but status 200 #workaround... data = r.read() if '' in data: error = IOError() error.code = 500 #print data raise error if r.code <> 200: fail('no code 200, error-code=%s' % r.code) return r def get_cookies(self): return dict((c.name,c.value) for c in self.cj) cookies = property(get_cookies) def assert_status(self,status, page, post): try : r = self.open_url(page, post) except IOError,e: self.assertEqual(e.code, status) else: self.fail('page was found "%s" (%s)' % (page, r.code )) def assert_404(self, page, post = None): self.assert_status(404, page, post) def assert_500(self, page, post = None): self.assert_status(500, page, post) def assert_303(self, page, redirect_to, post=None): try : r = self.open_url(page, post) except IOError,e: self.assertEqual(e.code, 303) self.assertEqual(e.headers['Location'], redirect_to) else: #print r self.fail('No 303!') def assert_exists(self, page, post = None): try : r = self.open_url(page, post) except IOError,e: self.fail('page was not found "%s" (%s)' % (page, e.code)) else: pass first_torrent_id = property(lambda self: ws.proxy.get_session_state()[0]) first_torrent = property(lambda self: get_status(self.first_torrent_id)) class TestNoAuth(TestWebUiBase): def test303(self): self.assert_303('/','/login') self.assert_303('','/login') self.assert_303('/index','/login') #self.assert_303('/torrent/pause/','/login') self.assert_303('/config','/login') self.assert_303('/torrent/info/','/login') def test404(self): self.assert_404('/torrent/info') self.assert_404('/garbage') #self.assert_404('/static/garbage') #self.assert_404('/template/static/garbage') self.assert_404('/torrent/pause/', post=1) def testOpen(self): self.assert_exists('/login') self.assert_exists('/about') def testStatic(self): self.assert_exists('/static/images/simple_line.jpg') self.assert_exists('/static/images/tango/up.png') #test 404 #test template-static class TestSession(TestWebUiBase): def testLogin(self): self.assert_303('/home','/login') #invalid pwd: self.assert_303('/login','/login?error=1',{'pwd':'invalid'}) #login self.assert_303('/login','/index',{'pwd':PWD}) #now i'm logged-in! #there are no sort-coockies yet so the default page is /index. self.assert_303('/home','/index') self.assert_exists('/index') self.assert_exists('/config') self.assert_exists('/torrent/add') self.assert_303('/','/index') self.assert_303('','/index') #logout self.assert_303('/logout','/login', post=1) #really logged out? self.assert_303('/','/login') self.assert_303('','/login') self.assert_303('/index','/login') self.assert_303('/torrent/add','/login') self.assert_exists('/about') def testRefresh(self): #starting pos self.assert_303('/login','/index',{'pwd':PWD}) r = self.open_url('/index') assert not 'auto_refresh' in self.cookies assert not 'auto_refresh_secs' in self.cookies assert not r.headers.has_key('Refresh') #on: self.assert_303('/refresh/on','/index', post=1) assert 'auto_refresh' in self.cookies assert 'auto_refresh_secs' in self.cookies self.assertEqual(self.cookies['auto_refresh'],'1') self.assertEqual(self.cookies['auto_refresh_secs'],'10') r = self.open_url('/index') assert r.headers['Refresh'] == '10 ; url=/index' #set: self.assert_303('/refresh/set','/index',{'refresh':'5'}) self.assertEqual(self.cookies['auto_refresh_secs'],'5') r = self.open_url('/index') assert r.headers['Refresh'] == '5 ; url=/index' self.assert_500('/refresh/set',{'refresh':'a string'}) #off: self.assert_303('/refresh/off','/index', post=1) self.assertEqual(self.cookies['auto_refresh'],'0') self.assertEqual(self.cookies['auto_refresh_secs'],'5') r = self.open_url('/index') assert not 'Refresh' in r.headers class TestIntegration(TestWebUiBase): initialized = False def setUp(self): TestWebUiBase.setUp(self) self.assert_303('/login','/index',{'pwd':PWD}) self.urls = sorted([ 'http://torrents.aelitis.com:88/torrents/azplatform2_1.13.zip.torrent', 'http://torrents.aelitis.com:88/torrents/azplugins_2.1.4.jar.torrent', 'http://torrents.aelitis.com:88/torrents/azautoseeder_0.1.1.jar.torrent' ]) torrent_ids = ws.proxy.get_session_state() #avoid hammering, investigate current torrent-list and do not re-add. #correct means : 3 torrent's in list (for now) if len(torrent_ids) <> 3: #delete all, nice use case for refactoring delete.. torrent_ids = ws.proxy.get_session_state() for torrent in torrent_ids: ws.proxy.remove_torrent([torrent], False, False) torrent_ids = ws.proxy.get_session_state() self.assertEqual(torrent_ids, []) #add 3 using url. for url in self.urls: self.assert_303('/torrent/add','/index',{'url':url,'torrent':None}) #added? self.torrent_ids = ws.proxy.get_session_state() self.assertEqual(len(self.torrent_ids), 3) else: #test correctness of existing-list #The setup makes 0.6 fail everything, added an else.. for url in self.urls: if ws.ENV.startswith('0.5'): self.assert_500('/torrent/add',{'url':url,'torrent':None}) else: self.assert_303('/torrent/add','/index',{'url':url,'torrent':None}) def testPauseResume(self): #pause all self.assert_303('/pause_all','/index', post=1) #pause worked? pause_status = [get_status(id)["user_paused"] for id in ws.proxy.get_session_state()] for paused in pause_status: self.assertEqual(paused, True) #resume all self.assert_303('/resume_all','/index', post=1) #resume worked? pause_status = [get_status(id)["user_paused"] for id in ws.proxy.get_session_state()] for paused in pause_status: self.assertEqual(paused,False) #pause again. self.assert_303('/pause_all','/index', post=1) torrent_id = self.first_torrent_id #single resume. self.assert_303('/torrent/start/%s' % torrent_id ,'/index', post=1) self.assertEqual(get_status(torrent_id)["user_paused"] ,False) #single pause self.assert_303('/torrent/stop/%s' % torrent_id,'/index', post=1) self.assertEqual(get_status(torrent_id)["user_paused"] , True) def testQueue(self): #find last: torrent_id = [id for id in ws.proxy.get_session_state() if (get_status(id)['queue_pos'] ==3 )][0] #queue torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 3) #up: self.assert_303('/torrent/queue/up/%s' % torrent_id,'/index', post=1) torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 2) self.assert_303('/torrent/queue/up/%s' % torrent_id,'/index', post=1) torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 1) self.assert_303('/torrent/queue/up/%s' % torrent_id,'/index', post=1) #upper limit torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 1) #down: self.assert_303('/torrent/queue/down/%s' % torrent_id,'/index', post=1) torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 2) self.assert_303('/torrent/queue/down/%s' % torrent_id,'/index', post=1) torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 3) self.assert_303('/torrent/queue/down/%s' % torrent_id,'/index', post=1) #down limit torrent = get_status(torrent_id) self.assertEqual(torrent['queue_pos'], 3) def testMeta(self): #info available? for torrent_id in ws.proxy.get_session_state(): self.assert_exists('/torrent/info/%s' % torrent_id) self.assert_exists('/torrent/delete/%s' % torrent_id) #no info: self.assert_500('/torrent/info/99999999') self.assert_500('/torrent/delete/99999999') def testAddRemove(self): #add a duplicate: self.assert_500('/torrent/add', post={'url':self.urls[0],'torrent':None}) #add a 4th using url #delete #add torrrent-file #./test01.torrent def test_do_redirect(self): self.assert_303('/home','/index') #1 self.assert_exists('/index?sort=download_rate&order=down') self.assert_303('/home','/index?sort=download_rate&order=down') assert self.cookies['sort'] == 'download_rate' assert self.cookies['order'] == 'down' #2 self.assert_exists('/index?sort=progress&order=up') self.assert_303('/home','/index?sort=progress&order=up') assert self.cookies['sort'] == 'progress' assert self.cookies['order'] == 'up' #redir after pause-POST? in /index. self.assert_exists('/index?sort=name&order=down') torrent_id = self.first_torrent_id self.assert_303('/torrent/stop/%s' % torrent_id, '/index?sort=name&order=down', post=1) #redir in details 1 self.assert_303('/torrent/stop/%s?redir=/torrent/info/%s' %(torrent_id,torrent_id) ,'/torrent/info/' + torrent_id, post = 1) #redir in details 2 self.assert_303('/torrent/stop/%s' % torrent_id ,'/torrent/info/' + torrent_id , post={'redir': '/torrent/info/' + torrent_id}) def testRemote(self): pass def test_redir_after_login(self): pass def testReannounce(self): torrent_id = self.first_torrent_id self.assert_303( '/torrent/reannounce/%(id)s?redir=/torrent/info/%(id)s' % {'id':torrent_id} ,'/torrent/info/' + torrent_id, post = 1) def testRecheck(self): #add test before writing code.. #RELEASE-->disable """ torrent_id = self.first_torrent_id self.assert_303( '/torrent/recheck/%(id)s?redir=/torrent/info/%(id)s' % {'id':torrent_id} ,'/torrent/info/' + torrent_id, post = 1) """ # if False: suiteFew = unittest.TestSuite() suiteFew.addTest(TestSession("testRefresh")) unittest.TextTestRunner(verbosity=2).run(suiteFew) elif False: suiteFew = unittest.TestSuite() suiteFew.addTest(TestIntegration("testDoRedirect")) unittest.TextTestRunner(verbosity=2).run(suiteFew) else: unittest.main()