1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
#
# 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 pytest_twisted
from twisted.internet.defer import maybeDeferred, succeed
from twisted.internet.task import Clock
import deluge.component as component
import deluge.ui.sessionproxy
from deluge.conftest import BaseTestCase
class Core:
def __init__(self):
self.reset()
def reset(self):
self.torrents = {}
self.torrents['a'] = {'key1': 1, 'key2': 2, 'key3': 3}
self.torrents['b'] = {'key1': 1, 'key2': 2, 'key3': 3}
self.torrents['c'] = {'key1': 1, 'key2': 2, 'key3': 3}
self.prev_status = {}
def get_session_state(self):
return maybeDeferred(self.torrents.keys)
def get_torrent_status(self, torrent_id, keys, diff=False):
if not keys:
keys = list(self.torrents[torrent_id])
if not diff:
ret = {}
for key in keys:
ret[key] = self.torrents[torrent_id][key]
return succeed(ret)
else:
ret = {}
if torrent_id in self.prev_status:
for key in keys:
if (
self.prev_status[torrent_id][key]
!= self.torrents[torrent_id][key]
):
ret[key] = self.torrents[torrent_id][key]
else:
ret = self.torrents[torrent_id]
self.prev_status[torrent_id] = dict(self.torrents[torrent_id])
return succeed(ret)
def get_torrents_status(self, filter_dict, keys, diff=False):
if not filter_dict:
filter_dict['id'] = list(self.torrents)
if not keys:
keys = list(self.torrents['a'])
if not diff:
if 'id' in filter_dict:
torrents = filter_dict['id']
ret = {}
for torrent in torrents:
ret[torrent] = {}
for key in keys:
ret[torrent][key] = self.torrents[torrent][key]
return succeed(ret)
else:
if 'id' in filter_dict:
torrents = filter_dict['id']
ret = {}
for torrent in torrents:
ret[torrent] = {}
if torrent in self.prev_status:
for key in self.prev_status[torrent]:
if (
self.prev_status[torrent][key]
!= self.torrents[torrent][key]
):
ret[torrent][key] = self.torrents[torrent][key]
else:
ret[torrent] = dict(self.torrents[torrent])
self.prev_status[torrent] = dict(self.torrents[torrent])
return succeed(ret)
class Client:
def __init__(self):
self.core = Core()
def __noop__(self, *args, **kwargs):
return None
def __getattr__(self, *args, **kwargs):
return self.__noop__
client = Client()
class TestSessionProxy(BaseTestCase):
def set_up(self):
self.clock = Clock()
self.patch(deluge.ui.sessionproxy, 'time', self.clock.seconds)
self.patch(deluge.ui.sessionproxy, 'client', client)
self.sp = deluge.ui.sessionproxy.SessionProxy()
client.core.reset()
d = self.sp.start()
def do_get_torrents_status(torrent_ids):
inital_keys = ['key1']
# Advance clock to expire the cache times
self.clock.advance(2)
return self.sp.get_torrents_status({'id': torrent_ids}, inital_keys)
d.addCallback(do_get_torrents_status)
return d
def tear_down(self):
return component.deregister(self.sp)
def test_startup(self):
assert client.core.torrents['a'] == self.sp.torrents['a'][1]
@pytest_twisted.ensureDeferred
async def test_get_torrent_status_no_change(self):
result = await self.sp.get_torrent_status('a', [])
assert result == client.core.torrents['a']
@pytest_twisted.ensureDeferred
async def test_get_torrent_status_change_with_cache(self):
client.core.torrents['a']['key1'] = 2
result = await self.sp.get_torrent_status('a', ['key1'])
assert result == {'key1': 1}
@pytest_twisted.ensureDeferred
async def test_get_torrent_status_change_without_cache(self):
client.core.torrents['a']['key1'] = 2
self.clock.advance(self.sp.cache_time + 0.1)
result = await self.sp.get_torrent_status('a', [])
assert result == client.core.torrents['a']
@pytest_twisted.ensureDeferred
async def test_get_torrent_status_key_not_updated(self):
self.clock.advance(self.sp.cache_time + 0.1)
self.sp.get_torrent_status('a', ['key1'])
client.core.torrents['a']['key2'] = 99
result = await self.sp.get_torrent_status('a', ['key2'])
assert result == {'key2': 99}
@pytest_twisted.ensureDeferred
async def test_get_torrents_status_key_not_updated(self):
self.clock.advance(self.sp.cache_time + 0.1)
self.sp.get_torrents_status({'id': ['a']}, ['key1'])
client.core.torrents['a']['key2'] = 99
result = await self.sp.get_torrents_status({'id': ['a']}, ['key2'])
assert result == {'a': {'key2': 99}}
|
Deluge
