pep/slixmpp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

itests: add another bunch of tests

Browse source
This commit is contained in:
mathieui 2020-12-06 18:58:35 +01:00
parent 12e3c9988d
commit 1836b2b887
6 changed files with 205 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
itests/test_disco.py Normal file
View file

@ -0,0 +1,33 @@
import unittest
from slixmpp.test.integration import SlixIntegration
class TestDisco(SlixIntegration):
async def asyncSetUp(self):
await super().asyncSetUp()
self.add_client(
self.envjid('CI_ACCOUNT1'),
self.envstr('CI_ACCOUNT1_PASSWORD'),
)
self.add_client(
self.envjid('CI_ACCOUNT2'),
self.envstr('CI_ACCOUNT2_PASSWORD'),
)
self.register_plugins(['xep_0030'])
await self.connect_clients()
async def test_features(self):
"""Check we can add, get and delete a feature"""
self.clients[0]['xep_0030'].add_feature('urn:xmpp:fake:0')
info = await self.clients[1]['xep_0030'].get_info(
self.clients[0].boundjid.full
)
self.assertIn('urn:xmpp:fake:0', info['disco_info']['features'])
self.clients[0]['xep_0030'].del_feature(feature='urn:xmpp:fake:0')
info = await self.clients[1]['xep_0030'].get_info(
self.clients[0].boundjid.full
)
self.assertNotIn('urn:xmpp:fake:0', info['disco_info']['features'])
suite = unittest.TestLoader().loadTestsFromTestCase(TestDisco)

55
itests/test_mam.py Normal file
View file

@ -0,0 +1,55 @@
import unittest
from random import randint
from slixmpp import JID
from slixmpp.test.integration import SlixIntegration
class TestMAM(SlixIntegration):
async def asyncSetUp(self):
await super().asyncSetUp()
self.add_client(
self.envjid('CI_ACCOUNT1'),
self.envstr('CI_ACCOUNT1_PASSWORD'),
)
self.add_client(
self.envjid('CI_ACCOUNT2'),
self.envstr('CI_ACCOUNT2_PASSWORD'),
)
self.register_plugins(['xep_0313'])
await self.connect_clients()
async def test_mam_retrieve(self):
"""Make sure we can get messages from our archive"""
# send messages first
tok = randint(1, 999999)
self.clients[0].make_message(mto=self.clients[1].boundjid, mbody='coucou').send()
await self.clients[1].wait_until('message')
self.clients[1].make_message(
mto=self.clients[0].boundjid,
mbody='coucou coucou %s' % tok,
).send()
await self.clients[0].wait_until('message')
# Get archive
retrieve = self.clients[0]['xep_0313'].retrieve(
with_jid=JID(self.envjid('CI_ACCOUNT2')),
iterator=True,
reverse=True,
rsm={'max': 2}
)
msgs = []
count = 0
async for rsm in retrieve:
for msg in rsm['mam']['results']:
msgs.append(
msg['mam_result']['forwarded']['stanza']
)
count += 1
if count >= 2:
break
self.assertEqual(msgs[0]['body'], 'coucou')
self.assertEqual(msgs[1]['body'], 'coucou coucou %s' % tok)
suite = unittest.TestLoader().loadTestsFromTestCase(TestMAM)

21
itests/test_ping.py Normal file
View file

@ -0,0 +1,21 @@
import unittest
from slixmpp.test.integration import SlixIntegration
class TestPing(SlixIntegration):
async def asyncSetUp(self):
await super().asyncSetUp()
self.add_client(
self.envjid('CI_ACCOUNT1'),
self.envstr('CI_ACCOUNT1_PASSWORD'),
)
self.register_plugins(['xep_0199'])
await self.connect_clients()
async def test_ping(self):
"""Check we can ping our own server"""
rtt = await self.clients[0]['xep_0199'].ping()
self.assertGreater(10, rtt)
suite = unittest.TestLoader().loadTestsFromTestCase(TestPing)

35
itests/test_privatestorage.py Normal file
View file

@ -0,0 +1,35 @@
import unittest
from slixmpp import ET
from slixmpp.test.integration import SlixIntegration
from slixmpp.plugins.xep_0048.stanza import Bookmarks
class TestPrivateStorage(SlixIntegration):
async def asyncSetUp(self):
self.add_client(
self.envjid('CI_ACCOUNT1'),
self.envstr('CI_ACCOUNT1_PASSWORD'),
)
self.register_plugins(['xep_0048', 'xep_0049'])
await self.connect_clients()
async def test_privatestorage(self):
"""Check we can set, get, and delete private in xml storage"""
# Set a bookmark using private storage
el = Bookmarks()
el.add_conference('test@example.com', 'toto')
await self.clients[0]['xep_0049'].store(
el,
)
result = await self.clients[0]['xep_0049'].retrieve('bookmarks')
self.assertEqual(str(result['private']['bookmarks']), str(el))
# Purge bookmarks
await self.clients[0]['xep_0049'].store(
Bookmarks(),
)
result = await self.clients[0]['xep_0049'].retrieve('bookmarks')
self.assertEqual(result['private']['bookmarks'], Bookmarks())
suite = unittest.TestLoader().loadTestsFromTestCase(TestPrivateStorage)

32
itests/test_reactions.py Normal file
View file

@ -0,0 +1,32 @@
import unittest
from slixmpp.test.integration import SlixIntegration
class TestReactions(SlixIntegration):
async def asyncSetUp(self):
self.add_client(
self.envjid('CI_ACCOUNT1'),
self.envstr('CI_ACCOUNT1_PASSWORD'),
)
self.add_client(
self.envjid('CI_ACCOUNT2'),
self.envstr('CI_ACCOUNT2_PASSWORD'),
)
self.register_plugins(['xep_0444'])
await self.connect_clients()
async def test_send_reaction(self):
"""Make sure we can send and receive reactions"""
self.clients[0]['xep_0444'].send_reactions(
self.clients[1].boundjid.full,
to_id='toto',
reactions=['🦙', '🦦'],
)
msg = await self.clients[1].wait_until('reactions')
self.assertEqual(
msg['reactions'].get_values(),
{'🦙', '🦦'},
)
self.assertEqual(msg['reactions']['id'], 'toto')
suite = unittest.TestLoader().loadTestsFromTestCase(TestReactions)

29
itests/test_retract.py Normal file
View file

@ -0,0 +1,29 @@
import unittest
from slixmpp.test.integration import SlixIntegration
class TestRetract(SlixIntegration):
async def asyncSetUp(self):
self.add_client(
self.envjid('CI_ACCOUNT1'),
self.envstr('CI_ACCOUNT1_PASSWORD'),
)
self.add_client(
self.envjid('CI_ACCOUNT2'),
self.envstr('CI_ACCOUNT2_PASSWORD'),
)
self.register_plugins(['xep_0424'])
await self.connect_clients()
async def test_retract_msg(self):
"""Try to retract a message"""
self.clients[0]['xep_0424'].send_retraction(
self.clients[1].boundjid.full,
id='toto',
fallback_text='Twas a mistake',
)
msg = await self.clients[1].wait_until('message')
self.assertEqual(msg['apply_to']['id'], 'toto')
self.assertTrue(msg['apply_to']['retract'])
suite = unittest.TestLoader().loadTestsFromTestCase(TestRetract)