github/SleekXMPP
1
0
Fork 0
mirror of https://github.com/correl/SleekXMPP.git synced 2024-11-24 03:00:15 +00:00
Code Issues Projects Releases Packages Wiki Activity

Merge branch 'develop-1.1' of github.com:fritzy/SleekXMPP into develop-1.1

Browse source
This commit is contained in:
Lance Stout 2011-12-15 12:04:14 -08:00
commit 8f9d1bcfe0
2 changed files with 91 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
sleekxmpp/xmlstream/xmlstream.py
View file

@ -1276,8 +1276,6 @@ class XMLStream(object):
:param xml: The :class:`~sleekxmpp.xmlstream.stanzabase.ElementBase`
stanza to analyze.
"""
log.debug("RECV: %s", tostring(xml, xmlns=self.default_ns,
stream=self))
# Apply any preprocessing filters.
xml = self.incoming_filter(xml)
@ -1291,6 +1289,9 @@ class XMLStream(object):
if stanza is None:
return
log.debug("RECV: %s", tostring(xml, xmlns=self.default_ns,
stream=self))
# Match the stanza against registered handlers. Handlers marked
# to run "in stream" will be executed immediately; the rest will
# be queued.

88
tests/test_stream_filters.py Normal file
View file

@ -0,0 +1,88 @@
import time
from sleekxmpp import Message
from sleekxmpp.test import *
from sleekxmpp.xmlstream.handler import *
from sleekxmpp.xmlstream.matcher import *
class TestFilters(SleekTest):
"""
Test using incoming and outgoing filters.
"""
def setUp(self):
self.stream_start()
def tearDown(self):
self.stream_close()
def testIncoming(self):
data = []
def in_filter(stanza):
if isinstance(stanza, Message):
if stanza['body'] == 'testing':
stanza['subject'] = stanza['body'] + ' filter'
print('>>> %s' % stanza['subject'])
return stanza
def on_message(msg):
print('<<< %s' % msg['subject'])
data.append(msg['subject'])
self.xmpp.add_filter('in', in_filter)
self.xmpp.add_event_handler('message', on_message)
self.recv("""
<message>
<body>no filter</body>
</message>
""")
self.recv("""
<message>
<body>testing</body>
</message>
""")
time.sleep(0.5)
self.assertEqual(data, ['', 'testing filter'],
'Incoming filter did not apply %s' % data)
def testOutgoing(self):
def out_filter(stanza):
if isinstance(stanza, Message):
if stanza['body'] == 'testing':
stanza['body'] = 'changed!'
return stanza
self.xmpp.add_filter('out', out_filter)
m1 = self.Message()
m1['body'] = 'testing'
m1.send()
m2 = self.Message()
m2['body'] = 'blah'
m2.send()
self.send("""
<message>
<body>changed!</body>
</message>
""")
self.send("""
<message>
<body>blah</body>
</message>
""")
suite = unittest.TestLoader().loadTestsFromTestCase(TestFilters)