summaryrefslogtreecommitdiff
path: root/itests
diff options
context:
space:
mode:
Diffstat (limited to 'itests')
-rw-r--r--itests/test_blocking.py32
-rw-r--r--itests/test_last_activity.py33
-rw-r--r--itests/test_user_avatar.py61
-rw-r--r--itests/test_vcard.py49
-rw-r--r--itests/test_vcard_avatar.py49
-rw-r--r--itests/test_version.py37
6 files changed, 261 insertions, 0 deletions
diff --git a/itests/test_blocking.py b/itests/test_blocking.py
new file mode 100644
index 00000000..7954c1dc
--- /dev/null
+++ b/itests/test_blocking.py
@@ -0,0 +1,32 @@
+import unittest
+from slixmpp import JID
+from slixmpp.test.integration import SlixIntegration
+
+
+class TestBlocking(SlixIntegration):
+ async def asyncSetUp(self):
+ await super().asyncSetUp()
+ self.add_client(
+ self.envjid('CI_ACCOUNT1'),
+ self.envstr('CI_ACCOUNT1_PASSWORD'),
+ )
+ self.register_plugins(['xep_0191'])
+ await self.connect_clients()
+
+ async def test_blocking(self):
+ """Check we can block, unblock, and list blocked"""
+ await self.clients[0]['xep_0191'].block(
+ [JID('toto@example.com'), JID('titi@example.com')]
+ )
+ blocked = {JID('toto@example.com'), JID('titi@example.com')}
+ iq = await self.clients[0]['xep_0191'].get_blocked()
+ self.assertEqual(iq['blocklist']['items'], blocked)
+
+ info = await self.clients[0]['xep_0191'].unblock(
+ blocked,
+ )
+ iq = await self.clients[0]['xep_0191'].get_blocked()
+ self.assertEqual(len(iq['blocklist']['items']), 0)
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestBlocking)
diff --git a/itests/test_last_activity.py b/itests/test_last_activity.py
new file mode 100644
index 00000000..3d36b4b8
--- /dev/null
+++ b/itests/test_last_activity.py
@@ -0,0 +1,33 @@
+import unittest
+from slixmpp.test.integration import SlixIntegration
+
+
+class TestLastActivity(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_0012'])
+ await self.connect_clients()
+
+ async def test_activity(self):
+ """Check we can set and get last activity"""
+ self.clients[0]['xep_0012'].set_last_activity(
+ status='coucou',
+ seconds=4242,
+ )
+ act = await self.clients[1]['xep_0012'].get_last_activity(
+ self.clients[0].boundjid.full
+ )
+ self.assertEqual(act['last_activity']['status'], 'coucou')
+ self.assertGreater(act['last_activity']['seconds'], 4241)
+ self.assertGreater(4250, act['last_activity']['seconds'])
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestLastActivity)
diff --git a/itests/test_user_avatar.py b/itests/test_user_avatar.py
new file mode 100644
index 00000000..193bbe72
--- /dev/null
+++ b/itests/test_user_avatar.py
@@ -0,0 +1,61 @@
+import asyncio
+import unittest
+from slixmpp import JID
+from slixmpp.test.integration import SlixIntegration
+
+
+class TestUserAvatar(SlixIntegration):
+ async def asyncSetUp(self):
+ await super().asyncSetUp()
+ self.add_client(
+ self.envjid('CI_ACCOUNT1'),
+ self.envstr('CI_ACCOUNT1_PASSWORD'),
+ )
+ self.register_plugins(['xep_0084'])
+ self.data = b'coucou coucou'
+ await self.connect_clients()
+
+ async def _clear_avatar(self):
+ """Utility for purging remote state"""
+ await self.clients[0]['xep_0084'].stop()
+ await self.clients[0]['xep_0084'].publish_avatar(b'')
+
+ async def test_set_avatar(self):
+ """Check we can set and get a PEP avatar and metadata"""
+ await self._clear_avatar()
+
+ await self.clients[0]['xep_0084'].publish_avatar(
+ self.data
+ )
+ metadata = {
+ 'id': self.clients[0]['xep_0084'].generate_id(self.data),
+ 'bytes': 13,
+ 'type': 'image/jpeg',
+ }
+ # Wait for metadata publish event
+ event = self.clients[0].wait_until('avatar_metadata_publish')
+ publish = self.clients[0]['xep_0084'].publish_avatar_metadata(
+ metadata,
+ )
+ res = await asyncio.gather(
+ event,
+ publish,
+ )
+ message = res[0]
+ recv_meta = message['pubsub_event']['items']['item']['avatar_metadata']
+ info = recv_meta['info']
+ self.assertEqual(info['bytes'], metadata['bytes'])
+ self.assertEqual(info['type'], metadata['type'])
+ self.assertEqual(info['id'], metadata['id'])
+
+ recv = await self.clients[0]['xep_0084'].retrieve_avatar(
+ JID(self.clients[0].boundjid.bare),
+ info['id']
+ )
+ avatar = recv['pubsub']['items']['item']['avatar_data']['value']
+ self.assertEqual(avatar, self.data)
+
+ await self._clear_avatar()
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestUserAvatar)
diff --git a/itests/test_vcard.py b/itests/test_vcard.py
new file mode 100644
index 00000000..800d5a5b
--- /dev/null
+++ b/itests/test_vcard.py
@@ -0,0 +1,49 @@
+import unittest
+from slixmpp.test.integration import SlixIntegration
+
+
+class TestVcardTemp(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_0054'])
+ await self.connect_clients()
+
+ async def _clear_vcard(self):
+ # cleanup
+ await self.clients[0]['xep_0054'].publish_vcard(
+ self.clients[0]['xep_0054'].make_vcard()
+ )
+
+ async def test_vcard(self):
+ """Check we can set and get a vcard"""
+ await self._clear_vcard()
+
+ # Check that vcard is empty
+ recv = await self.clients[1]['xep_0054'].get_vcard(
+ self.clients[0].boundjid.bare
+ )
+ self.assertEqual(recv['vcard_temp']['TITLE'], None)
+
+ vcard = self.clients[0]['xep_0054'].make_vcard()
+ vcard['TITLE'] = 'Coucou coucou'
+ await self.clients[0]['xep_0054'].publish_vcard(
+ vcard,
+ )
+ #
+ recv = await self.clients[1]['xep_0054'].get_vcard(
+ self.clients[0].boundjid.bare
+ )
+ self.assertEqual(recv['vcard_temp']['TITLE'], 'Coucou coucou')
+
+ await self._clear_vcard()
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestVcardTemp)
diff --git a/itests/test_vcard_avatar.py b/itests/test_vcard_avatar.py
new file mode 100644
index 00000000..284c08e1
--- /dev/null
+++ b/itests/test_vcard_avatar.py
@@ -0,0 +1,49 @@
+import asyncio
+import unittest
+from slixmpp import JID
+from slixmpp.test.integration import SlixIntegration
+from hashlib import sha1
+
+
+class TestVcardAvatar(SlixIntegration):
+ async def asyncSetUp(self):
+ await super().asyncSetUp()
+ self.add_client(
+ self.envjid('CI_ACCOUNT1'),
+ self.envstr('CI_ACCOUNT1_PASSWORD'),
+ )
+ self.register_plugins(['xep_0153'])
+ self.data = b'coucou coucou'
+ self.hashed_data = sha1(self.data).hexdigest()
+ await self.connect_clients()
+
+ async def _clear_avatar(self):
+ """Utility for purging remote state"""
+ await self.clients[0]['xep_0153'].set_avatar(avatar=b'')
+
+ async def test_set_avatar(self):
+ """Check we can set and get a PEP avatar and metadata"""
+ await self._clear_avatar()
+
+ event = self.clients[0].wait_until('vcard_avatar_update')
+ update = self.clients[0]['xep_0153'].set_avatar(
+ avatar=self.data
+ )
+ result = await asyncio.gather(
+ event,
+ update,
+ )
+ presence = result[0]
+ hash = presence['vcard_temp_update']['photo']
+ self.assertEqual(hash, self.hashed_data)
+
+ iq = await self.clients[0]['xep_0054'].get_vcard(
+ JID(self.clients[0].boundjid.bare)
+ )
+ photo = iq['vcard_temp']['PHOTO']['BINVAL']
+ self.assertEqual(photo, self.data)
+
+ await self._clear_avatar()
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestVcardAvatar)
diff --git a/itests/test_version.py b/itests/test_version.py
new file mode 100644
index 00000000..5b8e42fd
--- /dev/null
+++ b/itests/test_version.py
@@ -0,0 +1,37 @@
+import unittest
+from slixmpp.test.integration import SlixIntegration
+
+
+class TestVersion(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_0092'],
+ configs=[{
+ 'software_name': 'Slix Test',
+ 'version': '1.2.3.4',
+ 'os': 'I use arch btw',
+ }]
+ )
+ await self.connect_clients()
+
+ async def test_version(self):
+ """Check we can set and query software version info"""
+ iq = await self.clients[1]['xep_0092'].get_version(
+ self.clients[0].boundjid.full
+ )
+ version = iq['software_version']
+ self.assertEqual(version['name'], 'Slix Test')
+ self.assertEqual(version['version'], '1.2.3.4')
+ self.assertEqual(version['os'], 'I use arch btw')
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestVersion)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-22 10:28:42 +0000