codex-storage/deluge
1
0
Fork 0
mirror of https://github.com/codex-storage/deluge.git synced 2025-01-11 12:04:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Clean-up the tests and add some new ones

Browse Source
This commit is contained in:
Andrew Resch 2009-07-02 04:16:37 +00:00
parent 883e61909f
commit 0c0cfaf597
8 changed files with 70 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
deluge/tests/common.py Normal file
View File

@ -0,0 +1,8 @@
import tempfile
import deluge.configmanager
import deluge.log
deluge.log.setupLogger()
config_directory = tempfile.mkdtemp()
deluge.configmanager.set_config_dir(config_directory)

29
deluge/tests/test_alertmanager.py Normal file
View File

@ -0,0 +1,29 @@
from twisted.trial import unittest
import common
from deluge.core.alertmanager import AlertManager
from deluge.core.core import Core
class AlertManagerTestCase(unittest.TestCase):
def setUp(self):
self.core = Core()
self.am = AlertManager()
self.am.start()
def test_register_handler(self):
def handler(alert):
return
self.am.register_handler("dummy_alert", handler)
self.assertEquals(self.am.handlers["dummy_alert"], [handler])
def test_deregister_handler(self):
def handler(alert):
return
self.am.register_handler("dummy_alert", handler)
self.am.deregister_handler(handler)
self.assertEquals(self.am.handlers["dummy_alert"], [])

14
deluge/tests/test_authmanager.py Normal file
View File

@ -0,0 +1,14 @@
from twisted.trial import unittest
import common
from deluge.core.authmanager import AuthManager
class AuthManagerTestCase(unittest.TestCase):
def setUp(self):
self.auth = AuthManager()
self.auth.start()
def test_authorize(self):
from deluge.ui import common
self.assertEquals(self.auth.authorize(*common.get_localhost_auth()), 10)

95
deluge/tests/test_client.py
View File

@ -1,83 +1,26 @@
#------------------------------------------------------------------------------
#tests:
#------------------------------------------------------------------------------
import tempfile
import os
import signal
from deluge.ui.client import aclient, sclient
import common
def test_introspection():
print("*start introspection test*")
sclient.set_core_uri()
print("list_methods", sclient.list_methods())
print("sig of block_ip_range", sclient.methodSignature('block_ip_range'))
print("doc of block_ip_range", sclient.methodHelp('block_ip_range'))
from twisted.trial import unittest
def test_sync():
print("*start sync test*")
sclient.set_core_uri()
from deluge.ui.client import client
#get list of torrents and display the 1st.
torrent_ids = sclient.get_session_state()
print("session_state():", torrent_ids)
print("get_torrent_status(%s):" % torrent_ids[0],
sclient.get_torrent_status(torrent_ids[0], []))
sclient.pause_torrent(torrent_ids)
print("paused:", [
sclient.get_torrent_status(id, ['paused'])['paused']
for id in torrent_ids])
sclient.resume_torrent(torrent_ids)
print("resumed:", [
sclient.get_torrent_status(id, ['paused'])['paused']
for id in torrent_ids])
def test_async():
print("*start async test*")
torrent_ids = []
#callbacks:
def cb_session_state(temp_torrent_list):
print("session_state:" , temp_torrent_list)
torrent_ids.extend(temp_torrent_list)
def cb_torrent_status_full(status):
print("\ntorrent_status_full=", status)
def cb_torrent_status_paused(torrent_state):
print("paused=%s" % torrent_state['paused'])
#/callbacks
aclient.set_core_uri()
aclient.get_session_state(cb_session_state)
print("force_call 1")
aclient.force_call(block=True)
print("end force_call 1:", len(torrent_ids))
# Start a daemon to test with and wait a couple seconds to make sure it's started
client.start_daemon(58847, config_directory)
import time
time.sleep(2)
#has_callback+multicall
aclient.pause_torrent(torrent_ids)
aclient.force_call(block=True)
for id in torrent_ids:
aclient.get_torrent_status(cb_torrent_status_paused, id , ['paused'])
class ClientTestCase(unittest.TestCase):
def test_connect_no_credentials(self):
d = client.connect("localhost", 58847)
d.addCallback(self.assertEquals, 10)
aclient.get_torrent_status(cb_torrent_status_full, torrent_ids[0], [])
print("force_call 2")
aclient.force_call(block=True)
print("end force-call 2")
print("resume:")
aclient.resume_torrent(torrent_ids)
for id in torrent_ids:
aclient.get_torrent_status(cb_torrent_status_paused, id , ['paused'])
aclient.force_call(block=True)
test_introspection()
test_sync()
test_async()
def on_connect(result):
self.addCleanup(client.disconnect)
return result
d.addCallback(on_connect)
return d

80
deluge/tests/test_filters.py
View File

@ -1,80 +0,0 @@
#
# moving and refactoring torrent-filtering from labels-plugin to core.
#
KEYS = ["name","state", "label"]
#init:
from deluge.ui.client import sclient
sclient.set_core_uri()
torrent_id = sclient.get_session_state()[0]
torrent_id2 = sclient.get_session_state()[1]
#/init
def test_filter(filter):
status = sclient.get_torrents_status(filter, KEYS)
print len(status),status
print "#get_status_keys"
#both lines should return the same if all plugins are disabled.
#the 1st should be longer if the label plugin is enabled.
print sorted(sclient.get_torrent_status(torrent_id,[]).keys())
print sorted(sclient.get_status_keys())
print "#default, no filter argument."
test_filter(None)
if not (sclient.get_torrents_status({}, KEYS) == sclient.get_torrents_status(None, KEYS)):
raise Exception("should be equal")
print "#test keyword:"
test_filter({"keyword":["keyword1","prison"]})
print "#torrent_id filter:"
test_filter({"id":[torrent_id, torrent_id2]})
print "#filters on default status fields:"
print sclient.get_torrents_status({"state":["Paused","Downloading"]}, KEYS)
print sclient.get_torrents_status({"tracker_host":["aelitis.com"]}, KEYS)
print "#status fields from plugins:"
print "test&tpb:",len(sclient.get_torrents_status({"label":["test","tpb"]}, KEYS))
print "test:",len(sclient.get_torrents_status({"label":["test"]}, KEYS))
print "No Label:" , len(sclient.get_torrents_status({"label":[""]}, KEYS))
print "#registered filters, basic:"
print sclient.get_torrents_status({"keyword":["az"]}, KEYS)
print "#registered filters, overriude on 1 value(not yet)"
print sclient.get_torrents_status({"state":["Active"]}, KEYS)
print "#tree: Default (Active must be listed after Seeding)"
for field, items in sclient.get_filter_tree().iteritems():
print "*",field
for value, count in items:
print "-",value,count
print "#tree: Hide_zero (show=False)"
for field, items in sclient.get_filter_tree(False).iteritems():
print "*",field
for value, count in items:
print "-",value,count
print "#tree: Hide tracker"
for field, items in sclient.get_filter_tree(False, ["tracker_host"]).iteritems():
print "*",field
for value, count in items:
print "-",value,count
print "#tree: Hide all"
if sclient.get_filter_tree(False, ["tracker_host","label","state"]):
raise Exception("result should be {}")
print "hide-all :ok"
print "#must have an error here:"
try:
print sclient.get_torrents_status({"invalid-filter":[]}, KEYS)
print "WTF!"
except Exception, e:
print "ok, an exception was raised:", e ,e.message

9
deluge/tests/test_signalreceiver.py
View File

@ -1,9 +0,0 @@
from deluge.configmanager import ConfigManager
import deluge.xmlrpclib as xmlrpclib
config = ConfigManager("gtkui.conf")
client = xmlrpclib.ServerProxy("http://localhost:" + str(config["signal_port"]))
client.emit_signal("torrent_finished", "abc123")

13
deluge/tests/test_stats.py
View File

@ -1,13 +0,0 @@
#
# testing 123..
#
from deluge.ui.client import sclient
sclient.set_core_uri()
#/init
print "no-args:"
stats = sclient.get_stats()
for key in sorted(stats.keys()):
print key, ":", stats[key]

40
deluge/tests/test_tracker_icons.py
View File

@ -1,40 +0,0 @@
import time
import gobject
import os
from deluge.ui.tracker_icons import TrackerIcons
from deluge.common import get_default_config_dir
def del_old():
filename = os.path.join(get_default_config_dir("icons"),"legaltorrents.com.ico")
if os.path.exists(filename):
os.remove(filename)
def test_get():
del_old()
trackericons = TrackerIcons()
print trackericons.images
print trackericons.get("unknown2")
print trackericons.get("google.com")
print trackericons.get("legaltorrents.com")
time.sleep(5.0)
print trackericons.get("legaltorrents.com")
def callback1(value):
print "callback1:", value
return False
def test_async():
#test is broken :(,. but filtertreeview works.
del_old()
trackericons = TrackerIcons()
trackericons.get_async("legaltorrents.com",callback1)
print "here"
gobject.MainLoop()
test_get()
#test_async()