This is a tutorial for a very simple task list application. It is based on Pyramid Single File Tasks Tutorial. This tutorial assumes you have some basic knowledge of web development (HTTP, HTML, CSS) and python.
Avalanche relies on webob for HTTP (WSGI) Request/Response objects.
Avalanche is composed of two main parts. The core contains a very basic WSGI framework based on webapp2. This core contains a WSGI application, a class-based RequestHandler and URL routing.
Avalanche itself is built on top of this core framework to provide extra functionality. It uses Jinja2 as a template language (you should be familiar with Jinja2, this tutorial will not cover it).
This tutorial application uses SQLAlchemy. So for this tutorial you should also be familiar with it.
In this tutorial we first create an application and tests written using only avalanche.core (or “vanilla way”) this is similar to how most frameworks out there works. Then we re-write it on Avalanche “way”...
The complete code for this application can be found on avalanche repository in the folder avalanche/doc/tutorial/tasks.
Install the requirements:
$ pip install requirements.pip
This single file contains all application code.
The application contains a single model class. Each Task has a name, a boolean property closed to indicated if the Task has been completed. And time-stamp of when the task was created.
Using SQLAlchemy declarative we have:
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import func
from sqlalchemy.orm import sessionmaker
BaseModel = declarative_base()
class Task(BaseModel):
__tablename__ = 'task'
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
closed = Column(Boolean, default=False)
created = Column(DateTime, server_default=func.now())
TaskRequestHandler will be used as base class for all the handlers in the Task application. It is a subclass from core.RequestHandler.
In this application we will use Jinja2, so we added a jinja environment to the class to be used by all instances and a render method.
Apart from that it just add some string messages that will be used by the application.
import jinja2
from avalanche.core import RequestHandler
from avalanche.core import WSGIApplication
from avalanche.router import Route
class TaskRequestHandler(RequestHandler):
MSG_TASK_NEW = 'New task successfully added'
MSG_TASK_NAME = 'Please enter a name for the task'
MSG_TASK_CLOSED = 'Task was successfully closed'
JINJA_ENV = jinja2.Environment(
loader=jinja2.FileSystemLoader('templates'),
undefined=jinja2.DebugUndefined,
autoescape=True,
)
def render(self, template, **context):
"""Renders a template and writes the result to the response."""
template = self.JINJA_ENV.get_template(template)
uri_for = self.app.router.build # add uri_for to template
self.response.write(template.render(uri_for=uri_for, **context))
This handler will display the initial page with a list of tasks. It has an option to display all tasks or just the ones that are not completed.
class ListTasks(TaskRequestHandler):
"""list of tasks"""
def get(self):
db = self.app.Session()
show_closed = bool(self.request.GET.get('closed'))
tasks = db.query(Task).order_by(Task.created)
if not show_closed:
tasks = tasks.filter(Task.closed==False)
flash = self.request.GET.get('flash')
self.render('task_list.html', tasks=tasks, flash=flash,
show_closed=show_closed)
This handler will display a form to create a new task. The post method creates a new Task instance and saves it into the database. It also adds a message into the URL to be flashed out on the next page view.
class NewTask(TaskRequestHandler):
"""add a new task"""
def get(self):
flash = self.request.GET.get('flash')
self.render('task_new.html', flash=flash)
def post(self):
name = self.request.POST.get('task-name')
if name:
db = self.app.Session()
db.add(Task(name=name))
db.commit()
self.redirect_to('task-list', flash=self.MSG_TASK_NEW)
else:
# redisplay with error message
self.redirect_to('task-new', flash=self.MSG_TASK_NAME)
This handler’s post method will modify a task marking it as closed.
class CloseTask(TaskRequestHandler):
"""mark a task as closed"""
def post(self, task_id):
db = self.app.Session()
task = db.query(Task).get(task_id)
task.closed = True
db.commit()
self.redirect_to('task-list', flash=self.MSG_TASK_CLOSED)
Create a WSGI application mapping URI’s to handlers and adding some config values. This should really be in a different file but we will keep everything in one module to make it easier to follow the tutorial...
def tasks_app(Session):
routes = [
Route('/', ListTasks, name='task-list'),
Route('/new', NewTask, name='task-new'),
Route('/close/<task_id>', CloseTask, name='task-close'),
]
wsgi_app = WSGIApplication(routes, debug=True)
wsgi_app.Session = Session
return wsgi_app
if __name__ == '__main__': # pragma: no cover
from wsgiref.simple_server import make_server
engine = create_engine('sqlite:///tasks.db', echo=True)
BaseModel.metadata.create_all(engine)
app = tasks_app(Session=sessionmaker(bind=engine))
server = make_server('0.0.0.0', 8080, app)
print('Server started on port 8080\n\n')
server.serve_forever()
Complete source tasks_0.py
Create a folder name templates and the three Jinja template files. We won’t go through the templates in this tutorial.
base.html
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Avalanche Tutorial: Tasks</title>
<style>
body {
font-family: sans-serif;
font-size: 14px;
color: #3e4349;
}
h1, h2, h3, h4, h5, h6 {
font-family: Georgia;
color: #373839;
}
a {
color: #1b61d6;
text-decoration: none;
}
input {
font-size: 14px;
width: 400px;
border: 1px solid #bbbbbb;
padding: 5px;
}
.button {
font-size: 14px;
font-weight: bold;
width: auto;
background: #eeeeee;
padding: 5px 20px 5px 20px;
border: 1px solid #bbbbbb;
border-left: none;
border-right: none;
}
#flash, #notfound {
font-size: 16px;
width: 500px;
text-align: center;
background-color: #e1ecfe;
border-top: 2px solid #7a9eec;
border-bottom: 2px solid #7a9eec;
padding: 10px 20px 10px 20px;
}
#notfound {
background-color: #fbe3e4;
border-top: 2px solid #fbc2c4;
border-bottom: 2px solid #fbc2c4;
padding: 0 20px 30px 20px;
}
#tasks {
width: 500px;
}
#tasks li {
padding: 5px 0 5px 0;
border-bottom: 1px solid #bbbbbb;
}
#tasks li.last {
border-bottom: none;
}
#tasks .name {
width: 400px;
text-align: left;
display: inline-block;
}
#tasks .actions {
width: 80px;
text-align: right;
display: inline-block;
}
button[type=submit] {
background-color: transparent;
border: none;
color: blue;
padding: 0px;
cursor: pointer;
}
</style>
</head>
<body>
{% if flash %}
<div id="flash">
{{flash}}<br/>
</div>
{% endif %}
<div id="page">
{% block page %}{% endblock %}
</div>
</body>
</html>
task_list.html
{% extends "base.html" %}
{% block page %}
<h1>Task's List</h1>
{% if show_closed %}
<a href="{{uri_for('task-list')}}">hide closed</a>
{% else %}
<a href="{{uri_for('task-list', closed='1')}}">show closed</a>
{% endif %}
<ul id="tasks">
{% if tasks %}
{% for task in tasks %}
<li>
{% if task.closed %}
<span class="name" style="text-decoration:line-through">{{task.name}}</span>
{% else %}
<span class="name">{{task.name}}</span>
<span class="actions">
<form method="POST" action="{{uri_for('task-close', task_id=task.id)}}">
[ <button type="submit">close</button> ]
</form>
</span>
{% endif %}
</li>
{% endfor %}
{% else %}
<li>There are no open tasks</li>
{% endif %}
<li class="last">
<a href="{{uri_for('task-new')}}">Add task</a>
</li>
</ul>
{% endblock %}
task_new.html
{% extends "base.html" %}
{% block page %}
<h1>Add a new task</h1>
<form action="{{uri_for('task-new')}}" method="post">
<input type="text" maxlength="100" name="task-name">
<input type="submit" name="add" value="ADD" class="button">
</form>
{% endblock %}
The tests are going to use pytest as unit-test framework and WebTest (a helper to test WSGI applications). To install them:
~/tasks$ pip install pytest webtest
tests/test_tasks.py
import pytest
from webtest import TestApp
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from tasks_0 import BaseModel, tasks_app, Task, TaskRequestHandler
@pytest.fixture()
def engine(request):
engine = create_engine('sqlite:///:memory:', echo=False)
BaseModel.metadata.create_all(engine)
return engine
@pytest.fixture
def Session(request, engine):
session_cls = sessionmaker(bind=engine)
session_cls().query(Task).delete() # make sure test start with empty DB
return session_cls
@pytest.fixture
def client(request, Session):
app = tasks_app(Session)
return TestApp(app)
###############################################
class TestListTasks():
def test_get_display_not_closed(self, Session, client):
db = Session()
db.add(Task(name='my first task'))
db.add(Task(name='task 2', closed=True))
db.add(Task(name='task 3'))
db.commit()
response = client.get('/')
assert "Task's List" in response
assert "show closed" in response
assert "my first task" in response
assert "task 2" not in response
assert "task 3" in response
def test_get_display_all(self, Session, client):
db = Session()
db.add(Task(name='my first task'))
db.add(Task(name='task 2', closed=True))
db.add(Task(name='task 3'))
db.commit()
response = client.get('/?closed=1')
assert "Task's List" in response
assert "hide closed" in response
assert "my first task" in response
assert "task 2" in response
assert "task 3" in response
class TestNewTask():
def test_get(self, client):
response = client.get('/new')
assert "Add a new task" in response
def test_post_save(self, Session, client):
response_0 = client.post('/new', {'task-name':'task test xxx'})
# task is saved in DB
db = Session()
saved = db.query(Task).first()
assert saved
assert 'task test xxx' == saved.name
# page is redirected to list page
response = response_0.follow()
assert "Task's List" in response
# flash message in page
assert TaskRequestHandler.MSG_TASK_NEW in response
def test_post_error(self, Session, client):
response_0 = client.post('/new', {'wrong-name':'task test xxx'})
# task is not saved in DB
db = Session()
saved = db.query(Task).first()
assert not saved
# page is redirected to list page
response = response_0.follow()
assert "Add a new task" in response
# flash message in page
assert TaskRequestHandler.MSG_TASK_NAME in response
class TestCloseTask():
def test_post(self, Session, client):
db = Session()
task_before = Task(name="task abc")
db.add(task_before)
db.commit()
assert not task_before.closed
response_0 = client.post('/close/%s' % task_before.id)
# task is closed
db = Session()
task_after = db.query(Task).get(task_before.id)
assert task_after.closed
# page is redirected to list page
response = response_0.follow()
assert "Task's List" in response
# flash message in page
assert TaskRequestHandler.MSG_TASK_CLOSED in response
pytest.fixture is a pytest funcarg, this is an alternative (and better) approach instead of using xUnit test setUp and tearDown methods. In this case the funcarg will create a temporary database and a client for the application.
But these are very bad unit-tests. Most of the problems come from checking the HTML (string) response. These are actually black-box testing, unit-test by definition should be white-box testing.
writing tests is error prone
When you write a check for a string in a HTML document you expect that string to be in a determined position. If this same string happens to appear in a different position by another part of your page you are actually not testing anything. This is much more common that might seems at first.
Another problem is negative testing, that is testing that a page does not contains a string. This kind of test always pass when you get a blank page, and this is hardly ever what you expect.
no clear error message when test fails
Its normal that unit-tests will eventually fail, when it does it should be easy for the developer to pin-point what went wrong.
When test assertion fails it should tell you what value was expected and what it got. Testing a for a string in a web-page you will just get the whole HTML text. From the HTML text you will have to find the position where you expected the text to be and check what was the actual/incorrect output. This can be quite time consuming and annoying.
Another problem is that it impossible to tell in which stage things did not run as expected. In other words you can only write asserts for the final output.
test breaks too easily
A unit-test should fail only when the feature under test is broken. A page often contain several components/parts, sometimes a problem will break other unrelated tests. This also makes it harder to pin-point where is the problem.
tests are slower than necessary
The tests get slower because the application has to process components/parts of the page that are not under test.
Another reason is that checking for a string or regular expression in a text is much slower than a direct value comparison/assertion.
Using avalanche, we will modify only the request handlers from the previous step. The model, templates and static files will not be changed.
First step will be to change TaskRequestHandler to subclass avalanche.BaseHandler.
Avalanche uses the concept of a Renderer for its request handlers. A Renderer is a class that defines how to process some data to create a text output. i.e. avalanche.renderer.JinjaRenderer is used to render jinja templates.
So we can replace our old render method with a reference to renderer.
- from avalanche.core import RequestHandler
+ from avalanche import BaseHandler
+ from avalanche.renderer import JinjaRenderer
- class TaskRequestHandler(RequestHandler):
+ class TaskRequestHandler(BaseHandler):
- def render(self, template, **context):
- """Renders a template and writes the result to the response."""
- template = self.JINJA_ENV.get_template(template)
- uri_for = self.app.router.build # add uri_for to template
- self.response.write(template.render(uri_for=uri_for, **context))
+ renderer = JinjaRenderer(JINJA_ENV)
When using avalanche instead of writing the get method you should write a context-builder method. By default this context-builder is called a_get. This method should return a dictionary with values to be used by the renderer. Extra information like the template file name can be defined as a class attribute.
class ListTasks(TaskRequestHandler):
"""list all open tasks"""
template = 'task_list.html'
def a_get(self):
tasks = [] # TODO get real values
show_closed = True
return {
'tasks': tasks,
'show_closed': show_closed,
}
This way the output generation is decoupled from the processing logic. The framework is responsible for gluing the parts together.
In the handler processing logic it uses parameters received from the HTTP request. The problem with our previous code is that the logic is very much tied to how to read the params from a HTTP request.
Avalanche uses a decorator to do dependency-injection of the parameters into the function. For example:
+ from avalanche.params import UrlQueryParam
- def get(self):
- show_closed = bool(self.request.GET.get('closed'))
+ @UrlQueryParam('show_closed', 'closed', bool)
+ def a_get(self, show_closed=False):
The decorator does not modify the decorated function. It will just add some configuration to the class handler so that avalanche knows how to glue things together when processing a request.
But the method can be used directly by tests or other parts of the code, this effectively decouple the processing logic from a HTTP request.
It is common that a HTML page contains different parts of content that are not directly related. Looking at our application pages, we can see that all pages are composed of two parts or sections. The section containing the flash message and the main content of the page.
We could create a mix-in handler to deal with the flash messages:
class FlashMixinHandler(BaseHandler):
@UrlQueryParam('flash')
def ctx_flash(self, flash=None):
return {'flash': flash}
Since this will be used by all pages in the application we can add it to our base handler class.
Avalanche can handle more than one context-builder returning dictionaries to be used by the renderer. The list of builders can be specified by the attribute context_get
class TaskRequestHandler(FlashMixinHandler, BaseHandler):
context_get = ['a_get', 'ctx_flash',]
The handler code currently needs a reference to the Application to access the DB session. We can also move this code to a separate context-builder. Although the DB is not used directly in the templates this can be used to inject values in other context-builders (using avalanche.params.ContextParam).
class TaskRequestHandler(FlashMixinHandler, BaseHandler):
def db_session(self):
return {'db': self.app.Session()}
context_get = ['db_session', 'a_get', 'ctx_flash',]
So the complete code of our handler to list tasks:
class ListTasks(TaskRequestHandler):
"""list all open tasks"""
template = 'task_list.html'
@ContextParam('db')
@UrlQueryParam('show_closed', 'closed', bool)
def a_get(self, db, show_closed=False):
tasks = db.query(Task).order_by(Task.created)
if not show_closed:
tasks = tasks.filter(Task.closed==False)
return {
'tasks': tasks,
'show_closed': show_closed,
}
Note how there is no reference to the Request or Response objects!
To perform HTTP redirects the context-builder needs to return a RedirectTo object.
class NewTask(TaskRequestHandler):
"""add a new task"""
template = 'task_new.html'
def a_get(self): # pragma: no cover
pass
@ContextParam('db')
@PostGroupParam('data', 'task')
def a_post(self, db, data):
name = data.get('name')
if name:
db.add(Task(name=name))
db.commit()
return RedirectTo('task-list', flash=self.MSG_TASK_NEW)
else:
return RedirectTo('task-new', flash=self.MSG_TASK_NAME)
The source code for the third handler:
class CloseTask(TaskRequestHandler):
"""mark a task as closed"""
@ContextParam('db')
@UrlPathParam('task_id')
def a_post(self, db, task_id):
task = db.query(Task).get(int(task_id))
task.closed = True
db.commit()
return RedirectTo('task-list', flash=self.MSG_TASK_CLOSED)
avalanche.BaseHandler is not a subclass of avalanche.coreRequestHandler. It needs the help of avalanche.make_handler to create a WSGI app.
def tasks_app(Session):
route_raw = [('/', ListTasks, 'task-list'),
('/new', NewTask, 'task-new'),
('/close/<task_id>', CloseTask, 'task-close')
]
routes = []
for path, handler, name in route_raw:
handler_class = make_handler(RequestHandler, handler)
routes.append(Route(path, handler_class, name))
wsgi_app = WSGIApplication(routes, debug=True)
wsgi_app.Session = Session
return wsgi_app
Complete source tasks.py
The avalanche code makes it much easier to write “good” unit-tests for 2 reasons:
- not required to create Application, Request, Response objects to test handlers
- no assertion of string regexp in generated HTML pages
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from webtest import TestApp
from avalanche import RedirectTo
from tasks import BaseModel, Task, TaskRequestHandler
from tasks import ListTasks, NewTask, CloseTask
from tasks import FlashMixinHandler
from tasks import tasks_app
@pytest.fixture()
def engine(request):
engine = create_engine('sqlite:///:memory:', echo=False)
BaseModel.metadata.create_all(engine)
return engine
@pytest.fixture
def Session(request, engine):
session_cls = sessionmaker(bind=engine)
session_cls().query(Task).delete() # make sure test start with empty DB
return session_cls
###############################################
class TestFlahsHandler():
def test_ctx_flash(self):
handler = FlashMixinHandler()
ctx = handler.ctx_flash(flash='flash message')
assert ctx['flash'] == 'flash message'
class TestListTasks():
def test_get_display_not_closed(self, Session):
db = Session()
db.add(Task(name='my first task'))
db.add(Task(name='task 2', closed=True))
db.add(Task(name='task 3'))
db.commit()
handler = ListTasks()
ctx = handler.a_get(Session(), show_closed=False)
tasks = list(ctx['tasks'])
assert len(tasks) == 2 # closed task is not included
assert tasks[0].name == "my first task"
assert tasks[1].name == "task 3"
assert ctx['show_closed'] == False
def test_get_display_all(self, Session):
db = Session()
db.add(Task(name='my first task'))
db.add(Task(name='task 2', closed=True))
db.add(Task(name='task 3'))
db.commit()
handler = ListTasks()
ctx = handler.a_get(Session(), show_closed=True)
tasks = list(ctx['tasks'])
assert len(tasks) == 3
assert tasks[0].name == "my first task"
assert tasks[1].name == "task 2"
assert tasks[2].name == "task 3"
assert ctx['show_closed'] == True
class TestNewTask():
def test_get(self):
pass
def test_post_save(self, Session):
post_data = {'name':'task test xxx'}
handler = NewTask()
response = handler.a_post(Session(), post_data)
assert isinstance(response, RedirectTo)
assert 'task-list' == response.handler_name
assert {'flash':TaskRequestHandler.MSG_TASK_NEW} == response.params
# task is saved in DB
db = Session()
saved = db.query(Task).first()
assert saved
assert 'task test xxx' == saved.name
def test_post_error(self, Session):
post_data = {'wrong_name': 'task test xxx'}
handler = NewTask()
response = handler.a_post(Session(), data=post_data)
# page is redirected to new page
assert isinstance(response, RedirectTo)
assert 'task-new' == response.handler_name
assert {'flash':TaskRequestHandler.MSG_TASK_NAME} == response.params
# task is not saved in DB
saved = Session().query(Task).first()
assert not saved
class TestCloseTask():
def test_post(self, Session):
db = Session()
task_before = Task(name="task abc")
db.add(task_before)
db.commit()
assert not task_before.closed
handler = CloseTask()
response = handler.a_post(Session(), task_before.id)
# page is redirected to list page
assert isinstance(response, RedirectTo)
assert 'task-list' == response.handler_name
assert {'flash':TaskRequestHandler.MSG_TASK_CLOSED} == response.params
# task is closed
task_after = Session().query(Task).get(task_before.id)
assert task_after.closed
class TestTaskApp():
def test_app(self, Session):
app = tasks_app(Session)
client = TestApp(app)
assert "<h1>Task's List</h1>" in client.get('/')
The second version (using Avalanche) compared with the first version, is a bit longer, more complex and less straightforward! So is it worth? Your mileage may vary.
Beginners usually think a big blob with all your code is easier to understand than to use functions to split functionality... On the other side you have bloated code that makes uses of patterns adding complexity for no real reason...
Hopefully at least how avalanche produce more testable code should be clear.
In this simple example application it is hard to demonstrate how the code using avalanche is more re-usable... It is widely recognized the correlation that testable code leads to well -modularized and reusable code, so one goal leads to the other.