Commits

Anonymous committed 9dbd86d

* Basic container/file-based Session support working in Py3K. (#72)
* Further Python 3 fixes
* tests which require webetest are now all qualified since we don't have webtest for py3k yet
* don't really need webob for session tests

Comments (0)

Files changed (13)

 * Defer running of pkg_resources to look for external cache modules
   until requested. #66
 * memcached backend uses pylibmc.ThreadMappedPool to ensure thread-local
-  usage of pylibmc when that library is in use.
+  usage of pylibmc when that library is in use. (#60)
 * memcached backend also has ``memcache_module`` string argument, allows
   direct specification of the name of which memcache backend to use.
+* Basic container/file-based Session support working in Py3K. (#72)
+* Further Python 3 fixes
 
 Release 1.5.4 (6/16/2010)
 =========================

beaker/container.py

 """Container and Namespace classes"""
-import anydbm
+
+import beaker.util as util
+if util.py3k:
+    import dbm as anydbm
+else:
+    import anydbm
 import cPickle
 import logging
 import os
 import time
 
-import beaker.util as util
 from beaker.exceptions import CreationAbortedError, MissingCacheParameter
 from beaker.synchronization import _threading, file_synchronizer, \
      mutex_synchronizer, NameLock, null_synchronizer

beaker/session.py

 from datetime import datetime, timedelta
 
 from beaker.crypto import hmac as HMAC, hmac_sha1 as SHA1, md5
-from beaker.util import pickle
+from beaker.util import pickle, py3k
 
 from beaker import crypto
 from beaker.cache import clsmap
                     raise
 
     def _create_id(self):
-        self.id = md5(
-            md5("%f%s%f%s" % (time.time(), id({}), random.random(),
-                              getpid())).hexdigest(), 
-        ).hexdigest()
+        id_str = "%f%s%f%s" % (
+                    time.time(), 
+                    id({}), 
+                    random.random(),
+                    getpid()
+                ) 
+        if py3k:
+            self.id = md5(
+                            md5(
+                                id_str.encode('ascii')
+                            ).hexdigest().encode('ascii')
+                        ).hexdigest()
+        else:
+            self.id = md5(md5(id_str).hexdigest()).hexdigest()
+
         self.is_new = True
         self.last_accessed = None
         if self.use_cookies:
 __all__  = ["ThreadLocal", "Registry", "WeakValuedRegistry", "SyncDict",
             "encoded_path", "verify_directory"]
 
+def function_named(fn, name):
+    """Return a function with a given __name__.
+
+    Will assign to __name__ and return the original function if possible on
+    the Python implementation, otherwise a new function will be constructed.
+
+    """
+    fn.__name__ = name
+    return fn
+
+def skip_if(predicate, reason=None):
+    """Skip a test if predicate is true."""
+    reason = reason or predicate.__name__
+
+    from nose import SkipTest
+    def decorate(fn):
+        fn_name = fn.__name__
+        def maybe(*args, **kw):
+            if predicate():
+                msg = "'%s' skipped: %s" % (
+                    fn_name, reason)
+                raise SkipTest(msg)
+            else:
+                return fn(*args, **kw)
+        return function_named(maybe, fn_name)
+    return decorate
 
 def verify_directory(dir):
     """verifies and creates a directory.  tries to

tests/test_cache.py

 import tempfile
 import time
 from beaker.middleware import CacheMiddleware
+from beaker import util
 from beaker.cache import Cache
-from webtest import TestApp
+from nose import SkipTest
+from beaker.util import skip_if
+import base64
+import zlib
+try:
+    from webtest import TestApp
+except ImportError:
+    TestApp = None
 
 # Tarballs of the output of:
 # >>> from beaker.cache import Cache
 pRjbXgkAAAAAAFjVyc1Idc6U1lYGgbSmL0Mjpe248+PYjY87I91x/UGeb3udAAAAAACgfh+fAAAA
 AADgr/t5/sPFTZ5cb/38D19Lzn9pRHX/zR4CtEZ/o+nfiEX9N3kI0Gr9vWl/W0z0BwAAAAAAAAAA
 AAAAAAAAqPAFyOvcKA==
-""".decode('base64').decode('zlib')
+"""
+if util.py3k:
+    dbm_cache_tar = dbm_cache_tar.encode('ascii')
+dbm_cache_tar = zlib.decompress(base64.b64decode(dbm_cache_tar))
 
 # dumbdbm format
 dumbdbm_cache_tar = """\
 kudGVrKgushNkYuVc5VM/Rups5vjY3wErJU6nD+Z7fyFNFpEjIf4AFeef7Jq22TOZnzOpLiJLz0d
 CGyE+q/scHyMk/Wv+E79G0L9hzC7JSFMpv0PN0+J4rv7xNk+iTuKh07E6aXnB9Mao/7X/fExzt//
 FecS9R8C9v/r9rP+l49tubnk+e/z/J8JjvMfAAAAAAAAAADAn70DFJAAwQ==
-""".decode('base64').decode('zlib')
+"""
+if util.py3k:
+    dumbdbm_cache_tar = dumbdbm_cache_tar.encode('ascii')
+dumbdbm_cache_tar = zlib.decompress(base64.b64decode(dumbdbm_cache_tar))
 
 def simple_app(environ, start_response):
     clear = False
     assert 'howdy' == cache.get_value('key2', createfunc=create_func)
     assert called == {}
 
+@skip_if(lambda: TestApp is None, "webtest not installed")
 def test_increment():
     app = TestApp(CacheMiddleware(simple_app))
     res = app.get('/', extra_environ={'beaker.type':type, 'beaker.clear':True})
     res = app.get('/')
     assert 'current value is: 3' in res
 
+@skip_if(lambda: TestApp is None, "webtest not installed")
 def test_cache_manager():
     app = TestApp(CacheMiddleware(cache_manager_app))
     res = app.get('/')
                 continue
             dir = tempfile.mkdtemp()
             fd, name = tempfile.mkstemp(dir=dir)
-            fp = os.fdopen(fd, 'w')
+            fp = os.fdopen(fd, 'wb')
             fp.write(tar)
             fp.close()
             tar = tarfile.open(name)

tests/test_database.py

 # coding: utf-8
-from beaker.cache import clsmap, Cache
+from beaker.cache import clsmap, Cache, util
 from beaker.exceptions import InvalidCacheBackendError
 from beaker.middleware import CacheMiddleware
 from nose import SkipTest
-from webtest import TestApp
+
+try:
+    from webtest import TestApp
+except ImportError:
+    TestApp = None
 
 
 try:
     cache.remove_value(u'hiŏ')
     assert u'hiŏ' not in cache
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_increment():
     app = TestApp(CacheMiddleware(simple_app))
     res = app.get('/', extra_environ={'beaker.clear':True})
     res = app.get('/')
     assert 'current value is: 3' in res
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_cache_manager():
     app = TestApp(CacheMiddleware(cache_manager_app))
     res = app.get('/')

tests/test_domain_setting.py

 import os
 
 from beaker.middleware import SessionMiddleware
-from webtest import TestApp
+from nose import SkipTest
+try:
+    from webtest import TestApp
+except ImportError:
+    raise SkipTest("webtest not installed")
 
 def teardown():
     import shutil

tests/test_increment.py

 import os
 
 from beaker.middleware import SessionMiddleware
-from webtest import TestApp
+from nose import SkipTest
+try:
+    from webtest import TestApp
+except ImportError:
+    raise SkipTest("webtest not installed")
 
 
 def teardown():

tests/test_memcached.py

 # coding: utf-8
 import os
 
-from beaker.cache import clsmap, Cache
+from beaker.cache import clsmap, Cache, util
 from beaker.middleware import CacheMiddleware, SessionMiddleware
 from beaker.exceptions import InvalidCacheBackendError
 from nose import SkipTest
-from webtest import TestApp
 import unittest
 
 try:
+    from webtest import TestApp
+except ImportError:
+    TestApp = None
+
+try:
     from beaker.ext import memcached
     client = memcached._load_client()
 except InvalidCacheBackendError:
         yield "test_key wasn't cleared, is: %s\n" % \
             cm.get_cache('test')['test_key']
 
-
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_session():
     app = TestApp(SessionMiddleware(simple_session_app, data_dir='./cache', type='ext:memcached', url=mc_url))
     res = app.get('/')
     assert 'current value is: 3' in res
 
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_session_invalid():
     app = TestApp(SessionMiddleware(simple_session_app, data_dir='./cache', type='ext:memcached', url=mc_url))
     res = app.get('/invalid', headers=dict(Cookie='beaker.session.id=df7324911e246b70b5781c3c58328442; Path=/'))
     assert cache.has_key("hasspace")
     assert 42 == cache.get_value("hasspace")
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_increment():
     app = TestApp(CacheMiddleware(simple_app))
     res = app.get('/', extra_environ={'beaker.clear':True})
     res = app.get('/')
     assert 'current value is: 3' in res
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_cache_manager():
     app = TestApp(CacheMiddleware(cache_manager_app))
     res = app.get('/')
     assert 'test_key is: test value' in res
     assert 'test_key cleared' in res
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_store_none():
     app = TestApp(CacheMiddleware(using_none_app))
     res = app.get('/', extra_environ={'beaker.clear':True})

tests/test_sqla.py

 # coding: utf-8
-from beaker.cache import clsmap, Cache
+from beaker.cache import clsmap, Cache, util
 from beaker.exceptions import InvalidCacheBackendError
 from beaker.middleware import CacheMiddleware
 from nose import SkipTest
-from webtest import TestApp
+
+try:
+    from webtest import TestApp
+except ImportError:
+    TestApp = None
 
 try:
     clsmap['ext:sqla']._init_dependencies()
     cache.remove_value(u'hiŏ')
     assert u'hiŏ' not in cache
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_increment():
     app = TestApp(CacheMiddleware(simple_app))
     res = app.get('/', extra_environ={'beaker.clear': True})
     res = app.get('/')
     assert 'current value is: 3' in res
 
+@util.skip_if(lambda: TestApp is None, "webtest not installed")
 def test_cache_manager():
     app = TestApp(CacheMiddleware(cache_manager_app))
     res = app.get('/')