Hello Pyramid [Part 2] – Test First

Posted by – January 23, 2013

Currently, there are some testcases from the previous post. They need to be implemented controllers.

Model

There is an important thing I need to introduce first, SQLAlchemy. Not only recording into the database, transacion is also involved. If we take a look at a code template which is generated from a tool of Pyramid, there is the transaction that is different in my source code.

I start at declaring some fields.

Base = declarative_base()
class Todo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(512), nullable=False)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    done_at = Column(DateTime)
    priority = Column(Integer, default=5) # 1 => the most priority, 10 => not important now 
 
    def __init__(self, task, done_at=None, priority=1):
        super(Todo, self).__init__()
        self.task = task
        self.done_at = done_at
        self.priority = priority

There are the fields to identify some properties like id, what to do, when to be finished and priority.
Another confusing is transaction

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))

This code creates a database connection that passes ZopeTransactionExtension to scoped_session. ZopeTransactionExtension commits the transaction after finishing the processing the request. If there is any exception is raised then it will rollback.

The problem is if we follow an example on the site, it creates a transaction with transacton.manager. When it finishes work in transaction manager, it will commit so there is nothing to be rolled back. To add DBSession.remove() is for removing the data after testing finishes. The example in this doc doesn’t need to remove because it stores the data in sqlite which is a memory database. It removes automatically after testing.

Also, there is anothing to be added to make testing goes well. Let’s start at SQLAlchemy’s ORM method. Whatever it does through a SQLAlchemy’s session, it is like storing executed command to a stack and then execute all of them after returning in the function in the controller but what I expect is after saving, it should be retrived the result at that moment. Doesn’t need to wait until function returns.
To do that, just flush

DBSession.flush()

After flushing then todo’s instance is set some values such as there is a new todo that there is no id, it is a creating in the database. After flushing, id will be set as the id in row in the database. So if no flushing then no id to be set, we don’t know if it save passed. If fail, it will raise an exception that we don’t what happen exactly. It is just an error message from the database that is not friendly to the user. To catch the exception is out of the code’s scope also means we can’t catch the exception if no flushing.

Let’s back to the code. It’s not funny to flush every time after updating or deleting so add some helper methods and add some validation methods that SQLAlchemy doesn’t have.

class AppBase(object):
    _errors = None
 
    def save(self):
        if self.is_valid:
            try:
                DBSession.add(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def update(self):
        if self.is_valid:
            try:
                DBSession.merge(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def delete(self):
        try:
            DBSession.delete(self)
            DBSession.flush()
        except IntegrityError: # still dont have an idea to test this failure
            return False
        return True

Below is the validation added by myself.

    @property
    def errors(self):
        return self._errors
 
    @property
    def is_valid(self):
        return not bool(self._errors)
 
    def validate(self, validator, key, value):
        if not self._errors:
            self._errors = {}
        try:
            validator.to_python(value)
        except Invalid as e:
            if not self._errors.get(key):
                self._errors[key] = []
            self._errors[key].append(str(e))

There is a hooked method named ‘validates’ which is a decorator. It is invoked on value assinging.

    @validates('task')
    def validate_task(self, key, value):
        # validate
        return value

Task is a declared feild in Todo class and this method is in Todo class too.
The code below show how to extend this class.

Base = declarative_base(cls=AppBase)

And here is the whole code in models.py

from sqlalchemy import (
    Column,
    Integer,
    Text,
    String,
    DateTime,
    Integer
    )
 
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import text
from sqlalchemy.exc import IntegrityError
 
from sqlalchemy.orm import (
    scoped_session,
    sessionmaker,
    validates,
    )
 
from zope.sqlalchemy import ZopeTransactionExtension
 
from formencode import validators, Invalid
 
DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
#DBSession = scoped_session(sessionmaker())
 
class AppBase(object):
    _errors = None
 
    def save(self):
        if self.is_valid:
            try:
                DBSession.add(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def update(self):
        if self.is_valid:
            try:
                DBSession.merge(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def delete(self):
        try:
            DBSession.delete(self)
            DBSession.flush()
        except IntegrityError: # still dont have an idea to test this failure
            return False
        return True
 
    @property
    def errors(self):
        return self._errors
 
    @property
    def is_valid(self):
        return not bool(self._errors)
 
    def validate(self, validator, key, value):
        if not self._errors:
            self._errors = {}
        try:
            validator.to_python(value)
        except Invalid as e:
            if not self._errors.get(key):
                self._errors[key] = []
            self._errors[key].append(str(e))
 
Base = declarative_base(cls=AppBase)
 
class Todo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(512), nullable=False)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    done_at = Column(DateTime)
    priority = Column(Integer, default=5) # 1 => the most priority, 10 => not important now 
 
    def __init__(self, task, done_at=None, priority=1):
        super(Todo, self).__init__()
        self.task = task
        self.done_at = done_at
        self.priority = priority
 
    @validates('task')
    def validate_task(self, key, value):
        self.validate(validators.String(not_empty=True), key, value)
        return value
 
    @validates('priority')
    def validate_priority(self, key, value):
        self.validate(validators.Int(), key, value)
        return value

The validation helper I chosed is a href=”http://www.formencode.org/en/latest/” target=”_blank”>formencode. As in the example, just needs validators.String(not_empty=True) means not allowed empty value and validators.Int() which is needed only Int.

Controller

Python frameworks often use word ‘views’ instead controllers. I’m not used to it so I will call it ‘controller(s)’ then. Let’s create the controller pacakge and put todos.py follow the test in the previous post.

from todolist.controllers.todos import create

There is a tricky part with the template engine. Because a bug on the builtin template engine, ZPT(Chameleon). The issue is using macro with doctype cause an error so I changed to mako

Read

Here is for listing.

def get_todo_set():
    todos = DBSession.query(Todo).filter(Todo.done_at==None).order_by(Todo.priority.desc()).all()
    done_todos = DBSession.query(Todo).filter(Todo.done_at!=None).order_by(Todo.priority.desc()).all()
    return todos, done_todos
 
@view_config(route_name='todo_index', renderer='todos/index.mako')
def index(request):
    todos, done_todos = get_todo_set()
    return dict(todos=todos, done_todos=done_todos)

I add get_todo_set separately. It can be used anywhere. Then see the code in the testing. You will see

        self.assertEqual(len(response['todos']), 3)
        self.assertEqual(len(response['done_todos']), 1)
        # test ordering
        self.assertEqual(response['todos'][0].task, "First task")
        self.assertEqual(response['todos'][2].task, "Second task")

There is a checking to a response that is retrived back from the controller.
You can try seeing and compare the rest of the controlling mehtod with the testing.

@view_config, as a decorator set what to do. Paramerters in the code:

  • route_name for linking URI to a function in the controller
  • renderer for assign a template file. It can be either xml or json and it will render the content-type as it is assigned.
  • request_method to assign the http request method to be received.
  • xhr to allow only XmlHttpRequest

To test @view_config needs to be tested with functional test such as checking whether a status code is 200 or content-type is json.

Create
@view_config(route_name='todo_create', renderer='json', request_method='POST', xhr=True)
def create(request):
    todo = Todo(
        task=request.POST.get('task', None),
        priority=request.POST.get('priority', None)
    )
    if not todo.save():
        return {'errors': todo.errors}
    return {'id': todo.id, 'task': todo.task, 'priority':todo.priority, 'messages': '%s has been created' % todo.task }

In the code, it needs only ajax request and needs to ensure that definitely create a todo. As explained above, id will be set after creating. I like cheking like this instead of leave the exception raised because I can check every case can happen.

Update
@view_config(route_name='todo_update', renderer='json', request_method='POST', xhr=True)
def update(request):
    todo_id = request.matchdict['id']
    try:
        todo = DBSession.query(Todo).filter_by(id=todo_id).one()
    except NoResultFound:
        return {'errors': "No todo id: %s" % todo_id}
 
    todo.task=request.POST.get('task', None)
    todo.priority=request.POST.get('priority', None)
 
    if not todo.update():
        return {'errors': todo.errors}
 
    return {'task': todo.task, 'priority':todo.priority, 'messages': '%s has been updated' % todo.task}

The method to update is merge() by put an instance into DBSession, if id is already set then it will update. If not, it will create.

Delete
@view_config(route_name='todo_delete', renderer='json', request_method='POST', xhr=True)
def delete(request):
    todo_id = request.matchdict['id']
    try:
        todo = DBSession.query(Todo).filter_by(id=todo_id).one()
    except NoResultFound:
        return {'errors': "No todo id: %s" % todo_id}
    if not todo.delete():
        return {'errors': "%s can't be deleted" % todo.task}
 
    return {'id': todo.id, 'messages': '%s has been deleted' % todo.task}

And the last one is deleting with a couple of error messages.

Actually, I wanted to work the UI out based on ajax but it took too long and I wanted to show the testing so I left it.

Summary

I’m happy with integration test. The tool like nosetest for doing the code coverage includes unittest is more convenient or unittest can test cover all of I want. It might be I just started so some code might not look ok especially SQLAlchemy part.

You can Download source code or

hg clone https://bitbucket.org/punneng/pyramid-testfirst

I will find free time to try functional test.

0 Comments on Hello Pyramid [Part 2] – Test First

Respond

Respond

Comments

Comments