Commits

Alexander Shorin committed fec651a

Restructure test modules.

Comments (0)

Files changed (23)

tests/__init__.py

-# -*- coding: utf-8 -*-

tests/functional/__init__.py

-# -*- coding: utf-8 -*-

tests/helpers.py

-# -*- coding: utf-8 -*-
-
-import sys
-
-if sys.version_info >= (2, 7):
-    unittest = __import__('unittest')
-else:
-    unittest = __import__('unittest2')
-

tests/lib/__init__.py

-# -*- coding: utf-8 -*-

tests/lib/templates/message_for_test.html

-{{ message_for('message_for_test') }}

tests/lib/templates/ping.html

-{%- if user -%}
-pong from {{ user }}
-{%- else -%}
-pong
-{%- endif -%}

tests/lib/test_helpers.py

-# -*- coding: utf-8 -*-
-
-from tests.helpers import unittest
-from xmppflask.helpers import get_package_path
-
-class TestHelpers(unittest.TestCase):
-    def test_no_package_path(self):
-        import os
-        self.assertEquals(get_package_path('no_such_package(i_hope)'),
-                          os.getcwd())
-    
-    def test_safe_join(self):
-        from xmppflask import safe_join
-        from xmppflask.exceptions import NotFound
-
-        self.assertEquals(safe_join('a', 'b'),
-                          'a/b')
-        self.assertRaises(NotFound,
-                          lambda: safe_join('a', '../b'))
-        

tests/lib/test_routing.py

-# -*- coding: utf-8 -*-
-"""
-    xmppflask.routing test
-    ~~~~~~~~~~~~~~~~~~~~~~
-
-    :copyright: (c) 2011 Kostyantyn Rybnikov <k.bx@ya.ru>
-    :license: BSD.
-"""
-
-from tests.helpers import unittest
-
-from xmppflask.routing import Map, Rule
-from xmppflask.exceptions import NotFound
-
-class RoutingFunctionality(unittest.TestCase):
-    def test_simple(self):
-        map_ = Map([
-            Rule(u'ping', endpoint='ping'),
-            Rule(u'ping me', endpoint='ping_me'),
-            Rule(u'pong', endpoint='pong'),
-            Rule('weather in <string(maxlength=10):city>',
-                 endpoint='weather_in_city'),
-            Rule('weather in country <string(length=2):country_code>',
-                 endpoint='weather_in_country'),
-        ])
-        adapter = map_.bind()
-        self.assertEquals(adapter.match('ping'), ('ping', {}))
-        self.assertEquals(adapter.match('ping me'), ('ping_me', {}))
-        self.assertEquals(adapter.match('pong'), ('pong', {}))
-        self.assertEquals(adapter.match('weather in Kiev'),
-                          ('weather_in_city', {'city': 'Kiev'}))
-        self.assertEquals(adapter.match('weather in country UA'),
-                          ('weather_in_country', {'country_code': 'UA'}))
-
-    def test_basic_building(self):
-        map_ = Map([
-            Rule('ping', endpoint='ping'),
-            Rule('ping <user>', endpoint='ping_user'),
-            Rule('ping <user> <int:n> times',
-                 endpoint='ping_user_multiple_times'),
-        ])
-        adapter = map_.bind()
-
-        self.assertEquals(adapter.build('ping', {}), 
-                          'ping')
-        self.assertEquals(adapter.build('ping_user', {'user': 'k_bx'}),
-                          'ping k_bx')
-        self.assertEquals(adapter.build('ping_user_multiple_times', {'user': 'k_bx', 
-                                                      'n': 5}),
-                          'ping k_bx 5 times')
-
-    def test_defaults(self):
-        map_ = Map([
-            Rule('ping', defaults={'user': 'k_bx'}, endpoint='ping'),
-            Rule('ping <int:n> times', endpoint='ping_n_times')
-        ])
-        adapter = map_.bind()
-
-        self.assertEquals(adapter.match('ping'),
-                          ('ping', {'user': 'k_bx'}))
-        self.assertEquals(adapter.match('ping 7 times'),
-                          ('ping_n_times', {'n': 7}))
-        
-        self.assertEquals(adapter.build('ping'), 'ping')
-
-    def test_rule_repr_simple(self):
-        rule = Rule('ping')
-        self.assertEquals(repr(rule),
-                          '<Rule (unbound)>')
-
-    def test_rule_repr_binded(self):
-        map_ = Map([
-            Rule('ping <user> <int:n> times', endpoint='ping')
-        ])
-        self.assertEquals(
-            repr(map_._rules[0]),
-            '<Rule ping <user> <n> times -> ping>')
-
-    def test_any_converter(self):
-        map_ = Map([
-            Rule('ping <any(u"k_bx", u"kb", u"kost-bebix"):user>', endpoint='ping')
-        ])
-        adapter = map_.bind()
-        self.assertEquals(adapter.match('ping k_bx'),
-                          ('ping', {'user': 'k_bx'}))
-        
-
-    def test_suitable_for(self):
-        map_ = Map([
-            Rule('ping <any(u"k_bx", u"kb", u"kost-bebix"):user>', endpoint='ping'),
-            Rule('ping some <chicken> and <rabbit>', endpoint='ping_some_chicken_and_rabbit',
-                 defaults={'chicken': 'Chicken John'})
-        ])
-        adapter = map_.bind()
-        rule = map_._rules[0]
-        rule_2 = map_._rules[1]
-        self.assertEquals(rule.suitable_for({'user': 'kb'}),
-                          True)
-        self.assertEquals(rule.suitable_for({}),
-                          False)
-        self.assertEquals(rule_2.suitable_for({}),
-                          False)
-        self.assertEquals(rule_2.suitable_for({'rabbit': 'Rabbit Jack'}),
-                          True)
-        self.assertEquals(rule_2.suitable_for({'rabbit': 'Rabbit Jack',
-                                               'chicken': 'Chicken John'}),
-                          True)
-        self.assertEquals(rule_2.suitable_for({'rabbit': 'Rabbit Jack',
-                                               'chicken': 'Chicken Joe'}),
-                          False)
-
-    def test_aliases(self):
-        def ping():
-            return 'pong'
-
-        rmap = Map()
-        rmap.add(Rule('ping', endpoint=ping))
-        rmap.add(Rule('pong', endpoint=ping))
-        rmap.update()
-
-    def test_route_filtered_by_event_type(self):
-        rmap = Map()
-        rmap.add(Rule('ping', event_type='foo', endpoint='ping'))
-        rmap.add(Rule('pong', event_type='bar', endpoint='pong'))
-
-        adapter = rmap.bind()
-        self.assertEqual(adapter.match('ping', event_type='foo'), ('ping', {}))
-        self.assertEqual(adapter.match('pong', event_type='bar'), ('pong', {}))
-
-        self.assertRaises(NotFound, adapter.match, 'pong', event_type='foo')
-        self.assertRaises(NotFound, adapter.match, 'ping', event_type='bar')
-
-    def test_route_filtered_by_type(self):
-        rmap = Map()
-        rmap.add(Rule('ping', type='foo', endpoint='ping'))
-        rmap.add(Rule('pong', type='bar', endpoint='pong'))
-
-        adapter = rmap.bind()
-        self.assertEqual(adapter.match('ping', type='foo'), ('ping', {}))
-        self.assertEqual(adapter.match('pong', type='bar'), ('pong', {}))
-
-        self.assertRaises(NotFound, adapter.match, 'pong', type='foo')
-        self.assertRaises(NotFound, adapter.match, 'ping', type='bar')
-
-    def test_route_filtered_by_sender_jid(self):
-        rmap = Map()
-        rmap.add(Rule('ping', from_jid='.*@xmpp.ru', endpoint='ping'))
-        rmap.add(Rule('pong', from_jid='foo@bar', endpoint='pong'))
-
-        adapter = rmap.bind()
-        self.assertEqual(
-            adapter.match('ping', from_jid='foo@xmpp.ru'),
-            ('ping', {}))
-        self.assertEqual(
-            adapter.match('ping', from_jid='bar@xmpp.ru'),
-            ('ping', {}))
-        self.assertEqual(
-            adapter.match('pong', from_jid='foo@bar'),
-            ('pong', {}))
-        self.assertEqual(
-            adapter.match('pong', from_jid='foo@bar.ru'),
-            ('pong', {}))
-
-        self.assertRaises(NotFound, adapter.match, 'ping', from_jid='foo@bar')
-        self.assertRaises(NotFound, adapter.match, 'ping', from_jid='bar')
-        self.assertRaises(NotFound, adapter.match, 'ping', from_jid='bar@foo')
-        self.assertRaises(NotFound, adapter.match, 'pong', from_jid='foo@xmpp.ru')
-        self.assertRaises(NotFound, adapter.match, 'pong', from_jid='_foo@bar')

tests/lib/test_server.py

-# -*- coding: utf-8 -*-
-"""
-    XmppFlask Tests
-    ~~~~~~~~~~~~~~~
-
-    Test XmppFlask XMPPWSGI server.
-    
-    :copyright: (c) 2012 Alexander Shorin <kxepal@gmail.com>
-    :license: BSD.
-"""
-
-from tests.helpers import unittest
-from xmppflask import XmppFlask
-from xmppflask.server import XmppWsgiServer
-
-def counted(fun):
-    def wrapper(*args, **kwargs):
-        wrapper.times_called += 1
-        return fun(*args, **kwargs)
-    wrapper.times_called = 0
-    wrapper.__name__= fun.__name__
-    return wrapper
-
-
-class MockXmppWsgiServer(XmppWsgiServer):
-    connected = False
-    running = False
-    last_message = None
-
-    def _connect(self, *args, **kwargs):
-        self.connected = True
-
-    def serve_forever(self):
-        self.running = True
-
-    def dispatch_app_response(self, environ, response):
-        self.message_sent = False
-        self.presence_sent = False
-        self.iq_sent = False
-        return super(MockXmppWsgiServer, self).dispatch_app_response(environ, response)
-
-    def send_message(self, environ, body):
-        pass
-
-    def send_presence(self, environ, payload):
-        pass
-
-    def send_iq(self, environ, payload):
-        pass
-
-
-class XmppWsgiServerTestCase(unittest.TestCase):
-
-    def setUp(self):
-        app = XmppFlask('xmppflask.test')
-        self.server = MockXmppWsgiServer(app)
-        self.server.send_message = counted(self.server.send_message)
-        self.server.send_presence = counted(self.server.send_presence)
-        self.server.send_iq = counted(self.server.send_iq)
-
-    def test_connect(self):
-        self.server.connect('foo', 'bar')
-        assert self.server.connected
-
-    def test_dispatch_message_response(self):
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class('baz')
-
-        self.server.dispatch_app_response(environ, resp)
-
-        self.assertEqual(self.server.send_message.times_called, 1)
-
-    def test_dispatch_new_style_message_response(self):
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class([('message', 'baz')])
-
-        self.server.dispatch_app_response(environ, resp)
-
-        self.assertEqual(self.server.send_message.times_called, 1)
-
-    def test_dispatch_new_style_signle_message_response(self):
-        def pong():
-            return 'message', 'pong'
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class(pong())
-
-        self.server.dispatch_app_response(environ, resp)
-
-        self.assertEqual(self.server.send_message.times_called, 1)
-
-    def test_dispatch_presence_response(self):
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class([('presence', 'baz')])
-
-        self.server.dispatch_app_response(environ, resp)
-
-        self.assertEqual(self.server.send_presence.times_called, 1)
-
-    def test_dispatch_iq_response(self):
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class([('iq', 'baz')])
-
-        self.server.dispatch_app_response(environ, resp)
-
-        self.assertEqual(self.server.send_iq.times_called, 1)
-
-    def test_dispatch_weird_response(self):
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class([('foo', 'boo')])
-
-        self.assertRaises(ValueError,
-                          self.server.dispatch_app_response, environ, resp)
-
-    def test_dispatch_generator_of_responses(self):
-        def ping():
-            yield 'presence', '1'
-            yield 'presence', '2'
-            yield 'presence', '3'
-            yield 'PONG!'
-
-        environ = {'XMPP_JID': 'foo@bar'}
-        resp = self.server.app.response_class(ping())
-
-        self.server.dispatch_app_response(environ, resp)
-
-        self.assertEqual(self.server.send_message.times_called, 1)
-        self.assertEqual(self.server.send_presence.times_called, 3)

tests/lib/test_sessions.py

-# -*- coding: utf-8 -*-
-"""
-    XmppFlask Tests
-    ~~~~~~~~~~~~~~~
-
-    Test XmppFlask sessions.
-    
-    :copyright: (c) 2012 Alexander Shorin <kxepal@gmail.com>
-    :license: BSD.
-"""
-
-from tests.helpers import unittest
-import xmppflask
-
-class MemorySessionTestCase(unittest.TestCase):
-
-    def setUp(self):
-        from xmppflask.sessions import MemorySessionInterface
-        self.app = xmppflask.XmppFlask(__name__)
-        self.app.session_interface = MemorySessionInterface()
-
-    def test_session_basic(self):
-        """Basic test about Session initialization."""
-        environ = {'MESSAGE': 'ping', 'XMPP_JID': 'k.bx@ya.ru'}
-        @self.app.route(u'ping')
-        def ping():
-            from xmppflask.globals import session
-            if 'times_pinged' not in session:
-                session['times_pinged'] = 0
-            session['times_pinged'] += 1
-            return u'pong'
-        self.app(environ)
-
-        session = self.app.session_interface.storage[environ['XMPP_JID']]
-        self.assertEquals(session['times_pinged'], 1)
-
-    def test_session_should_keep_values(self):
-        """Session should keep stored values for whole his lifespan."""
-        environ = {'MESSAGE': 'ping', 'XMPP_JID': 'k.bx@ya.ru'}
-        @self.app.route(u'ping')
-        def ping():
-            from xmppflask.globals import session
-            if 'times_pinged' not in session:
-                session['times_pinged'] = 0
-            session['times_pinged'] += 1
-            return u'pong'
-        self.app(environ)
-        self.app(environ)
-        self.app(environ)
-
-        session = self.app.session_interface.storage[environ['XMPP_JID']]
-        self.assertEquals(session['times_pinged'], 3)
-
-    def test_sessionless_workflow(self):
-        """Test to be sure, that nullified sessions do not break anything."""
-        environ = {'MESSAGE': 'PING', 'XMPP_JID': 'k.bx@ya.ru'}
-        app = xmppflask.XmppFlask(__name__)
-        @app.route(u'PING')
-        def ping():
-            return u'PONG'
-        rv = app(environ)
-        rv = app(environ)
-        self.assertEquals(list(rv), ['PONG'])
-
-    def test_session_expires(self):
-        environ = {'MESSAGE': 'ping', 'XMPP_JID': 'k.bx@ya.ru'}
-        @self.app.route(u'ping')
-        def ping():
-            from xmppflask.globals import session
-            if 'times_pinged' not in session:
-                session['times_pinged'] = 0
-            session['times_pinged'] += 1
-            return u'pong'
-        self.app(environ)
-
-        session = self.app.session_interface.storage[environ['XMPP_JID']]
-        session._timestamp = 1 # tweak last update timestamp
-
-        self.app(environ)
-
-        # session was expired and all his data should be erased
-        self.assertEquals(session['times_pinged'], 1)

tests/lib/test_wrappers.py

-# -*- coding: utf-8 -*-
-"""
-    XmppFlask Tests
-    ~~~~~~~~~~~~~~~
-
-    Test XmppFlask Request/Response wrappers.
-    
-    :copyright: (c) 2012 Alexander Shorin <kxepal@gmail.com>
-    :license: BSD.
-"""
-
-from tests.helpers import unittest
-from xmppflask.wrappers import Response
-
-class ResponseTestCase(unittest.TestCase):
-
-    def test_init_by_string(self):
-        resp = Response('foo')
-        self.assertEqual(list(resp), ['foo'])
-
-    def test_init_by_iterable(self):
-        resp = Response([1, 2, 3])
-        self.assertEqual(list(resp), [1, 2, 3])
-
-        resp = Response(('foo', 'bar', 'baz'))
-        self.assertEqual(list(resp), ['foo', 'bar', 'baz'])
-
-        resp = Response(xrange(2))
-        self.assertEqual(list(resp), [0, 1])
-
-        resp = Response({'foo': 'bar'})
-        self.assertEqual(list(resp), ['foo'])
-
-    def test_single_pair_response(self):
-        resp = Response(('foo', 'bar'))
-        self.assertEqual(list(resp), [('foo', 'bar')])
-
-    def test_init_by_other(self):
-        class Foo(object):
-            def __str__(self):
-                return 'foo'
-
-        resp = Response(42)
-        self.assertEqual(list(resp), ['42'])
-
-        resp = Response(Foo())
-        self.assertEqual(list(resp), ['foo'])
-
-    def test_not_iterable_twice(self):
-        resp = Response('foo')
-        self.assertEqual(list(resp), ['foo'])
-        self.assertEqual(list(resp), [])
-
-    def test_concat_responses(self):
-        orig_resp = Response('foo')
-        resp = orig_resp(Response('bar'))
-        self.assertTrue(orig_resp is resp)
-
-        resp = resp(Response('baz'))
-
-        self.assertEqual(list(resp), ['foo', 'bar', 'baz'])
-
-    def test_lazy_iterator(self):
-        gen = (i for i in range(3))
-        resp = Response(42)(gen)
-
-        self.assertEqual(gen.next(), 0)
-        self.assertEqual(resp.next(), '42')
-        self.assertEqual(gen.next(), 1)
-        self.assertEqual(resp.next(), 2)
-        self.assertRaises(StopIteration, gen.next)

tests/lib/test_xmppflask.py

-# -*- coding: utf-8 -*-
-"""
-    XmppFlask Tests
-    ~~~~~~~~~~~~~~~
-
-    Test XmppFlask itself.
-    
-    :copyright: (c) 2011 Kostyantyn Rybnikov <k.bx@ya.ru>
-    :license: BSD.
-"""
-
-import os, sys
-
-from tests.helpers import unittest
-import xmppflask
-
-class BasicFunctionalityTestCase(unittest.TestCase):
-    def test_xmppwsgi_simple(self):
-        environ = {'MESSAGE': 'PING', 'XMPP_JID': 'k.bx@ya.ru'}
-        app = xmppflask.XmppFlask(__name__)
-        @app.route(u'PING')
-        def ping():
-            return u'PONG'
-        rv = app(environ)
-        self.assertEquals(list(rv), ['PONG'])
-    
-    def test_xmppwsgi_more_complex(self):
-        environ = {'MESSAGE': 'PING 10 times', 'XMPP_JID': 'k.bx@ya.ru'}
-        app = xmppflask.XmppFlask(__name__)
-        @app.route(u'PING <int:times> times')
-        def ping(times):
-            return u'PONG %s times' % times
-        rv = app(environ)
-        self.assertEquals(list(rv), ['PONG 10 times'])
-
-    def test_template_rendering(self):
-        from jinja2 import TemplateNotFound
-        from xmppflask import render_template
-
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-        app = xmppflask.XmppFlask(__name__)
-        @app.route(u'ping')
-        def ping():
-            return render_template('ping.html')
-        @app.route(u'ping <user>')
-        def ping_user(user):
-            return render_template('ping.html', user=user)
-
-        @app.route(u'template_not_found')
-        def template_not_found():
-            return render_template('template_not_found.html')
-        
-        environ['MESSAGE'] = 'ping'
-        rv = app(environ)
-        self.assertEquals(list(rv), ['pong'])
-        
-        environ['MESSAGE'] = 'ping k_bx'
-        rv = app(environ)
-        self.assertEquals(list(rv), ['pong from k_bx'])
-
-    def test_template_rendering_with_message_for(self):
-        from xmppflask import render_template
-        
-        app = xmppflask.XmppFlask(__name__)
-        
-        @app.route(u'message_for_test')
-        def message_for_test():
-            return render_template('message_for_test.html')
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-        
-        environ['MESSAGE'] = 'message_for_test'
-        rv = app(environ)
-        self.assertEquals(u''.join(rv), 'message_for_test')
-
-    def test_context_simple(self):
-        app = xmppflask.XmppFlask(__name__)
-        
-        @app.route(u'ping')
-        def message_for_test():
-            from xmppflask import g
-            g.some_val = 'some val'
-            return u'pong %s' % g.some_val
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-        
-        notification_queue = []
-        environ['MESSAGE'] = 'ping'
-        rv = app(environ, notification_queue)
-        self.assertEquals(u''.join(rv), 'pong some val')
-
-    def test_notification(self):
-        from xmppflask.notification import notify
-        app = xmppflask.XmppFlask(__name__)
-        
-        @app.route(u'ping')
-        def message_for_test():
-            notify('kost-bebix@ya.ru', 'ping route')
-            return u'pong'
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-        
-        notification_queue = []
-        environ['MESSAGE'] = 'ping'
-        rv = app(environ, notification_queue)
-        self.assertEquals(u''.join(rv), 'pong')
-        self.assertEquals(notification_queue, [('kost-bebix@ya.ru', 'ping route')])
-
-    def test_before_request(self):
-        from xmppflask import g
-        app = xmppflask.XmppFlask(__name__)
-
-        must_be_changed = {'status': 'not changed'}
-        
-        @app.before_request
-        def before_request():
-            g.db = 'db obj'
-
-        @app.teardown_request
-        def teardown_request(exception):
-            must_be_changed['status'] = 'changed'
-
-        @app.route(u'ping')
-        def message_for_test():
-            self.assertEquals(g.db, 'db obj')
-            return u'pong'
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-
-        environ['MESSAGE'] = 'ping'
-        rv = app(environ)
-        self.assertEquals(u''.join(rv), 'pong')
-        self.assertEquals(must_be_changed['status'], 'changed')
-
-    def test_static_url_path(self):
-        app = xmppflask.XmppFlask(__name__)
-        @app.route(u'ping')
-        def ping():
-            return u'pong'
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-
-        environ['MESSAGE'] = 'ping'
-        self.assertEquals(app.static_url_path,
-                          None)
-
-    def test_user_exception(self):
-        class TestException(Exception): pass
-        app = xmppflask.XmppFlask(__name__)
-        app.debug = True
-            
-        @app.route(u'ping')
-        def ping():
-            raise TestException('wrong!')
-        environ = {'XMPP_JID': 'k.bx@ya.ru'}
-
-        environ['MESSAGE'] = 'ping'
-        self.assertRaises(TestException, lambda: app(environ))
-
-    def test_handle_presences(self):
-        app = xmppflask.XmppFlask(__name__)
-
-        @app.route_presence()
-        def presence():
-            return 'got it'
-
-        environ = {'XMPP_JID': 'k.bx@ya.ru', 'XMPP_EVENT': 'presence'}
-
-        rv = app(environ)
-        self.assertEquals(u''.join(rv), 'got it')
-
-    def test_handle_presences_from_specific_domain(self):
-        app = xmppflask.XmppFlask(__name__)
-
-        @app.route_presence(from_jid='.*@ya.ru', type='available')
-        def presence():
-            return 'welcome back, %s!' % environ['XMPP_JID']
-
-        environ = {'XMPP_JID': 'k.bx@ya.ru',
-                   'XMPP_EVENT': 'presence',
-                   'type': 'available'}
-
-        rv = app(environ)
-        self.assertEquals(u''.join(rv),
-                          'welcome back, %s!' % environ['XMPP_JID'])
-
-    def test_presence_mismatch_doesnt_return_message_not_understood_resp(self):
-        app = xmppflask.XmppFlask(__name__)
-
-        @app.route_presence(from_jid='.*@ya.ru', type='available')
-        def presence():
-            return 'welcome back, %s!' % environ['XMPP_JID']
-
-        environ = {'XMPP_JID': 'foo@bar',
-                   'XMPP_EVENT': 'presence',
-                   'type': 'available'}
-
-        rv = app(environ)
-        self.assertEquals(u''.join(rv), '')

xmppflask/tests/__init__.py

+# -*- coding: utf-8 -*-

xmppflask/tests/functional/__init__.py

+# -*- coding: utf-8 -*-

xmppflask/tests/helpers.py

+# -*- coding: utf-8 -*-
+
+import sys
+
+if sys.version_info >= (2, 7):
+    unittest = __import__('unittest')
+else:
+    unittest = __import__('unittest2')
+

xmppflask/tests/templates/message_for_test.html

+{{ message_for('message_for_test') }}

xmppflask/tests/templates/ping.html

+{%- if user -%}
+pong from {{ user }}
+{%- else -%}
+pong
+{%- endif -%}

xmppflask/tests/test_helpers.py

+# -*- coding: utf-8 -*-
+
+from xmppflask.tests.helpers import unittest
+from xmppflask.helpers import get_package_path
+
+class TestHelpers(unittest.TestCase):
+    def test_no_package_path(self):
+        import os
+        self.assertEquals(get_package_path('no_such_package(i_hope)'),
+                          os.getcwd())
+    
+    def test_safe_join(self):
+        from xmppflask import safe_join
+        from xmppflask.exceptions import NotFound
+
+        self.assertEquals(safe_join('a', 'b'),
+                          'a/b')
+        self.assertRaises(NotFound,
+                          lambda: safe_join('a', '../b'))
+        

xmppflask/tests/test_routing.py

+# -*- coding: utf-8 -*-
+"""
+    xmppflask.routing test
+    ~~~~~~~~~~~~~~~~~~~~~~
+
+    :copyright: (c) 2011 Kostyantyn Rybnikov <k.bx@ya.ru>
+    :license: BSD.
+"""
+
+from xmppflask.tests.helpers import unittest
+
+from xmppflask.routing import Map, Rule
+from xmppflask.exceptions import NotFound
+
+class RoutingFunctionality(unittest.TestCase):
+    def test_simple(self):
+        map_ = Map([
+            Rule(u'ping', endpoint='ping'),
+            Rule(u'ping me', endpoint='ping_me'),
+            Rule(u'pong', endpoint='pong'),
+            Rule('weather in <string(maxlength=10):city>',
+                 endpoint='weather_in_city'),
+            Rule('weather in country <string(length=2):country_code>',
+                 endpoint='weather_in_country'),
+        ])
+        adapter = map_.bind()
+        self.assertEquals(adapter.match('ping'), ('ping', {}))
+        self.assertEquals(adapter.match('ping me'), ('ping_me', {}))
+        self.assertEquals(adapter.match('pong'), ('pong', {}))
+        self.assertEquals(adapter.match('weather in Kiev'),
+                          ('weather_in_city', {'city': 'Kiev'}))
+        self.assertEquals(adapter.match('weather in country UA'),
+                          ('weather_in_country', {'country_code': 'UA'}))
+
+    def test_basic_building(self):
+        map_ = Map([
+            Rule('ping', endpoint='ping'),
+            Rule('ping <user>', endpoint='ping_user'),
+            Rule('ping <user> <int:n> times',
+                 endpoint='ping_user_multiple_times'),
+        ])
+        adapter = map_.bind()
+
+        self.assertEquals(adapter.build('ping', {}), 
+                          'ping')
+        self.assertEquals(adapter.build('ping_user', {'user': 'k_bx'}),
+                          'ping k_bx')
+        self.assertEquals(adapter.build('ping_user_multiple_times', {'user': 'k_bx', 
+                                                      'n': 5}),
+                          'ping k_bx 5 times')
+
+    def test_defaults(self):
+        map_ = Map([
+            Rule('ping', defaults={'user': 'k_bx'}, endpoint='ping'),
+            Rule('ping <int:n> times', endpoint='ping_n_times')
+        ])
+        adapter = map_.bind()
+
+        self.assertEquals(adapter.match('ping'),
+                          ('ping', {'user': 'k_bx'}))
+        self.assertEquals(adapter.match('ping 7 times'),
+                          ('ping_n_times', {'n': 7}))
+        
+        self.assertEquals(adapter.build('ping'), 'ping')
+
+    def test_rule_repr_simple(self):
+        rule = Rule('ping')
+        self.assertEquals(repr(rule),
+                          '<Rule (unbound)>')
+
+    def test_rule_repr_binded(self):
+        map_ = Map([
+            Rule('ping <user> <int:n> times', endpoint='ping')
+        ])
+        self.assertEquals(
+            repr(map_._rules[0]),
+            '<Rule ping <user> <n> times -> ping>')
+
+    def test_any_converter(self):
+        map_ = Map([
+            Rule('ping <any(u"k_bx", u"kb", u"kost-bebix"):user>', endpoint='ping')
+        ])
+        adapter = map_.bind()
+        self.assertEquals(adapter.match('ping k_bx'),
+                          ('ping', {'user': 'k_bx'}))
+        
+
+    def test_suitable_for(self):
+        map_ = Map([
+            Rule('ping <any(u"k_bx", u"kb", u"kost-bebix"):user>', endpoint='ping'),
+            Rule('ping some <chicken> and <rabbit>', endpoint='ping_some_chicken_and_rabbit',
+                 defaults={'chicken': 'Chicken John'})
+        ])
+        adapter = map_.bind()
+        rule = map_._rules[0]
+        rule_2 = map_._rules[1]
+        self.assertEquals(rule.suitable_for({'user': 'kb'}),
+                          True)
+        self.assertEquals(rule.suitable_for({}),
+                          False)
+        self.assertEquals(rule_2.suitable_for({}),
+                          False)
+        self.assertEquals(rule_2.suitable_for({'rabbit': 'Rabbit Jack'}),
+                          True)
+        self.assertEquals(rule_2.suitable_for({'rabbit': 'Rabbit Jack',
+                                               'chicken': 'Chicken John'}),
+                          True)
+        self.assertEquals(rule_2.suitable_for({'rabbit': 'Rabbit Jack',
+                                               'chicken': 'Chicken Joe'}),
+                          False)
+
+    def test_aliases(self):
+        def ping():
+            return 'pong'
+
+        rmap = Map()
+        rmap.add(Rule('ping', endpoint=ping))
+        rmap.add(Rule('pong', endpoint=ping))
+        rmap.update()
+
+    def test_route_filtered_by_event_type(self):
+        rmap = Map()
+        rmap.add(Rule('ping', event_type='foo', endpoint='ping'))
+        rmap.add(Rule('pong', event_type='bar', endpoint='pong'))
+
+        adapter = rmap.bind()
+        self.assertEqual(adapter.match('ping', event_type='foo'), ('ping', {}))
+        self.assertEqual(adapter.match('pong', event_type='bar'), ('pong', {}))
+
+        self.assertRaises(NotFound, adapter.match, 'pong', event_type='foo')
+        self.assertRaises(NotFound, adapter.match, 'ping', event_type='bar')
+
+    def test_route_filtered_by_type(self):
+        rmap = Map()
+        rmap.add(Rule('ping', type='foo', endpoint='ping'))
+        rmap.add(Rule('pong', type='bar', endpoint='pong'))
+
+        adapter = rmap.bind()
+        self.assertEqual(adapter.match('ping', type='foo'), ('ping', {}))
+        self.assertEqual(adapter.match('pong', type='bar'), ('pong', {}))
+
+        self.assertRaises(NotFound, adapter.match, 'pong', type='foo')
+        self.assertRaises(NotFound, adapter.match, 'ping', type='bar')
+
+    def test_route_filtered_by_sender_jid(self):
+        rmap = Map()
+        rmap.add(Rule('ping', from_jid='.*@xmpp.ru', endpoint='ping'))
+        rmap.add(Rule('pong', from_jid='foo@bar', endpoint='pong'))
+
+        adapter = rmap.bind()
+        self.assertEqual(
+            adapter.match('ping', from_jid='foo@xmpp.ru'),
+            ('ping', {}))
+        self.assertEqual(
+            adapter.match('ping', from_jid='bar@xmpp.ru'),
+            ('ping', {}))
+        self.assertEqual(
+            adapter.match('pong', from_jid='foo@bar'),
+            ('pong', {}))
+        self.assertEqual(
+            adapter.match('pong', from_jid='foo@bar.ru'),
+            ('pong', {}))
+
+        self.assertRaises(NotFound, adapter.match, 'ping', from_jid='foo@bar')
+        self.assertRaises(NotFound, adapter.match, 'ping', from_jid='bar')
+        self.assertRaises(NotFound, adapter.match, 'ping', from_jid='bar@foo')
+        self.assertRaises(NotFound, adapter.match, 'pong', from_jid='foo@xmpp.ru')
+        self.assertRaises(NotFound, adapter.match, 'pong', from_jid='_foo@bar')

xmppflask/tests/test_server.py

+# -*- coding: utf-8 -*-
+"""
+    XmppFlask Tests
+    ~~~~~~~~~~~~~~~
+
+    Test XmppFlask XMPPWSGI server.
+    
+    :copyright: (c) 2012 Alexander Shorin <kxepal@gmail.com>
+    :license: BSD.
+"""
+
+from xmppflask.tests.helpers import unittest
+from xmppflask import XmppFlask
+from xmppflask.server import XmppWsgiServer
+
+def counted(fun):
+    def wrapper(*args, **kwargs):
+        wrapper.times_called += 1
+        return fun(*args, **kwargs)
+    wrapper.times_called = 0
+    wrapper.__name__= fun.__name__
+    return wrapper
+
+
+class MockXmppWsgiServer(XmppWsgiServer):
+    connected = False
+    running = False
+    last_message = None
+
+    def _connect(self, *args, **kwargs):
+        self.connected = True
+
+    def serve_forever(self):
+        self.running = True
+
+    def dispatch_app_response(self, environ, response):
+        self.message_sent = False
+        self.presence_sent = False
+        self.iq_sent = False
+        return super(MockXmppWsgiServer, self).dispatch_app_response(environ, response)
+
+    def send_message(self, environ, body):
+        pass
+
+    def send_presence(self, environ, payload):
+        pass
+
+    def send_iq(self, environ, payload):
+        pass
+
+
+class XmppWsgiServerTestCase(unittest.TestCase):
+
+    def setUp(self):
+        app = XmppFlask('xmppflask.test')
+        self.server = MockXmppWsgiServer(app)
+        self.server.send_message = counted(self.server.send_message)
+        self.server.send_presence = counted(self.server.send_presence)
+        self.server.send_iq = counted(self.server.send_iq)
+
+    def test_connect(self):
+        self.server.connect('foo', 'bar')
+        assert self.server.connected
+
+    def test_dispatch_message_response(self):
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class('baz')
+
+        self.server.dispatch_app_response(environ, resp)
+
+        self.assertEqual(self.server.send_message.times_called, 1)
+
+    def test_dispatch_new_style_message_response(self):
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class([('message', 'baz')])
+
+        self.server.dispatch_app_response(environ, resp)
+
+        self.assertEqual(self.server.send_message.times_called, 1)
+
+    def test_dispatch_new_style_signle_message_response(self):
+        def pong():
+            return 'message', 'pong'
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class(pong())
+
+        self.server.dispatch_app_response(environ, resp)
+
+        self.assertEqual(self.server.send_message.times_called, 1)
+
+    def test_dispatch_presence_response(self):
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class([('presence', 'baz')])
+
+        self.server.dispatch_app_response(environ, resp)
+
+        self.assertEqual(self.server.send_presence.times_called, 1)
+
+    def test_dispatch_iq_response(self):
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class([('iq', 'baz')])
+
+        self.server.dispatch_app_response(environ, resp)
+
+        self.assertEqual(self.server.send_iq.times_called, 1)
+
+    def test_dispatch_weird_response(self):
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class([('foo', 'boo')])
+
+        self.assertRaises(ValueError,
+                          self.server.dispatch_app_response, environ, resp)
+
+    def test_dispatch_generator_of_responses(self):
+        def ping():
+            yield 'presence', '1'
+            yield 'presence', '2'
+            yield 'presence', '3'
+            yield 'PONG!'
+
+        environ = {'XMPP_JID': 'foo@bar'}
+        resp = self.server.app.response_class(ping())
+
+        self.server.dispatch_app_response(environ, resp)
+
+        self.assertEqual(self.server.send_message.times_called, 1)
+        self.assertEqual(self.server.send_presence.times_called, 3)

xmppflask/tests/test_sessions.py

+# -*- coding: utf-8 -*-
+"""
+    XmppFlask Tests
+    ~~~~~~~~~~~~~~~
+
+    Test XmppFlask sessions.
+    
+    :copyright: (c) 2012 Alexander Shorin <kxepal@gmail.com>
+    :license: BSD.
+"""
+
+from xmppflask.tests.helpers import unittest
+import xmppflask
+
+class MemorySessionTestCase(unittest.TestCase):
+
+    def setUp(self):
+        from xmppflask.sessions import MemorySessionInterface
+        self.app = xmppflask.XmppFlask(__name__)
+        self.app.session_interface = MemorySessionInterface()
+
+    def test_session_basic(self):
+        """Basic test about Session initialization."""
+        environ = {'MESSAGE': 'ping', 'XMPP_JID': 'k.bx@ya.ru'}
+        @self.app.route(u'ping')
+        def ping():
+            from xmppflask.globals import session
+            if 'times_pinged' not in session:
+                session['times_pinged'] = 0
+            session['times_pinged'] += 1
+            return u'pong'
+        self.app(environ)
+
+        session = self.app.session_interface.storage[environ['XMPP_JID']]
+        self.assertEquals(session['times_pinged'], 1)
+
+    def test_session_should_keep_values(self):
+        """Session should keep stored values for whole his lifespan."""
+        environ = {'MESSAGE': 'ping', 'XMPP_JID': 'k.bx@ya.ru'}
+        @self.app.route(u'ping')
+        def ping():
+            from xmppflask.globals import session
+            if 'times_pinged' not in session:
+                session['times_pinged'] = 0
+            session['times_pinged'] += 1
+            return u'pong'
+        self.app(environ)
+        self.app(environ)
+        self.app(environ)
+
+        session = self.app.session_interface.storage[environ['XMPP_JID']]
+        self.assertEquals(session['times_pinged'], 3)
+
+    def test_sessionless_workflow(self):
+        """Test to be sure, that nullified sessions do not break anything."""
+        environ = {'MESSAGE': 'PING', 'XMPP_JID': 'k.bx@ya.ru'}
+        app = xmppflask.XmppFlask(__name__)
+        @app.route(u'PING')
+        def ping():
+            return u'PONG'
+        rv = app(environ)
+        rv = app(environ)
+        self.assertEquals(list(rv), ['PONG'])
+
+    def test_session_expires(self):
+        environ = {'MESSAGE': 'ping', 'XMPP_JID': 'k.bx@ya.ru'}
+        @self.app.route(u'ping')
+        def ping():
+            from xmppflask.globals import session
+            if 'times_pinged' not in session:
+                session['times_pinged'] = 0
+            session['times_pinged'] += 1
+            return u'pong'
+        self.app(environ)
+
+        session = self.app.session_interface.storage[environ['XMPP_JID']]
+        session._timestamp = 1 # tweak last update timestamp
+
+        self.app(environ)
+
+        # session was expired and all his data should be erased
+        self.assertEquals(session['times_pinged'], 1)

xmppflask/tests/test_wrappers.py

+# -*- coding: utf-8 -*-
+"""
+    XmppFlask Tests
+    ~~~~~~~~~~~~~~~
+
+    Test XmppFlask Request/Response wrappers.
+    
+    :copyright: (c) 2012 Alexander Shorin <kxepal@gmail.com>
+    :license: BSD.
+"""
+
+from xmppflask.tests.helpers import unittest
+from xmppflask.wrappers import Response
+
+class ResponseTestCase(unittest.TestCase):
+
+    def test_init_by_string(self):
+        resp = Response('foo')
+        self.assertEqual(list(resp), ['foo'])
+
+    def test_init_by_iterable(self):
+        resp = Response([1, 2, 3])
+        self.assertEqual(list(resp), [1, 2, 3])
+
+        resp = Response(('foo', 'bar', 'baz'))
+        self.assertEqual(list(resp), ['foo', 'bar', 'baz'])
+
+        resp = Response(xrange(2))
+        self.assertEqual(list(resp), [0, 1])
+
+        resp = Response({'foo': 'bar'})
+        self.assertEqual(list(resp), ['foo'])
+
+    def test_single_pair_response(self):
+        resp = Response(('foo', 'bar'))
+        self.assertEqual(list(resp), [('foo', 'bar')])
+
+    def test_init_by_other(self):
+        class Foo(object):
+            def __str__(self):
+                return 'foo'
+
+        resp = Response(42)
+        self.assertEqual(list(resp), ['42'])
+
+        resp = Response(Foo())
+        self.assertEqual(list(resp), ['foo'])
+
+    def test_not_iterable_twice(self):
+        resp = Response('foo')
+        self.assertEqual(list(resp), ['foo'])
+        self.assertEqual(list(resp), [])
+
+    def test_concat_responses(self):
+        orig_resp = Response('foo')
+        resp = orig_resp(Response('bar'))
+        self.assertTrue(orig_resp is resp)
+
+        resp = resp(Response('baz'))
+
+        self.assertEqual(list(resp), ['foo', 'bar', 'baz'])
+
+    def test_lazy_iterator(self):
+        gen = (i for i in range(3))
+        resp = Response(42)(gen)
+
+        self.assertEqual(gen.next(), 0)
+        self.assertEqual(resp.next(), '42')
+        self.assertEqual(gen.next(), 1)
+        self.assertEqual(resp.next(), 2)
+        self.assertRaises(StopIteration, gen.next)

xmppflask/tests/test_xmppflask.py

+# -*- coding: utf-8 -*-
+"""
+    XmppFlask Tests
+    ~~~~~~~~~~~~~~~
+
+    Test XmppFlask itself.
+    
+    :copyright: (c) 2011 Kostyantyn Rybnikov <k.bx@ya.ru>
+    :license: BSD.
+"""
+
+import os, sys
+
+from xmppflask.tests.helpers import unittest
+import xmppflask
+
+class BasicFunctionalityTestCase(unittest.TestCase):
+    def test_xmppwsgi_simple(self):
+        environ = {'MESSAGE': 'PING', 'XMPP_JID': 'k.bx@ya.ru'}
+        app = xmppflask.XmppFlask(__name__)
+        @app.route(u'PING')
+        def ping():
+            return u'PONG'
+        rv = app(environ)
+        self.assertEquals(list(rv), ['PONG'])
+    
+    def test_xmppwsgi_more_complex(self):
+        environ = {'MESSAGE': 'PING 10 times', 'XMPP_JID': 'k.bx@ya.ru'}
+        app = xmppflask.XmppFlask(__name__)
+        @app.route(u'PING <int:times> times')
+        def ping(times):
+            return u'PONG %s times' % times
+        rv = app(environ)
+        self.assertEquals(list(rv), ['PONG 10 times'])
+
+    def test_template_rendering(self):
+        from jinja2 import TemplateNotFound
+        from xmppflask import render_template
+
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+        app = xmppflask.XmppFlask(__name__)
+        @app.route(u'ping')
+        def ping():
+            return render_template('ping.html')
+        @app.route(u'ping <user>')
+        def ping_user(user):
+            return render_template('ping.html', user=user)
+
+        @app.route(u'template_not_found')
+        def template_not_found():
+            return render_template('template_not_found.html')
+        
+        environ['MESSAGE'] = 'ping'
+        rv = app(environ)
+        self.assertEquals(list(rv), ['pong'])
+        
+        environ['MESSAGE'] = 'ping k_bx'
+        rv = app(environ)
+        self.assertEquals(list(rv), ['pong from k_bx'])
+
+    def test_template_rendering_with_message_for(self):
+        from xmppflask import render_template
+        
+        app = xmppflask.XmppFlask(__name__)
+        
+        @app.route(u'message_for_test')
+        def message_for_test():
+            return render_template('message_for_test.html')
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+        
+        environ['MESSAGE'] = 'message_for_test'
+        rv = app(environ)
+        self.assertEquals(u''.join(rv), 'message_for_test')
+
+    def test_context_simple(self):
+        app = xmppflask.XmppFlask(__name__)
+        
+        @app.route(u'ping')
+        def message_for_test():
+            from xmppflask import g
+            g.some_val = 'some val'
+            return u'pong %s' % g.some_val
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+        
+        notification_queue = []
+        environ['MESSAGE'] = 'ping'
+        rv = app(environ, notification_queue)
+        self.assertEquals(u''.join(rv), 'pong some val')
+
+    def test_notification(self):
+        from xmppflask.notification import notify
+        app = xmppflask.XmppFlask(__name__)
+        
+        @app.route(u'ping')
+        def message_for_test():
+            notify('kost-bebix@ya.ru', 'ping route')
+            return u'pong'
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+        
+        notification_queue = []
+        environ['MESSAGE'] = 'ping'
+        rv = app(environ, notification_queue)
+        self.assertEquals(u''.join(rv), 'pong')
+        self.assertEquals(notification_queue, [('kost-bebix@ya.ru', 'ping route')])
+
+    def test_before_request(self):
+        from xmppflask import g
+        app = xmppflask.XmppFlask(__name__)
+
+        must_be_changed = {'status': 'not changed'}
+        
+        @app.before_request
+        def before_request():
+            g.db = 'db obj'
+
+        @app.teardown_request
+        def teardown_request(exception):
+            must_be_changed['status'] = 'changed'
+
+        @app.route(u'ping')
+        def message_for_test():
+            self.assertEquals(g.db, 'db obj')
+            return u'pong'
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+
+        environ['MESSAGE'] = 'ping'
+        rv = app(environ)
+        self.assertEquals(u''.join(rv), 'pong')
+        self.assertEquals(must_be_changed['status'], 'changed')
+
+    def test_static_url_path(self):
+        app = xmppflask.XmppFlask(__name__)
+        @app.route(u'ping')
+        def ping():
+            return u'pong'
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+
+        environ['MESSAGE'] = 'ping'
+        self.assertEquals(app.static_url_path,
+                          None)
+
+    def test_user_exception(self):
+        class TestException(Exception): pass
+        app = xmppflask.XmppFlask(__name__)
+        app.debug = True
+            
+        @app.route(u'ping')
+        def ping():
+            raise TestException('wrong!')
+        environ = {'XMPP_JID': 'k.bx@ya.ru'}
+
+        environ['MESSAGE'] = 'ping'
+        self.assertRaises(TestException, lambda: app(environ))
+
+    def test_handle_presences(self):
+        app = xmppflask.XmppFlask(__name__)
+
+        @app.route_presence()
+        def presence():
+            return 'got it'
+
+        environ = {'XMPP_JID': 'k.bx@ya.ru', 'XMPP_EVENT': 'presence'}
+
+        rv = app(environ)
+        self.assertEquals(u''.join(rv), 'got it')
+
+    def test_handle_presences_from_specific_domain(self):
+        app = xmppflask.XmppFlask(__name__)
+
+        @app.route_presence(from_jid='.*@ya.ru', type='available')
+        def presence():
+            return 'welcome back, %s!' % environ['XMPP_JID']
+
+        environ = {'XMPP_JID': 'k.bx@ya.ru',
+                   'XMPP_EVENT': 'presence',
+                   'type': 'available'}
+
+        rv = app(environ)
+        self.assertEquals(u''.join(rv),
+                          'welcome back, %s!' % environ['XMPP_JID'])
+
+    def test_presence_mismatch_doesnt_return_message_not_understood_resp(self):
+        app = xmppflask.XmppFlask(__name__)
+
+        @app.route_presence(from_jid='.*@ya.ru', type='available')
+        def presence():
+            return 'welcome back, %s!' % environ['XMPP_JID']
+
+        environ = {'XMPP_JID': 'foo@bar',
+                   'XMPP_EVENT': 'presence',
+                   'type': 'available'}
+
+        rv = app(environ)
+        self.assertEquals(u''.join(rv), '')