pep/slixmpp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Unit test reorganization.

Browse source 
Moved SleekTest to sleekxmpp.test.

Organized test suites by their focus.
- Suites focused on testing stanza objects are named test_stanza_X.py
- Suites focused on testing stream behavior are name test_stream_X.py
This commit is contained in:
Lance Stout 2010-10-07 10:58:13 -04:00
parent 21c32c6e1c
commit 0fffbb8200
22 changed files with 192 additions and 171 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
sleekxmpp/test/__init__.py Normal file
View file

@ -0,0 +1,10 @@
"""
SleekXMPP: The Sleek XMPP Library
Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
This file is part of SleekXMPP.
See the file LICENSE for copying permission.
"""
from sleekxmpp.test.mocksocket import TestSocket
from sleekxmpp.test.sleektest import *

139
sleekxmpp/test/mocksocket.py Normal file
View file

@ -0,0 +1,139 @@
"""
SleekXMPP: The Sleek XMPP Library
Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
This file is part of SleekXMPP.
See the file LICENSE for copying permission.
"""
import socket
try:
import queue
except ImportError:
import Queue as queue
class TestSocket(object):
"""
A dummy socket that reads and writes to queues instead
of an actual networking socket.
Methods:
next_sent -- Return the next sent stanza.
recv_data -- Make a stanza available to read next.
recv -- Read the next stanza from the socket.
send -- Write a stanza to the socket.
makefile -- Dummy call, returns self.
read -- Read the next stanza from the socket.
"""
def __init__(self, *args, **kwargs):
"""
Create a new test socket.
Arguments:
Same as arguments for socket.socket
"""
self.socket = socket.socket(*args, **kwargs)
self.recv_queue = queue.Queue()
self.send_queue = queue.Queue()
def __getattr__(self, name):
"""
Return attribute values of internal, dummy socket.
Some attributes and methods are disabled to prevent the
socket from connecting to the network.
Arguments:
name -- Name of the attribute requested.
"""
def dummy(*args):
"""Method to do nothing and prevent actual socket connections."""
return None
overrides = {'connect': dummy,
'close': dummy,
'shutdown': dummy}
return overrides.get(name, getattr(self.socket, name))
# ------------------------------------------------------------------
# Testing Interface
def next_sent(self, timeout=None):
"""
Get the next stanza that has been 'sent'.
Arguments:
timeout -- Optional timeout for waiting for a new value.
"""
args = {'block': False}
if timeout is not None:
args = {'block': True, 'timeout': timeout}
try:
return self.send_queue.get(**args)
except:
return None
def recv_data(self, data):
"""
Add data to the receiving queue.
Arguments:
data -- String data to 'write' to the socket to be received
by the XMPP client.
"""
self.recv_queue.put(data)
# ------------------------------------------------------------------
# Socket Interface
def recv(self, *args, **kwargs):
"""
Read a value from the received queue.
Arguments:
Placeholders. Same as for socket.Socket.recv.
"""
return self.read(block=True)
def send(self, data):
"""
Send data by placing it in the send queue.
Arguments:
data -- String value to write.
"""
self.send_queue.put(data)
# ------------------------------------------------------------------
# File Socket
def makefile(self, *args, **kwargs):
"""
File socket version to use with ElementTree.
Arguments:
Placeholders, same as socket.Socket.makefile()
"""
return self
def read(self, block=True, timeout=None, **kwargs):
"""
Implement the file socket interface.
Arguments:
block -- Indicate if the read should block until a
value is ready.
timeout -- Time in seconds a block should last before
returning None.
"""
if timeout is not None:
block = True
try:
return self.recv_queue.get(block, timeout)
except:
return None

144
tests/sleektest.py → sleekxmpp/test/sleektest.py
View file

@ -7,145 +7,15 @@
"""
import unittest
import socket
try:
import queue
except ImportError:
import Queue as queue
import sleekxmpp
from sleekxmpp import ClientXMPP, ComponentXMPP
from sleekxmpp.stanza import Message, Iq, Presence
from sleekxmpp.test import TestSocket
from sleekxmpp.xmlstream.stanzabase import registerStanzaPlugin, ET
from sleekxmpp.xmlstream.tostring import tostring
class TestSocket(object):
"""
A dummy socket that reads and writes to queues instead
of an actual networking socket.
Methods:
next_sent -- Return the next sent stanza.
recv_data -- Make a stanza available to read next.
recv -- Read the next stanza from the socket.
send -- Write a stanza to the socket.
makefile -- Dummy call, returns self.
read -- Read the next stanza from the socket.
"""
def __init__(self, *args, **kwargs):
"""
Create a new test socket.
Arguments:
Same as arguments for socket.socket
"""
self.socket = socket.socket(*args, **kwargs)
self.recv_queue = queue.Queue()
self.send_queue = queue.Queue()
def __getattr__(self, name):
"""
Return attribute values of internal, dummy socket.
Some attributes and methods are disabled to prevent the
socket from connecting to the network.
Arguments:
name -- Name of the attribute requested.
"""
def dummy(*args):
"""Method to do nothing and prevent actual socket connections."""
return None
overrides = {'connect': dummy,
'close': dummy,
'shutdown': dummy}
return overrides.get(name, getattr(self.socket, name))
# ------------------------------------------------------------------
# Testing Interface
def next_sent(self, timeout=None):
"""
Get the next stanza that has been 'sent'.
Arguments:
timeout -- Optional timeout for waiting for a new value.
"""
args = {'block': False}
if timeout is not None:
args = {'block': True, 'timeout': timeout}
try:
return self.send_queue.get(**args)
except:
return None
def recv_data(self, data):
"""
Add data to the receiving queue.
Arguments:
data -- String data to 'write' to the socket to be received
by the XMPP client.
"""
self.recv_queue.put(data)
# ------------------------------------------------------------------
# Socket Interface
def recv(self, *args, **kwargs):
"""
Read a value from the received queue.
Arguments:
Placeholders. Same as for socket.Socket.recv.
"""
return self.read(block=True)
def send(self, data):
"""
Send data by placing it in the send queue.
Arguments:
data -- String value to write.
"""
self.send_queue.put(data)
# ------------------------------------------------------------------
# File Socket
def makefile(self, *args, **kwargs):
"""
File socket version to use with ElementTree.
Arguments:
Placeholders, same as socket.Socket.makefile()
"""
return self
def read(self, block=True, timeout=None, **kwargs):
"""
Implement the file socket interface.
Arguments:
block -- Indicate if the read should block until a
value is ready.
timeout -- Time in seconds a block should last before
returning None.
"""
if timeout is not None:
block = True
try:
return self.recv_queue.get(block, timeout)
except:
return None
class SleekTest(unittest.TestCase):
"""
@ -402,12 +272,12 @@ class SleekTest(unittest.TestCase):
self.xmpp.socket.recv_data(data)
def stream_make_header(self, sto='',
sfrom='',
sid='',
stream_ns="http://etherx.jabber.org/streams",
default_ns="jabber:client",
version="1.0",
xml_header=True):
sfrom='',
sid='',
stream_ns="http://etherx.jabber.org/streams",
default_ns="jabber:client",
version="1.0",
xml_header=True):
"""
Create a stream header to be received by the test XMPP agent.

0
tests/__init__.py
View file

3
tests/test_events.py
View file

@ -1,6 +1,5 @@
import sleekxmpp
import time
from . sleektest import *
from sleekxmpp.test import *
class TestEvents(SleekTest):

4
tests/test_jid.py
View file

@ -1,6 +1,7 @@
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.xmlstream.jid import JID
class TestJIDClass(SleekTest):
def testJIDfromfull(self):
j = JID('user@someserver/some/resource')
@ -23,4 +24,5 @@ class TestJIDClass(SleekTest):
self.assertEqual(j.full, 'user@someserver/some/resource', "Full does not match")
self.assertEqual(str(j), 'user@someserver/some/resource', "String does not match")
suite = unittest.TestLoader().loadTestsFromTestCase(TestJIDClass)

4
tests/test_stanzabase.py → tests/test_stanza_base.py
View file

@ -1,7 +1,7 @@
from . sleektest import *
import sleekxmpp
from sleekxmpp.test import *
from sleekxmpp.xmlstream.stanzabase import ET, StanzaBase
class TestStanzaBase(SleekTest):
def testTo(self):

3
tests/test_elementbase.py → tests/test_stanza_element.py
View file

@ -1,6 +1,7 @@
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.xmlstream.stanzabase import ElementBase
class TestElementBase(SleekTest):
def testFixNs(self):

3
tests/test_errorstanzas.py → tests/test_stanza_error.py
View file

@ -1,4 +1,5 @@
from . sleektest import *
from sleekxmpp.test import *
class TestErrorStanzas(SleekTest):

24
tests/test_gmail.py → tests/test_stanza_gmail.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.gmail_notify as gmail
@ -8,10 +8,10 @@ class TestGmail(SleekTest):
registerStanzaPlugin(Iq, gmail.GmailQuery)
registerStanzaPlugin(Iq, gmail.MailBox)
registerStanzaPlugin(Iq, gmail.NewMail)
def testCreateQuery(self):
"""Testing querying Gmail for emails."""
iq = self.Iq()
iq['type'] = 'get'
iq['gmail']['search'] = 'is:starred'
@ -29,20 +29,20 @@ class TestGmail(SleekTest):
def testMailBox(self):
"""Testing reading from Gmail mailbox result"""
# Use the example from Google's documentation at
# http://code.google.com/apis/talk/jep_extensions/gmail.html#notifications
xml = ET.fromstring("""
<iq type="result">
<mailbox xmlns="google:mail:notify"
result-time='1118012394209'
url='http://mail.google.com/mail'
total-matched='95'
url='http://mail.google.com/mail'
total-matched='95'
total-estimate='0'>
<mail-thread-info tid='1172320964060972012'
<mail-thread-info tid='1172320964060972012'
participation='1'
messages='28'
date='1118012394209'
messages='28'
date='1118012394209'
url='http://mail.google.com/mail?view=cv'>
<senders>
<sender name='Me' address='romeo@gmail.com' originator='1' />
@ -63,7 +63,7 @@ class TestGmail(SleekTest):
self.failUnless(mailbox['url'] == 'http://mail.google.com/mail', "url doesn't match")
self.failUnless(mailbox['matched'] == '95', "total-matched incorrect")
self.failUnless(mailbox['estimate'] == False, "total-estimate incorrect")
self.failUnless(len(mailbox['threads']) == 1, "could not extract message threads")
self.failUnless(len(mailbox['threads']) == 1, "could not extract message threads")
thread = mailbox['threads'][0]
self.failUnless(thread['tid'] == '1172320964060972012', "thread tid doesn't match")
@ -81,8 +81,8 @@ class TestGmail(SleekTest):
self.failUnless(sender1['address'] == 'romeo@gmail.com', "sender address doesn't match")
self.failUnless(sender1['originator'] == True, "sender originator incorrect")
self.failUnless(sender1['unread'] == False, "sender unread incorrectly True")
sender2 = thread['senders'][2]
sender2 = thread['senders'][2]
self.failUnless(sender2['unread'] == True, "sender unread incorrectly False")
suite = unittest.TestLoader().loadTestsFromTestCase(TestGmail)

2
tests/test_iqstanzas.py → tests/test_stanza_iq.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.xmlstream.stanzabase import ET

2
tests/test_messagestanzas.py → tests/test_stanza_message.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.stanza.message import Message
from sleekxmpp.stanza.htmlim import HTMLIM

3
tests/test_presencestanzas.py → tests/test_stanza_presence.py
View file

@ -1,5 +1,4 @@
import sleekxmpp
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.stanza.presence import Presence

2
tests/test_roster.py → tests/test_stanza_roster.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.stanza.roster import Roster

2
tests/test_forms.py → tests/test_stanza_xep_0004.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.xep_0004 as xep_0004

2
tests/test_disco.py → tests/test_stanza_xep_0030.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.xep_0030 as xep_0030

2
tests/test_addresses.py → tests/test_stanza_xep_0033.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.xep_0033 as xep_0033

2
tests/test_pubsubstanzas.py → tests/test_stanza_xep_0060.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.xep_0004 as xep_0004
import sleekxmpp.plugins.stanza_pubsub as pubsub

2
tests/test_chatstates.py → tests/test_stanza_xep_0085.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.xep_0085 as xep_0085
class TestChatStates(SleekTest):

4
tests/test_streamtester.py → tests/test_stream.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
import sleekxmpp.plugins.xep_0033 as xep_0033
@ -55,6 +55,6 @@ class TestStreamTester(SleekTest):
def testSendStreamHeader(self):
"""Test that we can check a sent stream header."""
self.stream_start(mode='client', skip=False)
self.streamSendHeader(sto='localhost')
self.stream_send_header(sto='localhost')
suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamTester)

4
tests/test_handlers.py → tests/test_stream_handlers.py
View file

@ -1,8 +1,8 @@
from . sleektest import *
import sleekxmpp
from sleekxmpp.test import *
from sleekxmpp.xmlstream.handler import *
from sleekxmpp.xmlstream.matcher import *
class TestHandlers(SleekTest):
"""
Test using handlers and waiters.

2
tests/test_tostring.py
View file

@ -1,4 +1,4 @@
from . sleektest import *
from sleekxmpp.test import *
from sleekxmpp.stanza import Message
from sleekxmpp.xmlstream.stanzabase import ET
from sleekxmpp.xmlstream.tostring import tostring, xml_escape