summaryrefslogtreecommitdiff
path: root/slixmpp/xmlstream/matcher
diff options
context:
space:
mode:
Diffstat (limited to 'slixmpp/xmlstream/matcher')
-rw-r--r--slixmpp/xmlstream/matcher/base.py9
-rw-r--r--slixmpp/xmlstream/matcher/id.py8
-rw-r--r--slixmpp/xmlstream/matcher/idsender.py23
-rw-r--r--slixmpp/xmlstream/matcher/many.py5
-rw-r--r--slixmpp/xmlstream/matcher/stanzapath.py25
-rw-r--r--slixmpp/xmlstream/matcher/xmlmask.py28
-rw-r--r--slixmpp/xmlstream/matcher/xpath.py17
7 files changed, 69 insertions, 46 deletions
diff --git a/slixmpp/xmlstream/matcher/base.py b/slixmpp/xmlstream/matcher/base.py
index e2560d2b..552269c5 100644
--- a/slixmpp/xmlstream/matcher/base.py
+++ b/slixmpp/xmlstream/matcher/base.py
@@ -1,10 +1,13 @@
-
# slixmpp.xmlstream.matcher.base
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# Part of Slixmpp: The Slick XMPP Library
# :copyright: (c) 2011 Nathanael C. Fritz
# :license: MIT, see LICENSE for more details
+from typing import Any
+from slixmpp.xmlstream.stanzabase import StanzaBase
+
+
class MatcherBase(object):
"""
@@ -15,10 +18,10 @@ class MatcherBase(object):
:param criteria: Object to compare some aspect of a stanza against.
"""
- def __init__(self, criteria):
+ def __init__(self, criteria: Any):
self._criteria = criteria
- def match(self, xml):
+ def match(self, xml: StanzaBase) -> bool:
"""Check if a stanza matches the stored criteria.
Meant to be overridden.
diff --git a/slixmpp/xmlstream/matcher/id.py b/slixmpp/xmlstream/matcher/id.py
index 44df2c18..e4e7ad4e 100644
--- a/slixmpp/xmlstream/matcher/id.py
+++ b/slixmpp/xmlstream/matcher/id.py
@@ -5,6 +5,7 @@
# :copyright: (c) 2011 Nathanael C. Fritz
# :license: MIT, see LICENSE for more details
from slixmpp.xmlstream.matcher.base import MatcherBase
+from slixmpp.xmlstream.stanzabase import StanzaBase
class MatcherId(MatcherBase):
@@ -13,12 +14,13 @@ class MatcherId(MatcherBase):
The ID matcher selects stanzas that have the same stanza 'id'
interface value as the desired ID.
"""
+ _criteria: str
- def match(self, xml):
+ def match(self, xml: StanzaBase) -> bool:
"""Compare the given stanza's ``'id'`` attribute to the stored
``id`` value.
- :param xml: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase`
+ :param xml: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase`
stanza to compare against.
"""
- return xml['id'] == self._criteria
+ return bool(xml['id'] == self._criteria)
diff --git a/slixmpp/xmlstream/matcher/idsender.py b/slixmpp/xmlstream/matcher/idsender.py
index 8f5d0303..572f9d87 100644
--- a/slixmpp/xmlstream/matcher/idsender.py
+++ b/slixmpp/xmlstream/matcher/idsender.py
@@ -4,7 +4,19 @@
# Part of Slixmpp: The Slick XMPP Library
# :copyright: (c) 2011 Nathanael C. Fritz
# :license: MIT, see LICENSE for more details
+
from slixmpp.xmlstream.matcher.base import MatcherBase
+from slixmpp.xmlstream.stanzabase import StanzaBase
+from slixmpp.jid import JID
+from slixmpp.types import TypedDict
+
+from typing import Dict
+
+
+class CriteriaType(TypedDict):
+ self: JID
+ peer: JID
+ id: str
class MatchIDSender(MatcherBase):
@@ -14,25 +26,26 @@ class MatchIDSender(MatcherBase):
interface value as the desired ID, and that the 'from' value is one
of a set of approved entities that can respond to a request.
"""
+ _criteria: CriteriaType
- def match(self, xml):
+ def match(self, xml: StanzaBase) -> bool:
"""Compare the given stanza's ``'id'`` attribute to the stored
``id`` value, and verify the sender's JID.
- :param xml: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase`
+ :param xml: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase`
stanza to compare against.
"""
selfjid = self._criteria['self']
peerjid = self._criteria['peer']
- allowed = {}
+ allowed: Dict[str, bool] = {}
allowed[''] = True
allowed[selfjid.bare] = True
- allowed[selfjid.host] = True
+ allowed[selfjid.domain] = True
allowed[peerjid.full] = True
allowed[peerjid.bare] = True
- allowed[peerjid.host] = True
+ allowed[peerjid.domain] = True
_from = xml['from']
diff --git a/slixmpp/xmlstream/matcher/many.py b/slixmpp/xmlstream/matcher/many.py
index e8ad54e7..dae45463 100644
--- a/slixmpp/xmlstream/matcher/many.py
+++ b/slixmpp/xmlstream/matcher/many.py
@@ -3,7 +3,9 @@
# Copyright (C) 2010 Nathanael C. Fritz
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
+from typing import Iterable
from slixmpp.xmlstream.matcher.base import MatcherBase
+from slixmpp.xmlstream.stanzabase import StanzaBase
class MatchMany(MatcherBase):
@@ -18,8 +20,9 @@ class MatchMany(MatcherBase):
Methods:
match -- Overrides MatcherBase.match.
"""
+ _criteria: Iterable[MatcherBase]
- def match(self, xml):
+ def match(self, xml: StanzaBase) -> bool:
"""
Match a stanza against multiple criteria. The match is successful
if one of the criteria matches.
diff --git a/slixmpp/xmlstream/matcher/stanzapath.py b/slixmpp/xmlstream/matcher/stanzapath.py
index 1bf3fa8e..b7de3ee2 100644
--- a/slixmpp/xmlstream/matcher/stanzapath.py
+++ b/slixmpp/xmlstream/matcher/stanzapath.py
@@ -4,8 +4,9 @@
# Part of Slixmpp: The Slick XMPP Library
# :copyright: (c) 2011 Nathanael C. Fritz
# :license: MIT, see LICENSE for more details
+from typing import cast, List
from slixmpp.xmlstream.matcher.base import MatcherBase
-from slixmpp.xmlstream.stanzabase import fix_ns
+from slixmpp.xmlstream.stanzabase import fix_ns, StanzaBase
class StanzaPath(MatcherBase):
@@ -17,22 +18,28 @@ class StanzaPath(MatcherBase):
:param criteria: Object to compare some aspect of a stanza against.
"""
-
- def __init__(self, criteria):
- self._criteria = fix_ns(criteria, split=True,
- propagate_ns=False,
- default_ns='jabber:client')
+ _criteria: List[str]
+ _raw_criteria: str
+
+ def __init__(self, criteria: str):
+ self._criteria = cast(
+ List[str],
+ fix_ns(
+ criteria, split=True, propagate_ns=False,
+ default_ns='jabber:client'
+ )
+ )
self._raw_criteria = criteria
- def match(self, stanza):
+ def match(self, stanza: StanzaBase) -> bool:
"""
Compare a stanza against a "stanza path". A stanza path is similar to
an XPath expression, but uses the stanza's interfaces and plugins
instead of the underlying XML. See the documentation for the stanza
- :meth:`~slixmpp.xmlstream.stanzabase.ElementBase.match()` method
+ :meth:`~slixmpp.xmlstream.stanzabase.StanzaBase.match()` method
for more information.
- :param stanza: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase`
+ :param stanza: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase`
stanza to compare against.
"""
return stanza.match(self._criteria) or stanza.match(self._raw_criteria)
diff --git a/slixmpp/xmlstream/matcher/xmlmask.py b/slixmpp/xmlstream/matcher/xmlmask.py
index d50b706e..b63e0f05 100644
--- a/slixmpp/xmlstream/matcher/xmlmask.py
+++ b/slixmpp/xmlstream/matcher/xmlmask.py
@@ -1,4 +1,3 @@
-
# Slixmpp: The Slick XMPP Library
# Copyright (C) 2010 Nathanael C. Fritz
# This file is part of Slixmpp.
@@ -6,8 +5,9 @@
import logging
from xml.parsers.expat import ExpatError
+from xml.etree.ElementTree import Element
-from slixmpp.xmlstream.stanzabase import ET
+from slixmpp.xmlstream.stanzabase import ET, StanzaBase
from slixmpp.xmlstream.matcher.base import MatcherBase
@@ -33,32 +33,33 @@ class MatchXMLMask(MatcherBase):
:param criteria: Either an :class:`~xml.etree.ElementTree.Element` XML
object or XML string to use as a mask.
"""
+ _criteria: Element
- def __init__(self, criteria, default_ns='jabber:client'):
+ def __init__(self, criteria: str, default_ns: str = 'jabber:client'):
MatcherBase.__init__(self, criteria)
if isinstance(criteria, str):
- self._criteria = ET.fromstring(self._criteria)
+ self._criteria = ET.fromstring(criteria)
self.default_ns = default_ns
- def setDefaultNS(self, ns):
+ def setDefaultNS(self, ns: str) -> None:
"""Set the default namespace to use during comparisons.
:param ns: The new namespace to use as the default.
"""
self.default_ns = ns
- def match(self, xml):
+ def match(self, xml: StanzaBase) -> bool:
"""Compare a stanza object or XML object against the stored XML mask.
Overrides MatcherBase.match.
:param xml: The stanza object or XML object to compare against.
"""
- if hasattr(xml, 'xml'):
- xml = xml.xml
- return self._mask_cmp(xml, self._criteria, True)
+ real_xml = xml.xml
+ return self._mask_cmp(real_xml, self._criteria, True)
- def _mask_cmp(self, source, mask, use_ns=False, default_ns='__no_ns__'):
+ def _mask_cmp(self, source: Element, mask: Element, use_ns: bool = False,
+ default_ns: str = '__no_ns__') -> bool:
"""Compare an XML object against an XML mask.
:param source: The :class:`~xml.etree.ElementTree.Element` XML object
@@ -75,13 +76,6 @@ class MatchXMLMask(MatcherBase):
# If the element was not found. May happen during recursive calls.
return False
- # Convert the mask to an XML object if it is a string.
- if not hasattr(mask, 'attrib'):
- try:
- mask = ET.fromstring(mask)
- except ExpatError:
- log.warning("Expat error: %s\nIn parsing: %s", '', mask)
-
mask_ns_tag = "{%s}%s" % (self.default_ns, mask.tag)
if source.tag not in [mask.tag, mask_ns_tag]:
return False
diff --git a/slixmpp/xmlstream/matcher/xpath.py b/slixmpp/xmlstream/matcher/xpath.py
index b7503b73..bd41b60a 100644
--- a/slixmpp/xmlstream/matcher/xpath.py
+++ b/slixmpp/xmlstream/matcher/xpath.py
@@ -4,7 +4,8 @@
# Part of Slixmpp: The Slick XMPP Library
# :copyright: (c) 2011 Nathanael C. Fritz
# :license: MIT, see LICENSE for more details
-from slixmpp.xmlstream.stanzabase import ET, fix_ns
+from typing import cast
+from slixmpp.xmlstream.stanzabase import ET, fix_ns, StanzaBase
from slixmpp.xmlstream.matcher.base import MatcherBase
@@ -17,23 +18,23 @@ class MatchXPath(MatcherBase):
If the value of :data:`IGNORE_NS` is set to ``True``, then XPath
expressions will be matched without using namespaces.
"""
+ _criteria: str
- def __init__(self, criteria):
- self._criteria = fix_ns(criteria)
+ def __init__(self, criteria: str):
+ self._criteria = cast(str, fix_ns(criteria))
- def match(self, xml):
+ def match(self, xml: StanzaBase) -> bool:
"""
Compare a stanza's XML contents to an XPath expression.
If the value of :data:`IGNORE_NS` is set to ``True``, then XPath
expressions will be matched without using namespaces.
- :param xml: The :class:`~slixmpp.xmlstream.stanzabase.ElementBase`
+ :param xml: The :class:`~slixmpp.xmlstream.stanzabase.StanzaBase`
stanza to compare against.
"""
- if hasattr(xml, 'xml'):
- xml = xml.xml
+ real_xml = xml.xml
x = ET.Element('x')
- x.append(xml)
+ x.append(real_xml)
return x.find(self._criteria) is not None