[Repoze-checkins] r1113 - repoze.accelerator/trunk/repoze/accelerator
Tres Seaver
tseaver at palladion.com
Fri Jun 20 21:11:08 EDT 2008
Author: Tres Seaver <tseaver at palladion.com>
Date: Fri Jun 20 21:11:08 2008
New Revision: 1113
Log:
Assert / test API conformance.
Modified:
repoze.accelerator/trunk/repoze/accelerator/middleware.py
repoze.accelerator/trunk/repoze/accelerator/tests.py
Modified: repoze.accelerator/trunk/repoze/accelerator/middleware.py
==============================================================================
--- repoze.accelerator/trunk/repoze/accelerator/middleware.py (original)
+++ repoze.accelerator/trunk/repoze/accelerator/middleware.py Fri Jun 20 21:11:08 2008
@@ -5,6 +5,14 @@
from paste.request import parse_headers
from paste.response import header_value
+from repoze.accelerator.interfaces import IChunkHandler
+from repoze.accelerator.interfaces import IPolicy
+from repoze.accelerator.interfaces import IPolicyFactory
+from repoze.accelerator.interfaces import IStorage
+from repoze.accelerator.interfaces import IStorageFactory
+from repoze.accelerator.interfaces import implements
+from repoze.accelerator.interfaces import provides
+
class Accelerator:
def __init__(self, app, policy):
self.app = app
@@ -50,23 +58,41 @@
raise StopIteration
+
+
class NullPolicy:
+ """ Pass-through, caches nothing.
+ """
+ implements(IPolicy)
+ provides(IPolicyFactory)
+
+ def __init__(self, storage, config=None):
+ pass
+
def fetch(self, environ):
return None
+
def store(self, status, headers, environ):
pass
class NaivePolicy:
- def __init__(self, storage):
+ implements(IPolicy)
+ provides(IPolicyFactory)
+
+ def __init__(self, storage, config=None):
self.storage = storage
+ if config is None:
+ config = {}
+ allowed_methods = config.get('policy.allowed_methods', 'GET')
+ self.allowed_methods = filter(None, allowed_methods.split())
def _minimalCacheOK(self, headers, environ):
- if environ.get('REQUEST_METHOD', 'GET') != 'GET':
+ if environ.get('REQUEST_METHOD', 'GET') not in self.allowed_methods:
return False
for nocache in ('Pragma', 'Cache-Control'):
value = header_value(headers, nocache)
if value and 'no-cache' in value.lower():
- return
+ return False
return True
def store(self, status, headers, environ):
@@ -97,7 +123,10 @@
return self.storage.fetch(url)
class RAMStorage:
- def __init__(self, lock=threading.Lock()):
+ implements(IStorage)
+ provides(IStorageFactory)
+
+ def __init__(self, lock=threading.Lock(), config=None):
self.data = {}
self.lock = lock
@@ -106,6 +135,7 @@
storage = self
class SimpleHandler:
+ implements(IChunkHandler)
def write(self, chunk):
result.append(chunk)
@@ -122,7 +152,12 @@
return self.data.get(url)
def main(app, global_conf, **local_conf):
- storage = RAMStorage()
- policy = NaivePolicy(storage)
+
+ storage_factory = local_conf.get('storage', RAMStorage)
+ storage = storage_factory(config=local_conf)
+
+ policy_factory = local_conf.get('policy', NaivePolicy)
+ policy = policy_factory(storage, config=local_conf)
+
return Accelerator(app, policy)
Modified: repoze.accelerator/trunk/repoze/accelerator/tests.py
==============================================================================
--- repoze.accelerator/trunk/repoze/accelerator/tests.py (original)
+++ repoze.accelerator/trunk/repoze/accelerator/tests.py Fri Jun 20 21:11:08 2008
@@ -1,14 +1,37 @@
import unittest
+_MARKER = object()
+
class TestRAMStorage(unittest.TestCase):
def _getTargetClass(self):
from repoze.accelerator.middleware import RAMStorage
return RAMStorage
- def _makeOne(self, lock):
+ def _makeOne(self, lock, config=_MARKER):
klass = self._getTargetClass()
- return klass(lock)
-
+ if config is _MARKER:
+ return klass(lock)
+ return klass(lock, config=config)
+
+ def test_class_conforms_to_IStorage(self):
+ from repoze.accelerator.interfaces import verifyClass
+ from repoze.accelerator.interfaces import IStorage
+ verifyClass(IStorage, self._getTargetClass())
+
+ def test_class_provides_IStorageFactory(self):
+ from repoze.accelerator.interfaces import verifyObject
+ from repoze.accelerator.interfaces import IStorageFactory
+ verifyObject(IStorageFactory, self._getTargetClass())
+
+ def test_instance_conforms_to_IStorage(self):
+ from repoze.accelerator.interfaces import verifyObject
+ from repoze.accelerator.interfaces import IStorage
+ verifyObject(IStorage, self._makeOne(DummyLock()))
+
+ def test_ctor_accepts_config(self):
+ lock = DummyLock()
+ storage = self._makeOne(lock, config={})
+
def test_store_nonexistent(self):
lock = DummyLock()
storage = self._makeOne(lock)
@@ -42,7 +65,7 @@
lock = DummyLock()
storage = self._makeOne(lock)
self.assertEqual(storage.fetch('url'), None)
-
+
def test_fetch_existing(self):
lock = DummyLock()
storage = self._makeOne(lock)
@@ -54,9 +77,11 @@
from repoze.accelerator.middleware import NaivePolicy
return NaivePolicy
- def _makeOne(self, storage):
+ def _makeOne(self, storage, config=_MARKER):
klass = self._getTargetClass()
- return klass(storage)
+ if config is _MARKER:
+ return klass(storage)
+ return klass(storage, config=config)
def _makeEnviron(self):
return {
@@ -66,6 +91,25 @@
'REQUEST_METHOD': 'GET',
}
+ def test_class_conforms_to_IPolicy(self):
+ from repoze.accelerator.interfaces import verifyClass
+ from repoze.accelerator.interfaces import IPolicy
+ verifyClass(IPolicy, self._getTargetClass())
+
+ def test_class_provides_IPolicyFactory(self):
+ from repoze.accelerator.interfaces import verifyObject
+ from repoze.accelerator.interfaces import IPolicyFactory
+ verifyObject(IPolicyFactory, self._getTargetClass())
+
+ def test_instance_conforms_to_IPolicy(self):
+ from repoze.accelerator.interfaces import verifyObject
+ from repoze.accelerator.interfaces import IPolicy
+ verifyObject(IPolicy, self._makeOne(DummyStorage()))
+
+ def test_ctor_accepts_config(self):
+ storage = DummyStorage()
+ policy = self._makeOne(storage, config={})
+
def test_store_not_cacheable_post_request_method(self):
storage = DummyStorage()
policy = self._makeOne(storage)
@@ -73,7 +117,7 @@
environ['REQUEST_METHOD'] = 'POST'
result = policy.store('200 OK', [], environ)
self.assertEqual(result, None)
-
+
def test_store_not_cacheable_pragma_no_cache(self):
storage = DummyStorage()
policy = self._makeOne(storage)
@@ -105,6 +149,18 @@
result = policy.store('500 Error', [], environ)
self.assertEqual(result, None)
+ def test_store_allowed_request_method_cacheable(self):
+ storage = DummyStorage()
+ policy = self._makeOne(storage,
+ config={'policy.allowed_methods': 'GET FOO'})
+ environ = self._makeEnviron()
+ environ['REQUEST_METHOD'] = 'FOO'
+ result = policy.store('200 OK', [('header1', 'value1')], environ)
+ self.assertEqual(result, False)
+ self.assertEqual(storage.url, 'http://example.com')
+ self.assertEqual(storage.status, '200 OK')
+ self.assertEqual(storage.outheaders, [('header1', 'value1')])
+
def test_store_no_request_method_cacheable(self):
storage = DummyStorage()
policy = self._makeOne(storage)
@@ -133,7 +189,7 @@
environ['REQUEST_METHOD'] = 'POST'
result = policy.fetch(environ)
self.failIfEqual(result, 123)
-
+
def test_fetch_fails_pragma_no_cache(self):
storage = DummyStorage(result=123)
policy = self._makeOne(storage)
@@ -199,7 +255,7 @@
start_response = DummyStartResponse()
result = accelerator(environ, start_response)
self.assertRaises(RuntimeError, list, result)
-
+
def test_call_cantstore(self):
app = DummyApp(headers=[('a', 'b')])
policy = DummyPolicy(result=None)
@@ -227,7 +283,7 @@
self.assertEqual(start_response.exc_info, None)
self.assertEqual(policy.handler.chunks, ['hello', 'world'])
self.assertEqual(policy.handler.closed, True)
-
+
class DummyHandler:
def __init__(self):
@@ -249,7 +305,7 @@
self.status = status
self.headers = headers
self.call_start_response = True
-
+
def __call__(self, environ, start_response):
self.environ = environ
if self.call_start_response:
@@ -271,7 +327,7 @@
def __init__(self, result=None, writer=False):
self.result = result
self.writer = writer
-
+
def store(self, url, status, outheaders):
self.url = url
self.status = status
@@ -291,4 +347,4 @@
def release(self):
self.released += 1
-
+
More information about the Repoze-checkins
mailing list