Merge branch 'xep0055' into 'master'

Add XEP-0055 (Jabber Search)

See merge request poezio/slixmpp!204
This commit is contained in:
mathieui 2022-09-12 18:18:51 +00:00
commit 97b0c7ffac
7 changed files with 353 additions and 0 deletions

View file

@ -0,0 +1,18 @@
XEP-0055: Jabber search
=======================
.. module:: slixmpp.plugins.xep_0055
.. autoclass:: XEP_0055
:members:
:exclude-members: session_bind, plugin_init, plugin_end
Stanza elements
---------------
.. automodule:: slixmpp.plugins.xep_0055.stanza
:members:
:undoc-members:

View file

@ -23,6 +23,7 @@ __all__ = [
'xep_0049', # Private XML Storage 'xep_0049', # Private XML Storage
'xep_0050', # Ad-hoc Commands 'xep_0050', # Ad-hoc Commands
'xep_0054', # vcard-temp 'xep_0054', # vcard-temp
'xep_0055', # Jabber Search
'xep_0059', # Result Set Management 'xep_0059', # Result Set Management
'xep_0060', # Pubsub (Client) 'xep_0060', # Pubsub (Client)
'xep_0065', # SOCKS5 Bytestreams 'xep_0065', # SOCKS5 Bytestreams

View file

@ -0,0 +1,6 @@
from slixmpp.plugins.base import register_plugin
from .search import XEP_0055
register_plugin(XEP_0055)

View file

@ -0,0 +1,89 @@
import logging
from slixmpp import CoroutineCallback, StanzaPath, Iq, register_stanza_plugin
from slixmpp.plugins import BasePlugin
from slixmpp.xmlstream import StanzaBase
from . import stanza
class XEP_0055(BasePlugin):
"""
XEP-0055: Jabber Search
The config options are only useful for a "server-side" search feature,
and if the ``provide_search`` option is set to True.
API
===
``search_get_form``: customize the search form content (ie fields)
``search_query``: return search results
"""
name = "xep_0055"
description = "XEP-0055: Jabber search"
dependencies = {"xep_0004", "xep_0030"}
stanza = stanza
default_config = {
"form_fields": {"first", "last"},
"form_instructions": "",
"form_title": "",
"provide_search": True
}
def plugin_init(self):
register_stanza_plugin(Iq, stanza.Search)
register_stanza_plugin(stanza.Search, self.xmpp["xep_0004"].stanza.Form)
if self.provide_search:
self.xmpp["xep_0030"].add_feature(stanza.Search.namespace)
self.xmpp.register_handler(
CoroutineCallback(
"search",
StanzaPath("/iq/search"),
self._handle_search,
)
)
self.api.register(self._get_form, "search_get_form")
self.api.register(self._get_results, "search_query")
async def _handle_search(self, iq: StanzaBase):
if iq["search"]["form"].get_values():
reply = await self.api["search_query"](None, None, iq.get_from(), iq)
reply["search"]["form"]["type"] = "result"
else:
reply = await self.api["search_get_form"](None, None, iq.get_from(), iq)
reply["search"]["form"].add_field(
"FORM_TYPE", value=stanza.Search.namespace, ftype="hidden"
)
reply.send()
async def _get_form(self, jid, node, ifrom, iq):
reply = iq.reply()
form = reply["search"]["form"]
form["title"] = self.form_title
form["instructions"] = self.form_instructions
for field in self.form_fields:
form.add_field(field)
return reply
async def _get_results(self, jid, node, ifrom, iq):
reply = iq.reply()
form = reply["search"]["form"]
form["type"] = "result"
for field in self.form_fields:
form.add_reported(field)
return reply
def make_search_iq(self, **kwargs):
iq = self.xmpp.make_iq(itype="set", **kwargs)
iq["search"]["form"].set_type("submit")
iq["search"]["form"].add_field(
"FORM_TYPE", value=stanza.Search.namespace, ftype="hidden"
)
return iq
log = logging.getLogger(__name__)

View file

@ -0,0 +1,10 @@
from typing import Set, ClassVar
from slixmpp.xmlstream import ElementBase
class Search(ElementBase):
namespace = "jabber:iq:search"
name = "query"
plugin_attrib = "search"
interfaces: ClassVar[Set[str]] = set()

View file

@ -0,0 +1,59 @@
import unittest
from slixmpp import register_stanza_plugin, Iq
from slixmpp.test import SlixTest
from slixmpp.plugins.xep_0055 import stanza
class TestJabberSearch(SlixTest):
def setUp(self):
register_stanza_plugin(Iq, stanza.Search)
self.stream_start(plugins={"xep_0055"})
def testRequestSearchFields(self):
iq = self.Iq()
iq.set_from("juliet@capulet.com/balcony")
iq.set_to("characters.shakespeare.lit")
iq.set_type("get")
iq.enable("search")
iq["id"] = "0"
self.check(
iq,
"""
<iq type='get'
from='juliet@capulet.com/balcony'
to='characters.shakespeare.lit'>
<query xmlns='jabber:iq:search'/>
</iq>
""",
)
def testSendSearch(self):
iq = self.xmpp["xep_0055"].make_search_iq(
ifrom="juliet@capulet.com/balcony", ito="characters.shakespeare.lit"
)
iq["search"]["form"].add_field(var="x-gender", value="male")
self.check(
iq,
"""
<iq type='set'
from='juliet@capulet.com/balcony'
to='characters.shakespeare.lit'>
<query xmlns='jabber:iq:search'>
<x xmlns='jabber:x:data' type='submit'>
<field type='hidden' var='FORM_TYPE'>
<value>jabber:iq:search</value>
</field>
<field var='x-gender'>
<value>male</value>
</field>
</x>
</query>
</iq>
""",
use_values=False,
)
suite = unittest.TestLoader().loadTestsFromTestCase(TestJabberSearch)

View file

@ -0,0 +1,170 @@
import unittest
from slixmpp.test import SlixTest
class TestJabberSearch(SlixTest):
def setUp(self):
self.stream_start(
mode="component",
plugin_config={
"xep_0055": {
"form_fields": {"first", "last"},
"form_instructions": "INSTRUCTIONS",
"form_title": "User Directory Search",
}
},
jid="characters.shakespeare.lit",
plugins={"xep_0055"}
)
self.xmpp["xep_0055"].api.register(get_results, "search_query")
self.xmpp["xep_0055"].api.register(get_results, "search_query")
def tearDown(self):
self.stream_close()
def testRequestingSearchFields(self):
self.recv(
"""
<iq type='get'
from='juliet@capulet.com/balcony'
to='characters.shakespeare.lit'
id='search3'
xml:lang='en'>
<query xmlns='jabber:iq:search'/>
</iq>
"""
)
self.send(
"""
<iq type='result'
from='characters.shakespeare.lit'
to='juliet@capulet.com/balcony'
id='search3'
xml:lang='en'>
<query xmlns='jabber:iq:search'>
<x xmlns='jabber:x:data' type='form'>
<title>User Directory Search</title>
<instructions>INSTRUCTIONS</instructions>
<field type='hidden'
var='FORM_TYPE'>
<value>jabber:iq:search</value>
</field>
<field var='first'/>
<field var='last'/>
</x>
</query>
</iq>
""",
use_values=False,
)
def testSearchResult(self):
self.recv(
"""
<iq type='get'
from='juliet@capulet.com/balcony'
to='characters.shakespeare.lit'
id='search2'
xml:lang='en'>
<query xmlns='jabber:iq:search'>
<x xmlns='jabber:x:data' type='submit'>
<field type='hidden' var='FORM_TYPE'>
<value>jabber:iq:search</value>
</field>
<field var='last'>
<value>Montague</value>
</field>
</x>
</query>
</iq>
"""
)
self.send(
"""
<iq type='result'
from='characters.shakespeare.lit'
to='juliet@capulet.com/balcony'
id='search2'
xml:lang='en'>
<query xmlns='jabber:iq:search'>
<x xmlns='jabber:x:data' type='result'>
<field type='hidden' var='FORM_TYPE'>
<value>jabber:iq:search</value>
</field>
<reported>
<field var='first' label='Given Name' />
<field var='last' label='Family Name' />
</reported>
<item>
<field var='first'><value>Benvolio</value></field>
<field var='last'><value>Montague</value></field>
</item>
</x>
</query>
</iq>
""",
use_values=False, # TypeError: element indices must be integers without that
)
def testSearchNoResult(self):
self.xmpp["xep_0055"].api.register(get_results, "search_query")
self.recv(
"""
<iq type='get'
from='juliet@capulet.com/balcony'
to='characters.shakespeare.lit'
id='search2'
xml:lang='en'>
<query xmlns='jabber:iq:search'>
<x xmlns='jabber:x:data' type='submit'>
<field type='hidden' var='FORM_TYPE'>
<value>jabber:iq:search</value>
</field>
<field var='last'>
<value>Capulet</value>
</field>
</x>
</query>
</iq>
"""
)
self.send(
"""
<iq type='result'
from='characters.shakespeare.lit'
to='juliet@capulet.com/balcony'
id='search2'
xml:lang='en'>
<query xmlns='jabber:iq:search'>
<x xmlns='jabber:x:data' type='result'>
<field type='hidden' var='FORM_TYPE'>
<value>jabber:iq:search</value>
</field>
<reported>
<field var='first' label='Given Name' />
<field var='last' label='Family Name' />
</reported>
</x>
</query>
</iq>
""",
use_values=False, # TypeError: element indices must be integers without that
)
async def get_results(jid, node, ifrom, iq):
reply = iq.reply()
form = reply["search"]["form"]
form["type"] = "result"
form.add_reported("first", label="Given Name")
form.add_reported("last", label="Family Name")
d = iq["search"]["form"].get_values()
if d["last"] == "Montague":
form.add_item({"first": "Benvolio", "last": "Montague"})
return reply
suite = unittest.TestLoader().loadTestsFromTestCase(TestJabberSearch)