[Repoze-dev] Repoze CVS: __init__.py test_helper.py test_spike.py

[Chris McDonough]

Update of /home/repoze/cvs/repoze.zope2/repoze/zope2/tests
In directory laguna.palladion.com:/tmp/cvs-serv17144/repoze/zope2/tests

Added Files:
	__init__.py test_helper.py test_spike.py 
Log Message:
Add obob_helper and tests.


--- NEW FILE: test_helper.py ---
import unittest

class TestTopLevelFuncs(unittest.TestCase):
    def test_get_root(self):
        from repoze.zope2.obob_helper import get_root
        env = {}
        app = get_root(env, dummy_configure)
        self.failUnless(isinstance(app, DummyApplication))

    def test_get_connection(self):
        from repoze.zope2.obob_helper import get_connection
        env = {}
        conn = get_connection(env, dummy_configure)
        self.assertEqual(conn.debuginfo, env)
        self.failUnless(env['tm.cleanup'].has_key('closeconn'), env)

    def test_makeRequest(self):
        from repoze.zope2 import makeRequest
        environ = _makeEnviron()
        request = makeRequest(environ)
        self.assertEqual(request.stdin, environ['wsgi.input'])
        self.assertNotEqual(request.response, None)

class CloserTests(unittest.TestCase):
    def test_Closer(self):
        from repoze.zope2 import Closer
        jar = DummyConnection('whatever')
        a = Closer(jar)
        del a
        self.assertEqual(jar.closed, True)
    
class TestZope2ObobHelper(unittest.TestCase):
    def tearDown(self):
        from zope.security.management import endInteraction
        from zope.security.management import queryInteraction
        if queryInteraction() is not None:
            endInteraction()
        
    def _getTargetClass(self):
        from repoze.zope2.obob_helper import Zope2ObobHelper
        return Zope2ObobHelper

    def _makeOne(self, env=None):
        if env is None:
            env = _makeEnviron()
        return self._getTargetClass()(env)

    def test_ctor(self):
        environ = {}
        helper = self._makeOne(environ)
        self.assertEqual(helper.environ, environ)
        self.assertEqual(helper.request, None)

    def test_setup(self):
        helper = self._makeOne()
        helper.setup()
        self.assertEqual(helper.request['URL'], 'http://localhost:8080')

    def test_invoke(self):
        published = DummyPublishedObject()
        helper = self._makeOne()
        from repoze.zope2.obob_helper import makeRequest
        helper.request = makeRequest(_makeEnviron())
        helper.invoke(published)
        self.assertEqual(helper.request['URL'], 'http://localhost:8080')

    def test_teardown(self):
        from zope.security.management import queryInteraction
        from zope.security.management import newInteraction
        helper = self._makeOne()
        newInteraction()
        interaction = queryInteraction()
        helper.teardown()
        self.assertEqual(queryInteraction(), None)
            
class DummyPublishedObject:
    def __call__(self, request=None):
        self.request = request

class DummyApplication:
    def __init__(self, conn):
        self.conn = conn

class DummyConnection:
    closed = False
    debuginfo = None
    def __init__(self, db):
        self.db = db

    def close(self):
        self.closed = True

    def setDebugInfo(self, info):
        self.debuginfo = info

    def root(self):
        return {'Application':DummyApplication(self)}
        
class DummyDB:
    def open(self):
        return DummyConnection(self)

def dummy_configure():
    return DummyDB()

def _makeEnviron():
    from StringIO import StringIO
    stdin = StringIO()
    environ = {}
    environ['wsgi.input'] = stdin
    environ['SERVER_NAME'] = 'localhost'
    environ['SERVER_PORT'] = '8080'
    environ['PATH_INFO'] = '/foo'
    return environ

--- NEW FILE: test_spike.py ---
import os
import sys
import shutil
import tempfile
import unittest

class TestPublisher(unittest.TestCase):
    def _getTargetClass(self):
        from repoze.zope2 import Publisher
        return Publisher

    def _makeOne(self, result):
        db = DummyDB(result)
        appname = db.conn._root.appname
        debug = True
        realm = 'realm'
        publisher = self._getTargetClass()(db, appname, debug, realm)
        return publisher

    def test_call_return_basestring(self):
        publisher = self._makeOne('hello!')
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            result = publisher(environ, start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()
        self.assertEqual(result[0], 'hello!')
        self.assertEqual(L, [('200 OK', [('Content-Length', '6'),
                                         ('X-X', 'x')])])

    def test_call_return_iterable(self):
        publisher = self._makeOne(['hello!'])
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            result = publisher(environ, start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()
        self.assertEqual(result[0], 'hello!')
        self.assertEqual(L, [('200 OK', [('X-X', 'x')])])

    def test_call_return_None(self):
        publisher = self._makeOne(None)
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            result = publisher(environ, start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()
        self.assertEqual(result[0], '')
        self.assertEqual(L, [('200 OK', [('Content-Length', '0'),
                                         ('X-X', 'x')])]
                              )

    def test_call_return_response(self):
        publisher = self._makeOne('A_RESPONSE')
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            result = publisher(environ, start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()
        self.assertEqual(result[0], 'Response Body')
        self.assertEqual(L, [('200 OK', [('Content-Length', '13'),
                                         ('X-X', 'x')])])
    def test_call_raises_unhandled(self):
        class Unhandled(Exception):
            pass
        publisher = self._makeOne(Unhandled())
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            self.assertRaises(Unhandled, publisher, environ, start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()

    def test_call_handles_unauthorized(self):
        from zExceptions import Unauthorized
        publisher = self._makeOne(Unauthorized())
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            from paste.httpexceptions import HTTPUnauthorized
            self.assertRaises(HTTPUnauthorized, publisher, environ,
                              start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()

    def test_call_handles_redirect(self):
        from zExceptions import Redirect
        publisher = self._makeOne(Redirect('foo'))
        environ = _makeEnviron()
        L = []
        def start_response(status, headers):
            L.append((status, headers))
        try:
            from paste.httpexceptions import HTTPFound
            self.assertRaises(HTTPFound, publisher, environ,
                              start_response)
            db = publisher.db
            self.assertEqual(environ['tm.cleanup']['closeconn']._jar, db.conn)
        finally:
            environ['tm.cleanup'].clear()

class TestPublisherFactory(unittest.TestCase):
    def _makeOne(self, *arg, **kw):
        from repoze.zope2 import make_publisher
        return make_publisher(*arg, **kw)
        
    def dont_test_factory(self):
        # XXX this test takes forever because we init a filestorage
        # reenable when we allow for configuration
        tdir = tempfile.mkdtemp()
        try:
            global_conf = {'here':tdir}
            publisher = self._makeOne(global_conf, admin_username='fred')
            db = publisher.db
            conn = db.open()
            from OFS.Application import Application
            from OFS.DTMLMethod import DTMLMethod
            app = conn.root()[publisher.appname]
            self.failUnless(isinstance(app, Application))
            self.failUnless(isinstance(app.index_html, DTMLMethod))
            self.failUnless(app.acl_users.data.has_key('fred'))
        finally:
            shutil.rmtree(tdir)
            if locals().has_key('conn'):
                conn.close()

        self.assertEqual(publisher.appname, 'Application')
        self.assertEqual(publisher.debug_mode, False)
        self.assertEqual(publisher.realm, 'Zope')

class TestUtilFunctions(unittest.TestCase):

    def test_makeRequest(self):
        from repoze.zope2 import makeRequest
        environ = _makeEnviron()
        request = makeRequest(environ)
        self.assertEqual(request.stdin, environ['wsgi.input'])
        self.assertNotEqual(request.response, None)

    def test_Closer(self):
        from repoze.zope2 import Closer
        jar = DummyConnection('whatever')
        a = Closer(jar)
        del a
        self.assertEqual(jar.closed, True)

def _makeEnviron():
        from StringIO import StringIO
        stdin = StringIO()
        environ = {}
        environ['wsgi.input'] = stdin
        environ['SERVER_NAME'] = 'localhost'
        environ['SERVER_PORT'] = '8080'
        environ['PATH_INFO'] = '/foo'
        return environ

class DummyContent:
    """ """
    index_html = None
    def __init__(self, result):
        self.result = result
    def __call__(self, REQUEST):
        REQUEST.response.setHeader('x-x', 'x')
        if isinstance(self.result, Exception):
            raise self.result
        if self.result == 'A_RESPONSE':
            REQUEST.response.body = 'Response Body'
            return REQUEST.response
        return self.result
    def __getitem__(self, name):
        return self

class DummyApplication:
    def __init__(self, result):
        self.result = result
    def __bobo_traverse__(self, request, name):
        return DummyContent(self.result)

class DummyRoot:
    appname = 'thisone'
    def __init__(self, result):
        self.app = DummyApplication(result)
    def __getitem__(self, name):
        return self.app
        
class DummyConnection:
    closed = False
    def __init__(self, result):
        self._root = DummyRoot(result)
    def root(self):
        return self._root
    def close(self):
        self.closed = True
    def setDebugInfo(self, *info):
        self.info = info

class DummyDB:
    def __init__(self, result):
        self.conn = DummyConnection(result)
    def open(self):
        return self.conn


--- NEW FILE: __init__.py ---
# tests module

_______________________________________________
Repoze-dev mailing list
Repoze-dev at lists.repoze.org
http://lists.repoze.org/mailman/listinfo/repoze-dev