Commits

agentultra committed 1ca7f93 Merge

Merged upstream

Comments (0)

Files changed (8)

pylons/configuration.py

 import os
 import sys
 
-import venusian
-from marco.events import Events
 from paste.config import DispatchingConfig
 from paste.deploy.converters import asbool
 from routes import Mapper
 from webhelpers.mimehelper import MIMETypes
 
 
-from pylons.view import map_view
-
 request_defaults = dict(charset='utf-8', errors='replace',
                         decode_param_names=False, language='en-us')
 response_defaults = dict(content_type='text/html',
         'pylons.tmpl_context_attach_args': False,
     }
     
-    def begin(self):
-        """Intializes the configuration object for an application"""
-        self.events = Events()
-        self._scanner = venusian.Scanner(config=self)
-        self._delayed_views = {}
-        
-        # Ensure all the keys from defaults are present, load them if not
-        for key, val in copy.deepcopy(PylonsConfig.defaults).iteritems():
-            self.setdefault(key, val)
-    
-    def scan(self, package=None):
-        """Scan a package for Pylons decorated functions"""
-        if package:
-            self._scanner.scan(package, categories=('pylons',))
-        else:
-            self._scanner.scan(self._package, categories=('pylons',))
-    
-    def add_subscriber(self, func, event):
-        """Add an event subscriber"""
-        self.events.subscribe(func, event)
-    
-    def add_route(self, *args, **kwargs):
-        """Connect a route to the mapper
-        
-        This will create a Routes mapper on the 'routes.map' key on
-        this object if its not already present.
-        
-        """
-        if 'routes.map' not in self:
-            self['routes.map'] = Mapper()
-        mapper = self['routes.map']
-        args, kwargs = map_view(self, args, kwargs)
-        mapper.connect(*args, **kwargs)
-    
-    def add_route_view(self, route_name, view, attr=None, **predicates):
-        """Add a view for an existing route name"""
-        if route_name not in self._delayed_views:
-            raise Exception('The route "%s" has already been declared with its view' % route_name)
-        
-        dv = self._delayed_views[route_name]
-        dv.views.append((view, attr, predicates))
-    
-    def end(self):
-        """Finish tasks and things that accumulated during configuration"""
-        for route_name, dv in self._delayed_views.iteritems():
-            dv.finalize()
-    
     def init_app(self, global_conf, app_conf, package=None, paths=None):
         """Initialize configuration for the application
         
         """
         log.debug("Initializing configuration, package: '%s'", package)
         
-        # Backwards compat in case begin() was not called
-        if not hasattr(self, 'events'):
-            self.begin()
-        
         conf = global_conf.copy()
         conf.update(app_conf)
         conf.update(dict(app_conf=app_conf, global_conf=global_conf))

pylons/controllers/core.py

 from webob.exc import HTTPException, HTTPNotFound
 
 import pylons
-from pylons.events import NewResponse
 
 __all__ = ['WSGIController']
 
             if log_debug:
                 log.debug("Calling Response object to return WSGI data")
             
-            # Emit the NewResponse event
-            self._py_object.config.events.publish(NewResponse(response))
-            
             return response(environ, self.start_response)
         
         if log_debug:

pylons/events.py

-"""Events
-
-Pylons events are triggered in various stages during a Pylons application's execution
-and setup. They can be registered using the :func:`~pylons.events.subscriber`
-decorator or directly with the config object::
-    
-    config.add_subscriber(my_subscriber, NewRequest)
-
-"""
-import venusian
-
-def subscriber(*events):
-    """Register a function for an event, or multiple events
-    
-    Example::
-        
-        from pylons.events import subscriber, NewRequest
-        
-        @subscriber(NewRequest)
-        def check_user(event):
-            req = event.request
-            req.user_name = environ.get('REMOTE_USER')
-            return
-    
-    Event subscribers are not expected to return values, and any
-    values returned will be ignored.
-    
-    """
-    def func_wrapper(wrapped):
-        def callback(scanner, name, ob):
-            for event in events:
-                scanner.config.add_subscriber(wrapped, event)
-        venusian.attach(wrapped, callback, category='pylons')
-        return wrapped
-    return func_wrapper
-
-
-class NewRequest(object):
-    """ An instance of this class is emitted as an event whenever
-    Pylons begins to process a new request.  The instance has an
-    attribute, ``request``, which is the request object."""
-    def __init__(self, request):
-        self.request = request
-
-
-class NewResponse(object):
-    """ An instance of this class is emitted when a response is
-    recieved by Pylons, right before its sent out to the client. The
-    instance has an attribute, ``response``, which is the response
-    object."""
-    def __init__(self, response):
-        self.response = response
-
-
-class WSGIApplicationCreated(object):
-    """ An instance of this class is emitted when the PylonsApp
-    wsgi object is created. The instance has an attribute, 
-    ``application``, which is the WSGI application object."""
-    def __init__(self, application):
-        self.application = application

pylons/testutil.py

         
         environ.update(self.environ)
         py_obj.config = pylons.config._current_obj()
-        if not hasattr(py_obj.config, 'events'):
-            py_obj.config.begin()
         py_obj.request = req
         py_obj.response = response
         py_obj.tmpl_context = ContextObj()

pylons/view.py

-"""View functions and classes"""
-import logging
-
-import pkg_resources
-from webob.exc import HTTPException, HTTPNotFound
-
-log = logging.getLogger(__name__)
-
-
-def lookup_view(view, package_name=None):
-    """Load a view based on a resource specification"""
-    return pkg_resources.EntryPoint.parse('x=%s' % view).load(False)
-
-
-def map_view(config, args, kwargs):
-    """Given a responder name, handle looking it up and returning
-    a callable responder"""
-    view = kwargs.pop('view', None)
-    
-    # If its a string, determine if its a legacy controller name
-    # or a resource specification
-    if isinstance(view, basestring):
-        view_result = lookup_view(view)
-    else:
-        view_result = view
-    
-    # If it has an action, assume we can pick the action off it
-    if hasattr(view_result, '__bases__'):
-        kwargs['responder'] = class_responder(view_result, kwargs.get('action'))
-    else:
-        kwargs['responder'] = func_inst_responder(view_result, kwargs.get('action'))
-    return args, kwargs
-
-
-def class_responder(view, action=None):
-    """view_responder wraps a single class based view with a responder
-    interface
-    
-    If the action is a method, the class will be instantiated with a
-    request object, and the method will then be called with no
-    arguments.
-    
-    If there is no action specified, then the class will be called
-    after its instantiated (a __call__ should be implemented).
-    
-    The __init__ method may raise an HTTPException should it wish to
-    terminate calling, at which point it will be returned to the client.
-    
-    """
-    if action:
-        if not hasattr(view, action):
-            raise Exception("Unable to locate method %s on %r" % (action, view))
-        action = getattr(view, action)
-    
-    def view_wrapper(request):
-        """The view that is dispatched to by Pylons
-        
-        This wrapper implements the responder paradigm.
-        
-        """
-        # Instantiate the controller with the request
-        view_obj = view(request)
-        if action:
-            response = action(view_obj)
-        else:
-            response = view_obj()
-        return response
-    return view_wrapper
-
-
-def func_inst_responder(view, action=None):
-    """func_inst_responder wraps a function or instance
-    
-    If an action is supplied, it must be an attribute on the view that
-    is callable with a request, and returns a response.
-    
-    """
-    responder = view
-    if action:
-        responder =  getattr(view, action, None)
-        if responder is None:
-            raise Exception('No such action %r on view %r' % action, view)
-    return responder

pylons/wsgiapp.py

 import pylons
 import pylons.templating
 from pylons.controllers.util import Request, Response
-from pylons.events import NewRequest, NewResponse, WSGIApplicationCreated
 from pylons.i18n.translation import _get_translator
 from pylons.util import (AttribSafeContextObj, ContextObj, PylonsContext,
                          class_name_from_module_name)
         # Cache some options for use during requests
         self._session_key = self.environ_config.get('session', 'beaker.session')
         self._cache_key = self.environ_config.get('cache', 'beaker.cache')
-        
-        # Fire the app created event
-        self.config.events.publish(WSGIApplicationCreated(self))
     
     def __call__(self, environ, start_response):
         """Setup and handle a web request
         
         try:
             if response_obj:
-                self.config.events.publish(NewResponse(response))
                 return response(environ, start_response)
             elif response is not None:
                 return response
         if self._cache_key in environ:
             pylons_obj.cache = environ[self._cache_key]
         
-        # Publish the new request event
-        self.config.events.publish(NewRequest(req))
-        
         # Load the globals with the registry if around
         if 'paste.registry' in environ:
             self.register_globals(environ)

tests/test_units/test_events.py

-import os
-import sys
-
-from beaker.middleware import SessionMiddleware
-from paste.fixture import TestApp
-from paste.registry import RegistryManager
-from paste.deploy.converters import asbool
-from routes import Mapper
-from routes.middleware import RoutesMiddleware
-
-from nose.tools import raises
-
-from __init__ import test_root
-
-
-def make_app(global_conf, full_stack=True, static_files=True, **app_conf):
-    import pylons.configuration as configuration
-    from pylons.middleware import ErrorHandler, StatusCodeRedirect
-    from pylons.wsgiapp import PylonsApp
-    
-    import event_file
-    
-    root = os.path.dirname(os.path.abspath(__file__))
-    paths = dict(root=os.path.join(test_root, 'sample_controllers'), controllers=os.path.join(test_root, 'sample_controllers', 'controllers'))
-    sys.path.append(test_root)
-
-    config = configuration.PylonsConfig()
-    config.init_app(global_conf, app_conf, package='sample_controllers', paths=paths)
-    map = Mapper(directory=config['pylons.paths']['controllers'])
-    map.connect('/{controller}/{action}')
-    config['routes.map'] = map
-    
-    class AppGlobals(object): pass
-    config['pylons.app_globals'] = AppGlobals()
-    
-    config.scan(event_file)
-    
-    app = PylonsApp(config=config)
-    app = RoutesMiddleware(app, config['routes.map'], singleton=False)
-    app = SessionMiddleware(app, config)
-
-    if asbool(full_stack):
-        app = ErrorHandler(app, global_conf, **config['pylons.errorware'])
-        if asbool(config['debug']):
-            app = StatusCodeRedirect(app)
-        else:
-            app = StatusCodeRedirect(app, [401, 403, 404, 500])
-    app = RegistryManager(app)
-
-    app.config = config
-    return app
-
-class TestEvents(object):
-    def setUp(self):
-        self.app = TestApp(make_app({}))
-    
-    def test_basic_response(self):
-        response = self.app.get('/hello/index')
-        assert 'Hello World' in response
-        
-    def test_new_request_event(self):
-        response = self.app.get('/hello/index')
-        assert hasattr(response.req, 'reg')
-        assert response.req.reg == True
-    
-    def test_new_response_event(self):
-        response = self.app.get('/hello/index')
-        assert response.response.reg == True

tests/test_units/test_view.py

-# -*- coding: utf-8 -*-
-import os
-import re
-import sys
-
-from nose.tools import raises
-from paste.fixture import TestApp
-from routes import Mapper
-from routes.middleware import RoutesMiddleware
-from webob.exc import HTTPException
-
-from nose.tools import raises
-
-config = None
-
-class Smith(object):
-    def __init__(self, req):
-        self.req = req
-    
-    def __call__(self):
-        from pylons.controllers.util import Response
-        return Response('Hello Smith')
-
-class Doe(object):
-    def __init__(self, req):
-        self.req = req
-    
-    def index(self):
-        from pylons.controllers.util import Response
-        return Response('Hello Doe')
-    
-    def not_here(self):
-        from pylons.controllers.util import abort
-        abort(404)
-    
-    not_callable = 42
-        
-
-class Fawn(object):
-    def __init__(self, req):
-        from pylons.controllers.util import abort
-        abort(401)
-
-def plain_view(request):
-    from pylons.controllers.util import Response
-    return Response('Plain View')
-
-
-def make_app(global_conf, debug=True, **app_conf):
-    import pylons
-    import pylons.configuration as configuration
-    
-    from pylons import url
-    from pylons.controllers.util import abort, Response
-    from pylons.wsgiapp import PylonsApp
-    
-    global config
-    config = configuration.PylonsConfig()
-    config.begin()
-    if debug:
-        config['debug'] = True
-    config.add_route('/smith', view=Smith)
-    config.add_route('/doe', view=Doe, action='index')
-    config.add_route('/doe/not_here', view=Doe, action='not_here')
-    config.add_route('/doe/not_callable', view=Doe, action='not_callable')
-    config.add_route('/fawn', view=Fawn)
-    config.add_route('/plainview', view=plain_view)
-    config.add_route('/aview', view='sample_controllers.controllers.hello:a_view')
-    
-    app = PylonsApp(config=config)
-    app = RoutesMiddleware(app, config['routes.map'], singleton=False)
-    app.config = config
-    return app
-
-class TestWsgiApp(object):
-    def setUp(self):
-        self.app = TestApp(make_app({}))
-    
-    def test_class_view(self):
-        resp = self.app.get('/smith')
-        assert 'Hello Smith' in resp
-    
-    def test_index_view(self):
-        resp = self.app.get('/doe')
-        assert 'Hello Doe' in resp
-    
-    @raises(HTTPException)
-    def test_401_class_init(self):
-        resp = self.app.get('/fawn')
-    
-    def test_plain_view(self):
-        resp = self.app.get('/plainview')
-        assert 'Plain View' in resp
-    
-    def test_a_view(self):
-        resp = self.app.get('/aview')
-        assert 'A View' in resp
-    
-    @raises(HTTPException)
-    def test_404_class_method(self):
-        resp = self.app.get('/doe/not_here')
-
-    @raises(Exception)
-    def test_non_callable(self):
-        resp = self.app.get('/doe/not_callable')
-    
-    @raises(Exception)
-    def test_no_action(self):
-        import pylons.configuration as configuration
-        config = configuration.PylonsConfig()
-        config.begin()
-        config.add_route('/smith', view=Doe, action='no_action_here')