summaryrefslogtreecommitdiff
path: root/tests/test_stream_xep_0030.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_stream_xep_0030.py')
-rw-r--r--tests/test_stream_xep_0030.py528
1 files changed, 528 insertions, 0 deletions
diff --git a/tests/test_stream_xep_0030.py b/tests/test_stream_xep_0030.py
new file mode 100644
index 00000000..1f989745
--- /dev/null
+++ b/tests/test_stream_xep_0030.py
@@ -0,0 +1,528 @@
+import time
+import threading
+
+from sleekxmpp.test import *
+
+
+class TestStreamDisco(SleekTest):
+
+ """
+ Test using the XEP-0030 plugin.
+ """
+
+ def tearDown(self):
+ self.stream_close()
+
+ def testInfoEmptyDefaultNode(self):
+ """
+ Info query result from an entity MUST have at least one identity
+ and feature, namely http://jabber.org/protocol/disco#info.
+
+ Since the XEP-0030 plugin is loaded, a disco response should
+ be generated and not an error result.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ self.recv("""
+ <iq type="get" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info">
+ <identity category="client" type="bot" />
+ <feature var="http://jabber.org/protocol/disco#info" />
+ </query>
+ </iq>
+ """)
+
+ def testInfoEmptyDefaultNodeComponent(self):
+ """
+ Test requesting an empty, default node using a Component.
+ """
+ self.stream_start(mode='component',
+ jid='tester.localhost',
+ plugins=['xep_0030'])
+
+ self.recv("""
+ <iq type="get" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info">
+ <identity category="component" type="generic" />
+ <feature var="http://jabber.org/protocol/disco#info" />
+ </query>
+ </iq>
+ """)
+
+ def testInfoIncludeNode(self):
+ """
+ Results for info queries directed to a particular node MUST
+ include the node in the query response.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+
+ self.xmpp['xep_0030'].static.add_node(node='testing')
+
+ self.recv("""
+ <iq to="tester@localhost" type="get" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing">
+ </query>
+ </iq>""",
+ method='mask')
+
+ def testItemsIncludeNode(self):
+ """
+ Results for items queries directed to a particular node MUST
+ include the node in the query response.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+
+ self.xmpp['xep_0030'].static.add_node(node='testing')
+
+ self.recv("""
+ <iq to="tester@localhost" type="get" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing">
+ </query>
+ </iq>""",
+ method='mask')
+
+ def testDynamicInfoJID(self):
+ """
+ Test using a dynamic info handler for a particular JID.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ def dynamic_jid(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoInfo()
+ result['node'] = node
+ result.add_identity('client', 'console', name='Dynamic Info')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_info',
+ jid='tester@localhost',
+ handler=dynamic_jid)
+
+ self.recv("""
+ <iq type="get" id="test" to="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing">
+ <identity category="client"
+ type="console"
+ name="Dynamic Info" />
+ </query>
+ </iq>
+ """)
+
+ def testDynamicInfoGlobal(self):
+ """
+ Test using a dynamic info handler for all requests.
+ """
+ self.stream_start(mode='component',
+ jid='tester.localhost',
+ plugins=['xep_0030'])
+
+ def dynamic_global(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoInfo()
+ result['node'] = node
+ result.add_identity('component', 'generic', name='Dynamic Info')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_info',
+ handler=dynamic_global)
+
+ self.recv("""
+ <iq type="get" id="test"
+ to="user@tester.localhost"
+ from="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test"
+ to="tester@localhost"
+ from="user@tester.localhost">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing">
+ <identity category="component"
+ type="generic"
+ name="Dynamic Info" />
+ </query>
+ </iq>
+ """)
+
+ def testOverrideJIDInfoHandler(self):
+ """Test overriding a JID info handler."""
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ def dynamic_jid(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoInfo()
+ result['node'] = node
+ result.add_identity('client', 'console', name='Dynamic Info')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_info',
+ jid='tester@localhost',
+ handler=dynamic_jid)
+
+
+ self.xmpp['xep_0030'].make_static(jid='tester@localhost',
+ node='testing')
+
+ self.xmpp['xep_0030'].add_identity(jid='tester@localhost',
+ node='testing',
+ category='automation',
+ itype='command-list')
+
+ self.recv("""
+ <iq type="get" id="test" to="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing">
+ <identity category="automation"
+ type="command-list" />
+ </query>
+ </iq>
+ """)
+
+ def testOverrideGlobalInfoHandler(self):
+ """Test overriding the global JID info handler."""
+ self.stream_start(mode='component',
+ jid='tester.localhost',
+ plugins=['xep_0030'])
+
+ def dynamic_global(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoInfo()
+ result['node'] = node
+ result.add_identity('component', 'generic', name='Dynamic Info')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_info',
+ handler=dynamic_global)
+
+ self.xmpp['xep_0030'].make_static(jid='user@tester.localhost',
+ node='testing')
+
+ self.xmpp['xep_0030'].add_feature(jid='user@tester.localhost',
+ node='testing',
+ feature='urn:xmpp:ping')
+
+ self.recv("""
+ <iq type="get" id="test"
+ to="user@tester.localhost"
+ from="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test"
+ to="tester@localhost"
+ from="user@tester.localhost">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="testing">
+ <feature var="urn:xmpp:ping" />
+ </query>
+ </iq>
+ """)
+
+ def testGetInfoRemote(self):
+ """
+ Test sending a disco#info query to another entity
+ and receiving the result.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ events = set()
+
+ def handle_disco_info(iq):
+ events.add('disco_info')
+
+
+ self.xmpp.add_event_handler('disco_info', handle_disco_info)
+
+ t = threading.Thread(name="get_info",
+ target=self.xmpp['xep_0030'].get_info,
+ args=('user@localhost', 'foo'))
+ t.start()
+
+ self.send("""
+ <iq type="get" to="user@localhost" id="1">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="foo" />
+ </iq>
+ """)
+
+ self.recv("""
+ <iq type="result" to="tester@localhost" id="1">
+ <query xmlns="http://jabber.org/protocol/disco#info"
+ node="foo">
+ <identity category="client" type="bot" />
+ <feature var="urn:xmpp:ping" />
+ </query>
+ </iq>
+ """)
+
+ # Wait for disco#info request to be received.
+ t.join()
+
+ time.sleep(0.1)
+
+ self.assertEqual(events, set(('disco_info',)),
+ "Disco info event was not triggered: %s" % events)
+
+ def testDynamicItemsJID(self):
+ """
+ Test using a dynamic items handler for a particular JID.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ def dynamic_jid(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoItems()
+ result['node'] = node
+ result.add_item('tester@localhost', node='foo', name='JID')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_items',
+ jid='tester@localhost',
+ handler=dynamic_jid)
+
+ self.recv("""
+ <iq type="get" id="test" to="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing">
+ <item jid="tester@localhost" node="foo" name="JID" />
+ </query>
+ </iq>
+ """)
+
+ def testDynamicItemsGlobal(self):
+ """
+ Test using a dynamic items handler for all requests.
+ """
+ self.stream_start(mode='component',
+ jid='tester.localhost',
+ plugins=['xep_0030'])
+
+ def dynamic_global(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoItems()
+ result['node'] = node
+ result.add_item('tester@localhost', node='foo', name='Global')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_items',
+ handler=dynamic_global)
+
+ self.recv("""
+ <iq type="get" id="test"
+ to="user@tester.localhost"
+ from="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test"
+ to="tester@localhost"
+ from="user@tester.localhost">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing">
+ <item jid="tester@localhost" node="foo" name="Global" />
+ </query>
+ </iq>
+ """)
+
+ def testOverrideJIDItemsHandler(self):
+ """Test overriding a JID items handler."""
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ def dynamic_jid(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoItems()
+ result['node'] = node
+ result.add_item('tester@localhost', node='foo', name='Global')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_items',
+ jid='tester@localhost',
+ handler=dynamic_jid)
+
+
+ self.xmpp['xep_0030'].make_static(jid='tester@localhost',
+ node='testing')
+
+ self.xmpp['xep_0030'].add_item(jid='tester@localhost',
+ node='testing',
+ ijid='tester@localhost',
+ inode='foo',
+ name='Test')
+
+ self.recv("""
+ <iq type="get" id="test" to="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing">
+ <item jid="tester@localhost" node="foo" name="Test" />
+ </query>
+ </iq>
+ """)
+
+ def testOverrideGlobalItemsHandler(self):
+ """Test overriding the global JID items handler."""
+ self.stream_start(mode='component',
+ jid='tester.localhost',
+ plugins=['xep_0030'])
+
+ def dynamic_global(jid, node, iq):
+ result = self.xmpp['xep_0030'].stanza.DiscoItems()
+ result['node'] = node
+ result.add_item('tester.localhost', node='foo', name='Global')
+ return result
+
+ self.xmpp['xep_0030'].set_node_handler('get_items',
+ handler=dynamic_global)
+
+ self.xmpp['xep_0030'].make_static(jid='user@tester.localhost',
+ node='testing')
+
+ self.xmpp['xep_0030'].add_item(jid='user@tester.localhost',
+ node='testing',
+ ijid='user@tester.localhost',
+ inode='foo',
+ name='Test')
+
+ self.recv("""
+ <iq type="get" id="test"
+ to="user@tester.localhost"
+ from="tester@localhost">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing" />
+ </iq>
+ """)
+
+ self.send("""
+ <iq type="result" id="test"
+ to="tester@localhost"
+ from="user@tester.localhost">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="testing">
+ <item jid="user@tester.localhost" node="foo" name="Test" />
+ </query>
+ </iq>
+ """)
+
+ def testGetItemsRemote(self):
+ """
+ Test sending a disco#items query to another entity
+ and receiving the result.
+ """
+ self.stream_start(mode='client',
+ plugins=['xep_0030'])
+
+ events = set()
+ results = set()
+
+ def handle_disco_items(iq):
+ events.add('disco_items')
+ results.update(iq['disco_items']['items'])
+
+
+ self.xmpp.add_event_handler('disco_items', handle_disco_items)
+
+ t = threading.Thread(name="get_items",
+ target=self.xmpp['xep_0030'].get_items,
+ args=('user@localhost', 'foo'))
+ t.start()
+
+ self.send("""
+ <iq type="get" to="user@localhost" id="1">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="foo" />
+ </iq>
+ """)
+
+ self.recv("""
+ <iq type="result" to="tester@localhost" id="1">
+ <query xmlns="http://jabber.org/protocol/disco#items"
+ node="foo">
+ <item jid="user@localhost" node="bar" name="Test" />
+ <item jid="user@localhost" node="baz" name="Test 2" />
+ </query>
+ </iq>
+ """)
+
+ # Wait for disco#items request to be received.
+ t.join()
+
+ time.sleep(0.1)
+
+ items = set([('user@localhost', 'bar', 'Test'),
+ ('user@localhost', 'baz', 'Test 2')])
+ self.assertEqual(events, set(('disco_items',)),
+ "Disco items event was not triggered: %s" % events)
+ self.assertEqual(results, items,
+ "Unexpected items: %s" % results)
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamDisco)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-18 20:57:42 +0000