summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorChris McDonough <chrism@agendaless.com>2008-07-04 17:50:59 +0000
committerChris McDonough <chrism@agendaless.com>2008-07-04 17:50:59 +0000
commit8b2c67309a9a3f9cf709bf3003dc3275f17c709e (patch)
tree1b20d5b000e4333cf50b2c2305dbb03e8b312144
parentd6cb3c1b68e752922a149d37a8d208b7d01dbe14 (diff)
downloadpyramid-8b2c67309a9a3f9cf709bf3003dc3275f17c709e.tar.gz
pyramid-8b2c67309a9a3f9cf709bf3003dc3275f17c709e.tar.bz2
pyramid-8b2c67309a9a3f9cf709bf3003dc3275f17c709e.zip
Don't depend on ZODB; shuffle policy responsibilities around a little.
-rw-r--r--repoze/bfg/interfaces.py2
-rw-r--r--repoze/bfg/policy.py45
-rw-r--r--repoze/bfg/router.py6
-rw-r--r--repoze/bfg/tests/test_policy.py94
-rw-r--r--repoze/bfg/tests/test_zodb.py95
-rw-r--r--repoze/bfg/zodb.py39
-rw-r--r--setup.py4
7 files changed, 146 insertions, 139 deletions
diff --git a/repoze/bfg/interfaces.py b/repoze/bfg/interfaces.py
index 5e1a605ec..711ea44cc 100644
--- a/repoze/bfg/interfaces.py
+++ b/repoze/bfg/interfaces.py
@@ -5,6 +5,6 @@ class IWSGIApplication(Interface):
""" Represent a WSGI (PEP 333) application """
class IPolicy(Interface):
- def __call__(environ):
+ def __call__(root, environ):
""" Return a tuple in the form (context, name, subpath) """
diff --git a/repoze/bfg/policy.py b/repoze/bfg/policy.py
new file mode 100644
index 000000000..a25da9efd
--- /dev/null
+++ b/repoze/bfg/policy.py
@@ -0,0 +1,45 @@
+import urllib
+
+from zope.interface import implements
+
+from repoze.bfg.interfaces import IPolicy
+
+def split_path(path):
+ if path.startswith('/'):
+ path = path[1:]
+ if path.endswith('/'):
+ path = path[:-1]
+ clean=[]
+ for item in path.split('/'):
+ item = urllib.unquote(item) # deal with spaces in path segment
+ if not item or item=='.':
+ continue
+ elif item == '..':
+ del clean[-1]
+ else:
+ clean.append(item)
+ return clean
+
+class NaivePolicy:
+
+ implements(IPolicy)
+
+ def __call__(self, environ, root):
+ path = split_path(environ['PATH_INFO'])
+
+ ob = root
+ name = ''
+ while path:
+ element = pop(path)
+ try:
+ ob = ob[element]
+ except KeyError:
+ if path:
+ name = pop(path)
+ break
+
+ return ob, name, path
+
+def pop(path):
+ return path.pop(0)
+
diff --git a/repoze/bfg/router.py b/repoze/bfg/router.py
index 049430723..12d3d4ac4 100644
--- a/repoze/bfg/router.py
+++ b/repoze/bfg/router.py
@@ -2,12 +2,14 @@ from zope.component import getAdapter
from repoze.bfg.interfaces import IWSGIApplication
class Router:
- def __init__(self, app, policy):
+ def __init__(self, app, root_finder, policy):
self.app = app
+ self.root_finder = root_finder
self.policy = policy
def __call__(self, environ, start_response):
- context, name, subpath = self.policy(environ)
+ root = self.root_finder(environ)
+ context, name, subpath = self.policy(root, environ)
app = getAdapter(context, IWSGIApplication, name)
environ['repoze.bfg.context'] = context
environ['repoze.bfg.subpath'] = subpath
diff --git a/repoze/bfg/tests/test_policy.py b/repoze/bfg/tests/test_policy.py
new file mode 100644
index 000000000..e8752d1a1
--- /dev/null
+++ b/repoze/bfg/tests/test_policy.py
@@ -0,0 +1,94 @@
+import unittest
+
+class SplitPathTests(unittest.TestCase):
+ def _getFUT(self):
+ from repoze.bfg.policy import split_path
+ return split_path
+
+ def test_cleanPath_path_startswith_endswith(self):
+ f = self._getFUT()
+ self.assertEqual(f('/foo/'), ['foo'])
+
+ def test_cleanPath_empty_elements(self):
+ f = self._getFUT()
+ self.assertEqual(f('foo///'), ['foo'])
+
+ def test_cleanPath_onedot(self):
+ f = self._getFUT()
+ self.assertEqual(f('foo/./bar'), ['foo', 'bar'])
+
+ def test_cleanPath_twodots(self):
+ f = self._getFUT()
+ self.assertEqual(f('foo/../bar'), ['bar'])
+
+ def test_cleanPath_element_urllquoted(self):
+ f = self._getFUT()
+ self.assertEqual(f('/foo/space%20thing/bar'), ['foo', 'space thing',
+ 'bar'])
+
+class NaivePolicyTests(unittest.TestCase):
+ def _getTargetClass(self):
+ from repoze.bfg.policy import NaivePolicy
+ return NaivePolicy
+
+ def _makeOne(self, *arg, **kw):
+ klass = self._getTargetClass()
+ return klass(*arg, **kw)
+
+ def test_class_conforms_to_IPolicy(self):
+ from zope.interface.verify import verifyClass
+ from repoze.bfg.interfaces import IPolicy
+ verifyClass(IPolicy, self._getTargetClass())
+
+ def test_instance_conforms_to_IPolicy(self):
+ from zope.interface.verify import verifyObject
+ from repoze.bfg.interfaces import IPolicy
+ verifyObject(IPolicy, self._makeOne())
+
+ def test_call_nonkeyerror_raises(self):
+ policy = self._makeOne()
+ environ = {'PATH_INFO':'/foo'}
+ root = None
+ self.assertRaises(TypeError, policy, environ, root)
+
+ def test_call_withconn_getitem_emptypath_nosubpath(self):
+ policy = self._makeOne()
+ context = DummyContext()
+ environ = {'PATH_INFO':''}
+ root = context
+ ctx, name, subpath = policy(environ, root)
+ self.assertEqual(context, ctx)
+ self.assertEqual(name, '')
+ self.assertEqual(subpath, [])
+
+ def test_call_withconn_getitem_withpath_nosubpath(self):
+ policy = self._makeOne()
+ context = DummyContext()
+ context2 = DummyContext(context)
+ environ = {'PATH_INFO':'/foo/bar'}
+ root = context
+ ctx, name, subpath = policy(environ, root)
+ self.assertEqual(context, ctx)
+ self.assertEqual(name, 'bar')
+ self.assertEqual(subpath, [])
+
+ def test_call_withconn_getitem_withpath_withsubpath(self):
+ policy = self._makeOne()
+ context = DummyContext()
+ context2 = DummyContext(context)
+ environ = {'PATH_INFO':'/foo/bar/baz/buz'}
+ root = context
+ ctx, name, subpath = policy(environ, root)
+ self.assertEqual(context, ctx)
+ self.assertEqual(name, 'bar')
+ self.assertEqual(subpath, ['baz', 'buz'])
+
+class DummyContext:
+ def __init__(self, next=None):
+ self.next = next
+
+ def __getitem__(self, name):
+ if self.next is None:
+ raise KeyError, name
+ return self.next
+
diff --git a/repoze/bfg/tests/test_zodb.py b/repoze/bfg/tests/test_zodb.py
deleted file mode 100644
index d28622953..000000000
--- a/repoze/bfg/tests/test_zodb.py
+++ /dev/null
@@ -1,95 +0,0 @@
-import unittest
-
-class ZODBGetitemPolicyTests(unittest.TestCase):
- def _getTargetClass(self):
- from repoze.bfg.zodb import ZODBGetitemPolicy
- return ZODBGetitemPolicy
-
- def _makeOne(self, *arg, **kw):
- klass = self._getTargetClass()
- return klass(*arg, **kw)
-
- def test_class_conforms_to_IPolicy(self):
- from zope.interface.verify import verifyClass
- from repoze.bfg.interfaces import IPolicy
- verifyClass(IPolicy, self._getTargetClass())
-
- def test_instance_conforms_to_IPolicy(self):
- from zope.interface.verify import verifyObject
- from repoze.bfg.interfaces import IPolicy
- verifyObject(IPolicy, self._makeOne('dbname'))
-
- def test_call_noconn(self):
- mw = self._makeOne('dbname')
- environ = {}
- self.assertRaises(ValueError, mw, environ)
-
- def test_call_withconn_attributeerror(self):
- mw = self._makeOne('dbname')
- environ = {'repoze.zodbconn.dbname': DummyConnection(DummyNoGetitem()),
- 'PATH_INFO':''}
- self.assertRaises(AttributeError, mw, environ)
-
- def test_call_withconn_getitem_emptypath_nosubpath(self):
- mw = self._makeOne('dbname')
- context = DummyContext()
- environ = {'repoze.zodbconn.dbname': DummyConnection(context),
- 'PATH_INFO':''}
- ctx, name, subpath = mw(environ)
- self.assertEqual(context, ctx)
- self.assertEqual(name, '')
- self.assertEqual(subpath, [])
-
- def test_call_withconn_getitem_withpath_nosubpath(self):
- mw = self._makeOne('dbname')
- context = DummyContext()
- context2 = DummyContext(context)
- environ = {'repoze.zodbconn.dbname': DummyConnection(context2),
- 'PATH_INFO':'/foo/bar'}
- ctx, name, subpath = mw(environ)
- self.assertEqual(context, ctx)
- self.assertEqual(name, 'bar')
- self.assertEqual(subpath, [])
-
- def test_call_withconn_getitem_withpath_withsubpath(self):
- mw = self._makeOne('dbname')
- context = DummyContext()
- context2 = DummyContext(context)
- environ = {'repoze.zodbconn.dbname': DummyConnection(context2),
- 'PATH_INFO':'/foo/bar/baz/buz'}
- ctx, name, subpath = mw(environ)
- self.assertEqual(context, ctx)
- self.assertEqual(name, 'bar')
- self.assertEqual(subpath, ['baz', 'buz'])
-
- def test_call_withprefix(self):
- mw = self._makeOne('dbname', ['a', 'b'])
- context = DummyContext()
- context2 = DummyContext(context)
- context3 = DummyContext(context2)
- environ = {'repoze.zodbconn.dbname': DummyConnection(context3),
- 'PATH_INFO':'/foo/bar/baz/buz'}
- ctx, name, subpath = mw(environ)
- self.assertEqual(context, ctx)
- self.assertEqual(name, 'foo')
- self.assertEqual(subpath, ['bar', 'baz', 'buz'])
-
-class DummyNoGetitem:
- pass
-
-class DummyContext:
- def __init__(self, next=None):
- self.next = next
-
- def __getitem__(self, name):
- if self.next is None:
- raise KeyError, name
- return self.next
-
-class DummyConnection:
- def __init__(self, result):
- self.result = result
- def open(self):
- return self.result
-
-
diff --git a/repoze/bfg/zodb.py b/repoze/bfg/zodb.py
deleted file mode 100644
index c13d5fc58..000000000
--- a/repoze/bfg/zodb.py
+++ /dev/null
@@ -1,39 +0,0 @@
-from zope.interface import implements
-
-from repoze.zodbconn.middleware import get_conn
-
-from repoze.bfg.interfaces import IPolicy
-
-class ZODBGetitemPolicy:
-
- implements(IPolicy)
-
- def __init__(self, dbname, prefix=()):
- self.dbname = dbname
- self.prefix = prefix
- self.get_conn = get_conn
-
- def __call__(self, environ):
- conn = self.get_conn(environ, self.dbname)
- if conn is None:
- raise ValueError('No such connection %s' % self.dbname)
-
- path = environ['PATH_INFO'].split('/')
- path = list(self.prefix) + path
-
- ob = conn.open()
-
- name = ''
- while path:
- element = path.pop(0)
- try:
- ob = ob[element]
- except AttributeError, what:
- raise AttributeError(str(what[0]) + ' (element: '+element+')')
- except KeyError:
- if path:
- name = path.pop(0)
- break
-
- return ob, name, path
-
diff --git a/setup.py b/setup.py
index f01afe3ec..d53efb656 100644
--- a/setup.py
+++ b/setup.py
@@ -48,8 +48,8 @@ setup(name='repoze.bfg',
include_package_data=True,
namespace_packages=['repoze', 'repoze.bfg'],
zip_safe=False,
- tests_require = ['zope.interface', 'repoze.zodbconn'],
- install_requires=['zope.interface', 'repoze.zodbconn'],
+ tests_require = ['zope.interface'],
+ install_requires=['zope.interface'],
test_suite="repoze.bfg.tests",
entry_points = """\
"""