test_all.py

LINUX下

"""
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 '<!--ERROR-MARKER-->' 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()