test: add tests for plugin_misc_groupblog
This commit is contained in:
souliane
parent
f3b7bd1ec2
commit
d262aff438
4 changed files with 535 additions and 28 deletions
|
|
@ -23,6 +23,8 @@ from twisted.words.protocols.jabber import jid
|
|||
|
||||
class Const(object):
|
||||
|
||||
PROF_KEY_NONE = '@NONE@'
|
||||
|
||||
PROFILE = ['test_profile', 'test_profile2', 'test_profile3', 'test_profile4', 'test_profile5']
|
||||
JID_STR = [u"test@example.org/SàT", u"sender@example.net/house", u"sender@example.net/work", u"sender@server.net/res", u"xxx@server.net/res"]
|
||||
JID = [jid.JID(jid_s) for jid_s in JID_STR]
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ from sat.core import log_config
|
|||
log_config.satConfigure()
|
||||
|
||||
from sat.core import exceptions
|
||||
from constants import Const
|
||||
from constants import Const as C
|
||||
from wokkel.xmppim import RosterItem
|
||||
from sat.core.xmpp import SatRosterProtocol
|
||||
from sat.memory.memory import Params, Memory
|
||||
|
|
@ -123,7 +123,7 @@ class FakeSAT(object):
|
|||
def getJidNStream(self, profile_key):
|
||||
"""Convenient method to get jid and stream from profile key
|
||||
@return: tuple (jid, xmlstream) from profile, can be None"""
|
||||
return (Const.PROFILE_DICT[profile_key], None)
|
||||
return (C.PROFILE_DICT[profile_key], None)
|
||||
|
||||
def isConnected(self, profile):
|
||||
return True
|
||||
|
|
@ -134,7 +134,7 @@ class FakeSAT(object):
|
|||
been previously initialized with the method FakeSAT.getClient.
|
||||
@return: the sent message for given profile, or None"""
|
||||
try:
|
||||
return self.profiles[Const.PROFILE[profile_index]].xmlstream.sent.pop(0)
|
||||
return self.profiles[C.PROFILE[profile_index]].xmlstream.sent.pop(0)
|
||||
except IndexError:
|
||||
return None
|
||||
|
||||
|
|
@ -238,7 +238,7 @@ class FakeParams(Params):
|
|||
|
||||
def getProfileName(self, profile_key, return_profile_keys=False):
|
||||
if profile_key == '@DEFAULT@':
|
||||
return Const.PROFILE[0]
|
||||
return C.PROFILE[0]
|
||||
elif profile_key == '@NONE@':
|
||||
raise exceptions.ProfileNotSetError
|
||||
else:
|
||||
|
|
@ -345,8 +345,8 @@ class FakeClient(object):
|
|||
|
||||
def __init__(self, host, profile=None):
|
||||
self.host = host
|
||||
self.profile = profile if profile else Const.PROFILE[0]
|
||||
self.jid = Const.PROFILE_DICT[self.profile]
|
||||
self.profile = profile if profile else C.PROFILE[0]
|
||||
self.jid = C.PROFILE_DICT[self.profile]
|
||||
self.roster = FakeRosterProtocol(host, self)
|
||||
self.xmlstream = FakeXmlStream()
|
||||
|
||||
|
|
|
|||
|
|
@ -20,10 +20,16 @@
|
|||
|
||||
""" Helpers class for plugin dependencies """
|
||||
|
||||
from constants import Const
|
||||
from sat.plugins import plugin_xep_0045
|
||||
from twisted.internet import defer
|
||||
|
||||
from wokkel.muc import Room, User
|
||||
from wokkel.rsm import RSMResponse
|
||||
from wokkel.generic import parseXml
|
||||
from wokkel.disco import DiscoItem, DiscoItems
|
||||
|
||||
from constants import Const as C
|
||||
from sat.plugins import plugin_xep_0045
|
||||
from collections import OrderedDict
|
||||
|
||||
|
||||
class FakeMUCClient(object):
|
||||
|
|
@ -42,13 +48,13 @@ class FakeMUCClient(object):
|
|||
roster = {}
|
||||
|
||||
# ask the other profiles to fill our roster
|
||||
for i in xrange(0, len(Const.PROFILE)):
|
||||
other_profile = Const.PROFILE[i]
|
||||
for i in xrange(0, len(C.PROFILE)):
|
||||
other_profile = C.PROFILE[i]
|
||||
if other_profile == profile:
|
||||
continue
|
||||
try:
|
||||
other_room = self.plugin_parent.clients[other_profile].joined_rooms[roomJID.userhost()]
|
||||
roster.setdefault(other_room.nick, User(other_room.nick, Const.PROFILE_DICT[other_profile]))
|
||||
roster.setdefault(other_room.nick, User(other_room.nick, C.PROFILE_DICT[other_profile]))
|
||||
for other_nick in other_room.roster:
|
||||
roster.setdefault(other_nick, other_room.roster[other_nick])
|
||||
except (AttributeError, KeyError):
|
||||
|
|
@ -56,7 +62,7 @@ class FakeMUCClient(object):
|
|||
|
||||
# rename our nick if it already exists
|
||||
while nick in roster.keys():
|
||||
if Const.PROFILE_DICT[profile].userhost() == roster[nick].entity.userhost():
|
||||
if C.PROFILE_DICT[profile].userhost() == roster[nick].entity.userhost():
|
||||
break # same user with different resource --> same nickname
|
||||
nick = nick + "_"
|
||||
|
||||
|
|
@ -65,13 +71,13 @@ class FakeMUCClient(object):
|
|||
self.joined_rooms[roomJID.userhost()] = room
|
||||
|
||||
# fill the other rosters with the new entry
|
||||
for i in xrange(0, len(Const.PROFILE)):
|
||||
other_profile = Const.PROFILE[i]
|
||||
for i in xrange(0, len(C.PROFILE)):
|
||||
other_profile = C.PROFILE[i]
|
||||
if other_profile == profile:
|
||||
continue
|
||||
try:
|
||||
other_room = self.plugin_parent.clients[other_profile].joined_rooms[roomJID.userhost()]
|
||||
other_room.roster.setdefault(room.nick, User(room.nick, Const.PROFILE_DICT[profile]))
|
||||
other_room.roster.setdefault(room.nick, User(room.nick, C.PROFILE_DICT[profile]))
|
||||
except (AttributeError, KeyError):
|
||||
pass
|
||||
|
||||
|
|
@ -85,8 +91,8 @@ class FakeMUCClient(object):
|
|||
"""
|
||||
room = self.joined_rooms[roomJID.userhost()]
|
||||
# remove ourself from the other rosters
|
||||
for i in xrange(0, len(Const.PROFILE)):
|
||||
other_profile = Const.PROFILE[i]
|
||||
for i in xrange(0, len(C.PROFILE)):
|
||||
other_profile = C.PROFILE[i]
|
||||
if other_profile == profile:
|
||||
continue
|
||||
try:
|
||||
|
|
@ -103,7 +109,7 @@ class FakeXEP_0045(plugin_xep_0045.XEP_0045):
|
|||
def __init__(self, host):
|
||||
self.host = host
|
||||
self.clients = {}
|
||||
for profile in Const.PROFILE:
|
||||
for profile in C.PROFILE:
|
||||
self.clients[profile] = FakeMUCClient(self)
|
||||
|
||||
def join(self, room_jid, nick, options={}, profile_key='@DEFAULT@'):
|
||||
|
|
@ -124,9 +130,9 @@ class FakeXEP_0045(plugin_xep_0045.XEP_0045):
|
|||
def joinRoom(self, muc_index, user_index):
|
||||
"""Called by tests
|
||||
@return: the nickname of the user who joined room"""
|
||||
muc_jid = Const.MUC[muc_index]
|
||||
nick = Const.JID[user_index].user
|
||||
profile = Const.PROFILE[user_index]
|
||||
muc_jid = C.MUC[muc_index]
|
||||
nick = C.JID[user_index].user
|
||||
profile = C.PROFILE[user_index]
|
||||
self.join(muc_jid, nick, profile_key=profile)
|
||||
return self.getNick(muc_index, user_index)
|
||||
|
||||
|
|
@ -145,17 +151,17 @@ class FakeXEP_0045(plugin_xep_0045.XEP_0045):
|
|||
def leaveRoom(self, muc_index, user_index):
|
||||
"""Called by tests
|
||||
@return: the nickname of the user who left the room"""
|
||||
muc_jid = Const.MUC[muc_index]
|
||||
muc_jid = C.MUC[muc_index]
|
||||
nick = self.getNick(muc_index, user_index)
|
||||
profile = Const.PROFILE[user_index]
|
||||
profile = C.PROFILE[user_index]
|
||||
self.leave(muc_jid, profile_key=profile)
|
||||
return nick
|
||||
|
||||
def getRoom(self, muc_index, user_index):
|
||||
"""Called by tests
|
||||
@return: a wokkel.muc.Room instance"""
|
||||
profile = Const.PROFILE[user_index]
|
||||
muc_s = Const.MUC_STR[muc_index]
|
||||
profile = C.PROFILE[user_index]
|
||||
muc_s = C.MUC_STR[muc_index]
|
||||
try:
|
||||
return self.clients[profile].joined_rooms[muc_s]
|
||||
except (AttributeError, KeyError):
|
||||
|
|
@ -163,14 +169,14 @@ class FakeXEP_0045(plugin_xep_0045.XEP_0045):
|
|||
|
||||
def getNick(self, muc_index, user_index):
|
||||
try:
|
||||
return self.getRoomNick(Const.MUC_STR[muc_index], Const.PROFILE[user_index])
|
||||
return self.getRoomNick(C.MUC_STR[muc_index], C.PROFILE[user_index])
|
||||
except (KeyError, AttributeError):
|
||||
return ''
|
||||
|
||||
def getNickOfUser(self, muc_index, user_index, profile_index, secure=True):
|
||||
try:
|
||||
room = self.clients[Const.PROFILE[profile_index]].joined_rooms[Const.MUC_STR[muc_index]]
|
||||
return self.getRoomNickOfUser(room, Const.JID_STR[user_index])
|
||||
room = self.clients[C.PROFILE[profile_index]].joined_rooms[C.MUC_STR[muc_index]]
|
||||
return self.getRoomNickOfUser(room, C.JID_STR[user_index])
|
||||
except (KeyError, AttributeError):
|
||||
return None
|
||||
|
||||
|
|
@ -190,3 +196,84 @@ class FakeXEP_0249(object):
|
|||
@profile_key: %(doc_profile_key)s
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class FakeSatPubSubClient(object):
|
||||
|
||||
def __init__(self, host, parent_plugin):
|
||||
self.host = host
|
||||
self.parent_plugin = parent_plugin
|
||||
self.__items = OrderedDict()
|
||||
self.__rsm_responses = {}
|
||||
|
||||
def createNode(self, service, nodeIdentifier=None, options=None,
|
||||
sender=None):
|
||||
return defer.succeed(None)
|
||||
|
||||
def deleteNode(self, service, nodeIdentifier, sender=None):
|
||||
try:
|
||||
del self.__items[nodeIdentifier]
|
||||
except KeyError:
|
||||
pass
|
||||
return defer.succeed(None)
|
||||
|
||||
def subscribe(self, service, nodeIdentifier, subscriber,
|
||||
options=None, sender=None):
|
||||
return defer.succeed(None)
|
||||
|
||||
def unsubscribe(self, service, nodeIdentifier, subscriber,
|
||||
subscriptionIdentifier=None, sender=None):
|
||||
return defer.succeed(None)
|
||||
|
||||
def publish(self, service, nodeIdentifier, items=None, sender=None):
|
||||
node = self.__items.setdefault(nodeIdentifier, [])
|
||||
|
||||
def replace(item_obj):
|
||||
index = 0
|
||||
for current in node:
|
||||
if current['id'] == item_obj['id']:
|
||||
node[index] = item_obj
|
||||
return True
|
||||
index += 1
|
||||
return False
|
||||
|
||||
for item in items:
|
||||
item_obj = parseXml(item) if isinstance(item, unicode) else item
|
||||
if not replace(item_obj):
|
||||
node.append(item_obj)
|
||||
return defer.succeed(None)
|
||||
|
||||
def items(self, service, nodeIdentifier, maxItems=None, itemIdentifiers=None,
|
||||
subscriptionIdentifier=None, sender=None, ext_data=None):
|
||||
try:
|
||||
items = self.__items[nodeIdentifier]
|
||||
except KeyError:
|
||||
items = []
|
||||
if ext_data:
|
||||
assert('id' in ext_data)
|
||||
if 'rsm' in ext_data:
|
||||
args = (0, items[0]['id'], items[-1]['id']) if items else ()
|
||||
self.__rsm_responses[ext_data['id']] = RSMResponse(len(items), *args)
|
||||
return defer.succeed(items)
|
||||
|
||||
def retractItems(self, service, nodeIdentifier, itemIdentifiers, sender=None):
|
||||
node = self.__items[nodeIdentifier]
|
||||
for item in [item for item in node if item['id'] in itemIdentifiers]:
|
||||
node.remove(item)
|
||||
return defer.succeed(None)
|
||||
|
||||
def getRSMResponse(self, id):
|
||||
if id not in self.__rsm_responses:
|
||||
return {}
|
||||
result = self.__rsm_responses[id].toDict()
|
||||
del self.__rsm_responses[id]
|
||||
return result
|
||||
|
||||
def subscriptions(self, service, nodeIdentifier, sender=None):
|
||||
return defer.succeed([])
|
||||
|
||||
def service_getDiscoItems(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE):
|
||||
items = DiscoItems()
|
||||
for item in self.__items.keys():
|
||||
items.append(DiscoItem(service, item))
|
||||
return defer.succeed(items)
|
||||
|
|
|
|||
418
src/test/test_plugin_misc_groupblog.py
Normal file
418
src/test/test_plugin_misc_groupblog.py
Normal file
|
|
@ -0,0 +1,418 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# SAT: a jabber client
|
||||
# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014 Jérôme Poisson (goffi@goffi.org)
|
||||
# Copyright (C) 2013, 2014 Adrien Cossa (souliane@mailoo.org)
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
""" Plugin groupblogs """
|
||||
|
||||
from constants import Const as C
|
||||
from sat.test import helpers, helpers_plugins
|
||||
from sat.plugins import plugin_misc_groupblog
|
||||
from sat.plugins import plugin_xep_0060
|
||||
from sat.plugins import plugin_xep_0277
|
||||
from sat.plugins import plugin_xep_0163
|
||||
from sat.plugins import plugin_misc_text_syntaxes
|
||||
from twisted.internet import defer
|
||||
from twisted.words.protocols.jabber import jid
|
||||
|
||||
|
||||
NS_PUBSUB = 'http://jabber.org/protocol/pubsub'
|
||||
|
||||
DO_NOT_COUNT_COMMENTS = -1
|
||||
|
||||
SERVICE = u'pubsub.example.com'
|
||||
PUBLISHER = u'test@example.org'
|
||||
OTHER_PUBLISHER = u'other@xmpp.net'
|
||||
NODE_ID = u'urn:xmpp:groupblog:{publisher}'.format(publisher=PUBLISHER)
|
||||
OTHER_NODE_ID = u'urn:xmpp:groupblog:{publisher}'.format(publisher=OTHER_PUBLISHER)
|
||||
ITEM_ID_1 = u'c745a688-9b02-11e3-a1a3-c0143dd4fe51'
|
||||
COMMENT_ID_1 = u'd745a688-9b02-11e3-a1a3-c0143dd4fe52'
|
||||
COMMENT_ID_2 = u'e745a688-9b02-11e3-a1a3-c0143dd4fe53'
|
||||
|
||||
|
||||
def COMMENTS_NODE_ID(publisher=PUBLISHER):
|
||||
return u'urn:xmpp:comments:_{id}__urn:xmpp:groupblog:{publisher}'.format(id=ITEM_ID_1, publisher=publisher)
|
||||
|
||||
|
||||
def COMMENTS_NODE_URL(publisher=PUBLISHER):
|
||||
return u'xmpp:{service}?node={node}'.format(service=SERVICE, id=ITEM_ID_1,
|
||||
node=COMMENTS_NODE_ID(publisher).replace(':', '%3A').replace('@', '%40'))
|
||||
|
||||
|
||||
def ITEM(publisher=PUBLISHER):
|
||||
return u"""
|
||||
<item id='{id}' xmlns='{ns}'>
|
||||
<entry>
|
||||
<title type='text'>The Uses of This World</title>
|
||||
<id>{id}</id>
|
||||
<updated>2003-12-12T17:47:23Z</updated>
|
||||
<published>2003-12-12T17:47:23Z</published>
|
||||
<link href='{comments_node_url}' rel='replies' title='comments'/>
|
||||
<author>
|
||||
<name>{publisher}</name>
|
||||
</author>
|
||||
</entry>
|
||||
</item>
|
||||
""".format(ns=NS_PUBSUB, id=ITEM_ID_1, publisher=publisher, comments_node_url=COMMENTS_NODE_URL(publisher))
|
||||
|
||||
|
||||
def COMMENT(id_=COMMENT_ID_1):
|
||||
return u"""
|
||||
<item id='{id}' xmlns='{ns}'>
|
||||
<entry>
|
||||
<title type='text'>The Uses of This World</title>
|
||||
<id>{id}</id>
|
||||
<updated>2003-12-12T17:47:23Z</updated>
|
||||
<published>2003-12-12T17:47:23Z</published>
|
||||
<author>
|
||||
<name>{publisher}</name>
|
||||
</author>
|
||||
</entry>
|
||||
</item>
|
||||
""".format(ns=NS_PUBSUB, id=id_, publisher=PUBLISHER)
|
||||
|
||||
|
||||
def ITEM_DATA(id_=ITEM_ID_1, count=0):
|
||||
res = {'id': ITEM_ID_1,
|
||||
'type': 'main_item',
|
||||
'content': 'The Uses of This World',
|
||||
'author': PUBLISHER,
|
||||
'updated': '1071251243.0',
|
||||
'published': '1071251243.0',
|
||||
'service': SERVICE,
|
||||
'comments': COMMENTS_NODE_URL_1,
|
||||
'comments_service': SERVICE,
|
||||
'comments_node': COMMENTS_NODE_ID_1}
|
||||
if count != DO_NOT_COUNT_COMMENTS:
|
||||
res.update({'comments_count': unicode(count)})
|
||||
return res
|
||||
|
||||
|
||||
def COMMENT_DATA(id_=COMMENT_ID_1):
|
||||
return {'id': id_,
|
||||
'type': 'comment',
|
||||
'content': 'The Uses of This World',
|
||||
'author': PUBLISHER,
|
||||
'updated': '1071251243.0',
|
||||
'published': '1071251243.0',
|
||||
'service': SERVICE,
|
||||
'node': COMMENTS_NODE_ID_1,
|
||||
'verified_publisher': 'false'}
|
||||
|
||||
|
||||
COMMENTS_NODE_ID_1 = COMMENTS_NODE_ID()
|
||||
COMMENTS_NODE_ID_2 = COMMENTS_NODE_ID(OTHER_PUBLISHER)
|
||||
COMMENTS_NODE_URL_1 = COMMENTS_NODE_URL()
|
||||
COMMENTS_NODE_URL_2 = COMMENTS_NODE_URL(OTHER_PUBLISHER)
|
||||
ITEM_1 = ITEM()
|
||||
ITEM_2 = ITEM(OTHER_PUBLISHER)
|
||||
COMMENT_1 = COMMENT(COMMENT_ID_1)
|
||||
COMMENT_2 = COMMENT(COMMENT_ID_2)
|
||||
|
||||
|
||||
def ITEM_DATA_1(count=0):
|
||||
return ITEM_DATA(count=count)
|
||||
|
||||
COMMENT_DATA_1 = COMMENT_DATA()
|
||||
COMMENT_DATA_2 = COMMENT_DATA(COMMENT_ID_2)
|
||||
|
||||
|
||||
class XEP_groupblogTest(helpers.SatTestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.host = helpers.FakeSAT()
|
||||
self.plugin = plugin_misc_groupblog.GroupBlog(self.host)
|
||||
self.plugin._initialise = self._initialise
|
||||
self.host.plugins['XEP-0060'] = plugin_xep_0060.XEP_0060(self.host)
|
||||
self.host.plugins['XEP-0163'] = plugin_xep_0163.XEP_0163(self.host)
|
||||
self.host.plugins['TEXT-SYNTAXES'] = plugin_misc_text_syntaxes.TextSyntaxes(self.host)
|
||||
self.host.plugins['XEP-0277'] = plugin_xep_0277.XEP_0277(self.host)
|
||||
self.__initialised = False
|
||||
self._initialise(C.PROFILE[0])
|
||||
|
||||
def _initialise(self, profile_key):
|
||||
profile = profile_key
|
||||
client = self.host.getClient(profile)
|
||||
if not self.__initialised:
|
||||
client.item_access_pubsub = jid.JID(SERVICE)
|
||||
xep_0060 = self.host.plugins['XEP-0060']
|
||||
xep_0060.clients[profile] = helpers_plugins.FakeSatPubSubClient(self.host, xep_0060)
|
||||
xep_0060.clients[profile].parent = client
|
||||
self.psclient = xep_0060.clients[profile]
|
||||
helpers.FakeSAT.getDiscoItems = self.psclient.service_getDiscoItems
|
||||
self.__initialised = True
|
||||
return defer.succeed((profile, client))
|
||||
|
||||
def _addItem(self, profile, item, parent_node=None):
|
||||
self.host.plugins['XEP-0060'].clients[profile]._addItem(item, parent_node)
|
||||
|
||||
def test_sendGroupBlog(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.items(SERVICE, NODE_ID)
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
d.addCallback(lambda dummy: self.plugin.sendGroupBlog('PUBLIC', [], 'test', {}, C.PROFILE[0]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
return d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
|
||||
def test_deleteGroupBlog(self):
|
||||
pub_data = (SERVICE, NODE_ID, ITEM_ID_1)
|
||||
self.host.bridge.expectCall('personalEvent', C.JID_STR[0],
|
||||
"MICROBLOG_DELETE",
|
||||
{'type': 'main_item', 'id': ITEM_ID_1},
|
||||
C.PROFILE[0])
|
||||
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.deleteGroupBlog(pub_data, COMMENTS_NODE_URL_1, profile_key=C.PROFILE[0]))
|
||||
return d.addCallback(self.assertEqual, None)
|
||||
|
||||
def test_updateGroupBlog(self):
|
||||
pub_data = (SERVICE, NODE_ID, ITEM_ID_1)
|
||||
new_text = u"silfu23RFWUP)IWNOEIOEFÖ"
|
||||
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.updateGroupBlog(pub_data, COMMENTS_NODE_URL_1, new_text, {}, profile_key=C.PROFILE[0]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
return d.addCallback(lambda items: self.assertEqual(''.join(items[0].entry.title.children), new_text))
|
||||
|
||||
def test_sendGroupBlogComment(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.items(SERVICE, NODE_ID)
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
d.addCallback(lambda dummy: self.plugin.sendGroupBlogComment(COMMENTS_NODE_URL_1, 'test', {}, profile_key=C.PROFILE[0]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
return d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
|
||||
def test_getGroupBlogs(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogs(PUBLISHER, profile_key=C.PROFILE[0]))
|
||||
result = ([ITEM_DATA_1()], {'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogsNoCount(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogs(PUBLISHER, count_comments=False, profile_key=C.PROFILE[0]))
|
||||
result = ([ITEM_DATA_1(DO_NOT_COUNT_COMMENTS)], {'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogsWithIDs(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogs(PUBLISHER, [ITEM_ID_1], profile_key=C.PROFILE[0]))
|
||||
result = ([ITEM_DATA_1()], {'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogsWithRSM(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogs(PUBLISHER, rsm={'max': 1}, profile_key=C.PROFILE[0]))
|
||||
result = ([ITEM_DATA_1()], {'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogsWithComments(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1]))
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogsWithComments(PUBLISHER, [], profile_key=C.PROFILE[0]))
|
||||
result = ([(ITEM_DATA_1(1), ([COMMENT_DATA_1],
|
||||
{'count': '1', 'index': '0', 'first': COMMENT_ID_1, 'last': COMMENT_ID_1}))],
|
||||
{'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogsWithComments2(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogsWithComments(PUBLISHER, [], profile_key=C.PROFILE[0]))
|
||||
result = ([(ITEM_DATA_1(2), ([COMMENT_DATA_1, COMMENT_DATA_2],
|
||||
{'count': '2', 'index': '0', 'first': COMMENT_ID_1, 'last': COMMENT_ID_2}))],
|
||||
{'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})
|
||||
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogsAtom(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogsAtom(PUBLISHER, {'max': 1}, profile_key=C.PROFILE[0]))
|
||||
|
||||
def cb(atom):
|
||||
self.assertIsInstance(atom, unicode)
|
||||
self.assertTrue(atom.startswith('<?xml version="1.0" encoding="utf-8"?>'))
|
||||
|
||||
return d.addCallback(cb)
|
||||
|
||||
def test_getMassiveGroupBlogs(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.plugin.getMassiveGroupBlogs('JID', [jid.JID(PUBLISHER)], {'max': 1}, profile_key=C.PROFILE[0]))
|
||||
result = {PUBLISHER: ([ITEM_DATA_1()], {'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})}
|
||||
|
||||
def clean(res):
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@found@' + SERVICE]
|
||||
return res
|
||||
|
||||
d.addCallback(clean)
|
||||
d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getMassiveGroupBlogsWithComments(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.plugin.getMassiveGroupBlogs('JID', [jid.JID(PUBLISHER)], {'max': 1}, profile_key=C.PROFILE[0]))
|
||||
result = {PUBLISHER: ([ITEM_DATA_1(2)], {'count': '1', 'index': '0', 'first': ITEM_ID_1, 'last': ITEM_ID_1})}
|
||||
|
||||
def clean(res):
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@found@' + SERVICE]
|
||||
return res
|
||||
|
||||
d.addCallback(clean)
|
||||
d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_getGroupBlogComments(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1]))
|
||||
d.addCallback(lambda dummy: self.plugin.getGroupBlogComments(SERVICE, COMMENTS_NODE_ID_1, {'max': 1}, profile_key=C.PROFILE[0]))
|
||||
result = ([COMMENT_DATA_1], {'count': '1', 'index': '0', 'first': COMMENT_ID_1, 'last': COMMENT_ID_1})
|
||||
return d.addCallback(self.assertEqual, result)
|
||||
|
||||
def test_subscribeGroupBlog(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.plugin.subscribeGroupBlog(PUBLISHER, profile_key=C.PROFILE[0])
|
||||
return d.addCallback(self.assertEqual, None)
|
||||
|
||||
def test_massiveSubscribeGroupBlogs(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
d = self.plugin.massiveSubscribeGroupBlogs('JID', [jid.JID(PUBLISHER)], profile_key=C.PROFILE[0])
|
||||
|
||||
def clean(res):
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@found@' + SERVICE]
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@subscriptions@' + SERVICE]
|
||||
return res
|
||||
|
||||
d.addCallback(clean)
|
||||
return d.addCallback(self.assertEqual, None)
|
||||
|
||||
def test_deleteAllGroupBlogs(self):
|
||||
"""Delete our main node and associated comments node"""
|
||||
self._initialise(C.PROFILE[0])
|
||||
self.host.profiles[C.PROFILE[0]].roster.addItem(jid.JID(OTHER_PUBLISHER))
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, OTHER_NODE_ID, [ITEM_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_2, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, OTHER_NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_2))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
def clean(res):
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@found@' + SERVICE]
|
||||
return res
|
||||
|
||||
d.addCallback(lambda dummy: self.plugin.deleteAllGroupBlogs(C.PROFILE[0]))
|
||||
d.addCallback(clean)
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, OTHER_NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_2))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
return d
|
||||
|
||||
def test_deleteAllGroupBlogsComments(self):
|
||||
"""Delete the comments we posted on other node's"""
|
||||
self._initialise(C.PROFILE[0])
|
||||
self.host.profiles[C.PROFILE[0]].roster.addItem(jid.JID(OTHER_PUBLISHER))
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, OTHER_NODE_ID, [ITEM_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_2, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, OTHER_NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_2))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
def clean(res):
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@found@' + SERVICE]
|
||||
return res
|
||||
|
||||
d.addCallback(lambda dummy: self.plugin.deleteAllGroupBlogsComments(C.PROFILE[0]))
|
||||
d.addCallback(clean)
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, OTHER_NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_2))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
return d
|
||||
|
||||
def test_deleteAllGroupBlogsAndComments(self):
|
||||
self._initialise(C.PROFILE[0])
|
||||
self.host.profiles[C.PROFILE[0]].roster.addItem(jid.JID(OTHER_PUBLISHER))
|
||||
d = self.psclient.publish(SERVICE, NODE_ID, [ITEM_1])
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_1, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, OTHER_NODE_ID, [ITEM_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.publish(SERVICE, COMMENTS_NODE_ID_2, [COMMENT_1, COMMENT_2]))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, OTHER_NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_2))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 2))
|
||||
|
||||
def clean(res):
|
||||
del self.host.plugins['XEP-0060'].node_cache[C.PROFILE[0] + '@found@' + SERVICE]
|
||||
return res
|
||||
|
||||
d.addCallback(lambda dummy: self.plugin.deleteAllGroupBlogsAndComments(C.PROFILE[0]))
|
||||
d.addCallback(clean)
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_1))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, OTHER_NODE_ID))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 1))
|
||||
d.addCallback(lambda dummy: self.psclient.items(SERVICE, COMMENTS_NODE_ID_2))
|
||||
d.addCallback(lambda items: self.assertEqual(len(items), 0))
|
||||
return d
|
||||
Loading…
Reference in a new issue