openstack/deb-python-pecan
Code Issues Proposed changes

Merge pull request #78 from ryanpetrello/next

Browse Source 
A new implementation of the `$ pecan` command, including `serve`, `create`, and `shell`.
This commit is contained in:
markmcclain
2012-03-15 15:21:26 -07:00
parent 40913103ff 31fe5fb437
commit 8af6a4a60c
52 changed files with 802 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
MANIFEST.in
View File

@@ -1,2 +1,2 @@
recursive-include pecan/templates/project *
include pecan/templates/project/*
recursive-include pecan/scaffolds/base *
include pecan/scaffolds/base/*

10
docs/source/databases.rst
View File

@@ -1,14 +1,14 @@
.. _databases:
Working with Databases, Transactions, and ORM's
=============
===============================================
Out of the box, Pecan provides no opinionated support for working with databases,
but it's easy to hook into your ORM of choice with minimal effort. This article
details best practices for integrating the popular Python ORM, SQLAlchemy, into
your Pecan project.
``init_model`` and Preparing Your Model
----------------
---------------------------------------
Pecan's default quickstart project includes an empty stub directory for implementing
your model as you see fit::
@@ -96,7 +96,8 @@ Here's what a sample Pecan configuration file with database bindings might look
Session.remove()
Binding Within the Application
----------------
------------------------------
There are several approaches that can be taken to wrap your application's requests with calls
to appropriate model function calls. One approach is WSGI middleware. We also recommend
Pecan :ref:`hooks`. Pecan comes with ``TransactionHook``, a hook which can
@@ -146,7 +147,8 @@ manner:
Also note that there is a useful ``@after_commit`` decorator provided in :ref:`pecan_decorators`.
Splitting Reads and Writes
----------------
--------------------------
Employing the strategy above with ``TransactionHook`` makes it very simple to split database
reads and writes based upon HTTP methods (i.e., GET/HEAD requests are read-only and would potentially
be routed to a read-only database slave, while POST/PUT/DELETE requests require writing, and

2
docs/source/hooks.rst
View File

@@ -45,7 +45,7 @@ was chosen by Pecan's routing.
``on_error`` is passed a shared state object **and** the original exception.
Attaching Hooks
--------------
---------------
Hooks can be attached in a project-wide manner by specifying a list of hooks
in your project's ``app.py`` file::

4
docs/source/sessions.rst
View File

@@ -1,7 +1,7 @@
.. _session:
Working with Sessions and User Authentication
=============
=============================================
Out of the box, Pecan provides no opinionated support for managing user sessions,
but it's easy to hook into your session framework of choice with minimal
effort.
@@ -10,7 +10,7 @@ This article details best practices for integrating the popular session
framework, `Beaker <http://beaker.groovie.org>`_, into your Pecan project.
Setting up Session Management
----------------
-----------------------------
There are several approaches that can be taken to set up session management.
One approach is WSGI middleware. Another is Pecan :ref:`hooks`.

12
pecan/commands/__init__.py
View File

@@ -1,8 +1,4 @@
"""
PasteScript commands for Pecan.
"""
from runner import CommandRunner # noqa
from create import CreateCommand # noqa
from shell import ShellCommand # noqa
from serve import ServeCommand # noqa
from base import CommandRunner, BaseCommand
from serve import ServeCommand
from shell import ShellCommand
from create import CreateCommand

144
pecan/commands/base.py
View File

@@ -1,49 +1,119 @@
"""
PasteScript base command for Pecan.
"""
from pecan import load_app
from paste.script import command as paste_command
import pkg_resources
import os.path
import argparse
import logging
import sys
from warnings import warn
log = logging.getLogger(__name__)
class Command(paste_command.Command):
"""
Base class for Pecan commands.
class HelpfulArgumentParser(argparse.ArgumentParser):
This provides some standard functionality for interacting with Pecan
applications and handles some of the basic PasteScript command cruft.
def error(self, message): # pragma: nocover
"""error(message: string)
See ``paste.script.command.Command`` for more information.
"""
Prints a usage message incorporating the message to stderr and
exits.
# command information
group_name = 'Pecan'
summary = ''
If you override this in a subclass, it should not return -- it
should either exit or raise an exception.
"""
self.print_help(sys.stderr)
self._print_message('\n')
self.exit(2, '%s: %s\n' % (self.prog, message))
# command parser
parser = paste_command.Command.standard_parser()
class CommandManager(object):
""" Used to discover `pecan.command` entry points. """
def __init__(self):
self.commands = {}
self.load_commands()
def load_commands(self):
for ep in pkg_resources.iter_entry_points('pecan.command'):
log.debug('%s loading plugin %s', self.__class__.__name__, ep)
try:
cmd = ep.load()
assert hasattr(cmd, 'run')
except Exception, e: # pragma: nocover
warn("Unable to load plugin %s: %s" % (ep, e), RuntimeWarning)
continue
self.add({ep.name: cmd})
def add(self, cmd):
self.commands.update(cmd)
class CommandRunner(object):
""" Dispatches `pecan` command execution requests. """
def __init__(self):
self.manager = CommandManager()
self.parser = HelpfulArgumentParser(
version='Pecan %s' % self.version,
add_help=True
)
self.parse_sub_commands()
def parse_sub_commands(self):
subparsers = self.parser.add_subparsers(
dest='command_name',
metavar='command'
)
for name, cmd in self.commands.items():
sub = subparsers.add_parser(
name,
help=cmd.summary
)
for arg in getattr(cmd, 'arguments', tuple()):
arg = arg.copy()
sub.add_argument(arg.pop('command'), **arg)
def run(self, args):
ns = self.parser.parse_args(args)
self.commands[ns.command_name]().run(ns)
@classmethod
def handle_command_line(cls): # pragma: nocover
runner = CommandRunner()
runner.run(sys.argv[1:])
@property
def version(self):
try:
return paste_command.Command.run(self, args)
except paste_command.BadCommand, ex:
ex.args[0] = self.parser.error(ex.args[0])
raise
dist = pkg_resources.get_distribution('Pecan')
if os.path.dirname(os.path.dirname(__file__)) == dist.location:
return dist.version # pragma: nocover
else:
return '(development)'
except: # pragma: nocover
return '(development)'
@property
def commands(self):
return self.manager.commands
class BaseCommand(object):
""" Base class for Pecan commands. """
class __metaclass__(type):
@property
def summary(cls):
return cls.__doc__.strip().splitlines()[0].rstrip('.')
arguments = ({
'command': 'config_file',
'help': 'a Pecan configuration file'
},)
def run(self, args):
self.args = args
def load_app(self):
return load_app(self.validate_file(self.args))
def logging_file_config(self, config_file):
if os.path.splitext(config_file)[1].lower() == '.ini':
paste_command.Command.logging_file_config(self, config_file)
def validate_file(self, argv):
if not argv or not os.path.isfile(argv[0]):
raise paste_command.BadCommand(
'This command needs a valid config file.'
)
return argv[0]
def command(self):
pass
from pecan import load_app
if not os.path.isfile(self.args.config_file):
raise RuntimeError('`%s` is not a file.' % self.args.config_file)
return load_app(self.args.config_file)

84
pecan/commands/create.py
View File

@@ -1,44 +1,60 @@
"""
PasteScript create command for Pecan.
Create command for Pecan
"""
from paste.script.create_distro import CreateDistroCommand
import pkg_resources
import logging
from warnings import warn
from pecan.commands import BaseCommand
from pecan.scaffolds import DEFAULT_SCAFFOLD
from base import Command
from pecan.templates import DEFAULT_TEMPLATE
import copy
import sys
log = logging.getLogger(__name__)
class CreateCommand(CreateDistroCommand, Command):
class ScaffoldManager(object):
""" Used to discover `pecan.scaffold` entry points. """
def __init__(self):
self.scaffolds = {}
self.load_scaffolds()
def load_scaffolds(self):
for ep in pkg_resources.iter_entry_points('pecan.scaffold'):
log.debug('%s loading scaffold %s', self.__class__.__name__, ep)
try:
cmd = ep.load()
assert hasattr(cmd, 'copy_to')
except Exception, e: # pragma: nocover
warn(
"Unable to load scaffold %s: %s" % (ep, e), RuntimeWarning
)
continue
self.add({ep.name: cmd})
def add(self, cmd):
self.scaffolds.update(cmd)
class CreateCommand(BaseCommand):
"""
Creates the file layout for a new Pecan distribution.
For a template to show up when using this command, its name must begin
with "pecan-". Although not required, it should also include the "Pecan"
egg plugin for user convenience.
Creates the file layout for a new Pecan scaffolded project.
"""
# command information
summary = __doc__.strip().splitlines()[0].rstrip('.')
description = None
manager = ScaffoldManager()
def command(self):
if not self.options.list_templates:
if not self.options.templates:
self.options.templates = [DEFAULT_TEMPLATE]
try:
return CreateDistroCommand.command(self)
except LookupError, ex:
sys.stderr.write('%s\n\n' % ex)
CreateDistroCommand.list_templates(self)
return 2
arguments = ({
'command': 'project_name',
'help': 'the (package) name of the new project'
}, {
'metavar': 'template_name',
'command': 'template_name',
'help': 'a registered Pecan template',
'nargs': '?',
'default': DEFAULT_SCAFFOLD,
'choices': manager.scaffolds.keys()
})
def all_entry_points(self):
entry_points = []
for entry in CreateDistroCommand.all_entry_points(self):
if entry.name.startswith('pecan-'):
entry = copy.copy(entry)
entry_points.append(entry)
entry.name = entry.name[6:]
return entry_points
def run(self, args):
super(CreateCommand, self).run(args)
self.manager.scaffolds[args.template_name]().copy_to(
args.project_name
)

131
pecan/commands/runner.py
View File

@@ -1,131 +0,0 @@
"""
PasteScript command runner.
"""
from paste.script import command as paste_command
import optparse
import os
import pkg_resources
import sys
import warnings
class CommandRunner(object):
"""
Dispatches command execution requests.
This is a custom PasteScript command runner that is specific to Pecan
commands. For a command to show up, its name must begin with "pecan-".
It is also recommended that its group name be set to "Pecan" so that it
shows up under that group when using ``paster`` directly.
"""
def __init__(self):
# set up the parser
self.parser = optparse.OptionParser(
add_help_option=False,
version='Pecan %s' % self.get_version(),
usage='%prog [options] COMMAND [command_options]'
)
self.parser.disable_interspersed_args()
self.parser.add_option('-h', '--help',
action='store_true',
dest='show_help',
help='show detailed help message')
# suppress BaseException.message warnings for BadCommand
if sys.version_info < (2, 7):
warnings.filterwarnings(
'ignore',
'BaseException\.message has been deprecated as of Python 2\.6',
DeprecationWarning,
paste_command.__name__.replace('.', '\\.'))
# register Pecan as a system plugin when using the custom runner
paste_command.system_plugins.append('Pecan')
def get_command_template(self, command_names):
if not command_names:
max_length = 10
else:
max_length = max([len(name) for name in command_names])
return ' %%-%ds %%s\n' % max_length
def get_commands(self):
commands = {}
for name, command in paste_command.get_commands().iteritems():
if name.startswith('pecan-'):
commands[name[6:]] = command.load()
return commands
def get_version(self):
try:
dist = pkg_resources.get_distribution('Pecan')
if os.path.dirname(os.path.dirname(__file__)) == dist.location:
return dist.version
else:
return '(development)'
except:
return '(development)'
def print_usage(self, file=sys.stdout):
self.parser.print_help(file=file)
file.write('\n')
command_groups = {}
commands = self.get_commands()
if not commands:
file.write('No commands registered.\n')
return
command_template = self.get_command_template(commands.keys())
for name, command in commands.iteritems():
group_name = command.group_name
if group_name.lower() == 'pecan':
group_name = ''
command_groups.setdefault(group_name, {})[name] = command
command_groups = sorted(command_groups.items())
for i, (group, commands) in enumerate(command_groups):
file.write('%s:\n' % (group or 'Commands'))
for name, command in sorted(commands.items()):
file.write(command_template % (name, command.summary))
if i + 1 < len(command_groups):
file.write('\n')
def print_known_commands(self, file=sys.stderr):
commands = self.get_commands()
command_names = sorted(commands.keys())
if not command_names:
file.write('No commands registered.\n')
return
file.write('Known commands:\n')
command_template = self.get_command_template(command_names)
for name in command_names:
file.write(command_template % (name, commands[name].summary))
def run(self, args):
options, args = self.parser.parse_args(args)
if not args:
self.print_usage()
return 0
command_name = args.pop(0)
commands = self.get_commands()
if command_name not in commands:
sys.stderr.write('Command %s not known\n\n' % command_name)
self.print_known_commands()
return 1
else:
command = commands[command_name](command_name)
if options.show_help:
return command.run(['-h'])
else:
return command.run(args)
@classmethod
def handle_command_line(cls):
try:
runner = CommandRunner()
exit_code = runner.run(sys.argv[1:])
except paste_command.BadCommand, ex:
sys.stderr.write('%s\n' % ex)
exit_code = ex.exit_code
sys.exit(exit_code)

78
pecan/commands/serve.py
View File

@@ -1,64 +1,38 @@
"""
PasteScript serve command for Pecan.
Serve command for Pecan.
"""
from paste.script.serve import ServeCommand as _ServeCommand
from base import Command
import re
import os
from pecan.commands import BaseCommand
class ServeCommand(_ServeCommand, Command):
class ServeCommand(BaseCommand):
"""
Serves a Pecan web application.
This command serves a Pecan web application using the provided
configuration file for the server and application.
If start/stop/restart is given, then --daemon is implied, and it will
start (normal operation), stop (--stop-daemon), or do both.
"""
# command information
usage = 'CONFIG_FILE [start|stop|restart|status]'
summary = __doc__.strip().splitlines()[0].rstrip('.')
description = '\n'.join(
map(lambda s: s.rstrip(), __doc__.strip().splitlines()[2:])
)
def run(self, args):
super(ServeCommand, self).run(args)
app = self.load_app()
self.serve(app, app.config)
# command options/arguments
max_args = 2
# command parser
parser = _ServeCommand.parser
parser.remove_option('-n')
parser.remove_option('-s')
parser.remove_option('--server-name')
# configure scheme regex
_scheme_re = re.compile(r'.*')
def command(self):
# set defaults for removed options
setattr(self.options, 'app_name', None)
setattr(self.options, 'server', None)
setattr(self.options, 'server_name', None)
# run the base command
_ServeCommand.command(self)
def loadserver(self, server_spec, name, relative_to, **kw):
return (lambda app: WSGIRefServer(app.config.server.host, app.config.server.port, app))
def loadapp(self, app_spec, name, relative_to, **kw):
return self.load_app()
def WSGIRefServer(host, port, app, **options):
"""
A very simple approach for a WSGI server.
"""
from wsgiref.simple_server import make_server
port = int(port)
srv = make_server(host, port, app, **options)
srv.serve_forever()
def serve(self, app, conf):
"""
A very simple approach for a WSGI server.
"""
from wsgiref.simple_server import make_server
host, port = conf.server.host, int(conf.server.port)
srv = make_server(host, port, app)
print 'Starting server in PID %s' % os.getpid()
if host == '0.0.0.0':
print 'serving on 0.0.0.0:%s, view at http://127.0.0.1:%s' % \
(port, port)
else:
print "serving on http://%s:%s" % (host, port)
try:
srv.serve_forever()
except KeyboardInterrupt:
# allow CTRL+C to shutdown
pass

19
pecan/commands/shell.py
View File

@@ -1,27 +1,18 @@
"""
PasteScript shell command for Pecan.
Shell command for Pecan.
"""
from pecan.commands import BaseCommand
from webtest import TestApp
from base import Command
import sys
class ShellCommand(Command):
class ShellCommand(BaseCommand):
"""
Open an interactive shell with the Pecan app loaded.
"""
# command information
usage = 'CONFIG_NAME'
summary = __doc__.strip().splitlines()[0].rstrip('.')
# command options/arguments
min_args = 1
max_args = 1
def command(self):
def run(self, args):
super(ShellCommand, self).run(args)
# load the application
app = self.load_app()

40
pecan/compat.py Normal file
View File

@@ -0,0 +1,40 @@
import sys
# True if we are running on Python 3.
PY3 = sys.version_info[0] == 3
if PY3: # pragma: no cover
text_type = str
else:
text_type = unicode
def bytes_(s, encoding='latin-1', errors='strict'):
""" If ``s`` is an instance of ``text_type``, return
``s.encode(encoding, errors)``, otherwise return ``s``"""
if isinstance(s, text_type): # pragma: no cover
return s.encode(encoding, errors)
return s
if PY3: # pragma: no cover
def native_(s, encoding='latin-1', errors='strict'):
""" If ``s`` is an instance of ``text_type``, return
``s``, otherwise return ``str(s, encoding, errors)``"""
if isinstance(s, text_type):
return s
return str(s, encoding, errors)
else:
def native_(s, encoding='latin-1', errors='strict'): # noqa
""" If ``s`` is an instance of ``text_type``, return
``s.encode(encoding, errors)``, otherwise return ``str(s)``"""
if isinstance(s, text_type):
return s.encode(encoding, errors)
return str(s)
native_.__doc__ = """
Python 3: If ``s`` is an instance of ``text_type``, return ``s``, otherwise
return ``str(s, encoding, errors)``
Python 2: If ``s`` is an instance of ``text_type``, return
``s.encode(encoding, errors)``, otherwise return ``str(s)``
"""

122
pecan/scaffolds/__init__.py Normal file
View File

@@ -0,0 +1,122 @@
import sys
import os
import re
import pkg_resources
from string import Template
from pecan.compat import native_, bytes_
DEFAULT_SCAFFOLD = 'base'
_bad_chars_re = re.compile('[^a-zA-Z0-9_]')
class PecanScaffold(object):
"""
A base Pecan scaffold. New scaffolded implementations should extend this
class and define a ``_scaffold_dir`` attribute, e.g.,
class CoolAddOnScaffold(PecanScaffold):
_scaffold_dir = ('package', os.path.join('scaffolds', 'scaffold_name'))
...where...
pkg_resources.resource_listdir(_scaffold_dir[0], _scaffold_dir[1]))
...points to some scaffold directory root.
"""
def normalize_output_dir(self, dest):
return os.path.abspath(os.path.normpath(dest))
def normalize_pkg_name(self, dest):
return _bad_chars_re.sub('', dest.lower())
def copy_to(self, dest, **kw):
output_dir = self.normalize_output_dir(dest)
pkg_name = self.normalize_pkg_name(dest)
copy_dir(self._scaffold_dir, output_dir, {'package': pkg_name}, **kw)
class BaseScaffold(PecanScaffold):
_scaffold_dir = ('pecan', os.path.join('scaffolds', 'base'))
def copy_dir(source, dest, variables, out_=sys.stdout, i=0):
"""
Copies the ``source`` directory to the ``dest`` directory, where
``source`` is some tuple representing an installed package and a
subdirectory in the package, e.g.,
('pecan', os.path.join('scaffolds', 'base'))
('pecan_extension', os.path.join('scaffolds', 'scaffold_name'))
``variables``: A dictionary of variables to use in any substitutions.
Substitution is performed via ``string.Template``.
``out_``: File object to write to (default is sys.stdout).
"""
def out(msg):
out_.write('%s%s' % (' ' * (i * 2), msg))
out_.write('\n')
out_.flush()
names = sorted(pkg_resources.resource_listdir(source[0], source[1]))
if not os.path.exists(dest):
out('Creating %s' % dest)
makedirs(dest)
else:
out('%s already exists' % dest)
return
for name in names:
full = '/'.join([source[1], name])
dest_full = os.path.join(dest, substitute_filename(name, variables))
sub_file = False
if dest_full.endswith('_tmpl'):
dest_full = dest_full[:-5]
sub_file = True
if pkg_resources.resource_isdir(source[0], full):
out('Recursing into %s' % os.path.basename(full))
copy_dir((source[0], full), dest_full, variables, out_, i + 1)
continue
else:
content = pkg_resources.resource_string(source[0], full)
if sub_file:
content = render_template(content, variables)
if content is None:
continue # pragma: no cover
out('Copying %s to %s' % (full, dest_full))
f = open(dest_full, 'wb')
f.write(content)
f.close()
def makedirs(directory):
""" Resursively create a named directory. """
parent = os.path.dirname(os.path.abspath(directory))
if not os.path.exists(parent):
makedirs(parent)
os.mkdir(directory)
def substitute_filename(fn, variables):
""" Substitute +variables+ in file directory names. """
for var, value in variables.items():
fn = fn.replace('+%s+' % var, str(value))
return fn
def render_template(content, variables):
"""
Return a bytestring representing a templated file based on the
input (content) and the variable names defined (vars).
"""
fsenc = sys.getfilesystemencoding()
content = native_(content, fsenc)
return bytes_(Template(content).substitute(variables), fsenc)

0
pecan/templates/project/+package+/__init__.py → pecan/scaffolds/base/+package+/__init__.py
View File

0
pecan/templates/project/+package+/app.py_tmpl → pecan/scaffolds/base/+package+/app.py_tmpl
View File

0
pecan/templates/project/+package+/controllers/__init__.py → pecan/scaffolds/base/+package+/controllers/__init__.py
View File

0
pecan/templates/project/+package+/controllers/root.py → pecan/scaffolds/base/+package+/controllers/root.py
View File

0
pecan/templates/project/+package+/model/__init__.py → pecan/scaffolds/base/+package+/model/__init__.py
View File

0
pecan/templates/project/+package+/templates/error.html → pecan/scaffolds/base/+package+/templates/error.html
View File

0
pecan/templates/project/+package+/templates/index.html → pecan/scaffolds/base/+package+/templates/index.html
View File

0
pecan/templates/project/+package+/templates/layout.html → pecan/scaffolds/base/+package+/templates/layout.html
View File

0
pecan/templates/project/+package+/tests/__init__.py_tmpl → pecan/scaffolds/base/+package+/tests/__init__.py_tmpl
View File

0
pecan/templates/project/+package+/tests/config.py_tmpl → pecan/scaffolds/base/+package+/tests/config.py_tmpl
View File

0
pecan/templates/project/+package+/tests/test_functional.py_tmpl → pecan/scaffolds/base/+package+/tests/test_functional.py_tmpl
View File

0
pecan/templates/project/+package+/tests/test_units.py → pecan/scaffolds/base/+package+/tests/test_units.py
View File

0
pecan/templates/project/MANIFEST.in → pecan/scaffolds/base/MANIFEST.in
View File

0
pecan/templates/project/config.py_tmpl → pecan/scaffolds/base/config.py_tmpl
View File

0
pecan/templates/project/public/css/style.css → pecan/scaffolds/base/public/css/style.css
View File

0
pecan/templates/project/public/images/logo.png → pecan/scaffolds/base/public/images/logo.png
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 34 KiB

After

Width:  |  Height:  |  Size: 34 KiB

0
pecan/templates/project/setup.cfg_tmpl → pecan/scaffolds/base/setup.cfg_tmpl
View File

3
pecan/templates/project/setup.py_tmpl → pecan/scaffolds/base/setup.py_tmpl
View File

@@ -7,7 +7,7 @@ except ImportError:
from setuptools import setup, find_packages
setup(
name = '${project}',
name = '${package}',
version = '0.1',
description = '',
author = '',
@@ -17,7 +17,6 @@ setup(
],
test_suite = '${package}',
zip_safe = False,
paster_plugins = ${egg_plugins},
include_package_data = True,
packages = find_packages(exclude=['ez_setup'])
)

9
pecan/templates/__init__.py
View File

@@ -1,9 +0,0 @@
from paste.script.templates import Template
DEFAULT_TEMPLATE = 'base'
class BaseTemplate(Template):
summary = 'Template for creating a basic Pecan project'
_template_dir = 'project'
egg_plugins = ['Pecan']

0
pecan/tests/test_config/bad/importerror.py → pecan/tests/config_fixtures/bad/importerror.py
View File

0
pecan/tests/test_config/bad/module_and_underscore.py → pecan/tests/config_fixtures/bad/module_and_underscore.py
View File

0
pecan/tests/test_config/bad/syntaxerror.py → pecan/tests/config_fixtures/bad/syntaxerror.py
View File

0
pecan/tests/test_config/config.py → pecan/tests/config_fixtures/config.py
View File

0
pecan/tests/test_config/empty.py → pecan/tests/config_fixtures/empty.py
View File

0
pecan/tests/test_config/foobar.py → pecan/tests/config_fixtures/foobar.py
View File

0
pecan/tests/test_config/forcedict.py → pecan/tests/config_fixtures/forcedict.py
View File

0
pecan/tests/scaffold_fixtures/__init__.py Normal file
View File

1
pecan/tests/scaffold_fixtures/content_sub/bar/spam.txt_tmpl Normal file
View File

@@ -0,0 +1 @@
Pecan ${package}

1
pecan/tests/scaffold_fixtures/content_sub/foo_tmpl Normal file
View File

@@ -0,0 +1 @@
YAR ${package}

1
pecan/tests/scaffold_fixtures/file_sub/bar_+package+/spam.txt Normal file
View File

@@ -0,0 +1 @@
Pecan

1
pecan/tests/scaffold_fixtures/file_sub/foo_+package+ Normal file
View File

@@ -0,0 +1 @@
YAR

1
pecan/tests/scaffold_fixtures/simple/bar/spam.txt Normal file
View File

@@ -0,0 +1 @@
Pecan

1
pecan/tests/scaffold_fixtures/simple/foo Normal file
View File

@@ -0,0 +1 @@
YAR

28
pecan/tests/test_base.py
View File

@@ -1,11 +1,12 @@
import sys
import warnings
from paste.translogger import TransLogger
from webtest import TestApp
if sys.version_info < (2, 7):
import unittest2 as unittest
else:
import unittest
import unittest # noqa
from pecan import (
Pecan, expose, request, response, redirect, abort, make_app,
@@ -172,9 +173,11 @@ class TestLookups(unittest.TestCase):
def _lookup(self, someID):
return 'Bad arg spec'
app = TestApp(Pecan(RootController()))
r = app.get('/foo/bar', expect_errors=True)
assert r.status_int == 404
with warnings.catch_warnings():
warnings.simplefilter("ignore")
app = TestApp(Pecan(RootController()))
r = app.get('/foo/bar', expect_errors=True)
assert r.status_int == 404
class TestControllerArguments(unittest.TestCase):
@@ -409,7 +412,10 @@ class TestControllerArguments(unittest.TestCase):
assert r.body == 'optional: 7'
def test_optional_arg_with_multiple_url_encoded_dictionary_kwargs(self):
r = self.app_.post('/optional', {'id': 'Some%20Number', 'dummy': 'dummy'})
r = self.app_.post('/optional', {
'id': 'Some%20Number',
'dummy': 'dummy'
})
assert r.status_int == 200
assert r.body == 'optional: Some%20Number'
@@ -572,7 +578,9 @@ class TestControllerArguments(unittest.TestCase):
assert r.body == 'variable_kwargs: dummy=dummy, id=2'
def test_multiple_variable_kwargs_with_explicit_encoded_kwargs(self):
r = self.app_.get('/variable_kwargs?id=Two%21&dummy=This%20is%20a%20test')
r = self.app_.get(
'/variable_kwargs?id=Two%21&dummy=This%20is%20a%20test'
)
assert r.status_int == 200
assert r.body == 'variable_kwargs: dummy=This is a test, id=Two!'
@@ -895,8 +903,10 @@ class TestFileTypeExtensions(unittest.TestCase):
assert r.status_int == 200
assert r.body == '/'
r = app.get('/index.txt', expect_errors=True)
assert r.status_int == 404
with warnings.catch_warnings():
warnings.simplefilter("ignore")
r = app.get('/index.txt', expect_errors=True)
assert r.status_int == 404
class TestCanonicalRouting(unittest.TestCase):
@@ -964,7 +974,7 @@ class TestCanonicalRouting(unittest.TestCase):
def test_posts_fail(self):
try:
r = self.app_.post('/sub', dict(foo=1))
self.app_.post('/sub', dict(foo=1))
raise Exception("Post should fail")
except Exception, e:
assert isinstance(e, RuntimeError)

53
pecan/tests/test_commands.py Normal file
View File

@@ -0,0 +1,53 @@
import unittest
class TestCommandManager(unittest.TestCase):
def test_commands(self):
from pecan.commands import ServeCommand, ShellCommand, CreateCommand
from pecan.commands.base import CommandManager
m = CommandManager()
assert m.commands['serve'] == ServeCommand
assert m.commands['shell'] == ShellCommand
assert m.commands['create'] == CreateCommand
class TestCommandRunner(unittest.TestCase):
def test_commands(self):
from pecan.commands import (
ServeCommand, ShellCommand, CreateCommand, CommandRunner
)
runner = CommandRunner()
assert runner.commands['serve'] == ServeCommand
assert runner.commands['shell'] == ShellCommand
assert runner.commands['create'] == CreateCommand
def test_run(self):
from pecan.commands import CommandRunner
runner = CommandRunner()
with self.assertRaises(RuntimeError):
runner.run(['serve', 'missing_file.py'])
class TestCreateCommand(unittest.TestCase):
def test_run(self):
from pecan.commands import CreateCommand
class FakeArg(object):
project_name = 'default'
template_name = 'default'
class FakeScaffold(object):
def copy_to(self, project_name):
assert project_name == 'default'
class FakeManager(object):
scaffolds = {
'default': FakeScaffold
}
c = CreateCommand()
c.manager = FakeManager()
c.run(FakeArg())

29
pecan/tests/test_conf.py
View File

@@ -20,7 +20,7 @@ class TestConf(TestCase):
conf = configuration.initconf()
conf.update(configuration.conf_from_file(os.path.join(
__here__,
'test_config/config.py'
'config_fixtures/config.py'
)))
self.assertEqual(conf.app.root, None)
@@ -37,7 +37,7 @@ class TestConf(TestCase):
conf = configuration.initconf()
conf.update(configuration.conf_from_file(os.path.join(
__here__,
'test_config/empty.py'
'config_fixtures/empty.py'
)))
self.assertEqual(conf.app.root, None)
@@ -53,7 +53,7 @@ class TestConf(TestCase):
conf = configuration.initconf()
conf.update(configuration.conf_from_file(os.path.join(
__here__,
'test_config/forcedict.py'
'config_fixtures/forcedict.py'
)))
self.assertEqual(conf.app.root, None)
@@ -93,16 +93,16 @@ class TestConf(TestCase):
def test_config_from_file(self):
from pecan import configuration
path = os.path.join(
os.path.dirname(__file__), 'test_config', 'config.py'
os.path.dirname(__file__), 'config_fixtures', 'config.py'
)
conf = configuration.conf_from_file(path)
configuration.conf_from_file(path)
def test_config_illegal_ids(self):
from pecan import configuration
conf = configuration.Config({})
conf.update(configuration.conf_from_file(os.path.join(
__here__,
'test_config/bad/module_and_underscore.py'
'config_fixtures/bad/module_and_underscore.py'
)))
self.assertEqual([], list(conf))
@@ -112,7 +112,7 @@ class TestConf(TestCase):
configuration.Config({})
self.assertRaises(IOError, configuration.conf_from_file, os.path.join(
__here__,
'test_config',
'config_fixtures',
*path
))
@@ -123,7 +123,7 @@ class TestConf(TestCase):
self.assertRaises(IOError, configuration.conf_from_file, os.path.join(
__here__,
'test_config',
'config_fixtures',
*path
))
@@ -135,7 +135,7 @@ class TestConf(TestCase):
self.assertRaises(
SyntaxError,
configuration.conf_from_file,
os.path.join(__here__, 'test_config', *path)
os.path.join(__here__, 'config_fixtures', *path)
)
def test_config_with_bad_import(self):
@@ -148,7 +148,7 @@ class TestConf(TestCase):
configuration.conf_from_file,
os.path.join(
__here__,
'test_config',
'config_fixtures',
*path
)
)
@@ -238,7 +238,10 @@ class TestGlobalConfig(TestCase):
def tearDown(self):
from pecan import configuration
configuration.set_config(dict(configuration.initconf()), overwrite=True)
configuration.set_config(
dict(configuration.initconf()),
overwrite=True
)
def test_paint_from_dict(self):
from pecan import configuration
@@ -255,7 +258,7 @@ class TestGlobalConfig(TestCase):
from pecan import configuration
configuration.set_config(os.path.join(
__here__,
'test_config/foobar.py'
'config_fixtures/foobar.py'
))
assert dict(configuration._runtime_conf) != {'foo': 'bar'}
assert configuration._runtime_conf.foo == 'bar'
@@ -264,7 +267,7 @@ class TestGlobalConfig(TestCase):
from pecan import configuration
configuration.set_config(os.path.join(
__here__,
'test_config/foobar.py',
'config_fixtures/foobar.py',
),
overwrite=True
)

29
pecan/tests/test_rest.py
View File

@@ -1,7 +1,8 @@
from pecan import abort, expose, make_app, request, response
from pecan import abort, expose, make_app, response
from pecan.rest import RestController
from unittest import TestCase
from webtest import TestApp
import warnings
try:
from simplejson import dumps, loads
except:
@@ -208,21 +209,27 @@ class TestRestController(TestCase):
assert r.body == 'OPTIONS'
# test the "other" custom action
r = app.request('/things/other', method='MISC', status=405)
assert r.status_int == 405
with warnings.catch_warnings():
warnings.simplefilter("ignore")
r = app.request('/things/other', method='MISC', status=405)
assert r.status_int == 405
# test the "other" custom action with the _method parameter
r = app.post('/things/other', {'_method': 'MISC'}, status=405)
assert r.status_int == 405
# test the "others" custom action
r = app.request('/things/others/', method='MISC')
assert r.status_int == 200
assert r.body == 'OTHERS'
with warnings.catch_warnings():
warnings.simplefilter("ignore")
r = app.request('/things/others/', method='MISC')
assert r.status_int == 200
assert r.body == 'OTHERS'
# test the "others" custom action missing trailing slash
r = app.request('/things/others', method='MISC', status=302)
assert r.status_int == 302
with warnings.catch_warnings():
warnings.simplefilter("ignore")
r = app.request('/things/others', method='MISC', status=302)
assert r.status_int == 302
# test the "others" custom action with the _method parameter
r = app.get('/things/others/?_method=MISC')
@@ -609,8 +616,10 @@ class TestRestController(TestCase):
assert r.status_int == 404
# test "RESET" custom action
r = app.request('/things', method='RESET', status=404)
assert r.status_int == 404
with warnings.catch_warnings():
warnings.simplefilter("ignore")
r = app.request('/things', method='RESET', status=404)
assert r.status_int == 404
def test_custom_delete(self):

309
pecan/tests/test_scaffolds.py Normal file
View File

@@ -0,0 +1,309 @@
import os
import sys
import tempfile
import shutil
import subprocess
import pkg_resources
import httplib
import urllib2
import time
from cStringIO import StringIO
import pecan
if sys.version_info < (2, 7):
import unittest2 as unittest
else:
import unittest # noqa
def has_internet():
try:
urllib2.urlopen('http://google.com', timeout=1)
return True
except urllib2.URLError:
pass # pragma: no cover
return False
class TestPecanScaffold(unittest.TestCase):
def test_normalize_pkg_name(self):
from pecan.scaffolds import PecanScaffold
s = PecanScaffold()
assert s.normalize_pkg_name('sam') == 'sam'
assert s.normalize_pkg_name('sam1') == 'sam1'
assert s.normalize_pkg_name('sam_') == 'sam_'
assert s.normalize_pkg_name('Sam') == 'sam'
assert s.normalize_pkg_name('SAM') == 'sam'
assert s.normalize_pkg_name('sam ') == 'sam'
assert s.normalize_pkg_name(' sam') == 'sam'
assert s.normalize_pkg_name('sam$') == 'sam'
assert s.normalize_pkg_name('sam-sam') == 'samsam'
class TestScaffoldUtils(unittest.TestCase):
def setUp(self):
self.scaffold_destination = tempfile.mkdtemp()
self.out = sys.stdout
sys.stdout = StringIO()
def tearDown(self):
shutil.rmtree(self.scaffold_destination)
sys.stdout = self.out
def test_copy_dir(self):
from pecan.scaffolds import PecanScaffold
class SimpleScaffold(PecanScaffold):
_scaffold_dir = ('pecan', os.path.join(
'tests', 'scaffold_fixtures', 'simple'
))
SimpleScaffold().copy_to(os.path.join(
self.scaffold_destination,
'someapp'
), out_=StringIO())
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'someapp', 'foo'
))
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'someapp', 'bar', 'spam.txt'
))
assert open(os.path.join(
self.scaffold_destination, 'someapp', 'foo'
), 'r').read().strip() == 'YAR'
assert open(os.path.join(
self.scaffold_destination, 'someapp', 'foo'
), 'r').read().strip() == 'YAR'
def test_destination_directory_levels_deep(self):
from pecan.scaffolds import copy_dir
f = StringIO()
copy_dir(('pecan', os.path.join(
'tests', 'scaffold_fixtures', 'simple'
)),
os.path.join(self.scaffold_destination, 'some', 'app'),
{},
out_=f
)
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'some', 'app', 'foo')
)
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'some', 'app', 'bar', 'spam.txt')
)
assert open(os.path.join(
self.scaffold_destination, 'some', 'app', 'foo'
), 'r').read().strip() == 'YAR'
assert open(os.path.join(
self.scaffold_destination, 'some', 'app', 'bar', 'spam.txt'
), 'r').read().strip() == 'Pecan'
def test_destination_directory_already_exists(self):
from pecan.scaffolds import copy_dir
from cStringIO import StringIO
f = StringIO()
copy_dir(('pecan', os.path.join(
'tests', 'scaffold_fixtures', 'simple'
)),
os.path.join(self.scaffold_destination),
{},
out_=f
)
assert 'already exists' in f.getvalue()
def test_copy_dir_with_filename_substitution(self):
from pecan.scaffolds import copy_dir
copy_dir(('pecan', os.path.join(
'tests', 'scaffold_fixtures', 'file_sub'
)),
os.path.join(
self.scaffold_destination, 'someapp'
),
{'package': 'thingy'},
out_=StringIO()
)
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'someapp', 'foo_thingy')
)
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'someapp', 'bar_thingy', 'spam.txt')
)
assert open(os.path.join(
self.scaffold_destination, 'someapp', 'foo_thingy'
), 'r').read().strip() == 'YAR'
assert open(os.path.join(
self.scaffold_destination, 'someapp', 'bar_thingy', 'spam.txt'
), 'r').read().strip() == 'Pecan'
def test_copy_dir_with_file_content_substitution(self):
from pecan.scaffolds import copy_dir
copy_dir(('pecan', os.path.join(
'tests', 'scaffold_fixtures', 'content_sub'
)),
os.path.join(
self.scaffold_destination, 'someapp'
),
{'package': 'thingy'},
out_=StringIO()
)
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'someapp', 'foo')
)
assert os.path.isfile(os.path.join(
self.scaffold_destination, 'someapp', 'bar', 'spam.txt')
)
assert open(os.path.join(
self.scaffold_destination, 'someapp', 'foo'
), 'r').read().strip() == 'YAR thingy'
assert open(os.path.join(
self.scaffold_destination, 'someapp', 'bar', 'spam.txt'
), 'r').read().strip() == 'Pecan thingy'
class TestTemplateBuilds(unittest.TestCase):
"""
Used to build and test the templated quickstart project(s).
"""
install_dir = tempfile.mkdtemp()
cwd = os.getcwd()
def setUp(self):
# Make a temp install location and record the cwd
self.install()
def tearDown(self):
shutil.rmtree(self.install_dir)
os.chdir(self.cwd)
def install(self):
# Create a new virtualenv in the temp install location
import virtualenv
virtualenv.create_environment(
self.install_dir,
site_packages=False
)
# chdir into the pecan source
os.chdir(pkg_resources.get_distribution('pecan').location)
py_exe = os.path.join(self.install_dir, 'bin', 'python')
pecan_exe = os.path.join(self.install_dir, 'bin', 'pecan')
# env/bin/python setup.py develop (pecan)
subprocess.check_call([
py_exe,
'setup.py',
'develop'
])
# create the templated project
os.chdir(self.install_dir)
subprocess.check_call([pecan_exe, 'create', 'Testing123'])
# move into the new project directory and install
os.chdir('Testing123')
subprocess.check_call([
py_exe,
'setup.py',
'develop'
])
def poll(self, proc):
limit = 5
for i in range(limit):
time.sleep(1)
proc.poll()
# Make sure it's running
if proc.returncode is None:
break
elif i == limit: # pragma: no cover
raise RuntimeError("pecan serve config.py didn't start.")
@unittest.skipUnless(has_internet(), 'Internet connectivity unavailable.')
@unittest.skipUnless(
getattr(pecan, '__run_all_tests__', False) is True,
'Skipping (slow). To run, `$ python setup.py test --functional.`'
)
def test_project_pecan_serve_command(self):
pecan_exe = os.path.join(self.install_dir, 'bin', 'pecan')
# Start the server
proc = subprocess.Popen([
pecan_exe,
'serve',
'config.py'
])
try:
self.poll(proc)
# ...and that it's serving (valid) content...
conn = httplib.HTTPConnection('localhost:8080')
conn.request('GET', '/')
resp = conn.getresponse()
assert resp.status == 200
assert 'This is a sample Pecan project.' in resp.read()
finally:
proc.terminate()
@unittest.skipUnless(has_internet(), 'Internet connectivity unavailable.')
@unittest.skipUnless(
getattr(pecan, '__run_all_tests__', False) is True,
'Skipping (slow). To run, `$ python setup.py test --functional.`'
)
def test_project_pecan_shell_command(self):
pecan_exe = os.path.join(self.install_dir, 'bin', 'pecan')
# Start the server
proc = subprocess.Popen([
pecan_exe,
'shell',
'config.py'
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.PIPE
)
self.poll(proc)
out, _ = proc.communicate(
'{"model" : model, "conf" : conf, "app" : app}'
)
assert 'testing123.model' in out
assert 'Config(' in out
assert 'webtest.app.TestApp' in out
try:
# just in case stdin doesn't close
proc.terminate()
except:
pass
@unittest.skipUnless(has_internet(), 'Internet connectivity unavailable.')
@unittest.skipUnless(
getattr(pecan, '__run_all_tests__', False) is True,
'Skipping (slow). To run, `$ python setup.py test --functional.`'
)
def test_project_tests_command(self):
py_exe = os.path.join(self.install_dir, 'bin', 'python')
# Run the tests
proc = subprocess.Popen([
py_exe,
'setup.py',
'test'
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
proc.wait()
assert proc.stderr.read().splitlines()[-1].strip() == 'OK'

165
pecan/tests/test_templates.py
View File

@@ -1,165 +0,0 @@
import os
import sys
import tempfile
import shutil
import subprocess
import pkg_resources
import httplib
import urllib2
import time
import pecan
if sys.version_info < (2, 7):
import unittest2 as unittest
else:
import unittest
def has_internet():
try:
response = urllib2.urlopen('http://google.com', timeout=1)
return True
except urllib2.URLError:
pass # pragma: no cover
return False
class TestTemplateBuilds(unittest.TestCase):
"""
Used to build and test the templated quickstart project(s).
"""
install_dir = tempfile.mkdtemp()
cwd = os.getcwd()
def setUp(self):
# Make a temp install location and record the cwd
self.install()
def tearDown(self):
shutil.rmtree(self.install_dir)
os.chdir(self.cwd)
def install(self):
# Create a new virtualenv in the temp install location
import virtualenv
virtualenv.create_environment(
self.install_dir,
site_packages=False
)
# chdir into the pecan source
os.chdir(pkg_resources.get_distribution('pecan').location)
py_exe = os.path.join(self.install_dir, 'bin', 'python')
pecan_exe = os.path.join(self.install_dir, 'bin', 'pecan')
# env/bin/python setup.py develop (pecan)
subprocess.check_call([
py_exe,
'setup.py',
'develop'
])
# create the templated project
os.chdir(self.install_dir)
subprocess.check_call([pecan_exe, 'create', 'Testing123'])
# move into the new project directory and install
os.chdir('Testing123')
subprocess.check_call([
py_exe,
'setup.py',
'develop'
])
def poll(self, proc):
limit = 5
for i in range(limit):
time.sleep(1)
proc.poll()
# Make sure it's running
if proc.returncode is None:
break
elif i == limit: # pragma: no cover
raise RuntimeError("pecan serve config.py didn't start.")
@unittest.skipUnless(has_internet(), 'Internet connectivity unavailable.')
@unittest.skipUnless(
getattr(pecan, '__run_all_tests__', False) is True,
'Skipping (really slow). To run, `$ python setup.py test --functional.`'
)
def test_project_pecan_serve_command(self):
pecan_exe = os.path.join(self.install_dir, 'bin', 'pecan')
# Start the server
proc = subprocess.Popen([
pecan_exe,
'serve',
'config.py'
])
try:
self.poll(proc)
# ...and that it's serving (valid) content...
conn = httplib.HTTPConnection('localhost:8080')
conn.request('GET', '/')
resp = conn.getresponse()
assert resp.status == 200
assert 'This is a sample Pecan project.' in resp.read()
finally:
proc.terminate()
@unittest.skipUnless(has_internet(), 'Internet connectivity unavailable.')
@unittest.skipUnless(
getattr(pecan, '__run_all_tests__', False) is True,
'Skipping (really slow). To run, `$ python setup.py test --functional.`'
)
def test_project_pecan_shell_command(self):
from pecan.testing import load_test_app
pecan_exe = os.path.join(self.install_dir, 'bin', 'pecan')
# Start the server
proc = subprocess.Popen([
pecan_exe,
'shell',
'config.py'
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.PIPE
)
self.poll(proc)
out, _ = proc.communicate('{"model" : model, "conf" : conf, "app" : app}')
assert 'testing123.model' in out
assert 'Config(' in out
assert 'webtest.app.TestApp' in out
try:
# just in case stdin doesn't close
proc.terminate()
except:
pass
@unittest.skipUnless(has_internet(), 'Internet connectivity unavailable.')
@unittest.skipUnless(
getattr(pecan, '__run_all_tests__', False) is True,
'Skipping (really slow). To run, `$ python setup.py test --functional.`'
)
def test_project_tests_command(self):
py_exe = os.path.join(self.install_dir, 'bin', 'python')
# Run the tests
proc = subprocess.Popen([
py_exe,
'setup.py',
'test'
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
proc.wait()
assert proc.stderr.read().splitlines()[-1].strip() == 'OK'

24
setup.py
View File

@@ -1,5 +1,5 @@
import sys
from setuptools import setup, command, find_packages
from setuptools import setup, find_packages
from setuptools.command.test import test as TestCommand
version = '0.1.0'
@@ -13,18 +13,22 @@ requirements = [
"simplegeneric >= 0.7",
"Mako >= 0.4.0",
"Paste >= 1.7.5.1",
"PasteScript >= 1.7.3",
"WebTest >= 1.2.2"
]
try:
import json
import json # noqa
except:
try:
import simplejson
import simplejson # noqa
except:
requirements.append("simplejson >= 2.1.1")
try:
import argparse # noqa
except:
requirements.append('argparse')
tests_require = requirements + ['virtualenv']
if sys.version_info < (2, 7):
tests_require += ['unittest2']
@@ -91,12 +95,12 @@ setup(
test_suite='pecan',
cmdclass={'test': test},
entry_points="""
[paste.paster_command]
pecan-serve = pecan.commands:ServeCommand
pecan-shell = pecan.commands:ShellCommand
pecan-create = pecan.commands:CreateCommand
[paste.paster_create_template]
pecan-base = pecan.templates:BaseTemplate
[pecan.command]
serve = pecan.commands:ServeCommand
shell = pecan.commands:ShellCommand
create = pecan.commands:CreateCommand
[pecan.scaffold]
base = pecan.scaffolds:BaseScaffold
[console_scripts]
pecan = pecan.commands:CommandRunner.handle_command_line
""",