diff options
Diffstat (limited to 'deluge/tests/test_web_api.py')
| -rw-r--r-- | deluge/tests/test_web_api.py | 202 |
1 files changed, 202 insertions, 0 deletions
diff --git a/deluge/tests/test_web_api.py b/deluge/tests/test_web_api.py new file mode 100644 index 0000000..814fecf --- /dev/null +++ b/deluge/tests/test_web_api.py @@ -0,0 +1,202 @@ +# +# Copyright (C) 2016 bendikro <bro.devel+deluge@gmail.com> +# +# This file is part of Deluge and is licensed under GNU General Public License 3.0, or later, with +# the additional special exception to link portions of this program with the OpenSSL library. +# See LICENSE for more details. +# + +import json +from io import BytesIO + +import pytest +import pytest_twisted +from twisted.internet import defer, reactor +from twisted.web.client import Agent, FileBodyProducer +from twisted.web.http_headers import Headers +from twisted.web.static import File + +import deluge.component as component + +from . import common +from .common_web import WebServerTestBase + +common.disable_new_release_check() + + +class TestWebAPI(WebServerTestBase): + @pytest.mark.xfail(reason='This just logs an error at the moment.') + async def test_connect_invalid_host(self): + with pytest.raises(Exception): + await self.deluge_web.web_api.connect('id') + + def test_connect(self, client): + d = self.deluge_web.web_api.connect(self.host_id) + + def on_connect(result): + assert type(result) == tuple + assert len(result) > 0 + return result + + d.addCallback(on_connect) + d.addErrback(self.fail) + return d + + def test_disconnect(self): + d = self.deluge_web.web_api.connect(self.host_id) + + @defer.inlineCallbacks + def on_connect(result): + assert self.deluge_web.web_api.connected() + yield self.deluge_web.web_api.disconnect() + assert not self.deluge_web.web_api.connected() + + d.addCallback(on_connect) + d.addErrback(self.fail) + return d + + def test_get_config(self): + config = self.deluge_web.web_api.get_config() + assert self.deluge_web.port == config['port'] + + def test_set_config(self): + config = self.deluge_web.web_api.get_config() + config['pwd_salt'] = 'new_salt' + config['pwd_sha1'] = 'new_sha' + config['sessions'] = { + '233f23632af0a74748bc5dd1d8717564748877baa16420e6898e17e8aa365e6e': { + 'login': 'skrot', + 'expires': 1460030877.0, + 'level': 10, + } + } + self.deluge_web.web_api.set_config(config) + web_config = component.get('DelugeWeb').config.config + assert config['pwd_salt'] != web_config['pwd_salt'] + assert config['pwd_sha1'] != web_config['pwd_sha1'] + assert config['sessions'] != web_config['sessions'] + + @defer.inlineCallbacks + def get_host_status(self): + host = list(self.deluge_web.web_api._get_host(self.host_id)) + host[3] = 'Online' + host[4] = '2.0.0.dev562' + status = yield self.deluge_web.web_api.get_host_status(self.host_id) + assert status == tuple(status) + + def test_get_host(self): + assert not self.deluge_web.web_api._get_host('invalid_id') + conn = list(self.deluge_web.web_api.hostlist.get_hosts_info()[0]) + assert self.deluge_web.web_api._get_host(conn[0]) == conn[0:4] + + def test_add_host(self): + conn = ['abcdef', '10.0.0.1', 0, 'user123', 'pass123'] + assert not self.deluge_web.web_api._get_host(conn[0]) + # Add valid host + result, host_id = self.deluge_web.web_api.add_host( + conn[1], conn[2], conn[3], conn[4] + ) + assert result + conn[0] = host_id + assert self.deluge_web.web_api._get_host(conn[0]) == conn[0:4] + + # Add already existing host + ret = self.deluge_web.web_api.add_host(conn[1], conn[2], conn[3], conn[4]) + assert ret == (False, 'Host details already in hostlist') + + # Add invalid port + conn[2] = 'bad port' + ret = self.deluge_web.web_api.add_host(conn[1], conn[2], conn[3], conn[4]) + assert ret == (False, 'Invalid port. Must be an integer') + + def test_remove_host(self): + conn = ['connection_id', '', 0, '', ''] + self.deluge_web.web_api.hostlist.config['hosts'].append(conn) + assert self.deluge_web.web_api._get_host(conn[0]) == conn[0:4] + # Remove valid host + assert self.deluge_web.web_api.remove_host(conn[0]) + assert not self.deluge_web.web_api._get_host(conn[0]) + # Remove non-existing host + assert not self.deluge_web.web_api.remove_host(conn[0]) + + def test_get_torrent_info(self): + filename = common.get_test_data_file('test.torrent') + ret = self.deluge_web.web_api.get_torrent_info(filename) + assert ret['name'] == 'azcvsupdater_2.6.2.jar' + assert ret['info_hash'] == 'ab570cdd5a17ea1b61e970bb72047de141bce173' + assert 'files_tree' in ret + + def test_get_torrent_info_with_md5(self): + filename = common.get_test_data_file('md5sum.torrent') + ret = self.deluge_web.web_api.get_torrent_info(filename) + # JSON dumping happens during response creation in normal usage + # JSON serialization may fail if any of the dictionary items are byte arrays rather than strings + ret = json.loads(json.dumps(ret)) + assert ret['name'] == 'test' + assert ret['info_hash'] == 'f6408ba9944cf9fe01b547b28f336b3ee6ec32c5' + assert 'files_tree' in ret + + def test_get_magnet_info(self): + ret = self.deluge_web.web_api.get_magnet_info( + 'magnet:?xt=urn:btih:SU5225URMTUEQLDXQWRB2EQWN6KLTYKN' + ) + assert ret['name'] == '953bad769164e8482c7785a21d12166f94b9e14d' + assert ret['info_hash'] == '953bad769164e8482c7785a21d12166f94b9e14d' + assert 'files_tree' in ret + + @pytest_twisted.inlineCallbacks + def test_get_torrent_files(self): + yield self.deluge_web.web_api.connect(self.host_id) + filename = common.get_test_data_file('test.torrent') + torrents = [ + {'path': filename, 'options': {'download_location': '/home/deluge/'}} + ] + yield self.deluge_web.web_api.add_torrents(torrents) + ret = yield self.deluge_web.web_api.get_torrent_files( + 'ab570cdd5a17ea1b61e970bb72047de141bce173' + ) + assert ret['type'] == 'dir' + assert ret['contents'] == { + 'azcvsupdater_2.6.2.jar': { + 'priority': 4, + 'index': 0, + 'offset': 0, + 'progress': 0.0, + 'path': 'azcvsupdater_2.6.2.jar', + 'type': 'file', + 'size': 307949, + } + } + + @pytest_twisted.inlineCallbacks + def test_download_torrent_from_url(self): + filename = 'ubuntu-9.04-desktop-i386.iso.torrent' + self.deluge_web.top_level.putChild( + filename.encode(), File(common.get_test_data_file(filename)) + ) + url = 'http://localhost:%d/%s' % (self.deluge_web.port, filename) + res = yield self.deluge_web.web_api.download_torrent_from_url(url) + assert res.endswith(filename) + + @pytest_twisted.inlineCallbacks + def test_invalid_json(self): + """ + If json_api._send_response does not return server.NOT_DONE_YET + this error is thrown when json is invalid: + exceptions.RuntimeError: Request.write called on a request after Request.finish was called. + + """ + agent = Agent(reactor) + bad_body = b'{ method": "auth.login" }' + d = yield agent.request( + b'POST', + b'http://127.0.0.1:%i/json' % self.deluge_web.port, + Headers( + { + b'User-Agent': [b'Twisted Web Client Example'], + b'Content-Type': [b'application/json'], + } + ), + FileBodyProducer(BytesIO(bad_body)), + ) + yield d |