Mercurial > libervia-backend

comparison sat/test/helpers_plugins.py @ 2562:26edcf3a30eb

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
core, setup: huge cleaning: - moved directories from src and frontends/src to sat and sat_frontends, which is the recommanded naming convention - move twisted directory to root - removed all hacks from setup.py, and added missing dependencies, it is now clean - use https URL for website in setup.py - removed "Environment :: X11 Applications :: GTK", as wix is deprecated and removed - renamed sat.sh to sat and fixed its installation - added python_requires to specify Python version needed - replaced glib2reactor which use deprecated code by gtk3reactor sat can now be installed directly from virtualenv without using --system-site-packages anymore \o/
author Goffi <goffi@goffi.org>
date Mon, 02 Apr 2018 19:44:50 +0200
parents src/test/helpers_plugins.py@0046283a285d
children 56f94936df1e
comparison
equal deleted inserted replaced
2561:bd30dc3ffe5a 2562:26edcf3a30eb
1 #!/usr/bin/env python2
2 # -*- coding: utf-8 -*-
3
4 # SAT: a jabber client
5 # Copyright (C) 2009-2018 Jérôme Poisson (goffi@goffi.org)
6 # Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org)
7
8 # This program is free software: you can redistribute it and/or modify
9 # it under the terms of the GNU Affero General Public License as published by
10 # the Free Software Foundation, either version 3 of the License, or
11 # (at your option) any later version.
12
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 # GNU Affero General Public License for more details.
17
18 # You should have received a copy of the GNU Affero General Public License
19 # along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21 """ Helpers class for plugin dependencies """
22
23 from twisted.internet import defer
24
25 from wokkel.muc import Room, User
26 from wokkel.generic import parseXml
27 from wokkel.disco import DiscoItem, DiscoItems
28
29 # temporary until the changes are integrated to Wokkel
30 from sat_tmp.wokkel.rsm import RSMResponse
31
32 from constants import Const as C
33 from sat.plugins import plugin_xep_0045
34 from collections import OrderedDict
35
36
37 class FakeMUCClient(object):
38 def __init__(self, plugin_parent):
39 self.plugin_parent = plugin_parent
40 self.host = plugin_parent.host
41 self.joined_rooms = {}
42
43 def join(self, room_jid, nick, options=None, profile_key=C.PROF_KEY_NONE):
44 """
45 @param room_jid: the room JID
46 @param nick: nick to be used in the room
47 @param options: joining options
48 @param profile_key: the profile key of the user joining the room
49 @return: the deferred joined wokkel.muc.Room instance
50 """
51 profile = self.host.memory.getProfileName(profile_key)
52 roster = {}
53
54 # ask the other profiles to fill our roster
55 for i in xrange(0, len(C.PROFILE)):
56 other_profile = C.PROFILE[i]
57 if other_profile == profile:
58 continue
59 try:
60 other_room = self.plugin_parent.clients[other_profile].joined_rooms[room_jid]
61 roster.setdefault(other_room.nick, User(other_room.nick, C.PROFILE_DICT[other_profile]))
62 for other_nick in other_room.roster:
63 roster.setdefault(other_nick, other_room.roster[other_nick])
64 except (AttributeError, KeyError):
65 pass
66
67 # rename our nick if it already exists
68 while nick in roster.keys():
69 if C.PROFILE_DICT[profile].userhost() == roster[nick].entity.userhost():
70 break # same user with different resource --> same nickname
71 nick = nick + "_"
72
73 room = Room(room_jid, nick)
74 room.roster = roster
75 self.joined_rooms[room_jid] = room
76
77 # fill the other rosters with the new entry
78 for i in xrange(0, len(C.PROFILE)):
79 other_profile = C.PROFILE[i]
80 if other_profile == profile:
81 continue
82 try:
83 other_room = self.plugin_parent.clients[other_profile].joined_rooms[room_jid]
84 other_room.roster.setdefault(room.nick, User(room.nick, C.PROFILE_DICT[profile]))
85 except (AttributeError, KeyError):
86 pass
87
88 return defer.succeed(room)
89
90 def leave(self, roomJID, profile_key=C.PROF_KEY_NONE):
91 """
92 @param roomJID: the room JID
93 @param profile_key: the profile key of the user joining the room
94 @return: a dummy deferred
95 """
96 profile = self.host.memory.getProfileName(profile_key)
97 room = self.joined_rooms[roomJID]
98 # remove ourself from the other rosters
99 for i in xrange(0, len(C.PROFILE)):
100 other_profile = C.PROFILE[i]
101 if other_profile == profile:
102 continue
103 try:
104 other_room = self.plugin_parent.clients[other_profile].joined_rooms[roomJID]
105 del other_room.roster[room.nick]
106 except (AttributeError, KeyError):
107 pass
108 del self.joined_rooms[roomJID]
109 return defer.Deferred()
110
111
112 class FakeXEP_0045(plugin_xep_0045.XEP_0045):
113
114 def __init__(self, host):
115 self.host = host
116 self.clients = {}
117 for profile in C.PROFILE:
118 self.clients[profile] = FakeMUCClient(self)
119
120 def join(self, room_jid, nick, options={}, profile_key='@DEFAULT@'):
121 """
122 @param roomJID: the room JID
123 @param nick: nick to be used in the room
124 @param options: ignore
125 @param profile_key: the profile of the user joining the room
126 @return: the deferred joined wokkel.muc.Room instance or None
127 """
128 profile = self.host.memory.getProfileName(profile_key)
129 if room_jid in self.clients[profile].joined_rooms:
130 return defer.succeed(None)
131 room = self.clients[profile].join(room_jid, nick, profile_key=profile)
132 return room
133
134 def joinRoom(self, muc_index, user_index):
135 """Called by tests
136 @return: the nickname of the user who joined room"""
137 muc_jid = C.MUC[muc_index]
138 nick = C.JID[user_index].user
139 profile = C.PROFILE[user_index]
140 self.join(muc_jid, nick, profile_key=profile)
141 return self.getNick(muc_index, user_index)
142
143 def leave(self, room_jid, profile_key='@DEFAULT@'):
144 """
145 @param roomJID: the room JID
146 @param profile_key: the profile of the user leaving the room
147 @return: a dummy deferred
148 """
149 profile = self.host.memory.getProfileName(profile_key)
150 if room_jid not in self.clients[profile].joined_rooms:
151 raise plugin_xep_0045.UnknownRoom("This room has not been joined")
152 return self.clients[profile].leave(room_jid, profile)
153
154 def leaveRoom(self, muc_index, user_index):
155 """Called by tests
156 @return: the nickname of the user who left the room"""
157 muc_jid = C.MUC[muc_index]
158 nick = self.getNick(muc_index, user_index)
159 profile = C.PROFILE[user_index]
160 self.leave(muc_jid, profile_key=profile)
161 return nick
162
163 def getRoom(self, muc_index, user_index):
164 """Called by tests
165 @return: a wokkel.muc.Room instance"""
166 profile = C.PROFILE[user_index]
167 muc_jid = C.MUC[muc_index]
168 try:
169 return self.clients[profile].joined_rooms[muc_jid]
170 except (AttributeError, KeyError):
171 return None
172
173 def getNick(self, muc_index, user_index):
174 try:
175 return self.getRoomNick(C.MUC[muc_index], C.PROFILE[user_index])
176 except (KeyError, AttributeError):
177 return ''
178
179 def getNickOfUser(self, muc_index, user_index, profile_index, secure=True):
180 try:
181 room = self.clients[C.PROFILE[profile_index]].joined_rooms[C.MUC[muc_index]]
182 return self.getRoomNickOfUser(room, C.JID[user_index])
183 except (KeyError, AttributeError):
184 return None
185
186
187 class FakeXEP_0249(object):
188
189 def __init__(self, host):
190 self.host = host
191
192 def invite(self, target, room, options={}, profile_key='@DEFAULT@'):
193 """
194 Invite a user to a room. To accept the invitation from a test,
195 just call FakeXEP_0045.joinRoom (no need to have a dedicated method).
196 @param target: jid of the user to invite
197 @param room: jid of the room where the user is invited
198 @options: attribute with extra info (reason, password) as in #XEP-0249
199 @profile_key: %(doc_profile_key)s
200 """
201 pass
202
203
204 class FakeSatPubSubClient(object):
205
206 def __init__(self, host, parent_plugin):
207 self.host = host
208 self.parent_plugin = parent_plugin
209 self.__items = OrderedDict()
210 self.__rsm_responses = {}
211
212 def createNode(self, service, nodeIdentifier=None, options=None,
213 sender=None):
214 return defer.succeed(None)
215
216 def deleteNode(self, service, nodeIdentifier, sender=None):
217 try:
218 del self.__items[nodeIdentifier]
219 except KeyError:
220 pass
221 return defer.succeed(None)
222
223 def subscribe(self, service, nodeIdentifier, subscriber,
224 options=None, sender=None):
225 return defer.succeed(None)
226
227 def unsubscribe(self, service, nodeIdentifier, subscriber,
228 subscriptionIdentifier=None, sender=None):
229 return defer.succeed(None)
230
231 def publish(self, service, nodeIdentifier, items=None, sender=None):
232 node = self.__items.setdefault(nodeIdentifier, [])
233
234 def replace(item_obj):
235 index = 0
236 for current in node:
237 if current['id'] == item_obj['id']:
238 node[index] = item_obj
239 return True
240 index += 1
241 return False
242
243 for item in items:
244 item_obj = parseXml(item) if isinstance(item, unicode) else item
245 if not replace(item_obj):
246 node.append(item_obj)
247 return defer.succeed(None)
248
249 def items(self, service, nodeIdentifier, maxItems=None, itemIdentifiers=None,
250 subscriptionIdentifier=None, sender=None, ext_data=None):
251 try:
252 items = self.__items[nodeIdentifier]
253 except KeyError:
254 items = []
255 if ext_data:
256 assert('id' in ext_data)
257 if 'rsm' in ext_data:
258 args = (0, items[0]['id'], items[-1]['id']) if items else ()
259 self.__rsm_responses[ext_data['id']] = RSMResponse(len(items), *args)
260 return defer.succeed(items)
261
262 def retractItems(self, service, nodeIdentifier, itemIdentifiers, sender=None):
263 node = self.__items[nodeIdentifier]
264 for item in [item for item in node if item['id'] in itemIdentifiers]:
265 node.remove(item)
266 return defer.succeed(None)
267
268 def getRSMResponse(self, id):
269 if id not in self.__rsm_responses:
270 return {}
271 result = self.__rsm_responses[id].toDict()
272 del self.__rsm_responses[id]
273 return result
274
275 def subscriptions(self, service, nodeIdentifier, sender=None):
276 return defer.succeed([])
277
278 def service_getDiscoItems(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE):
279 items = DiscoItems()
280 for item in self.__items.keys():
281 items.append(DiscoItem(service, item))
282 return defer.succeed(items)