774 lines
		
	
	
		
			24 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			774 lines
		
	
	
		
			24 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
from pecan import make_app, expose, request, redirect
 | 
						|
from pecan.core import state
 | 
						|
from pecan.hooks import PecanHook, TransactionHook, HookController
 | 
						|
from pecan.decorators import transactional
 | 
						|
from formencode import Schema, validators
 | 
						|
from webtest import TestApp
 | 
						|
 | 
						|
 | 
						|
class TestHooks(object):
 | 
						|
    
 | 
						|
    def test_basic_single_hook(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
        
 | 
						|
        class SimpleHook(PecanHook):
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route')
 | 
						|
            
 | 
						|
            def before(self, state):
 | 
						|
                run_hook.append('before')
 | 
						|
 | 
						|
            def after(self, state):
 | 
						|
                run_hook.append('after')
 | 
						|
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error')
 | 
						|
        
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[SimpleHook()]))
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 4
 | 
						|
        assert run_hook[0] == 'on_route'
 | 
						|
        assert run_hook[1] == 'before'
 | 
						|
        assert run_hook[2] == 'inside'
 | 
						|
        assert run_hook[3] == 'after'
 | 
						|
    
 | 
						|
    def test_basic_multi_hook(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
        
 | 
						|
        class SimpleHook(PecanHook):
 | 
						|
            def __init__(self, id):
 | 
						|
                self.id = str(id)
 | 
						|
            
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route'+self.id)
 | 
						|
            
 | 
						|
            def before(self, state):
 | 
						|
                run_hook.append('before'+self.id)
 | 
						|
 | 
						|
            def after(self, state):
 | 
						|
                run_hook.append('after'+self.id)
 | 
						|
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error'+self.id)
 | 
						|
        
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[
 | 
						|
            SimpleHook(1), SimpleHook(2), SimpleHook(3)
 | 
						|
        ]))
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 10
 | 
						|
        assert run_hook[0] == 'on_route1'
 | 
						|
        assert run_hook[1] == 'on_route2'
 | 
						|
        assert run_hook[2] == 'on_route3'
 | 
						|
        assert run_hook[3] == 'before1'
 | 
						|
        assert run_hook[4] == 'before2'
 | 
						|
        assert run_hook[5] == 'before3'
 | 
						|
        assert run_hook[6] == 'inside'
 | 
						|
        assert run_hook[7] == 'after3'
 | 
						|
        assert run_hook[8] == 'after2'
 | 
						|
        assert run_hook[9] == 'after1'
 | 
						|
 | 
						|
    def test_partial_hooks(self):
 | 
						|
        run_hook = []
 | 
						|
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello World!'
 | 
						|
 | 
						|
            @expose()
 | 
						|
            def causeerror(self):
 | 
						|
                return [][1]
 | 
						|
 | 
						|
        class ErrorHook(PecanHook):
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error')
 | 
						|
 | 
						|
        class OnRouteHook(PecanHook):
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route')
 | 
						|
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[
 | 
						|
            ErrorHook(), OnRouteHook()
 | 
						|
        ]))
 | 
						|
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello World!'
 | 
						|
 | 
						|
        assert len(run_hook) == 2
 | 
						|
        assert run_hook[0] == 'on_route'
 | 
						|
        assert run_hook[1] == 'inside'
 | 
						|
 | 
						|
        run_hook = []
 | 
						|
        try:
 | 
						|
            response = app.get('/causeerror')
 | 
						|
        except Exception, e:
 | 
						|
            assert isinstance(e, IndexError)
 | 
						|
 | 
						|
        assert len(run_hook) == 2
 | 
						|
        assert run_hook[0] == 'on_route'
 | 
						|
        assert run_hook[1] == 'error'
 | 
						|
 | 
						|
    def test_prioritized_hooks(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
        
 | 
						|
        class SimpleHook(PecanHook):
 | 
						|
            def __init__(self, id):
 | 
						|
                self.id = str(id)
 | 
						|
            
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route'+self.id)
 | 
						|
            
 | 
						|
            def before(self, state):
 | 
						|
                run_hook.append('before'+self.id)
 | 
						|
 | 
						|
            def after(self, state):
 | 
						|
                run_hook.append('after'+self.id)
 | 
						|
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error'+self.id)
 | 
						|
        
 | 
						|
        papp = make_app(RootController(), hooks=[
 | 
						|
            SimpleHook(1), SimpleHook(2), SimpleHook(3)
 | 
						|
        ])
 | 
						|
        app = TestApp(papp)
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 10
 | 
						|
        assert run_hook[0] == 'on_route1'
 | 
						|
        assert run_hook[1] == 'on_route2'
 | 
						|
        assert run_hook[2] == 'on_route3'
 | 
						|
        assert run_hook[3] == 'before1'
 | 
						|
        assert run_hook[4] == 'before2'
 | 
						|
        assert run_hook[5] == 'before3'
 | 
						|
        assert run_hook[6] == 'inside'
 | 
						|
        assert run_hook[7] == 'after3'
 | 
						|
        assert run_hook[8] == 'after2'
 | 
						|
        assert run_hook[9] == 'after1'
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        state.app.hooks[0].priority = 3
 | 
						|
        state.app.hooks[1].priority = 2
 | 
						|
        state.app.hooks[2].priority = 1
 | 
						|
        
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 10
 | 
						|
        assert run_hook[0] == 'on_route3'
 | 
						|
        assert run_hook[1] == 'on_route2'
 | 
						|
        assert run_hook[2] == 'on_route1'
 | 
						|
        assert run_hook[3] == 'before3'
 | 
						|
        assert run_hook[4] == 'before2'
 | 
						|
        assert run_hook[5] == 'before1'
 | 
						|
        assert run_hook[6] == 'inside'
 | 
						|
        assert run_hook[7] == 'after1'
 | 
						|
        assert run_hook[8] == 'after2'
 | 
						|
        assert run_hook[9] == 'after3'
 | 
						|
    
 | 
						|
    def test_transaction_hook(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
            
 | 
						|
            @expose()
 | 
						|
            def redirect(self):
 | 
						|
                redirect('/')            
 | 
						|
            
 | 
						|
            @expose()
 | 
						|
            def error(self):
 | 
						|
                return [][1]
 | 
						|
        
 | 
						|
        def gen(event):
 | 
						|
            return lambda: run_hook.append(event)
 | 
						|
        
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[
 | 
						|
            TransactionHook(
 | 
						|
                start    = gen('start'),
 | 
						|
                start_ro = gen('start_ro'),
 | 
						|
                commit   = gen('commit'),
 | 
						|
                rollback = gen('rollback'),
 | 
						|
                clear    = gen('clear')
 | 
						|
            )
 | 
						|
        ]))
 | 
						|
        
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'inside'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.post('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 4
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'inside'
 | 
						|
        assert run_hook[2] == 'commit'
 | 
						|
        assert run_hook[3] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for GET /redirect
 | 
						|
        # This controller should always be non-transactional
 | 
						|
        #
 | 
						|
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.get('/redirect')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 2
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for POST /redirect
 | 
						|
        # This controller should always be transactional,
 | 
						|
        # even in the case of redirects
 | 
						|
        #
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.post('/redirect')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'commit'
 | 
						|
        assert run_hook[2] == 'clear'        
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        try:
 | 
						|
            response = app.post('/error')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
        
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
    def test_transaction_hook_with_transactional_decorator(self):
 | 
						|
        run_hook = []
 | 
						|
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
 | 
						|
            @expose()
 | 
						|
            def redirect(self):
 | 
						|
                redirect('/')
 | 
						|
                
 | 
						|
            @expose()
 | 
						|
            @transactional()
 | 
						|
            def redirect_transactional(self):
 | 
						|
                redirect('/')
 | 
						|
                
 | 
						|
            @expose()
 | 
						|
            @transactional(False)
 | 
						|
            def redirect_rollback(self):
 | 
						|
                redirect('/')
 | 
						|
 | 
						|
            @expose()
 | 
						|
            def error(self):
 | 
						|
                return [][1]
 | 
						|
                
 | 
						|
            @expose()
 | 
						|
            @transactional(False)
 | 
						|
            def error_rollback(self):
 | 
						|
                return [][1]                
 | 
						|
                
 | 
						|
            @expose()
 | 
						|
            @transactional()
 | 
						|
            def error_transactional(self):
 | 
						|
                return [][1]                
 | 
						|
 | 
						|
        def gen(event):
 | 
						|
            return lambda: run_hook.append(event)
 | 
						|
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[
 | 
						|
            TransactionHook(
 | 
						|
                start    = gen('start'),
 | 
						|
                start_ro = gen('start_ro'),
 | 
						|
                commit   = gen('commit'),
 | 
						|
                rollback = gen('rollback'),
 | 
						|
                clear    = gen('clear')
 | 
						|
            )
 | 
						|
        ]))
 | 
						|
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'inside'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
 | 
						|
        run_hook = []
 | 
						|
 | 
						|
        # test hooks for /
 | 
						|
 | 
						|
        response = app.post('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
 | 
						|
        assert len(run_hook) == 4
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'inside'
 | 
						|
        assert run_hook[2] == 'commit'
 | 
						|
        assert run_hook[3] == 'clear'
 | 
						|
 | 
						|
        #
 | 
						|
        # test hooks for GET /redirect
 | 
						|
        # This controller should always be non-transactional
 | 
						|
        #
 | 
						|
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.get('/redirect')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 2
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for POST /redirect
 | 
						|
        # This controller should always be transactional,
 | 
						|
        # even in the case of redirects
 | 
						|
        #
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.post('/redirect')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'commit'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for GET /redirect_transactional
 | 
						|
        # This controller should always be transactional,
 | 
						|
        # even in the case of redirects
 | 
						|
        #
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.get('/redirect_transactional')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 5
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'clear'        
 | 
						|
        assert run_hook[2] == 'start'
 | 
						|
        assert run_hook[3] == 'commit'
 | 
						|
        assert run_hook[4] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for POST /redirect_transactional
 | 
						|
        # This controller should always be transactional,
 | 
						|
        # even in the case of redirects
 | 
						|
        #
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.post('/redirect_transactional')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'commit'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for GET /redirect_rollback
 | 
						|
        # This controller should always be transactional,
 | 
						|
        # *except* in the case of redirects
 | 
						|
        #
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.get('/redirect_rollback')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 5
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'clear'        
 | 
						|
        assert run_hook[2] == 'start'        
 | 
						|
        assert run_hook[3] == 'rollback'
 | 
						|
        assert run_hook[4] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # test hooks for POST /redirect_rollback
 | 
						|
        # This controller should always be transactional,
 | 
						|
        # *except* in the case of redirects
 | 
						|
        #
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.post('/redirect_rollback')
 | 
						|
        assert response.status_int == 302
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        #
 | 
						|
        # Exceptions (other than HTTPFound) should *always*
 | 
						|
        # rollback no matter what
 | 
						|
        #
 | 
						|
        run_hook = []        
 | 
						|
        
 | 
						|
        try:
 | 
						|
            response = app.post('/error')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        run_hook = []        
 | 
						|
        
 | 
						|
        try:
 | 
						|
            response = app.get('/error')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
 | 
						|
        assert len(run_hook) == 2
 | 
						|
        assert run_hook[0] == 'start_ro'
 | 
						|
        assert run_hook[1] == 'clear'
 | 
						|
        
 | 
						|
        run_hook = []        
 | 
						|
        
 | 
						|
        try:
 | 
						|
            response = app.post('/error_transactional')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        run_hook = []        
 | 
						|
        
 | 
						|
        try:
 | 
						|
            response = app.get('/error_transactional')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        run_hook = []        
 | 
						|
        
 | 
						|
        try:
 | 
						|
            response = app.post('/error_rollback')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
        
 | 
						|
        run_hook = []        
 | 
						|
        
 | 
						|
        try:
 | 
						|
            response = app.get('/error_rollback')
 | 
						|
        except IndexError:
 | 
						|
            pass
 | 
						|
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'start'
 | 
						|
        assert run_hook[1] == 'rollback'
 | 
						|
        assert run_hook[2] == 'clear'
 | 
						|
    
 | 
						|
    def test_basic_isolated_hook(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class SimpleHook(PecanHook):
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route')
 | 
						|
            
 | 
						|
            def before(self, state):
 | 
						|
                run_hook.append('before')
 | 
						|
 | 
						|
            def after(self, state):
 | 
						|
                run_hook.append('after')
 | 
						|
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error')
 | 
						|
 | 
						|
        class SubSubController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside_sub_sub')
 | 
						|
                return 'Deep inside here!'
 | 
						|
        
 | 
						|
        class SubController(HookController):
 | 
						|
            __hooks__ = [SimpleHook()]
 | 
						|
            
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside_sub')
 | 
						|
                return 'Inside here!'
 | 
						|
            
 | 
						|
            sub = SubSubController()
 | 
						|
        
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
            
 | 
						|
            sub = SubController()
 | 
						|
                
 | 
						|
        app = TestApp(make_app(RootController()))
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 1
 | 
						|
        assert run_hook[0] == 'inside'
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.get('/sub/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Inside here!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'before'
 | 
						|
        assert run_hook[1] == 'inside_sub'
 | 
						|
        assert run_hook[2] == 'after'
 | 
						|
 | 
						|
        run_hook = []
 | 
						|
        response = app.get('/sub/sub/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Deep inside here!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 3
 | 
						|
        assert run_hook[0] == 'before'
 | 
						|
        assert run_hook[1] == 'inside_sub_sub'
 | 
						|
        assert run_hook[2] == 'after'
 | 
						|
    
 | 
						|
    def test_isolated_hook_with_global_hook(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class SimpleHook(PecanHook):
 | 
						|
            def __init__(self, id):
 | 
						|
                self.id = str(id)
 | 
						|
            
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route'+self.id)
 | 
						|
            
 | 
						|
            def before(self, state):
 | 
						|
                run_hook.append('before'+self.id)
 | 
						|
 | 
						|
            def after(self, state):
 | 
						|
                run_hook.append('after'+self.id)
 | 
						|
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error'+self.id)
 | 
						|
        
 | 
						|
        class SubController(HookController):
 | 
						|
            __hooks__ = [SimpleHook(2)]
 | 
						|
            
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside_sub')
 | 
						|
                return 'Inside here!'
 | 
						|
        
 | 
						|
        class RootController(object):
 | 
						|
            @expose()
 | 
						|
            def index(self):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'Hello, World!'
 | 
						|
            
 | 
						|
            sub = SubController()
 | 
						|
                
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[SimpleHook(1)]))
 | 
						|
        response = app.get('/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Hello, World!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 4
 | 
						|
        assert run_hook[0] == 'on_route1'
 | 
						|
        assert run_hook[1] == 'before1'
 | 
						|
        assert run_hook[2] == 'inside'
 | 
						|
        assert run_hook[3] == 'after1'
 | 
						|
        
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        response = app.get('/sub/')
 | 
						|
        assert response.status_int == 200
 | 
						|
        assert response.body == 'Inside here!'
 | 
						|
        
 | 
						|
        assert len(run_hook) == 6
 | 
						|
        assert run_hook[0] == 'on_route1'
 | 
						|
        assert run_hook[1] == 'before2'
 | 
						|
        assert run_hook[2] == 'before1'
 | 
						|
        assert run_hook[3] == 'inside_sub'
 | 
						|
        assert run_hook[4] == 'after1'
 | 
						|
        assert run_hook[5] == 'after2'
 | 
						|
    
 | 
						|
    def test_hooks_with_validation(self):
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        class RegistrationSchema(Schema):
 | 
						|
            first_name         = validators.String(not_empty=True)
 | 
						|
            last_name          = validators.String(not_empty=True)
 | 
						|
            email              = validators.Email()
 | 
						|
            username           = validators.PlainText()
 | 
						|
            password           = validators.String()
 | 
						|
            password_confirm   = validators.String()
 | 
						|
            age                = validators.Int()
 | 
						|
            chained_validators = [
 | 
						|
                validators.FieldsMatch('password', 'password_confirm')
 | 
						|
            ]
 | 
						|
        
 | 
						|
        class SimpleHook(PecanHook):
 | 
						|
            def on_route(self, state):
 | 
						|
                run_hook.append('on_route')
 | 
						|
            
 | 
						|
            def before(self, state):
 | 
						|
                run_hook.append('before')
 | 
						|
 | 
						|
            def after(self, state):
 | 
						|
                run_hook.append('after')
 | 
						|
 | 
						|
            def on_error(self, state, e):
 | 
						|
                run_hook.append('error')
 | 
						|
        
 | 
						|
        class RootController(object):            
 | 
						|
            @expose()
 | 
						|
            def errors(self, *args, **kwargs):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return 'errors'
 | 
						|
            
 | 
						|
            @expose(schema=RegistrationSchema())
 | 
						|
            def index(self, first_name, 
 | 
						|
                            last_name, 
 | 
						|
                            email, 
 | 
						|
                            username, 
 | 
						|
                            password,
 | 
						|
                            password_confirm,
 | 
						|
                            age):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return str(len(request.pecan['validation_errors']) > 0)
 | 
						|
            
 | 
						|
            @expose(schema=RegistrationSchema(), error_handler='/errors')
 | 
						|
            def with_handler(self, first_name, 
 | 
						|
                            last_name, 
 | 
						|
                            email, 
 | 
						|
                            username, 
 | 
						|
                            password,
 | 
						|
                            password_confirm,
 | 
						|
                            age):
 | 
						|
                run_hook.append('inside')
 | 
						|
                return str(len(request.pecan['validation_errors']) > 0)
 | 
						|
        
 | 
						|
        # test that the hooks get properly run with no validation errors        
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[SimpleHook()]))
 | 
						|
        r = app.post('/', dict(
 | 
						|
            first_name       = 'Jonathan',
 | 
						|
            last_name        = 'LaCour',
 | 
						|
            email            = 'jonathan@cleverdevil.org',
 | 
						|
            username         = 'jlacour',
 | 
						|
            password         = '123456',
 | 
						|
            password_confirm = '123456',
 | 
						|
            age              = '31'
 | 
						|
        ))
 | 
						|
        assert r.status_int == 200
 | 
						|
        assert r.body == 'False'
 | 
						|
        assert len(run_hook) == 4
 | 
						|
        assert run_hook[0] == 'on_route'
 | 
						|
        assert run_hook[1] == 'before'
 | 
						|
        assert run_hook[2] == 'inside'
 | 
						|
        assert run_hook[3] == 'after'
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        # test that the hooks get properly run with validation errors        
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[SimpleHook()]))
 | 
						|
        r = app.post('/', dict(
 | 
						|
            first_name       = 'Jonathan',
 | 
						|
            last_name        = 'LaCour',
 | 
						|
            email            = 'jonathan@cleverdevil.org',
 | 
						|
            username         = 'jlacour',
 | 
						|
            password         = '654321',
 | 
						|
            password_confirm = '123456',
 | 
						|
            age              = '31'
 | 
						|
        ))
 | 
						|
        assert r.status_int == 200
 | 
						|
        assert r.body == 'True'
 | 
						|
        assert len(run_hook) == 4
 | 
						|
        assert run_hook[0] == 'on_route'
 | 
						|
        assert run_hook[1] == 'before'
 | 
						|
        assert run_hook[2] == 'inside'
 | 
						|
        assert run_hook[3] == 'after'
 | 
						|
        run_hook = []
 | 
						|
        
 | 
						|
        # test that the hooks get properly run with validation errors 
 | 
						|
        # and an error handler       
 | 
						|
        app = TestApp(make_app(RootController(), hooks=[SimpleHook()]))
 | 
						|
        r = app.post('/with_handler', dict(
 | 
						|
            first_name       = 'Jonathan',
 | 
						|
            last_name        = 'LaCour',
 | 
						|
            email            = 'jonathan@cleverdevil.org',
 | 
						|
            username         = 'jlacour',
 | 
						|
            password         = '654321',
 | 
						|
            password_confirm = '123456',
 | 
						|
            age              = '31'
 | 
						|
        ))
 | 
						|
        assert r.status_int == 200
 | 
						|
        assert r.body == 'errors'
 | 
						|
        assert len(run_hook) == 7
 | 
						|
        assert run_hook[0] == 'on_route'
 | 
						|
        assert run_hook[1] == 'before'
 | 
						|
        assert run_hook[2] == 'after'
 | 
						|
        assert run_hook[3] == 'on_route'
 | 
						|
        assert run_hook[4] == 'before'
 | 
						|
        assert run_hook[5] == 'inside'
 | 
						|
        assert run_hook[6] == 'after'
 |