Merge pull request #2985 from mmerickel/plaster
migrate pyramid to use plaster
| | |
| | | unreleased |
| | | ========== |
| | | |
| | | Features |
| | | -------- |
| | | Major Features |
| | | -------------- |
| | | |
| | | - The file format used by all ``p*`` command line scripts such as ``pserve`` |
| | | and ``pshell``, as well as the ``pyramid.paster.bootstrap`` function |
| | | is now replaceable thanks to a new dependency on |
| | | `plaster <http://docs.pylonsproject.org/projects/plaster/en/latest/>`_. |
| | | |
| | | For now, Pyramid is still shipping with integrated support for the |
| | | PasteDeploy INI format by depending on the ``plaster_pastedeploy`` binding. |
| | | |
| | | See https://github.com/Pylons/pyramid/pull/2985 |
| | | |
| | | - Added an execution policy hook to the request pipeline. An execution |
| | | policy has the ability to control creation and execution of the request |
| | | objects before they enter rest of the pipeline. This means for a given |
| | | request that the policy may create more than one request for retry |
| | | purposes. See https://github.com/Pylons/pyramid/pull/2964 |
| | | objects before they enter the rest of the pipeline. This means for a single |
| | | request environ the policy may create more than one request object. |
| | | |
| | | The first library to use this feature is |
| | | `pyramid_retry |
| | | <http://docs.pylonsproject.org/projects/pyramid-retry/en/latest/>`_. |
| | | |
| | | See https://github.com/Pylons/pyramid/pull/2964 |
| | | |
| | | Features |
| | | -------- |
| | | |
| | | - Support an ``open_url`` config setting in the ``pserve`` section of the |
| | | config file. This url is used to open a web browser when ``pserve --browser`` |
| | |
| | | |
| | | .. autofunction:: bootstrap |
| | | |
| | | .. autofunction:: get_app(config_uri, name=None, options=None) |
| | | .. autofunction:: get_app |
| | | |
| | | .. autofunction:: get_appsettings(config_uri, name=None, options=None) |
| | | .. autofunction:: get_appsettings |
| | | |
| | | .. autofunction:: setup_logging(config_uri, global_conf=None) |
| | | .. autofunction:: setup_logging |
| | |
| | | :term:`WSGI` components together declaratively within an ``.ini`` |
| | | file. It was developed by Ian Bicking. |
| | | |
| | | plaster |
| | | `plaster <http://docs.pylonsproject.org/projects/plaster/en/latest/>`_ is |
| | | a library used by :app:`Pyramid` which acts as an abstraction between |
| | | command-line scripts and the file format used to load the :term:`WSGI` |
| | | components and application settings. By default :app:`Pyramid` ships |
| | | with the ``plaster_pastedeploy`` library installed which provides |
| | | integrated support for loading a :term:`PasteDeploy` INI file. |
| | | |
| | | Chameleon |
| | | `chameleon <https://chameleon.readthedocs.org/en/latest/>`_ is an |
| | | attribute language template compiler which supports the :term:`ZPT` |
| | |
| | | PasteDeploy |
| | | ----------- |
| | | |
| | | :term:`PasteDeploy` is the system that Pyramid uses to allow :term:`deployment |
| | | settings` to be specified using an ``.ini`` configuration file format. It also |
| | | allows the ``pserve`` command to work. Its configuration format provides a |
| | | convenient place to define application :term:`deployment settings` and WSGI |
| | | server settings, and its server runner allows you to stop and start a Pyramid |
| | | application easily. |
| | | :term:`plaster` is the system that Pyramid uses to load settings from configuration files. The most common format for these files is an ``.ini`` format structured in a way defined by :term:`PasteDeploy`. The format supports mechanisms to define WSGI app :term:`deployment settings`, WSGI server settings and logging. This allows the ``pserve`` command to work, allowing you to stop and start a Pyramid application easily. |
| | | |
| | | .. _pastedeploy_entry_points: |
| | | |
| | |
| | | file. The values in a ``[DEFAULT]`` section will be passed to your |
| | | application's ``main`` function as ``global_config`` (see the reference to the |
| | | ``main`` function in :ref:`init_py`). |
| | | |
| | | Alternative Configuration File Formats |
| | | -------------------------------------- |
| | | |
| | | It is possible to use different file formats with :app:`Pyramid` if you do not like :term:`PasteDeploy`. Under the hood all command-line scripts such as ``pserve`` and ``pshell`` pass the ``config_uri`` (e.g. ``development.ini`` or ``production.ini``) to the :term:`plaster` library which performs a lookup for an appropriate parser. For ``.ini`` files it uses PasteDeploy but you can register your own configuration formats that plaster will find instead. |
| | |
| | | begin to run and serve an application using the information contained |
| | | within the ``development.ini`` file. |
| | | |
| | | #. The framework finds a section named either ``[app:main]``, |
| | | #. ``pserve`` passes the ``development.ini`` path to :term:`plaster` which |
| | | finds an available configuration loader that recognizes the ``ini`` format. |
| | | |
| | | #. :term:`plaster` finds the ``plaster_pastedeploy`` library which binds |
| | | the :term:`PasteDeploy` library and returns a parser that can understand |
| | | the format. |
| | | |
| | | #. The :term:`PasteDeploy` finds a section named either ``[app:main]``, |
| | | ``[pipeline:main]``, or ``[composite:main]`` in the ``.ini`` file. This |
| | | section represents the configuration of a :term:`WSGI` application that will |
| | | be served. If you're using a simple application (e.g., ``[app:main]``), the |
| | |
| | | import os |
| | | |
| | | from paste.deploy import ( |
| | | loadapp, |
| | | appconfig, |
| | | ) |
| | | |
| | | from pyramid.scripting import prepare |
| | | from pyramid.scripts.common import setup_logging # noqa, api |
| | | from pyramid.scripts.common import get_config_loader |
| | | |
| | | def get_app(config_uri, name=None, options=None, loadapp=loadapp): |
| | | def setup_logging(config_uri, global_conf=None): |
| | | """ |
| | | Set up Python logging with the filename specified via ``config_uri`` |
| | | (a string in the form ``filename#sectionname``). |
| | | |
| | | Extra defaults can optionally be specified as a dict in ``global_conf``. |
| | | """ |
| | | loader = get_config_loader(config_uri) |
| | | loader.setup_logging(global_conf) |
| | | |
| | | def get_app(config_uri, name=None, options=None): |
| | | """ Return the WSGI application named ``name`` in the PasteDeploy |
| | | config file specified by ``config_uri``. |
| | | |
| | |
| | | |
| | | If the ``name`` is None, this will attempt to parse the name from |
| | | the ``config_uri`` string expecting the format ``inifile#name``. |
| | | If no name is found, the name will default to "main".""" |
| | | path, section = _getpathsec(config_uri, name) |
| | | config_name = 'config:%s' % path |
| | | here_dir = os.getcwd() |
| | | If no name is found, the name will default to "main". |
| | | |
| | | app = loadapp( |
| | | config_name, |
| | | name=section, |
| | | relative_to=here_dir, |
| | | global_conf=options) |
| | | """ |
| | | loader = get_config_loader(config_uri) |
| | | return loader.get_wsgi_app(name, options) |
| | | |
| | | return app |
| | | |
| | | def get_appsettings(config_uri, name=None, options=None, appconfig=appconfig): |
| | | def get_appsettings(config_uri, name=None, options=None): |
| | | """ Return a dictionary representing the key/value pairs in an ``app`` |
| | | section within the file represented by ``config_uri``. |
| | | |
| | |
| | | |
| | | If the ``name`` is None, this will attempt to parse the name from |
| | | the ``config_uri`` string expecting the format ``inifile#name``. |
| | | If no name is found, the name will default to "main".""" |
| | | path, section = _getpathsec(config_uri, name) |
| | | config_name = 'config:%s' % path |
| | | here_dir = os.getcwd() |
| | | return appconfig( |
| | | config_name, |
| | | name=section, |
| | | relative_to=here_dir, |
| | | global_conf=options) |
| | | If no name is found, the name will default to "main". |
| | | |
| | | def _getpathsec(config_uri, name): |
| | | if '#' in config_uri: |
| | | path, section = config_uri.split('#', 1) |
| | | else: |
| | | path, section = config_uri, 'main' |
| | | if name: |
| | | section = name |
| | | return path, section |
| | | """ |
| | | loader = get_config_loader(config_uri) |
| | | return loader.get_wsgi_app_settings(name, options) |
| | | |
| | | def bootstrap(config_uri, request=None, options=None): |
| | | """ Load a WSGI application from the PasteDeploy config file specified |
| | |
| | | import os |
| | | from pyramid.compat import configparser |
| | | from logging.config import fileConfig |
| | | import plaster |
| | | |
| | | def parse_vars(args): |
| | | """ |
| | |
| | | result[name] = value |
| | | return result |
| | | |
| | | def setup_logging(config_uri, global_conf=None, |
| | | fileConfig=fileConfig, |
| | | configparser=configparser): |
| | | def get_config_loader(config_uri): |
| | | """ |
| | | Set up logging via :func:`logging.config.fileConfig` with the filename |
| | | specified via ``config_uri`` (a string in the form |
| | | ``filename#sectionname``). |
| | | Find a ``plaster.ILoader`` object supporting the "wsgi" protocol. |
| | | |
| | | ConfigParser defaults are specified for the special ``__file__`` |
| | | and ``here`` variables, similar to PasteDeploy config loading. |
| | | Extra defaults can optionally be specified as a dict in ``global_conf``. |
| | | """ |
| | | path = config_uri.split('#', 1)[0] |
| | | parser = configparser.ConfigParser() |
| | | parser.read([path]) |
| | | if parser.has_section('loggers'): |
| | | config_file = os.path.abspath(path) |
| | | full_global_conf = dict( |
| | | __file__=config_file, |
| | | here=os.path.dirname(config_file)) |
| | | if global_conf: |
| | | full_global_conf.update(global_conf) |
| | | return fileConfig(config_file, full_global_conf) |
| | | return plaster.get_loader(config_uri, protocols=['wsgi']) |
| | |
| | | |
| | | from pyramid.compat import url_unquote |
| | | from pyramid.request import Request |
| | | from pyramid.paster import get_app |
| | | from pyramid.scripts.common import get_config_loader |
| | | from pyramid.scripts.common import parse_vars |
| | | from pyramid.scripts.common import setup_logging |
| | | |
| | | def main(argv=sys.argv, quiet=False): |
| | | command = PRequestCommand(argv, quiet) |
| | |
| | | "passed here.", |
| | | ) |
| | | |
| | | get_app = staticmethod(get_app) |
| | | _get_config_loader = staticmethod(get_config_loader) |
| | | stdin = sys.stdin |
| | | |
| | | def __init__(self, argv, quiet=False): |
| | |
| | | if not self.quiet: |
| | | print(msg) |
| | | |
| | | def configure_logging(self, app_spec): |
| | | setup_logging(app_spec) |
| | | |
| | | def run(self): |
| | | if not self.args.config_uri or not self.args.path_info: |
| | | self.out('You must provide at least two arguments') |
| | | return 2 |
| | | app_spec = self.args.config_uri |
| | | config_uri = self.args.config_uri |
| | | config_vars = parse_vars(self.args.config_vars) |
| | | path = self.args.path_info |
| | | |
| | | self.configure_logging(app_spec) |
| | | loader = self._get_config_loader(config_uri) |
| | | loader.setup_logging(config_vars) |
| | | |
| | | app = loader.get_wsgi_app(self.args.app_name, config_vars) |
| | | |
| | | if not path.startswith('/'): |
| | | path = '/' + path |
| | |
| | | return 2 |
| | | name, value = item.split(':', 1) |
| | | headers[name] = value.strip() |
| | | |
| | | app = self.get_app(app_spec, self.args.app_name, |
| | | options=parse_vars(self.args.config_vars)) |
| | | |
| | | request_method = (self.args.method or 'GET').upper() |
| | | |
| | |
| | | from zope.interface import Interface |
| | | |
| | | from pyramid.paster import bootstrap |
| | | from pyramid.compat import (string_types, configparser) |
| | | from pyramid.compat import string_types |
| | | from pyramid.interfaces import IRouteRequest |
| | | from pyramid.config import not_ |
| | | |
| | | from pyramid.scripts.common import get_config_loader |
| | | from pyramid.scripts.common import parse_vars |
| | | from pyramid.static import static_view |
| | | from pyramid.view import _find_views |
| | |
| | | (route.name, route_intr['external_url'], UNKNOWN_KEY, ANY_KEY) |
| | | ] |
| | | |
| | | |
| | | route_request_methods = route_intr['request_methods'] |
| | | view_intr = registry.introspector.related(route_intr) |
| | | |
| | |
| | | will be assumed. Example: 'proutes myapp.ini'. |
| | | |
| | | """ |
| | | bootstrap = (bootstrap,) |
| | | bootstrap = staticmethod(bootstrap) # testing |
| | | get_config_loader = staticmethod(get_config_loader) # testing |
| | | stdout = sys.stdout |
| | | ConfigParser = configparser.ConfigParser # testing |
| | | parser = argparse.ArgumentParser( |
| | | description=textwrap.dedent(description), |
| | | formatter_class=argparse.RawDescriptionHelpFormatter, |
| | |
| | | |
| | | return True |
| | | |
| | | def proutes_file_config(self, filename): |
| | | config = self.ConfigParser() |
| | | config.read(filename) |
| | | try: |
| | | items = config.items('proutes') |
| | | for k, v in items: |
| | | if 'format' == k: |
| | | cols = re.split(r'[,|\s\n]+', v) |
| | | self.column_format = [x.strip() for x in cols] |
| | | |
| | | except configparser.NoSectionError: |
| | | return |
| | | def proutes_file_config(self, loader, global_conf=None): |
| | | settings = loader.get_settings('proutes', global_conf) |
| | | format = settings.get('format') |
| | | if format: |
| | | cols = re.split(r'[,|\s\n]+', format) |
| | | self.column_format = [x.strip() for x in cols] |
| | | |
| | | def out(self, msg): # pragma: no cover |
| | | if not self.quiet: |
| | |
| | | return 2 |
| | | |
| | | config_uri = self.args.config_uri |
| | | env = self.bootstrap[0](config_uri, options=parse_vars(self.args.config_vars)) |
| | | config_vars = parse_vars(self.args.config_vars) |
| | | loader = self.get_config_loader(config_uri) |
| | | loader.setup_logging(config_vars) |
| | | self.proutes_file_config(loader, config_vars) |
| | | |
| | | env = self.bootstrap(config_uri, options=config_vars) |
| | | registry = env['registry'] |
| | | mapper = self._get_mapper(registry) |
| | | |
| | | self.proutes_file_config(config_uri) |
| | | |
| | | if self.args.format: |
| | | columns = self.args.format.split(',') |
| | |
| | | import webbrowser |
| | | |
| | | import hupper |
| | | from paste.deploy import ( |
| | | loadapp, |
| | | loadserver, |
| | | ) |
| | | |
| | | from pyramid.compat import PY2 |
| | | from pyramid.compat import configparser |
| | | |
| | | from pyramid.scripts.common import get_config_loader |
| | | from pyramid.scripts.common import parse_vars |
| | | from pyramid.scripts.common import setup_logging |
| | | from pyramid.path import AssetResolver |
| | | from pyramid.settings import aslist |
| | | |
| | |
| | | "passed here.", |
| | | ) |
| | | |
| | | ConfigParser = configparser.ConfigParser # testing |
| | | loadapp = staticmethod(loadapp) # testing |
| | | loadserver = staticmethod(loadserver) # testing |
| | | _get_config_loader = staticmethod(get_config_loader) # for testing |
| | | |
| | | open_url = None |
| | | |
| | |
| | | if self.args.verbose > 0: |
| | | print(msg) |
| | | |
| | | def get_config_vars(self): |
| | | restvars = self.args.config_vars |
| | | return parse_vars(restvars) |
| | | def get_config_path(self, loader): |
| | | return os.path.abspath(loader.uri.path) |
| | | |
| | | def pserve_file_config(self, filename, global_conf=None): |
| | | here = os.path.abspath(os.path.dirname(filename)) |
| | | defaults = {} |
| | | if global_conf: |
| | | defaults.update(global_conf) |
| | | defaults['here'] = here |
| | | |
| | | config = self.ConfigParser(defaults=defaults) |
| | | config.optionxform = str |
| | | config.read(filename) |
| | | try: |
| | | items = dict(config.items('pserve')) |
| | | except configparser.NoSectionError: |
| | | return |
| | | |
| | | watch_files = aslist(items.get('watch_files', ''), flatten=False) |
| | | def pserve_file_config(self, loader, global_conf=None): |
| | | settings = loader.get_settings('pserve', global_conf) |
| | | config_path = self.get_config_path(loader) |
| | | here = os.path.dirname(config_path) |
| | | watch_files = aslist(settings.get('watch_files', ''), flatten=False) |
| | | |
| | | # track file paths relative to the ini file |
| | | resolver = AssetResolver(package=None) |
| | |
| | | self.watch_files.add(os.path.abspath(file)) |
| | | |
| | | # attempt to determine the url of the server |
| | | open_url = items.get('open_url') |
| | | open_url = settings.get('open_url') |
| | | if open_url: |
| | | self.open_url = open_url |
| | | |
| | | def _guess_server_url(self, filename, server_name, |
| | | global_conf=None): # pragma: no cover |
| | | def guess_server_url(self, loader, server_name, global_conf=None): |
| | | server_name = server_name or 'main' |
| | | here = os.path.abspath(os.path.dirname(filename)) |
| | | defaults = {} |
| | | if global_conf: |
| | | defaults.update(global_conf) |
| | | defaults['here'] = here |
| | | |
| | | config = self.ConfigParser(defaults=defaults) |
| | | config.optionxform = str |
| | | config.read(filename) |
| | | try: |
| | | items = dict(config.items('server:' + server_name)) |
| | | except configparser.NoSectionError: |
| | | return |
| | | |
| | | if 'port' in items: |
| | | return 'http://127.0.0.1:{port}'.format(**items) |
| | | settings = loader.get_settings('server:' + server_name, global_conf) |
| | | if 'port' in settings: |
| | | return 'http://127.0.0.1:{port}'.format(**settings) |
| | | |
| | | def run(self): # pragma: no cover |
| | | if not self.args.config_uri: |
| | | self.out('You must give a config file') |
| | | return 2 |
| | | config_uri = self.args.config_uri |
| | | config_vars = parse_vars(self.args.config_vars) |
| | | app_spec = self.args.config_uri |
| | | |
| | | vars = self.get_config_vars() |
| | | app_name = self.args.app_name |
| | | |
| | | base = os.getcwd() |
| | | if not self._scheme_re.search(app_spec): |
| | | config_path = os.path.join(base, app_spec) |
| | | app_spec = 'config:' + app_spec |
| | | else: |
| | | config_path = None |
| | | loader = self._get_config_loader(config_uri) |
| | | loader.setup_logging(config_vars) |
| | | |
| | | self.pserve_file_config(loader, global_conf=config_vars) |
| | | |
| | | server_name = self.args.server_name |
| | | if self.args.server: |
| | | server_spec = 'egg:pyramid' |
| | |
| | | else: |
| | | server_spec = app_spec |
| | | |
| | | server_loader = loader |
| | | if server_spec != app_spec: |
| | | server_loader = self.get_config_loader(server_spec) |
| | | |
| | | # do not open the browser on each reload so check hupper first |
| | | if self.args.browser and not hupper.is_active(): |
| | | self.pserve_file_config(config_path, global_conf=vars) |
| | | url = self.open_url |
| | | |
| | | # do not guess the url if the server is sourced from a different |
| | | # location than the config_path |
| | | if not url and server_spec == app_spec: |
| | | url = self._guess_server_url(config_path, server_name, vars) |
| | | if not url: |
| | | url = self.guess_server_url( |
| | | server_loader, server_name, config_vars) |
| | | |
| | | if not url: |
| | | self.out('WARNING: could not determine the server\'s url to ' |
| | |
| | | ) |
| | | return 0 |
| | | |
| | | if config_path: |
| | | setup_logging(config_path, global_conf=vars) |
| | | self.pserve_file_config(config_path, global_conf=vars) |
| | | self.watch_files.add(config_path) |
| | | config_path = self.get_config_path(loader) |
| | | self.watch_files.add(config_path) |
| | | |
| | | server_path = self.get_config_path(server_loader) |
| | | self.watch_files.add(server_path) |
| | | |
| | | if hupper.is_active(): |
| | | reloader = hupper.get_reloader() |
| | | reloader.watch_files(list(self.watch_files)) |
| | | |
| | | server = self.loadserver( |
| | | server_spec, name=server_name, relative_to=base, global_conf=vars) |
| | | server = server_loader.get_wsgi_server(server_name, config_vars) |
| | | |
| | | app = self.loadapp( |
| | | app_spec, name=app_name, relative_to=base, global_conf=vars) |
| | | app = loader.get_wsgi_app(app_name, config_vars) |
| | | |
| | | if self.args.verbose > 0: |
| | | if hasattr(os, 'getpid'): |
| | |
| | | import textwrap |
| | | import pkg_resources |
| | | |
| | | from pyramid.compat import configparser |
| | | from pyramid.compat import exec_ |
| | | from pyramid.util import DottedNameResolver |
| | | from pyramid.paster import bootstrap |
| | | |
| | | from pyramid.settings import aslist |
| | | |
| | | from pyramid.scripts.common import get_config_loader |
| | | from pyramid.scripts.common import parse_vars |
| | | from pyramid.scripts.common import setup_logging |
| | | |
| | | def main(argv=sys.argv, quiet=False): |
| | | command = PShellCommand(argv, quiet) |
| | |
| | | than one Pyramid application within it, the loader will use the |
| | | last one. |
| | | """ |
| | | bootstrap = (bootstrap,) # for testing |
| | | bootstrap = staticmethod(bootstrap) # for testing |
| | | get_config_loader = staticmethod(get_config_loader) # for testing |
| | | pkg_resources = pkg_resources # for testing |
| | | |
| | | parser = argparse.ArgumentParser( |
| | |
| | | "passed here.", |
| | | ) |
| | | |
| | | ConfigParser = configparser.ConfigParser # testing |
| | | default_runner = python_shell_runner # testing |
| | | |
| | | loaded_objects = {} |
| | |
| | | self.quiet = quiet |
| | | self.args = self.parser.parse_args(argv[1:]) |
| | | |
| | | def pshell_file_config(self, filename): |
| | | config = self.ConfigParser() |
| | | config.optionxform = str |
| | | config.read(filename) |
| | | try: |
| | | items = config.items('pshell') |
| | | except configparser.NoSectionError: |
| | | return |
| | | |
| | | def pshell_file_config(self, loader, defaults): |
| | | settings = loader.get_settings('pshell', defaults) |
| | | resolver = DottedNameResolver(None) |
| | | self.loaded_objects = {} |
| | | self.object_help = {} |
| | | self.setup = None |
| | | for k, v in items: |
| | | for k, v in settings.items(): |
| | | if k == 'setup': |
| | | self.setup = v |
| | | elif k == 'default_shell': |
| | |
| | | self.out('Requires a config file argument') |
| | | return 2 |
| | | config_uri = self.args.config_uri |
| | | config_file = config_uri.split('#', 1)[0] |
| | | setup_logging(config_file) |
| | | self.pshell_file_config(config_file) |
| | | config_vars = parse_vars(self.args.config_vars) |
| | | loader = self.get_config_loader(config_uri) |
| | | loader.setup_logging(config_vars) |
| | | self.pshell_file_config(loader, config_vars) |
| | | |
| | | # bootstrap the environ |
| | | env = self.bootstrap[0](config_uri, |
| | | options=parse_vars(self.args.config_vars)) |
| | | env = self.bootstrap(config_uri, options=config_vars) |
| | | |
| | | # remove the closer from the env |
| | | self.closer = env.pop('closer') |
| | |
| | | from pyramid.tweens import MAIN |
| | | from pyramid.tweens import INGRESS |
| | | from pyramid.paster import bootstrap |
| | | from pyramid.paster import setup_logging |
| | | from pyramid.scripts.common import parse_vars |
| | | |
| | | def main(argv=sys.argv, quiet=False): |
| | |
| | | ) |
| | | |
| | | stdout = sys.stdout |
| | | bootstrap = (bootstrap,) # testing |
| | | bootstrap = staticmethod(bootstrap) # testing |
| | | setup_logging = staticmethod(setup_logging) # testing |
| | | |
| | | def __init__(self, argv, quiet=False): |
| | | self.quiet = quiet |
| | |
| | | self.out('Requires a config file argument') |
| | | return 2 |
| | | config_uri = self.args.config_uri |
| | | env = self.bootstrap[0](config_uri, options=parse_vars(self.args.config_vars)) |
| | | config_vars = parse_vars(self.args.config_vars) |
| | | self.setup_logging(config_uri, global_conf=config_vars) |
| | | env = self.bootstrap(config_uri, options=config_vars) |
| | | registry = env['registry'] |
| | | tweens = self._get_tweens(registry) |
| | | if tweens is not None: |
| | |
| | | |
| | | from pyramid.interfaces import IMultiView |
| | | from pyramid.paster import bootstrap |
| | | from pyramid.paster import setup_logging |
| | | from pyramid.request import Request |
| | | from pyramid.scripts.common import parse_vars |
| | | from pyramid.view import _find_views |
| | |
| | | ) |
| | | |
| | | |
| | | bootstrap = (bootstrap,) # testing |
| | | bootstrap = staticmethod(bootstrap) # testing |
| | | setup_logging = staticmethod(setup_logging) # testing |
| | | |
| | | def __init__(self, argv, quiet=False): |
| | | self.quiet = quiet |
| | |
| | | self.out('Command requires a config file arg and a url arg') |
| | | return 2 |
| | | config_uri = self.args.config_uri |
| | | config_vars = parse_vars(self.args.config_vars) |
| | | url = self.args.url |
| | | |
| | | self.setup_logging(config_uri, global_conf=config_vars) |
| | | |
| | | if not url.startswith('/'): |
| | | url = '/%s' % url |
| | | request = Request.blank(url) |
| | | env = self.bootstrap[0](config_uri, options=parse_vars(self.args.config_vars), |
| | | request=request) |
| | | env = self.bootstrap(config_uri, options=config_vars, request=request) |
| | | view = self._find_view(request) |
| | | self.out('') |
| | | self.out("URL = %s" % url) |
| | |
| | | import os |
| | | import unittest |
| | | from pyramid.tests.test_scripts.dummy import DummyLoader |
| | | |
| | | here = os.path.dirname(__file__) |
| | | |
| | | class Test_get_app(unittest.TestCase): |
| | | def _callFUT(self, config_file, section_name, **kw): |
| | | from pyramid.paster import get_app |
| | | return get_app(config_file, section_name, **kw) |
| | | def _callFUT(self, config_file, section_name, options=None, _loader=None): |
| | | import pyramid.paster |
| | | old_loader = pyramid.paster.get_config_loader |
| | | try: |
| | | if _loader is not None: |
| | | pyramid.paster.get_config_loader = _loader |
| | | return pyramid.paster.get_app(config_file, section_name, |
| | | options=options) |
| | | finally: |
| | | pyramid.paster.get_config_loader = old_loader |
| | | |
| | | def test_it(self): |
| | | app = DummyApp() |
| | | loadapp = DummyLoadWSGI(app) |
| | | result = self._callFUT('/foo/bar/myapp.ini', 'myapp', loadapp=loadapp) |
| | | self.assertEqual(loadapp.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(loadapp.section_name, 'myapp') |
| | | self.assertEqual(loadapp.relative_to, os.getcwd()) |
| | | self.assertEqual(result, app) |
| | | |
| | | def test_it_with_hash(self): |
| | | app = DummyApp() |
| | | loadapp = DummyLoadWSGI(app) |
| | | loader = DummyLoader(app=app) |
| | | result = self._callFUT( |
| | | '/foo/bar/myapp.ini#myapp', None, loadapp=loadapp |
| | | ) |
| | | self.assertEqual(loadapp.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(loadapp.section_name, 'myapp') |
| | | self.assertEqual(loadapp.relative_to, os.getcwd()) |
| | | self.assertEqual(result, app) |
| | | |
| | | def test_it_with_hash_and_name_override(self): |
| | | app = DummyApp() |
| | | loadapp = DummyLoadWSGI(app) |
| | | result = self._callFUT( |
| | | '/foo/bar/myapp.ini#myapp', 'yourapp', loadapp=loadapp |
| | | ) |
| | | self.assertEqual(loadapp.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(loadapp.section_name, 'yourapp') |
| | | self.assertEqual(loadapp.relative_to, os.getcwd()) |
| | | self.assertEqual(result, app) |
| | | |
| | | def test_it_with_options(self): |
| | | app = DummyApp() |
| | | loadapp = DummyLoadWSGI(app) |
| | | options = {'a':1} |
| | | result = self._callFUT( |
| | | '/foo/bar/myapp.ini#myapp', |
| | | 'yourapp', |
| | | loadapp=loadapp, |
| | | options=options, |
| | | ) |
| | | self.assertEqual(loadapp.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(loadapp.section_name, 'yourapp') |
| | | self.assertEqual(loadapp.relative_to, os.getcwd()) |
| | | self.assertEqual(loadapp.kw, {'global_conf':options}) |
| | | '/foo/bar/myapp.ini', 'myapp', options={'a': 'b'}, |
| | | _loader=loader) |
| | | self.assertEqual(loader.uri.path, '/foo/bar/myapp.ini') |
| | | self.assertEqual(len(loader.calls), 1) |
| | | self.assertEqual(loader.calls[0]['op'], 'app') |
| | | self.assertEqual(loader.calls[0]['name'], 'myapp') |
| | | self.assertEqual(loader.calls[0]['defaults'], {'a': 'b'}) |
| | | self.assertEqual(result, app) |
| | | |
| | | def test_it_with_dummyapp_requiring_options(self): |
| | |
| | | self.assertEqual(app.settings['foo'], 'baz') |
| | | |
| | | class Test_get_appsettings(unittest.TestCase): |
| | | def _callFUT(self, config_file, section_name, **kw): |
| | | from pyramid.paster import get_appsettings |
| | | return get_appsettings(config_file, section_name, **kw) |
| | | def _callFUT(self, config_file, section_name, options=None, _loader=None): |
| | | import pyramid.paster |
| | | old_loader = pyramid.paster.get_config_loader |
| | | try: |
| | | if _loader is not None: |
| | | pyramid.paster.get_config_loader = _loader |
| | | return pyramid.paster.get_appsettings(config_file, section_name, |
| | | options=options) |
| | | finally: |
| | | pyramid.paster.get_config_loader = old_loader |
| | | |
| | | def test_it(self): |
| | | values = {'a':1} |
| | | appconfig = DummyLoadWSGI(values) |
| | | result = self._callFUT('/foo/bar/myapp.ini', 'myapp', |
| | | appconfig=appconfig) |
| | | self.assertEqual(appconfig.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(appconfig.section_name, 'myapp') |
| | | self.assertEqual(appconfig.relative_to, os.getcwd()) |
| | | self.assertEqual(result, values) |
| | | |
| | | def test_it_with_hash(self): |
| | | values = {'a':1} |
| | | appconfig = DummyLoadWSGI(values) |
| | | result = self._callFUT('/foo/bar/myapp.ini#myapp', None, |
| | | appconfig=appconfig) |
| | | self.assertEqual(appconfig.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(appconfig.section_name, 'myapp') |
| | | self.assertEqual(appconfig.relative_to, os.getcwd()) |
| | | self.assertEqual(result, values) |
| | | |
| | | def test_it_with_hash_and_name_override(self): |
| | | values = {'a':1} |
| | | appconfig = DummyLoadWSGI(values) |
| | | result = self._callFUT('/foo/bar/myapp.ini#myapp', 'yourapp', |
| | | appconfig=appconfig) |
| | | self.assertEqual(appconfig.config_name, 'config:/foo/bar/myapp.ini') |
| | | self.assertEqual(appconfig.section_name, 'yourapp') |
| | | self.assertEqual(appconfig.relative_to, os.getcwd()) |
| | | values = {'a': 1} |
| | | loader = DummyLoader(app_settings=values) |
| | | result = self._callFUT( |
| | | '/foo/bar/myapp.ini', 'myapp', options={'a': 'b'}, |
| | | _loader=loader) |
| | | self.assertEqual(loader.uri.path, '/foo/bar/myapp.ini') |
| | | self.assertEqual(len(loader.calls), 1) |
| | | self.assertEqual(loader.calls[0]['op'], 'app_settings') |
| | | self.assertEqual(loader.calls[0]['name'], 'myapp') |
| | | self.assertEqual(loader.calls[0]['defaults'], {'a': 'b'}) |
| | | self.assertEqual(result, values) |
| | | |
| | | def test_it_with_dummyapp_requiring_options(self): |
| | |
| | | self.assertEqual(result['foo'], 'baz') |
| | | |
| | | class Test_setup_logging(unittest.TestCase): |
| | | def _callFUT(self, config_file, global_conf=None): |
| | | from pyramid.paster import setup_logging |
| | | dummy_cp = DummyConfigParserModule |
| | | return setup_logging( |
| | | config_uri=config_file, |
| | | global_conf=global_conf, |
| | | fileConfig=self.fileConfig, |
| | | configparser=dummy_cp, |
| | | ) |
| | | def _callFUT(self, config_file, global_conf=None, _loader=None): |
| | | import pyramid.paster |
| | | old_loader = pyramid.paster.get_config_loader |
| | | try: |
| | | if _loader is not None: |
| | | pyramid.paster.get_config_loader = _loader |
| | | return pyramid.paster.setup_logging(config_file, global_conf) |
| | | finally: |
| | | pyramid.paster.get_config_loader = old_loader |
| | | |
| | | def test_it_no_global_conf(self): |
| | | config_file, dict = self._callFUT('/abc') |
| | | # os.path.abspath is a sop to Windows |
| | | self.assertEqual(config_file, os.path.abspath('/abc')) |
| | | self.assertEqual(dict['__file__'], os.path.abspath('/abc')) |
| | | self.assertEqual(dict['here'], os.path.abspath('/')) |
| | | loader = DummyLoader() |
| | | self._callFUT('/abc.ini', _loader=loader) |
| | | self.assertEqual(loader.uri.path, '/abc.ini') |
| | | self.assertEqual(len(loader.calls), 1) |
| | | self.assertEqual(loader.calls[0]['op'], 'logging') |
| | | self.assertEqual(loader.calls[0]['defaults'], None) |
| | | |
| | | def test_it_global_conf_empty(self): |
| | | config_file, dict = self._callFUT('/abc', global_conf={}) |
| | | # os.path.abspath is a sop to Windows |
| | | self.assertEqual(config_file, os.path.abspath('/abc')) |
| | | self.assertEqual(dict['__file__'], os.path.abspath('/abc')) |
| | | self.assertEqual(dict['here'], os.path.abspath('/')) |
| | | loader = DummyLoader() |
| | | self._callFUT('/abc.ini', global_conf={}, _loader=loader) |
| | | self.assertEqual(loader.uri.path, '/abc.ini') |
| | | self.assertEqual(len(loader.calls), 1) |
| | | self.assertEqual(loader.calls[0]['op'], 'logging') |
| | | self.assertEqual(loader.calls[0]['defaults'], {}) |
| | | |
| | | def test_it_global_conf_not_empty(self): |
| | | config_file, dict = self._callFUT('/abc', global_conf={'key': 'val'}) |
| | | # os.path.abspath is a sop to Windows |
| | | self.assertEqual(config_file, os.path.abspath('/abc')) |
| | | self.assertEqual(dict['__file__'], os.path.abspath('/abc')) |
| | | self.assertEqual(dict['here'], os.path.abspath('/')) |
| | | self.assertEqual(dict['key'], 'val') |
| | | |
| | | def fileConfig(self, config_file, dict): |
| | | return config_file, dict |
| | | loader = DummyLoader() |
| | | self._callFUT('/abc.ini', global_conf={'key': 'val'}, _loader=loader) |
| | | self.assertEqual(loader.uri.path, '/abc.ini') |
| | | self.assertEqual(len(loader.calls), 1) |
| | | self.assertEqual(loader.calls[0]['op'], 'logging') |
| | | self.assertEqual(loader.calls[0]['defaults'], {'key': 'val'}) |
| | | |
| | | class Test_bootstrap(unittest.TestCase): |
| | | def _callFUT(self, config_uri, request=None): |
| | |
| | | |
| | | dummy_registry = DummyRegistry() |
| | | |
| | | class DummyLoadWSGI: |
| | | def __init__(self, result): |
| | | self.result = result |
| | | |
| | | def __call__(self, config_name, name=None, relative_to=None, **kw): |
| | | self.config_name = config_name |
| | | self.section_name = name |
| | | self.relative_to = relative_to |
| | | self.kw = kw |
| | | return self.result |
| | | |
| | | class DummyApp: |
| | | def __init__(self): |
| | | self.registry = dummy_registry |
| | |
| | | def __init__(self, environ): |
| | | self.environ = environ |
| | | self.matchdict = {} |
| | | |
| | | class DummyConfigParser(object): |
| | | def read(self, x): |
| | | pass |
| | | |
| | | def has_section(self, name): |
| | | return True |
| | | |
| | | class DummyConfigParserModule(object): |
| | | ConfigParser = DummyConfigParser |
| | |
| | | self.views = [(None, view, None) for view in views] |
| | | self.__request_attrs__ = attrs |
| | | |
| | | class DummyConfigParser(object): |
| | | def __init__(self, result, defaults=None): |
| | | self.result = result |
| | | self.defaults = defaults |
| | | |
| | | def read(self, filename): |
| | | self.filename = filename |
| | | |
| | | def items(self, section): |
| | | self.section = section |
| | | if self.result is None: |
| | | from pyramid.compat import configparser |
| | | raise configparser.NoSectionError(section) |
| | | return self.result |
| | | |
| | | class DummyConfigParserFactory(object): |
| | | items = None |
| | | |
| | | def __call__(self, defaults=None): |
| | | self.defaults = defaults |
| | | self.parser = DummyConfigParser(self.items, defaults) |
| | | return self.parser |
| | | |
| | | class DummyCloser(object): |
| | | def __call__(self): |
| | | self.called = True |
| | |
| | | |
| | | def iter_entry_points(self, name): |
| | | return self.entry_points |
| | | |
| | | |
| | | class dummy_setup_logging(object): |
| | | def __call__(self, config_uri, global_conf): |
| | | self.config_uri = config_uri |
| | | self.defaults = global_conf |
| | | |
| | | |
| | | class DummyLoader(object): |
| | | def __init__(self, settings=None, app_settings=None, app=None, server=None): |
| | | if not settings: |
| | | settings = {} |
| | | if not app_settings: |
| | | app_settings = {} |
| | | self.settings = settings |
| | | self.app_settings = app_settings |
| | | self.app = app |
| | | self.server = server |
| | | self.calls = [] |
| | | |
| | | def __call__(self, uri): |
| | | import plaster |
| | | self.uri = plaster.parse_uri(uri) |
| | | return self |
| | | |
| | | def add_call(self, op, name, defaults): |
| | | self.calls.append({'op': op, 'name': name, 'defaults': defaults}) |
| | | |
| | | def get_settings(self, name=None, defaults=None): |
| | | self.add_call('settings', name, defaults) |
| | | return self.settings.get(name, {}) |
| | | |
| | | def get_wsgi_app(self, name=None, defaults=None): |
| | | self.add_call('app', name, defaults) |
| | | return self.app |
| | | |
| | | def get_wsgi_app_settings(self, name=None, defaults=None): |
| | | self.add_call('app_settings', name, defaults) |
| | | return self.app_settings |
| | | |
| | | def get_wsgi_server(self, name=None, defaults=None): |
| | | self.add_call('server', name, defaults) |
| | | return self.server |
| | | |
| | | def setup_logging(self, defaults): |
| | | self.add_call('logging', None, defaults) |
| | | self.defaults = defaults |
| | |
| | | import unittest |
| | | from pyramid.tests.test_scripts import dummy |
| | | |
| | | class TestPRequestCommand(unittest.TestCase): |
| | | def _getTargetClass(self): |
| | |
| | | |
| | | def _makeOne(self, argv, headers=None): |
| | | cmd = self._getTargetClass()(argv) |
| | | cmd.get_app = self.get_app |
| | | self._headers = headers or [] |
| | | self._out = [] |
| | | cmd.out = self.out |
| | | return cmd |
| | | |
| | | def get_app(self, spec, app_name=None, options=None): |
| | | self._spec = spec |
| | | self._app_name = app_name |
| | | self._options = options or {} |
| | | |
| | | def helloworld(environ, start_request): |
| | | self._environ = environ |
| | | self._path_info = environ['PATH_INFO'] |
| | | start_request('200 OK', self._headers) |
| | | start_request('200 OK', headers or []) |
| | | return [b'abc'] |
| | | return helloworld |
| | | self.loader = dummy.DummyLoader(app=helloworld) |
| | | self._out = [] |
| | | cmd._get_config_loader = self.loader |
| | | cmd.out = self.out |
| | | return cmd |
| | | |
| | | def out(self, msg): |
| | | self._out.append(msg) |
| | |
| | | [('Content-Type', 'text/html; charset=UTF-8')]) |
| | | command.run() |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self.loader.uri.path, 'development.ini') |
| | | self.assertEqual(self.loader.calls[0]['op'], 'logging') |
| | | self.assertEqual(self.loader.calls[1]['op'], 'app') |
| | | self.assertEqual(self.loader.calls[1]['name'], None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_path_doesnt_start_with_slash(self): |
| | |
| | | [('Content-Type', 'text/html; charset=UTF-8')]) |
| | | command.run() |
| | | self.assertEqual(self._path_info, '/abc') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self.loader.uri.path, 'development.ini') |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_has_bad_config_header(self): |
| | |
| | | command.run() |
| | | self.assertEqual(self._environ['HTTP_NAME'], 'value') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_w_basic_auth(self): |
| | |
| | | self.assertEqual(self._environ['HTTP_AUTHORIZATION'], |
| | | 'Basic dXNlcjpwYXNzd29yZA==') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_has_content_type_header_var(self): |
| | |
| | | command.run() |
| | | self.assertEqual(self._environ['CONTENT_TYPE'], 'app/foo') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_has_multiple_header_vars(self): |
| | |
| | | self.assertEqual(self._environ['HTTP_NAME'], 'value') |
| | | self.assertEqual(self._environ['HTTP_NAME2'], 'value2') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_method_get(self): |
| | |
| | | command.run() |
| | | self.assertEqual(self._environ['REQUEST_METHOD'], 'GET') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_method_post(self): |
| | |
| | | self.assertEqual(self._environ['CONTENT_LENGTH'], '-1') |
| | | self.assertEqual(self._environ['wsgi.input'], stdin) |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_method_put(self): |
| | |
| | | self.assertEqual(self._environ['CONTENT_LENGTH'], '-1') |
| | | self.assertEqual(self._environ['wsgi.input'], stdin) |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_method_patch(self): |
| | |
| | | self.assertEqual(self._environ['CONTENT_LENGTH'], '-1') |
| | | self.assertEqual(self._environ['wsgi.input'], stdin) |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_method_propfind(self): |
| | |
| | | command.run() |
| | | self.assertEqual(self._environ['REQUEST_METHOD'], 'PROPFIND') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_method_options(self): |
| | |
| | | command.run() |
| | | self.assertEqual(self._environ['REQUEST_METHOD'], 'OPTIONS') |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_with_query_string(self): |
| | |
| | | command.run() |
| | | self.assertEqual(self._environ['QUERY_STRING'], 'a=1&b=2&c') |
| | | self.assertEqual(self._path_info, '/abc') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual(self._out, ['abc']) |
| | | |
| | | def test_command_display_headers(self): |
| | |
| | | [('Content-Type', 'text/html; charset=UTF-8')]) |
| | | command.run() |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | self.assertEqual( |
| | | self._out, |
| | | ['200 OK', 'Content-Type: text/html; charset=UTF-8', 'abc']) |
| | |
| | | headers=[('Content-Type', 'image/jpeg')]) |
| | | command.run() |
| | | self.assertEqual(self._path_info, '/') |
| | | self.assertEqual(self._spec, 'development.ini') |
| | | self.assertEqual(self._app_name, None) |
| | | |
| | | self.assertEqual(self._out, [b'abc']) |
| | | |
| | | def test_command_method_configures_logging(self): |
| | | command = self._makeOne(['', 'development.ini', '/']) |
| | | called_args = [] |
| | | |
| | | def configure_logging(app_spec): |
| | | called_args.append(app_spec) |
| | | |
| | | command.configure_logging = configure_logging |
| | | command.run() |
| | | self.assertEqual(called_args, ['development.ini']) |
| | | self.assertEqual(self.loader.calls[0]['op'], 'logging') |
| | | |
| | | |
| | | class Test_main(unittest.TestCase): |
| | |
| | | |
| | | def _makeOne(self): |
| | | cmd = self._getTargetClass()([]) |
| | | cmd.bootstrap = (dummy.DummyBootstrap(),) |
| | | cmd.bootstrap = dummy.DummyBootstrap() |
| | | cmd.get_config_loader = dummy.DummyLoader() |
| | | cmd.args.config_uri = '/foo/bar/myapp.ini#myapp' |
| | | |
| | | return cmd |
| | |
| | | |
| | | def test_good_args(self): |
| | | cmd = self._getTargetClass()([]) |
| | | cmd.bootstrap = (dummy.DummyBootstrap(),) |
| | | cmd.bootstrap = dummy.DummyBootstrap() |
| | | cmd.get_config_loader = dummy.DummyLoader() |
| | | cmd.args.config_uri = '/foo/bar/myapp.ini#myapp' |
| | | cmd.args.config_args = ('a=1',) |
| | | route = dummy.DummyRoute('a', '/a') |
| | | mapper = dummy.DummyMapper(route) |
| | | cmd._get_mapper = lambda *arg: mapper |
| | | registry = self._makeRegistry() |
| | | cmd.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | cmd.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | L = [] |
| | | cmd.out = lambda msg: L.append(msg) |
| | | cmd.run() |
| | |
| | | |
| | | def test_bad_args(self): |
| | | cmd = self._getTargetClass()([]) |
| | | cmd.bootstrap = (dummy.DummyBootstrap(),) |
| | | cmd.bootstrap = dummy.DummyBootstrap() |
| | | cmd.get_config_loader = dummy.DummyLoader() |
| | | cmd.args.config_uri = '/foo/bar/myapp.ini#myapp' |
| | | cmd.args.config_vars = ('a',) |
| | | route = dummy.DummyRoute('a', '/a') |
| | |
| | | mapper = dummy.DummyMapper(route) |
| | | command._get_mapper = lambda *arg: mapper |
| | | registry = self._makeRegistry() |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | |
| | | L = [] |
| | | command.out = L.append |
| | |
| | | L = [] |
| | | command.out = L.append |
| | | registry = self._makeRegistry() |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command._get_mapper = lambda *arg: mapper |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command._get_mapper = lambda *arg: mapper |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command._get_mapper = lambda *arg: mapper |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command._get_mapper = lambda *arg: mapper |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command._get_mapper = lambda *arg: mapper |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 5) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config2.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config2.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command.args.format = 'method,name' |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | command.args.format = 'predicates,name,pattern' |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | expected = ( |
| | | "You provided invalid formats ['predicates'], " |
| | | "Available formats are ['name', 'pattern', 'view', 'method']" |
| | |
| | | |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | config_factory = dummy.DummyConfigParserFactory() |
| | | command.ConfigParser = config_factory |
| | | config_factory.items = [('format', 'method\nname')] |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | command.get_config_loader = dummy.DummyLoader( |
| | | {'proutes': {'format': 'method\nname'}}) |
| | | |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | |
| | | |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | config_factory = dummy.DummyConfigParserFactory() |
| | | command.ConfigParser = config_factory |
| | | config_factory.items = [('format', 'method name')] |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | command.get_config_loader = dummy.DummyLoader( |
| | | {'proutes': {'format': 'method name'}}) |
| | | |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | |
| | | |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | config_factory = dummy.DummyConfigParserFactory() |
| | | command.ConfigParser = config_factory |
| | | config_factory.items = [('format', 'method,name')] |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | command.get_config_loader = dummy.DummyLoader( |
| | | {'proutes': {'format': 'method,name'}}) |
| | | |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | |
| | | command = self._makeOne() |
| | | L = [] |
| | | command.out = L.append |
| | | command.bootstrap = (dummy.DummyBootstrap(registry=config.registry),) |
| | | command.bootstrap = dummy.DummyBootstrap(registry=config.registry) |
| | | result = command.run() |
| | | self.assertEqual(result, 0) |
| | | self.assertEqual(len(L), 3) |
| | |
| | | def setUp(self): |
| | | from pyramid.compat import NativeIO |
| | | self.out_ = NativeIO() |
| | | self.config_factory = dummy.DummyConfigParserFactory() |
| | | |
| | | def out(self, msg): |
| | | self.out_.write(msg) |
| | | |
| | | def _get_server(*args, **kwargs): |
| | | def server(app): |
| | | return '' |
| | | return server |
| | | |
| | | def _getTargetClass(self): |
| | | from pyramid.scripts.pserve import PServeCommand |
| | |
| | | effargs.extend(args) |
| | | cmd = self._getTargetClass()(effargs) |
| | | cmd.out = self.out |
| | | cmd.ConfigParser = self.config_factory |
| | | self.loader = dummy.DummyLoader() |
| | | cmd._get_config_loader = self.loader |
| | | return cmd |
| | | |
| | | def test_run_no_args(self): |
| | |
| | | self.assertEqual(result, 2) |
| | | self.assertEqual(self.out_.getvalue(), 'You must give a config file') |
| | | |
| | | def test_config_vars_no_command(self): |
| | | inst = self._makeOne() |
| | | inst.args.config_uri = 'foo' |
| | | inst.args.config_vars = ['a=1', 'b=2'] |
| | | result = inst.get_config_vars() |
| | | self.assertEqual(result, {'a': '1', 'b': '2'}) |
| | | |
| | | def test_parse_vars_good(self): |
| | | inst = self._makeOne('development.ini', 'a=1', 'b=2') |
| | | inst.loadserver = self._get_server |
| | | |
| | | app = dummy.DummyApp() |
| | | |
| | | def get_app(*args, **kwargs): |
| | | app.global_conf = kwargs.get('global_conf', None) |
| | | def get_app(name, global_conf): |
| | | app.name = name |
| | | app.global_conf = global_conf |
| | | return app |
| | | self.loader.get_wsgi_app = get_app |
| | | self.loader.server = lambda x: x |
| | | |
| | | inst.loadapp = get_app |
| | | inst.run() |
| | | self.assertEqual(app.global_conf, {'a': '1', 'b': '2'}) |
| | | |
| | | def test_parse_vars_bad(self): |
| | | inst = self._makeOne('development.ini', 'a') |
| | | inst.loadserver = self._get_server |
| | | self.assertRaises(ValueError, inst.run) |
| | | |
| | | def test_config_file_finds_watch_files(self): |
| | | inst = self._makeOne('development.ini') |
| | | self.config_factory.items = [( |
| | | 'watch_files', |
| | | 'foo\n/baz\npyramid.tests.test_scripts:*.py', |
| | | )] |
| | | inst.pserve_file_config('/base/path.ini', global_conf={'a': '1'}) |
| | | self.assertEqual(self.config_factory.defaults, { |
| | | loader = self.loader('/base/path.ini') |
| | | loader.settings = {'pserve': { |
| | | 'watch_files': 'foo\n/baz\npyramid.tests.test_scripts:*.py', |
| | | }} |
| | | inst.pserve_file_config(loader, global_conf={'a': '1'}) |
| | | self.assertEqual(loader.calls[0]['defaults'], { |
| | | 'a': '1', |
| | | 'here': os.path.abspath('/base'), |
| | | }) |
| | | self.assertEqual(inst.watch_files, set([ |
| | | os.path.abspath('/base/foo'), |
| | |
| | | |
| | | def test_config_file_finds_open_url(self): |
| | | inst = self._makeOne('development.ini') |
| | | self.config_factory.items = [( |
| | | 'open_url', 'http://127.0.0.1:8080/', |
| | | )] |
| | | inst.pserve_file_config('/base/path.ini', global_conf={'a': '1'}) |
| | | self.assertEqual(self.config_factory.defaults, { |
| | | loader = self.loader('/base/path.ini') |
| | | loader.settings = {'pserve': { |
| | | 'open_url': 'http://127.0.0.1:8080/', |
| | | }} |
| | | inst.pserve_file_config(loader, global_conf={'a': '1'}) |
| | | self.assertEqual(loader.calls[0]['defaults'], { |
| | | 'a': '1', |
| | | 'here': os.path.abspath('/base'), |
| | | }) |
| | | self.assertEqual(inst.open_url, 'http://127.0.0.1:8080/') |
| | | |
| | | def test__guess_server_url(self): |
| | | def test_guess_server_url(self): |
| | | inst = self._makeOne('development.ini') |
| | | self.config_factory.items = [( |
| | | 'port', '8080', |
| | | )] |
| | | url = inst._guess_server_url( |
| | | '/base/path.ini', 'main', global_conf={'a': '1'}) |
| | | self.assertEqual(self.config_factory.defaults, { |
| | | loader = self.loader('/base/path.ini') |
| | | loader.settings = {'server:foo': { |
| | | 'port': '8080', |
| | | }} |
| | | url = inst.guess_server_url(loader, 'foo', global_conf={'a': '1'}) |
| | | self.assertEqual(loader.calls[0]['defaults'], { |
| | | 'a': '1', |
| | | 'here': os.path.abspath('/base'), |
| | | }) |
| | | self.assertEqual(self.config_factory.parser.section, 'server:main') |
| | | self.assertEqual(url, 'http://127.0.0.1:8080') |
| | | |
| | | def test_reload_call_hupper_with_correct_args(self): |
| | |
| | | from pyramid.scripts.pshell import PShellCommand |
| | | return PShellCommand |
| | | |
| | | def _makeOne(self, patch_bootstrap=True, patch_config=True, |
| | | def _makeOne(self, patch_bootstrap=True, patch_loader=True, |
| | | patch_args=True, patch_options=True): |
| | | cmd = self._getTargetClass()([]) |
| | | |
| | | if patch_bootstrap: |
| | | self.bootstrap = dummy.DummyBootstrap() |
| | | cmd.bootstrap = (self.bootstrap,) |
| | | if patch_config: |
| | | self.config_factory = dummy.DummyConfigParserFactory() |
| | | cmd.ConfigParser = self.config_factory |
| | | cmd.bootstrap = self.bootstrap |
| | | if patch_loader: |
| | | self.loader = dummy.DummyLoader() |
| | | cmd.get_config_loader = self.loader |
| | | if patch_args: |
| | | class Args(object): pass |
| | | self.args = Args() |
| | |
| | | |
| | | command.default_runner = shell |
| | | command.run() |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':self.bootstrap.app, 'root':self.bootstrap.root, |
| | |
| | | self.assertEqual( |
| | | out_calls, ['could not find a shell named "unknown_python_shell"'] |
| | | ) |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertTrue(self.bootstrap.closer.called) |
| | | |
| | |
| | | command.args.python_shell = 'ipython' |
| | | |
| | | command.run() |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':self.bootstrap.app, 'root':self.bootstrap.root, |
| | |
| | | command = self._makeOne() |
| | | model = dummy.Dummy() |
| | | user = dummy.Dummy() |
| | | self.config_factory.items = [('m', model), ('User', user)] |
| | | self.loader.settings = {'pshell': {'m': model, 'User': user}} |
| | | shell = dummy.DummyShell() |
| | | command.run(shell) |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':self.bootstrap.app, 'root':self.bootstrap.root, |
| | |
| | | env['a'] = 1 |
| | | env['root'] = 'root override' |
| | | env['none'] = None |
| | | self.config_factory.items = [('setup', setup)] |
| | | self.loader.settings = {'pshell': {'setup': setup}} |
| | | shell = dummy.DummyShell() |
| | | command.run(shell) |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':self.bootstrap.app, 'root':'root override', |
| | |
| | | 'python': dshell, |
| | | } |
| | | ) |
| | | self.config_factory.items = [ |
| | | ('default_shell', 'bpython python\nipython')] |
| | | self.loader.settings = {'pshell': { |
| | | 'default_shell': 'bpython python\nipython'}} |
| | | command.run() |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertTrue(dshell.called) |
| | | |
| | |
| | | env['a'] = 1 |
| | | env['m'] = 'model override' |
| | | env['root'] = 'root override' |
| | | self.config_factory.items = [('setup', setup), ('m', model)] |
| | | self.loader.settings = {'pshell': {'setup': setup, 'm': model}} |
| | | shell = dummy.DummyShell() |
| | | command.run(shell) |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':self.bootstrap.app, 'root':'root override', |
| | |
| | | env['a'] = 1 |
| | | env['root'] = 'root override' |
| | | model = dummy.Dummy() |
| | | self.config_factory.items = [('setup', 'abc'), |
| | | ('m', model)] |
| | | self.loader.settings = {'pshell': {'setup': 'abc', 'm': model}} |
| | | command.args.setup = setup |
| | | shell = dummy.DummyShell() |
| | | command.run(shell) |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':self.bootstrap.app, 'root':'root override', |
| | |
| | | def test_command_custom_section_override(self): |
| | | command = self._makeOne() |
| | | dummy_ = dummy.Dummy() |
| | | self.config_factory.items = [('app', dummy_), ('root', dummy_), |
| | | ('registry', dummy_), ('request', dummy_)] |
| | | self.loader.settings = {'pshell': { |
| | | 'app': dummy_, 'root': dummy_, 'registry': dummy_, |
| | | 'request': dummy_}} |
| | | shell = dummy.DummyShell() |
| | | command.run(shell) |
| | | self.assertTrue(self.config_factory.parser) |
| | | self.assertEqual(self.config_factory.parser.filename, |
| | | '/foo/bar/myapp.ini') |
| | | self.assertEqual(self.bootstrap.a[0], '/foo/bar/myapp.ini#myapp') |
| | | self.assertEqual(shell.env, { |
| | | 'app':dummy_, 'root':dummy_, 'registry':dummy_, 'request':dummy_, |
| | |
| | | |
| | | def _makeOne(self): |
| | | cmd = self._getTargetClass()([]) |
| | | cmd.bootstrap = (dummy.DummyBootstrap(),) |
| | | cmd.bootstrap = dummy.DummyBootstrap() |
| | | cmd.setup_logging = dummy.dummy_setup_logging() |
| | | cmd.args.config_uri = '/foo/bar/myapp.ini#myapp' |
| | | return cmd |
| | | |
| | |
| | | |
| | | def _makeOne(self, registry=None): |
| | | cmd = self._getTargetClass()([]) |
| | | cmd.bootstrap = (dummy.DummyBootstrap(registry=registry),) |
| | | cmd.bootstrap = dummy.DummyBootstrap(registry=registry) |
| | | cmd.setup_logging = dummy.dummy_setup_logging() |
| | | cmd.args.config_uri = '/foo/bar/myapp.ini#myapp' |
| | | return cmd |
| | | |
| | |
| | | 'venusian >= 1.0a3', # ``ignore`` |
| | | 'translationstring >= 0.4', # py3 compat |
| | | 'PasteDeploy >= 1.5.0', # py3 compat |
| | | 'plaster', |
| | | 'plaster_pastedeploy', |
| | | 'hupper', |
| | | ] |
| | | |