itests: add 0012, 0054, 0084, 0092, 0153, 0191 tests

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)

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)

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)

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)

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)

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)