2
0
Fork 0

feat: add install handler and related logic

- Mako is now a core dependency
- therefore no more 'email' extra
- add `get_install_handler()` method for app handler
- add `render_mako_template()` method for app handler
- add `resource_path()` method for app handler
- install handler thus far can:
  - confirm db connection
  - make appdir plus config/scripts:
    - wutta.conf
    - web.conf
    - upgrade.sh
  - upgrade db schema to create tables
  - from there web app can run, create admin user
- quick start docs now describe "generated code" option
This commit is contained in:
Lance Edgar 2024-11-24 10:13:56 -06:00
parent 49e77d7407
commit ceeff7e911
15 changed files with 1526 additions and 32 deletions

View file

@ -26,6 +26,7 @@
email.message email.message
enum enum
exc exc
install
people people
progress progress
testing testing

View file

@ -0,0 +1,6 @@
``wuttjamaican.install``
========================
.. automodule:: wuttjamaican.install
:members:

View file

@ -29,11 +29,13 @@ templates_path = ['_templates']
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store'] exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
intersphinx_mapping = { intersphinx_mapping = {
'mako': ('https://docs.makotemplates.org/en/latest/', None),
'packaging': ('https://packaging.python.org/en/latest/', None), 'packaging': ('https://packaging.python.org/en/latest/', None),
'python': ('https://docs.python.org/3/', None), 'python': ('https://docs.python.org/3/', None),
'python-configuration': ('https://python-configuration.readthedocs.io/en/latest/', None), 'python-configuration': ('https://python-configuration.readthedocs.io/en/latest/', None),
'rattail': ('https://rattailproject.org/docs/rattail/', None), 'rattail': ('https://rattailproject.org/docs/rattail/', None),
'rattail-manual': ('https://rattailproject.org/docs/rattail-manual/', None), 'rattail-manual': ('https://rattailproject.org/docs/rattail-manual/', None),
'rich': ('https://rich.readthedocs.io/en/latest/', None),
'sqlalchemy': ('http://docs.sqlalchemy.org/en/latest/', None), 'sqlalchemy': ('http://docs.sqlalchemy.org/en/latest/', None),
'wutta-continuum': ('https://rattailproject.org/docs/wutta-continuum/', None), 'wutta-continuum': ('https://rattailproject.org/docs/wutta-continuum/', None),
} }

View file

@ -145,6 +145,12 @@ Glossary
Similar to a "plugin" concept but only *one* handler may be used Similar to a "plugin" concept but only *one* handler may be used
for a given purpose. See also :doc:`narr/handlers/index`. for a given purpose. See also :doc:`narr/handlers/index`.
install handler
The :term:`handler` responsible for installing a new instance of
the :term:`app`.
Default is :class:`~wuttjamaican.install.InstallHandler`.
package package
Generally refers to a proper Python package, i.e. a collection of Generally refers to a proper Python package, i.e. a collection of
modules etc. which is installed via ``pip``. See also modules etc. which is installed via ``pip``. See also

View file

@ -2,6 +2,56 @@
Quick Start Quick Start
=========== ===========
We have two varieties of "quick start" instructions:
* :ref:`quick-start-generated`
* :ref:`quick-start-manual`
.. _quick-start-generated:
From Generated Code
-------------------
Note that this section describes an app based on WuttaWeb (i.e. not
just WuttJamaican).
There is a tool to `generate new project code`_, on the Rattail Demo
site. Use it to download a ZIP file (e.g. ``poser.zip``) for your
project.
.. _generate new project code: https://demo.rattailproject.org/generated-projects/new/wutta
Make a local :term:`virtual environment` for your project.
Also make a new e.g. ``poser`` database in PostgreSQL (or MySQL).
Unzip and install the source code to the virtual environment, then run
the app installer:
.. code-block:: sh
mkdir -p ~/src
unzip ~/Downloads/poser.zip -d ~/src
cd /path/to/venv
bin/pip install -e ~/src/poser
bin/poser install
Assuming all goes well, you can run the web app with:
.. code-block:: sh
bin/pserve --reload file+ini:app/web.conf
And browse it at http://localhost:9080
.. _quick-start-manual:
From Scratch
------------
This shows the *minimum* use case, basically how to make/use the This shows the *minimum* use case, basically how to make/use the
:term:`config object` and :term:`app handler`. :term:`config object` and :term:`app handler`.
@ -67,7 +117,7 @@ For more info see:
.. _db-setup: .. _db-setup:
Database Setup Database Setup
============== ~~~~~~~~~~~~~~
You should already have the package installed (see previous section). You should already have the package installed (see previous section).

View file

@ -28,6 +28,7 @@ requires-python = ">= 3.8"
dependencies = [ dependencies = [
'importlib-metadata; python_version < "3.10"', 'importlib-metadata; python_version < "3.10"',
"importlib_resources ; python_version < '3.9'", "importlib_resources ; python_version < '3.9'",
"Mako",
"progress", "progress",
"python-configuration", "python-configuration",
"typer", "typer",
@ -36,7 +37,6 @@ dependencies = [
[project.optional-dependencies] [project.optional-dependencies]
db = ["SQLAlchemy<2", "alembic", "alembic-postgresql-enum", "passlib"] db = ["SQLAlchemy<2", "alembic", "alembic-postgresql-enum", "passlib"]
email = ["Mako"]
docs = ["Sphinx", "sphinxcontrib-programoutput", "enum-tools[sphinx]", "furo"] docs = ["Sphinx", "sphinxcontrib-programoutput", "enum-tools[sphinx]", "furo"]
tests = ["pytest-cov", "tox"] tests = ["pytest-cov", "tox"]

View file

@ -30,8 +30,8 @@ import sys
import warnings import warnings
from wuttjamaican.util import (load_entry_points, load_object, from wuttjamaican.util import (load_entry_points, load_object,
make_title, make_uuid, parse_bool, make_title, make_uuid,
progress_loop) progress_loop, resource_path)
class AppHandler: class AppHandler:
@ -82,6 +82,7 @@ class AppHandler:
default_enum_spec = 'wuttjamaican.enum' default_enum_spec = 'wuttjamaican.enum'
default_auth_handler_spec = 'wuttjamaican.auth:AuthHandler' default_auth_handler_spec = 'wuttjamaican.auth:AuthHandler'
default_email_handler_spec = 'wuttjamaican.email:EmailHandler' default_email_handler_spec = 'wuttjamaican.email:EmailHandler'
default_install_handler_spec = 'wuttjamaican.install:InstallHandler'
default_people_handler_spec = 'wuttjamaican.people:PeopleHandler' default_people_handler_spec = 'wuttjamaican.people:PeopleHandler'
def __init__(self, config): def __init__(self, config):
@ -418,20 +419,52 @@ class AppHandler:
:param subfolders: Optional list of subfolder names to create :param subfolders: Optional list of subfolder names to create
within the app dir. If not specified, defaults will be: within the app dir. If not specified, defaults will be:
``['data', 'log', 'work']``. ``['cache', 'data', 'log', 'work']``.
""" """
appdir = path appdir = path
if not os.path.exists(appdir): if not os.path.exists(appdir):
os.makedirs(appdir) os.makedirs(appdir)
if not subfolders: if not subfolders:
subfolders = ['data', 'log', 'work'] subfolders = ['cache', 'data', 'log', 'work']
for name in subfolders: for name in subfolders:
path = os.path.join(appdir, name) path = os.path.join(appdir, name)
if not os.path.exists(path): if not os.path.exists(path):
os.mkdir(path) os.mkdir(path)
def render_mako_template(
self,
template,
context,
output_path=None,
):
"""
Convenience method to render a Mako template.
:param template: :class:`~mako:mako.template.Template`
instance.
:param context: Dict of context for the template.
:param output_path: Optional path to which output should be
written.
:returns: Rendered output as string.
"""
output = template.render(**context)
if output_path:
with open(output_path, 'wt') as f:
f.write(output)
return output
def resource_path(self, path):
"""
Convenience wrapper for
:func:`wuttjamaican.util.resource_path()`.
"""
return resource_path(path)
def make_session(self, **kwargs): def make_session(self, **kwargs):
""" """
Creates a new SQLAlchemy session for the app DB. By default Creates a new SQLAlchemy session for the app DB. By default
@ -637,6 +670,19 @@ class AppHandler:
self.handlers['email'] = factory(self.config, **kwargs) self.handlers['email'] = factory(self.config, **kwargs)
return self.handlers['email'] return self.handlers['email']
def get_install_handler(self, **kwargs):
"""
Get the configured :term:`install handler`.
:rtype: :class:`~wuttjamaican.install.handler.InstallHandler`
"""
if 'install' not in self.handlers:
spec = self.config.get(f'{self.appname}.install.handler',
default=self.default_install_handler_spec)
factory = self.load_object(spec)
self.handlers['install'] = factory(self.config, **kwargs)
return self.handlers['install']
def get_people_handler(self, **kwargs): def get_people_handler(self, **kwargs):
""" """
Get the configured "people" :term:`handler`. Get the configured "people" :term:`handler`.
@ -727,7 +773,7 @@ class GenericHandler:
instance of :class:`~wuttjamaican.conf.WuttaConfig`. instance of :class:`~wuttjamaican.conf.WuttaConfig`.
""" """
def __init__(self, config, **kwargs): def __init__(self, config):
self.config = config self.config = config
self.app = self.config.get_app() self.app = self.config.get_app()

583
src/wuttjamaican/install.py Normal file
View file

@ -0,0 +1,583 @@
# -*- coding: utf-8; -*-
################################################################################
#
# WuttJamaican -- Base package for Wutta Framework
# Copyright © 2023-2024 Lance Edgar
#
# This file is part of Wutta Framework.
#
# Wutta Framework is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# Wutta Framework is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
Install Handler
"""
import os
import shutil
import stat
import subprocess
import sys
import rich
from mako.lookup import TemplateLookup
from wuttjamaican.app import GenericHandler
class InstallHandler(GenericHandler):
"""
Base class and default implementation for the :term:`install
handler`.
See also
:meth:`~wuttjamaican.app.AppHandler.get_install_handler()`.
The installer runs interactively via command line, prompting the
user for various config settings etc.
If installation completes okay the exit code is 0, but if not:
* exit code 1 indicates user canceled
* exit code 2 indicates sanity check failed
* other codes possible if errors occur
Usually an app will define e.g. ``poser install`` command which
would invoke the install handler's :meth:`run()` method::
app = config.get_app()
install = app.get_install_handler(pkg_name='poser')
install.run()
Note that these first 4 attributes may be specified via
constructor kwargs:
.. attribute:: pkg_name
Python package name for the app, e.g. ``poser``.
.. attribute:: app_title
Display title for the app, e.g. "Poser".
.. attribute:: pypi_name
Package distribution name, e.g. for PyPI. If not specified one
will be guessed.
.. attribute:: egg_name
Egg name for the app. If not specified one will be guessed.
"""
pkg_name = 'poser'
app_title = None
pypi_name = None
egg_name = None
def __init__(self, config, **kwargs):
super().__init__(config)
# nb. caller may specify pkg_name etc.
self.__dict__.update(kwargs)
# some package names we can generate by default
if not self.app_title:
self.app_title = self.pkg_name
if not self.pypi_name:
self.pypi_name = self.app_title
if not self.egg_name:
self.egg_name = self.pypi_name.replace('-', '_')
def run(self):
"""
Run the interactive command-line installer.
This does the following:
* check for ``prompt_toolkit`` and maybe ask to install it
* define the template lookup paths, for making config files
* call :meth:`show_welcome()`
* call :meth:`sanity_check()`
* call :meth:`do_install_steps()`
* call :meth:`show_goodbye()`
Although if a problem is encountered then not all calls may
happen.
"""
self.require_prompt_toolkit()
paths = [
self.app.resource_path('wuttjamaican:templates/install'),
]
try:
paths.insert(0, self.app.resource_path(f'{self.pkg_name}:templates/install'))
except ModuleNotFoundError:
pass
self.templates = TemplateLookup(directories=paths)
self.show_welcome()
self.sanity_check()
self.schema_installed = False
self.do_install_steps()
self.show_goodbye()
def show_welcome(self):
"""
Show the intro/welcome message, and prompt user to begin the
install.
This is normally called by :meth:`run()`.
"""
self.rprint("\n\t[blue]Welcome to {}![/blue]".format(self.app.get_title()))
self.rprint("\n\tThis tool will install and configure the app.")
self.rprint("\n\t[italic]NB. You should already have created the database in PostgreSQL or MySQL.[/italic]")
# shall we continue?
if not self.prompt_bool("continue?", True):
self.rprint()
sys.exit(1)
def sanity_check(self):
"""
Perform various sanity checks before doing the install. If
any problem is found the installer should exit with code 2.
This is normally called by :meth:`run()`.
"""
# appdir must not yet exist
appdir = os.path.join(sys.prefix, 'app')
if os.path.exists(appdir):
self.rprint(f"\n\t[bold red]appdir already exists:[/bold red] {appdir}\n")
sys.exit(2)
def do_install_steps(self):
"""
Perform the real installation steps.
This method is called by :meth:`run()` and does the following:
* call :meth:`get_dbinfo()` to get DB info from user, and test connection
* call :meth:`make_template_context()` to use when generating output
* call :meth:`make_appdir()` to create app dir with config files
* call :meth:`install_db_schema()` to (optionally) create tables in DB
"""
# prompt user for db info
dbinfo = self.get_dbinfo()
# get context for generated app files
context = self.make_template_context(dbinfo)
# make the appdir
self.make_appdir(context)
# install db schema if user likes
self.schema_installed = self.install_db_schema(dbinfo['dburl'])
def get_dbinfo(self):
"""
Collect DB connection info from the user, and test the
connection. If connection fails, exit the install.
This method is normally called by :meth:`do_install_steps()`.
:returns: Dict of DB info collected from user.
"""
dbinfo = {}
# get db info
dbinfo['dbtype'] = self.prompt_generic('db type', 'postgresql')
dbinfo['dbhost'] = self.prompt_generic('db host', 'localhost')
default_port = '3306' if dbinfo['dbtype'] == 'mysql' else '5432'
dbinfo['dbport'] = self.prompt_generic('db port', default_port)
dbinfo['dbname'] = self.prompt_generic('db name', self.pkg_name)
dbinfo['dbuser'] = self.prompt_generic('db user', self.pkg_name)
# get db password
dbinfo['dbpass'] = None
while not dbinfo['dbpass']:
dbinfo['dbpass'] = self.prompt_generic('db pass', is_password=True)
# test db connection
self.rprint("\n\ttesting db connection... ", end='')
dbinfo['dburl'] = self.make_db_url(dbinfo['dbtype'],
dbinfo['dbhost'],
dbinfo['dbport'],
dbinfo['dbname'],
dbinfo['dbuser'],
dbinfo['dbpass'])
error = self.test_db_connection(dbinfo['dburl'])
if error:
self.rprint("[bold red]cannot connect![/bold red] ..error was:")
self.rprint("\n{}".format(error))
self.rprint("\n\t[bold yellow]aborting mission[/bold yellow]\n")
sys.exit(1)
self.rprint("[bold green]good[/bold green]")
return dbinfo
def make_db_url(self, dbtype, dbhost, dbport, dbname, dbuser, dbpass):
from sqlalchemy.engine import URL
if dbtype == 'mysql':
drivername = 'mysql+mysqlconnector'
else:
drivername = 'postgresql+psycopg2'
return URL.create(drivername=drivername,
username=dbuser,
password=dbpass,
host=dbhost,
port=dbport,
database=dbname)
def test_db_connection(self, url):
import sqlalchemy as sa
engine = sa.create_engine(url)
# check for random table; does not matter if it exists, we
# just need to test interaction and this is a neutral way
try:
sa.inspect(engine).has_table('whatever')
except Exception as error:
return str(error)
def make_template_context(self, dbinfo, **kwargs):
"""
This must return a dict to be used as global template context
when generating output (e.g. config) files.
This method is normally called by :meth:`do_install_steps()`.
The ``context`` returned is then passed to
:meth:`render_mako_template()`.
:param dbinfo: Dict of DB connection info as obtained from
:meth:`get_dbinfo()`.
:param \**kwargs: Extra template context.
:returns: Dict for global template context.
The context dict will include:
* ``envdir`` - value from :data:`python:sys.prefix`
* ``envname`` - "last" dirname from ``sys.prefix``
* ``pkg_name`` - value from :attr:`pkg_name`
* ``app_title`` - value from :attr:`app_title`
* ``pypi_name`` - value from :attr:`pypi_name`
* ``egg_name`` - value from :attr:`egg_name`
* ``appdir`` - ``app`` folder under ``sys.prefix``
* ``db_url`` - value from ``dbinfo['dburl']``
"""
envname = os.path.basename(sys.prefix)
appdir = os.path.join(sys.prefix, 'app')
context = {
'envdir': sys.prefix,
'envname': envname,
'pkg_name': self.pkg_name,
'app_title': self.app_title,
'pypi_name': self.pypi_name,
'appdir': appdir,
'db_url': dbinfo['dburl'],
'egg_name': self.egg_name,
}
context.update(kwargs)
return context
def make_appdir(self, context, appdir=None):
"""
Create the app folder structure and generate config files.
This method is normally called by :meth:`do_install_steps()`.
:param context: Template context dict, i.e. from
:meth:`make_template_context()`.
The default logic will create a structure as follows, assuming
``/venv`` is the path to the virtual environment:
.. code-block:: none
/venv/
app/
cache/
data/
log/
work/
wutta.conf
web.conf
upgrade.sh
File templates for this come from
``wuttjamaican:templates/install`` by default.
"""
# app handler makes appdir proper
appdir = appdir or self.app.get_appdir()
self.app.make_appdir(appdir)
# but then we also generate some files...
# wutta.conf
self.make_config_file('wutta.conf.mako',
os.path.join(appdir, 'wutta.conf'),
**context)
# web.conf
web_context = dict(context)
web_context.setdefault('beaker_key', context.get('pkg_name', 'poser'))
web_context.setdefault('beaker_secret', 'TODO_YOU_SHOULD_CHANGE_THIS')
web_context.setdefault('pyramid_host', '0.0.0.0')
web_context.setdefault('pyramid_port', '9080')
self.make_config_file('web.conf.mako',
os.path.join(appdir, 'web.conf'),
**web_context)
# upgrade.sh
template = self.templates.get_template('upgrade.sh.mako')
output_path = os.path.join(appdir, 'upgrade.sh')
self.render_mako_template(template, context,
output_path=output_path)
os.chmod(output_path, stat.S_IRWXU
| stat.S_IRGRP
| stat.S_IXGRP
| stat.S_IROTH
| stat.S_IXOTH)
self.rprint(f"\n\tappdir created at: [bold green]{appdir}[/bold green]")
def render_mako_template(
self,
template,
context,
output_path=None,
):
"""
Convenience wrapper around
:meth:`~wuttjamaican.app.AppHandler.render_mako_template()`.
:param template: :class:`~mako:mako.template.Template`
instance, or name of one to fetch via lookup.
This method allows specifying the template by name, in which
case the real template object is fetched via lookup.
Other args etc. are the same as for the wrapped app handler
method.
"""
if isinstance(template, str):
template = self.templates.get_template(template)
return self.app.render_mako_template(template, context,
output_path=output_path)
def make_config_file(self, template, output_path, **kwargs):
"""
Write a new config file to the given path, using the given
template and context.
:param template: :class:`~mako:mako.template.Template`
instance, or name of one to fetch via lookup.
:param output_path: Path to which output should be written.
:param \**kwargs: Extra context for the template.
Some context will be provided automatically for the template,
but these may be overridden via the ``**kwargs``:
* ``app_title`` - value from
:meth:`~wuttjamaican.app.AppHandler.get_title()`.
* ``appdir`` - value from
:meth:`~wuttjamaican.app.AppHandler.get_appdir()`.
* ``db_url`` - poser/dummy value
* ``os`` - reference to :mod:`os` module
This method is mostly about sorting out the context dict.
Once it does that it calls :meth:`render_mako_template()`.
"""
context = {
'app_title': self.app.get_title(),
'appdir': self.app.get_appdir(),
'db_url': 'postresql://user:pass@localhost/poser',
'os': os,
}
context.update(kwargs)
self.render_mako_template(template, context,
output_path=output_path)
return output_path
def install_db_schema(self, db_url, appdir=None):
"""
First prompt the user, but if they agree then apply all
Alembic migrations to the configured database.
This method is normally called by :meth:`do_install_steps()`.
The end result should be a complete schema, ready for the app
to use.
:param db_url: :class:`sqlalchemy:sqlalchemy.engine.URL`
instance.
"""
from alembic.util.messaging import obfuscate_url_pw
if not self.prompt_bool("install db schema?", True):
return False
self.rprint()
# install db schema
appdir = appdir or self.app.get_appdir()
cmd = [os.path.join(sys.prefix, 'bin', 'alembic'),
'-c', os.path.join(appdir, 'wutta.conf'),
'upgrade', 'heads']
subprocess.check_call(cmd)
self.rprint("\n\tdb schema installed to: [bold green]{}[/bold green]".format(
obfuscate_url_pw(db_url)))
return True
def show_goodbye(self):
"""
Show the final message; this assumes setup completed okay.
This is normally called by :meth:`run()`.
"""
self.rprint("\n\t[bold green]initial setup is complete![/bold green]")
if self.schema_installed:
self.rprint("\n\tyou can run the web app with:")
self.rprint(f"\n\t[blue]cd {sys.prefix}[/blue]")
self.rprint("\t[blue]bin/pserve file+ini:app/web.conf[/blue]")
self.rprint()
##############################
# console utility functions
##############################
def require_prompt_toolkit(self, answer=None):
try:
import prompt_toolkit
except ImportError:
value = answer or input("\nprompt_toolkit is not installed. shall i install it? [Yn] ")
value = value.strip()
if value and not self.config.parse_bool(value):
sys.stderr.write("prompt_toolkit is required; aborting\n")
sys.exit(1)
subprocess.check_call([sys.executable, '-m', 'pip',
'install', 'prompt_toolkit'])
# nb. this should now succeed
import prompt_toolkit
def rprint(self, *args, **kwargs):
"""
Convenience wrapper for :func:`rich:rich.print()`.
"""
rich.print(*args, **kwargs)
def get_prompt_style(self):
from prompt_toolkit.styles import Style
# message formatting styles
return Style.from_dict({
'': '',
'bold': 'bold',
})
def prompt_generic(
self,
info,
default=None,
is_password=False,
is_bool=False,
required=False,
):
"""
Prompt the user to get their input.
See also :meth:`prompt_bool()`.
:param info: String to display (in bold) as prompt text.
:param default: Default value to assume if user just presses
Enter without providing a value.
:param is_bool: Whether the prompt is for a boolean (Y/N)
value, vs. a normal text value.
:param is_password: Whether the prompt is for a "password" or
other sensitive text value. (User input will be masked.)
:param required: Whether the value is required (user must
provide a value before continuing).
:returns: String value provided by the user (or the default),
unless ``is_bool`` was requested in which case ``True`` or
``False``.
"""
from prompt_toolkit import prompt
# build prompt message
message = [
('', '\n'),
('class:bold', info),
]
if default is not None:
if is_bool:
message.append(('', ' [{}]: '.format('Y' if default else 'N')))
else:
message.append(('', ' [{}]: '.format(default)))
else:
message.append(('', ': '))
# prompt user for input
style = self.get_prompt_style()
try:
text = prompt(message, style=style, is_password=is_password)
except (KeyboardInterrupt, EOFError):
self.rprint("\n\t[bold yellow]operation canceled by user[/bold yellow]\n",
file=sys.stderr)
sys.exit(1)
if is_bool:
if text == '':
return default
elif text.upper() == 'Y':
return True
elif text.upper() == 'N':
return False
self.rprint("\n\t[bold yellow]ambiguous, please try again[/bold yellow]\n")
return self.prompt_generic(info, default, is_bool=True)
if required and not text and not default:
return self.prompt_generic(info, default, is_password=is_password,
required=True)
return text or default
def prompt_bool(self, info, default=None):
"""
Prompt the user for a boolean (Y/N) value.
Convenience wrapper around :meth:`prompt_generic()` with
``is_bool=True``..
:returns: ``True`` or ``False``.
"""
return self.prompt_generic(info, is_bool=True, default=default)

View file

@ -0,0 +1,29 @@
#!/bin/sh -e
<%text>##################################################</%text>
#
# ${app_title} - upgrade script
#
<%text>##################################################</%text>
if [ "$1" = "--verbose" ]; then
VERBOSE='--verbose'
QUIET=
else
VERBOSE=
QUIET='--quiet'
fi
cd ${envdir}
PIP='bin/pip'
ALEMBIC='bin/alembic'
# upgrade pip and friends
$PIP install $QUIET --disable-pip-version-check --upgrade pip
$PIP install $QUIET --upgrade setuptools wheel
# upgrade app proper
$PIP install $QUIET --upgrade --upgrade-strategy eager '${pypi_name}'
# migrate schema
$ALEMBIC -c app/wutta.conf upgrade heads

View file

@ -0,0 +1,90 @@
## -*- mode: conf; -*-
<%text>############################################################</%text>
#
# ${app_title} - web app config
#
<%text>############################################################</%text>
<%text>##############################</%text>
# wutta
<%text>##############################</%text>
${self.section_wutta_config()}
<%text>##############################</%text>
# pyramid
<%text>##############################</%text>
${self.section_app_main()}
${self.section_server_main()}
<%text>##############################</%text>
# logging
<%text>##############################</%text>
${self.sectiongroup_logging()}
######################################################################
## section templates below
######################################################################
<%def name="section_wutta_config()">
[wutta.config]
require = %(here)s/wutta.conf
</%def>
<%def name="section_app_main()">
[app:main]
#use = egg:wuttaweb
use = egg:${egg_name}
pyramid.reload_templates = true
pyramid.debug_all = true
pyramid.default_locale_name = en
#pyramid.includes = pyramid_debugtoolbar
beaker.session.type = file
beaker.session.data_dir = %(here)s/cache/sessions/data
beaker.session.lock_dir = %(here)s/cache/sessions/lock
beaker.session.secret = ${beaker_secret}
beaker.session.key = ${beaker_key}
exclog.extra_info = true
# required for wuttaweb
wutta.config = %(__file__)s
</%def>
<%def name="section_server_main()">
[server:main]
use = egg:waitress#main
host = ${pyramid_host}
port = ${pyramid_port}
# NOTE: this is needed for local reverse proxy stuff to work with HTTPS
# https://docs.pylonsproject.org/projects/waitress/en/latest/reverse-proxy.html
# https://docs.pylonsproject.org/projects/waitress/en/latest/arguments.html
trusted_proxy = 127.0.0.1
trusted_proxy_headers = x-forwarded-for x-forwarded-host x-forwarded-proto x-forwarded-port
clear_untrusted_proxy_headers = True
# TODO: leave this empty if proxy serves as root site, e.g. https://wutta.example.com/
# url_prefix =
# TODO: or, if proxy serves as subpath of root site, e.g. https://wutta.example.com/backend/
# url_prefix = /backend
</%def>
<%def name="sectiongroup_logging()">
[handler_console]
level = INFO
[handler_file]
args = (${repr(os.path.join(appdir, 'log', 'web.log'))}, 'a', 1000000, 100, 'utf_8')
</%def>

View file

@ -0,0 +1,160 @@
## -*- mode: conf; -*-
<%text>############################################################</%text>
#
# ${app_title} - base config
#
<%text>############################################################</%text>
<%text>##############################</%text>
# wutta
<%text>##############################</%text>
${self.section_wutta()}
${self.section_wutta_config()}
${self.section_wutta_db()}
${self.section_wutta_mail()}
${self.section_wutta_upgrades()}
<%text>##############################</%text>
# alembic
<%text>##############################</%text>
${self.section_alembic()}
<%text>##############################</%text>
# logging
<%text>##############################</%text>
${self.sectiongroup_logging()}
######################################################################
## section templates below
######################################################################
<%def name="section_wutta()">
[wutta]
app_title = ${app_title}
</%def>
<%def name="section_wutta_config()">
[wutta.config]
#require = /etc/wutta/wutta.conf
configure_logging = true
usedb = true
preferdb = true
</%def>
<%def name="section_wutta_db()">
[wutta.db]
default.url = ${db_url}
## TODO
## versioning.enabled = true
</%def>
<%def name="section_wutta_mail()">
[wutta.mail]
# this is the global email shutoff switch
#send_emails = false
# recommended setup is to always talk to postfix on localhost and then
# it can handle any need complexities, e.g. sending to relay
smtp.server = localhost
# by default only email templates from wuttjamaican are used
templates = wuttjamaican:templates/mail
## TODO
## # this is the "default" email profile, from which all others initially
## # inherit, but most/all profiles will override these values
## default.prefix = [${app_title}]
## default.from = wutta@localhost
## default.to = root@localhost
# nb. in test environment it can be useful to disable by default, and
# then selectively enable certain (e.g. feedback, upgrade) emails
#default.enabled = false
</%def>
<%def name="section_wutta_upgrades()">
## TODO
## [wutta.upgrades]
## command = ${os.path.join(appdir, 'upgrade.sh')} --verbose
## files = ${os.path.join(appdir, 'data', 'upgrades')}
</%def>
<%def name="section_alembic()">
[alembic]
script_location = wuttjamaican.db:alembic
version_locations = ${pkg_name}.db:alembic/versions wuttjamaican.db:alembic/versions
</%def>
<%def name="sectiongroup_logging()">
[loggers]
keys = root, beaker, exc_logger, sqlalchemy, txn
[handlers]
keys = file, console, email
[formatters]
keys = generic, console
[logger_root]
handlers = file, console
level = DEBUG
[logger_beaker]
qualname = beaker
handlers =
level = INFO
[logger_exc_logger]
qualname = exc_logger
handlers = email
level = ERROR
[logger_sqlalchemy]
qualname = sqlalchemy.engine
handlers =
# handlers = file
# level = INFO
[logger_txn]
qualname = txn
handlers =
level = INFO
[handler_file]
class = handlers.RotatingFileHandler
args = (${repr(os.path.join(appdir, 'log', 'wutta.log'))}, 'a', 1000000, 100, 'utf_8')
formatter = generic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
formatter = console
# formatter = generic
# level = INFO
# level = WARNING
[handler_email]
class = handlers.SMTPHandler
args = ('localhost', 'wutta@localhost', ['root@localhost'], "[${app_title}] Logging")
formatter = generic
level = ERROR
[formatter_generic]
format = %(asctime)s %(levelname)-5.5s [%(name)s][%(threadName)s] %(funcName)s: %(message)s
datefmt = %Y-%m-%d %H:%M:%S
[formatter_console]
format = %(levelname)-5.5s [%(name)s][%(threadName)s] %(funcName)s: %(message)s
</%def>

View file

@ -65,7 +65,7 @@ class FileTestCase(TestCase):
def setup_file_config(self): # pragma: no cover def setup_file_config(self): # pragma: no cover
""" """ """ """
warnings.warn("FileConfigTestCase.setup_file_config() is deprecated; " warnings.warn("FileTestCase.setup_file_config() is deprecated; "
"please use setup_files() instead", "please use setup_files() instead",
DeprecationWarning, stacklevel=2) DeprecationWarning, stacklevel=2)
self.setup_files() self.setup_files()
@ -82,7 +82,7 @@ class FileTestCase(TestCase):
def teardown_file_config(self): # pragma: no cover def teardown_file_config(self): # pragma: no cover
""" """ """ """
warnings.warn("FileConfigTestCase.teardown_file_config() is deprecated; " warnings.warn("FileTestCase.teardown_file_config() is deprecated; "
"please use teardown_files() instead", "please use teardown_files() instead",
DeprecationWarning, stacklevel=2) DeprecationWarning, stacklevel=2)
self.teardown_files() self.teardown_files()
@ -112,6 +112,58 @@ class FileTestCase(TestCase):
FileConfigTestCase = FileTestCase FileConfigTestCase = FileTestCase
class ConfigTestCase(FileTestCase):
"""
Base class for test suites requiring a config object.
It inherits from :class:`FileTestCase` so also has the
file-related methods.
The running test has these attributes:
.. attribute:: config
Reference to the config object.
.. attribute:: app
Reference to the app handler.
.. note::
If you subclass this directly and need to override
setup/teardown, please be sure to call the corresponding
methods for this class.
"""
def setUp(self):
""" """
self.setup_config()
def setup_config(self):
"""
Perform config setup operations for the test.
"""
self.setup_files()
self.config = self.make_config()
self.app = self.config.get_app()
def tearDown(self):
""" """
self.teardown_config()
def teardown_config(self):
"""
Perform config teardown operations for the test.
"""
self.teardown_files()
def make_config(self, **kwargs):
""" """
return WuttaConfig(**kwargs)
# TODO: this should inherit from ConfigTestCase
class DataTestCase(FileTestCase): class DataTestCase(FileTestCase):
""" """
Base class for test suites requiring a full (typical) database. Base class for test suites requiring a full (typical) database.

View file

@ -9,9 +9,10 @@ from unittest import TestCase
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
import pytest import pytest
from mako.template import Template
import wuttjamaican.enum import wuttjamaican.enum
from wuttjamaican import app from wuttjamaican import app as mod
from wuttjamaican.progress import ProgressBase from wuttjamaican.progress import ProgressBase
from wuttjamaican.conf import WuttaConfig from wuttjamaican.conf import WuttaConfig
from wuttjamaican.util import UNSPECIFIED from wuttjamaican.util import UNSPECIFIED
@ -23,7 +24,7 @@ class TestAppHandler(FileTestCase):
def setUp(self): def setUp(self):
self.setup_files() self.setup_files()
self.config = WuttaConfig(appname='wuttatest') self.config = WuttaConfig(appname='wuttatest')
self.app = app.AppHandler(self.config) self.app = mod.AppHandler(self.config)
self.config.app = self.app self.config.app = self.app
def test_init(self): def test_init(self):
@ -83,7 +84,7 @@ class TestAppHandler(FileTestCase):
self.assertFalse(os.path.exists(appdir)) self.assertFalse(os.path.exists(appdir))
self.app.make_appdir(appdir) self.app.make_appdir(appdir)
self.assertTrue(os.path.exists(appdir)) self.assertTrue(os.path.exists(appdir))
self.assertEqual(len(os.listdir(appdir)), 3) self.assertEqual(len(os.listdir(appdir)), 4)
shutil.rmtree(tempdir) shutil.rmtree(tempdir)
# subfolders still added if appdir already exists # subfolders still added if appdir already exists
@ -91,9 +92,28 @@ class TestAppHandler(FileTestCase):
self.assertTrue(os.path.exists(tempdir)) self.assertTrue(os.path.exists(tempdir))
self.assertEqual(len(os.listdir(tempdir)), 0) self.assertEqual(len(os.listdir(tempdir)), 0)
self.app.make_appdir(tempdir) self.app.make_appdir(tempdir)
self.assertEqual(len(os.listdir(tempdir)), 3) self.assertEqual(len(os.listdir(tempdir)), 4)
shutil.rmtree(tempdir) shutil.rmtree(tempdir)
def test_render_mako_template(self):
output_conf = self.write_file('output.conf', '')
template = Template("""\
[wutta]
app_title = WuttaTest
""")
output = self.app.render_mako_template(template, {}, output_path=output_conf)
self.assertEqual(output, """\
[wutta]
app_title = WuttaTest
""")
with open(output_conf, 'rt') as f:
self.assertEqual(f.read(), output)
def test_resource_path(self):
result = self.app.resource_path('wuttjamaican:templates')
self.assertEqual(result, os.path.join(os.path.dirname(mod.__file__), 'templates'))
def test_make_session(self): def test_make_session(self):
try: try:
from wuttjamaican import db from wuttjamaican import db
@ -411,16 +431,17 @@ class TestAppHandler(FileTestCase):
self.assertIsInstance(auth, AuthHandler) self.assertIsInstance(auth, AuthHandler)
def test_get_email_handler(self): def test_get_email_handler(self):
try:
import mako
except ImportError:
pytest.skip("test not relevant without mako")
from wuttjamaican.email import EmailHandler from wuttjamaican.email import EmailHandler
mail = self.app.get_email_handler() mail = self.app.get_email_handler()
self.assertIsInstance(mail, EmailHandler) self.assertIsInstance(mail, EmailHandler)
def test_get_install_handler(self):
from wuttjamaican.install import InstallHandler
install = self.app.get_install_handler()
self.assertIsInstance(install, InstallHandler)
def test_get_people_handler(self): def test_get_people_handler(self):
from wuttjamaican.people import PeopleHandler from wuttjamaican.people import PeopleHandler
@ -428,11 +449,6 @@ class TestAppHandler(FileTestCase):
self.assertIsInstance(people, PeopleHandler) self.assertIsInstance(people, PeopleHandler)
def test_send_email(self): def test_send_email(self):
try:
import mako
except ImportError:
pytest.skip("test not relevant without mako")
from wuttjamaican.email import EmailHandler from wuttjamaican.email import EmailHandler
with patch.object(EmailHandler, 'send_email') as send_email: with patch.object(EmailHandler, 'send_email') as send_email:
@ -444,13 +460,13 @@ class TestAppProvider(TestCase):
def setUp(self): def setUp(self):
self.config = WuttaConfig(appname='wuttatest') self.config = WuttaConfig(appname='wuttatest')
self.app = app.AppHandler(self.config) self.app = mod.AppHandler(self.config)
self.config._app = self.app self.config._app = self.app
def test_constructor(self): def test_constructor(self):
# config object is expected # config object is expected
provider = app.AppProvider(self.config) provider = mod.AppProvider(self.config)
self.assertIs(provider.config, self.config) self.assertIs(provider.config, self.config)
self.assertIs(provider.app, self.app) self.assertIs(provider.app, self.app)
self.assertEqual(provider.appname, 'wuttatest') self.assertEqual(provider.appname, 'wuttatest')
@ -458,13 +474,13 @@ class TestAppProvider(TestCase):
# but can pass app handler instead # but can pass app handler instead
with warnings.catch_warnings(): with warnings.catch_warnings():
warnings.filterwarnings('ignore', category=DeprecationWarning) warnings.filterwarnings('ignore', category=DeprecationWarning)
provider = app.AppProvider(self.app) provider = mod.AppProvider(self.app)
self.assertIs(provider.config, self.config) self.assertIs(provider.config, self.config)
self.assertIs(provider.app, self.app) self.assertIs(provider.app, self.app)
def test_get_all_providers(self): def test_get_all_providers(self):
class FakeProvider(app.AppProvider): class FakeProvider(mod.AppProvider):
pass pass
# nb. we specify *classes* here # nb. we specify *classes* here
@ -482,7 +498,7 @@ class TestAppProvider(TestCase):
def test_hasattr(self): def test_hasattr(self):
class FakeProvider(app.AppProvider): class FakeProvider(mod.AppProvider):
def fake_foo(self): def fake_foo(self):
pass pass
@ -499,7 +515,7 @@ class TestAppProvider(TestCase):
# now we test that providers are loaded... # now we test that providers are loaded...
class FakeProvider(app.AppProvider): class FakeProvider(mod.AppProvider):
def fake_foo(self): def fake_foo(self):
return 42 return 42
@ -541,11 +557,11 @@ class TestGenericHandler(TestCase):
def setUp(self): def setUp(self):
self.config = WuttaConfig(appname='wuttatest') self.config = WuttaConfig(appname='wuttatest')
self.app = app.AppHandler(self.config) self.app = mod.AppHandler(self.config)
self.config._app = self.app self.config._app = self.app
def test_constructor(self): def test_constructor(self):
handler = app.GenericHandler(self.config) handler = mod.GenericHandler(self.config)
self.assertIs(handler.config, self.config) self.assertIs(handler.config, self.config)
self.assertIs(handler.app, self.app) self.assertIs(handler.app, self.app)
self.assertEqual(handler.appname, 'wuttatest') self.assertEqual(handler.appname, 'wuttatest')

452
tests/test_install.py Normal file
View file

@ -0,0 +1,452 @@
# -*- coding: utf-8; -*-
import os
import sys
from unittest.mock import patch, MagicMock
import pytest
from mako.lookup import TemplateLookup
from wuttjamaican import install as mod
from wuttjamaican.testing import ConfigTestCase
class TestInstallHandler(ConfigTestCase):
def make_handler(self, **kwargs):
return mod.InstallHandler(self.config, **kwargs)
def test_constructor(self):
handler = self.make_handler()
self.assertEqual(handler.pkg_name, 'poser')
self.assertEqual(handler.app_title, 'poser')
self.assertEqual(handler.pypi_name, 'poser')
self.assertEqual(handler.egg_name, 'poser')
def test_run(self):
handler = self.make_handler()
with patch.object(handler, 'show_welcome') as show_welcome:
with patch.object(handler, 'sanity_check') as sanity_check:
with patch.object(handler, 'do_install_steps') as do_install_steps:
handler.run()
show_welcome.assert_called_once_with()
sanity_check.assert_called_once_with()
do_install_steps.assert_called_once_with()
def test_show_welcome(self):
handler = self.make_handler()
with patch.object(mod, 'sys') as sys:
with patch.object(handler, 'rprint') as rprint:
with patch.object(handler, 'prompt_bool') as prompt_bool:
# user continues
prompt_bool.return_value = True
handler.show_welcome()
self.assertFalse(sys.exit.called)
# user aborts
prompt_bool.return_value = False
handler.show_welcome()
sys.exit.assert_called_once_with(1)
def test_sanity_check(self):
handler = self.make_handler()
with patch.object(mod, 'sys') as sys:
with patch.object(mod, 'os') as os:
with patch.object(handler, 'rprint') as rprint:
# pretend appdir does not exist
os.path.exists.return_value = False
handler.sanity_check()
self.assertFalse(sys.exit.called)
# pretend appdir does exist
os.path.exists.return_value = True
handler.sanity_check()
sys.exit.assert_called_once_with(2)
def test_do_install_steps(self):
handler = self.make_handler()
handler.templates = TemplateLookup(directories=[
self.app.resource_path('wuttjamaican:templates/install'),
])
dbinfo = {
'dburl': f'sqlite:///{self.tempdir}/poser.sqlite',
}
orig_import = __import__
mock_prompt = MagicMock()
def mock_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'prompt_toolkit':
if fromlist == ('prompt',):
return MagicMock(prompt=mock_prompt)
return orig_import(name, globals, locals, fromlist, level)
with patch('builtins.__import__', side_effect=mock_import):
with patch.object(handler, 'get_dbinfo', return_value=dbinfo):
with patch.object(handler, 'install_db_schema') as install_db_schema:
# nb. just for sanity/coverage
install_db_schema.return_value = True
self.assertFalse(hasattr(handler, 'schema_installed'))
handler.do_install_steps()
self.assertTrue(handler.schema_installed)
install_db_schema.assert_called_once_with(dbinfo['dburl'])
def test_get_dbinfo(self):
try:
import sqlalchemy
except ImportError:
pytest.skip("test is not relevant without sqlalchemy")
handler = self.make_handler()
def prompt_generic(info, default=None, is_password=False):
if info in ('db name', 'db user'):
return 'poser'
if is_password:
return 'seekrit'
return default
with patch.object(mod, 'sys') as sys:
with patch.object(handler, 'prompt_generic', side_effect=prompt_generic):
with patch.object(handler, 'test_db_connection') as test_db_connection:
with patch.object(handler, 'rprint') as rprint:
# bad dbinfo
test_db_connection.return_value = "bad dbinfo"
sys.exit.side_effect = RuntimeError
self.assertRaises(RuntimeError, handler.get_dbinfo)
sys.exit.assert_called_once_with(1)
# good dbinfo
sys.exit.reset_mock()
test_db_connection.return_value = None
dbinfo = handler.get_dbinfo()
self.assertFalse(sys.exit.called)
rprint.assert_called_with("[bold green]good[/bold green]")
self.assertEqual(str(dbinfo['dburl']),
'postgresql+psycopg2://poser:seekrit@localhost:5432/poser')
def test_make_db_url(self):
try:
import sqlalchemy
except ImportError:
pytest.skip("test is not relevant without sqlalchemy")
handler = self.make_handler()
url = handler.make_db_url('postgresql', 'localhost', '5432', 'poser', 'poser', 'seekrit')
self.assertEqual(str(url), 'postgresql+psycopg2://poser:seekrit@localhost:5432/poser')
url = handler.make_db_url('mysql', 'localhost', '3306', 'poser', 'poser', 'seekrit')
self.assertEqual(str(url), 'mysql+mysqlconnector://poser:seekrit@localhost:3306/poser')
def test_test_db_connection(self):
try:
import sqlalchemy as sa
except ImportError:
pytest.skip("test is not relevant without sqlalchemy")
handler = self.make_handler()
# db does not exist
result = handler.test_db_connection('sqlite:///bad/url/should/not/exist')
self.assertIn('unable to open database file', result)
# db is setup
url = f'sqlite:///{self.tempdir}/db.sqlite'
engine = sa.create_engine(url)
with engine.begin() as cxn:
cxn.execute(sa.text("create table whatever (id int primary key);"))
self.assertIsNone(handler.test_db_connection(url))
def test_make_template_context(self):
handler = self.make_handler()
dbinfo = {'dburl': 'sqlite:///poser.sqlite'}
context = handler.make_template_context(dbinfo)
self.assertEqual(context['envdir'], sys.prefix)
self.assertEqual(context['pkg_name'], 'poser')
self.assertEqual(context['app_title'], 'poser')
self.assertEqual(context['pypi_name'], 'poser')
self.assertEqual(context['egg_name'], 'poser')
self.assertEqual(context['appdir'], os.path.join(sys.prefix, 'app'))
self.assertEqual(context['db_url'], 'sqlite:///poser.sqlite')
def test_make_appdir(self):
handler = self.make_handler()
handler.templates = TemplateLookup(directories=[
self.app.resource_path('wuttjamaican:templates/install'),
])
dbinfo = {'dburl': 'sqlite:///poser.sqlite'}
context = handler.make_template_context(dbinfo)
handler.make_appdir(context, appdir=self.tempdir)
wutta_conf = os.path.join(self.tempdir, 'wutta.conf')
with open(wutta_conf, 'rt') as f:
self.assertIn('default.url = sqlite:///poser.sqlite', f.read())
def test_install_db_schema(self):
try:
import sqlalchemy as sa
except ImportError:
pytest.skip("test is not relevant without sqlalchemy")
handler = self.make_handler()
db_url = f'sqlite:///{self.tempdir}/poser.sqlite'
wutta_conf = self.write_file('wutta.conf', f"""
[wutta.db]
default.url = {db_url}
""")
# convert to proper URL object
db_url = sa.create_engine(db_url).url
with patch.object(mod, 'subprocess') as subprocess:
# user declines offer to install schema
with patch.object(handler, 'prompt_bool', return_value=False):
self.assertFalse(handler.install_db_schema(db_url, appdir=self.tempdir))
# user agrees to install schema
with patch.object(handler, 'prompt_bool', return_value=True):
self.assertTrue(handler.install_db_schema(db_url, appdir=self.tempdir))
subprocess.check_call.assert_called_once_with([
os.path.join(sys.prefix, 'bin', 'alembic'),
'-c', wutta_conf, 'upgrade', 'heads'])
def test_show_goodbye(self):
handler = self.make_handler()
with patch.object(handler, 'rprint') as rprint:
handler.schema_installed = True
handler.show_goodbye()
rprint.assert_any_call("\n\t[bold green]initial setup is complete![/bold green]")
rprint.assert_any_call("\t[blue]bin/pserve file+ini:app/web.conf[/blue]")
def test_require_prompt_toolkit_installed(self):
# nb. this assumes we *do* have prompt_toolkit installed
handler = self.make_handler()
with patch.object(mod, 'subprocess') as subprocess:
handler.require_prompt_toolkit(answer='Y')
self.assertFalse(subprocess.check_call.called)
def test_require_prompt_toolkit_missing(self):
handler = self.make_handler()
orig_import = __import__
stuff = {'attempts': 0}
def mock_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'prompt_toolkit':
# nb. pretend this is not installed
raise ImportError
return orig_import(name, globals, locals, fromlist, level)
# prompt_toolkit not installed, and user declines offer to install
with patch('builtins.__import__', side_effect=mock_import):
with patch.object(mod, 'subprocess') as subprocess:
with patch.object(mod, 'sys') as sys:
sys.exit.side_effect = RuntimeError
self.assertRaises(RuntimeError, handler.require_prompt_toolkit, answer='N')
self.assertFalse(subprocess.check_call.called)
sys.stderr.write.assert_called_once_with("prompt_toolkit is required; aborting\n")
sys.exit.assert_called_once_with(1)
def test_require_prompt_toolkit_missing_then_installed(self):
handler = self.make_handler()
orig_import = __import__
stuff = {'attempts': 0}
def mock_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'prompt_toolkit':
stuff['attempts'] += 1
if stuff['attempts'] == 1:
# nb. pretend this is not installed
raise ImportError
return orig_import('prompt_toolkit')
return orig_import(name, globals, locals, fromlist, level)
# prompt_toolkit not installed, and user declines offer to install
with patch('builtins.__import__', side_effect=mock_import):
with patch.object(mod, 'subprocess') as subprocess:
with patch.object(mod, 'sys') as sys:
sys.executable = 'python'
handler.require_prompt_toolkit(answer='Y')
subprocess.check_call.assert_called_once_with(['python', '-m', 'pip',
'install', 'prompt_toolkit'])
self.assertFalse(sys.exit.called)
self.assertEqual(stuff['attempts'], 2)
def test_prompt_generic(self):
handler = self.make_handler()
style = handler.get_prompt_style()
orig_import = __import__
mock_prompt = MagicMock()
def mock_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'prompt_toolkit':
if fromlist == ('prompt',):
return MagicMock(prompt=mock_prompt)
return orig_import(name, globals, locals, fromlist, level)
with patch('builtins.__import__', side_effect=mock_import):
with patch.object(handler, 'get_prompt_style', return_value=style):
with patch.object(handler, 'rprint') as rprint:
# no input or default value
mock_prompt.return_value = ''
result = handler.prompt_generic('foo')
self.assertIsNone(result)
mock_prompt.assert_called_once_with([('', '\n'),
('class:bold', 'foo'),
('', ': ')],
style=style, is_password=False)
# fallback to default value
mock_prompt.reset_mock()
mock_prompt.return_value = ''
result = handler.prompt_generic('foo', default='baz')
self.assertEqual(result, 'baz')
mock_prompt.assert_called_once_with([('', '\n'),
('class:bold', 'foo'),
('', ' [baz]: ')],
style=style, is_password=False)
# text input value
mock_prompt.reset_mock()
mock_prompt.return_value = 'bar'
result = handler.prompt_generic('foo')
self.assertEqual(result, 'bar')
mock_prompt.assert_called_once_with([('', '\n'),
('class:bold', 'foo'),
('', ': ')],
style=style, is_password=False)
# bool value (no default; true input)
mock_prompt.reset_mock()
mock_prompt.return_value = 'Y'
result = handler.prompt_generic('foo', is_bool=True)
self.assertTrue(result)
mock_prompt.assert_called_once_with([('', '\n'),
('class:bold', 'foo'),
('', ': ')],
style=style, is_password=False)
# bool value (no default; false input)
mock_prompt.reset_mock()
mock_prompt.return_value = 'N'
result = handler.prompt_generic('foo', is_bool=True)
self.assertFalse(result)
mock_prompt.assert_called_once_with([('', '\n'),
('class:bold', 'foo'),
('', ': ')],
style=style, is_password=False)
# bool value (default; no input)
mock_prompt.reset_mock()
mock_prompt.return_value = ''
result = handler.prompt_generic('foo', is_bool=True, default=True)
self.assertTrue(result)
mock_prompt.assert_called_once_with([('', '\n'),
('class:bold', 'foo'),
('', ' [Y]: ')],
style=style, is_password=False)
# bool value (bad input)
mock_prompt.reset_mock()
counter = {'attempts': 0}
def omg(*args, **kwargs):
counter['attempts'] += 1
if counter['attempts'] == 1:
# nb. bad input first time we ask
return 'doesnotmakesense'
# nb. but good input after that
return 'N'
mock_prompt.side_effect = omg
result = handler.prompt_generic('foo', is_bool=True)
self.assertFalse(result)
# nb. user was prompted twice
self.assertEqual(mock_prompt.call_count, 2)
# Ctrl+C
mock_prompt.reset_mock()
mock_prompt.side_effect = KeyboardInterrupt
with patch.object(mod, 'sys') as sys:
sys.exit.side_effect = RuntimeError
self.assertRaises(RuntimeError, handler.prompt_generic, 'foo')
sys.exit.assert_called_once_with(1)
# Ctrl+D
mock_prompt.reset_mock()
mock_prompt.side_effect = EOFError
with patch.object(mod, 'sys') as sys:
sys.exit.side_effect = RuntimeError
self.assertRaises(RuntimeError, handler.prompt_generic, 'foo')
sys.exit.assert_called_once_with(1)
# missing required value
mock_prompt.reset_mock()
counter = {'attempts': 0}
def omg(*args, **kwargs):
counter['attempts'] += 1
if counter['attempts'] == 1:
# nb. no input first time we ask
return ''
# nb. but good input after that
return 'bar'
mock_prompt.side_effect = omg
result = handler.prompt_generic('foo', required=True)
self.assertEqual(result, 'bar')
# nb. user was prompted twice
self.assertEqual(mock_prompt.call_count, 2)
def test_prompt_bool(self):
handler = self.make_handler()
orig_import = __import__
mock_prompt = MagicMock()
def mock_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'prompt_toolkit':
if fromlist == ('prompt',):
return MagicMock(prompt=mock_prompt)
return orig_import(name, globals, locals, fromlist, level)
with patch('builtins.__import__', side_effect=mock_import):
with patch.object(handler, 'rprint') as rprint:
# no default; true input
mock_prompt.reset_mock()
mock_prompt.return_value = 'Y'
result = handler.prompt_bool('foo')
self.assertTrue(result)
mock_prompt.assert_called_once()
# no default; false input
mock_prompt.reset_mock()
mock_prompt.return_value = 'N'
result = handler.prompt_bool('foo')
self.assertFalse(result)
mock_prompt.assert_called_once()
# default; no input
mock_prompt.reset_mock()
mock_prompt.return_value = ''
result = handler.prompt_bool('foo', default=True)
self.assertTrue(result)
mock_prompt.assert_called_once()
# bad input
mock_prompt.reset_mock()
counter = {'attempts': 0}
def omg(*args, **kwargs):
counter['attempts'] += 1
if counter['attempts'] == 1:
# nb. bad input first time we ask
return 'doesnotmakesense'
# nb. but good input after that
return 'N'
mock_prompt.side_effect = omg
result = handler.prompt_bool('foo')
self.assertFalse(result)
# nb. user was prompted twice
self.assertEqual(mock_prompt.call_count, 2)

View file

@ -3,7 +3,8 @@
envlist = py38, py39, py310, py311, nox envlist = py38, py39, py310, py311, nox
[testenv] [testenv]
extras = db,email,docs,tests extras = db,docs,tests
deps = prompt_toolkit
commands = pytest {posargs} commands = pytest {posargs}
[testenv:nox] [testenv:nox]