Hello Pyramid [Part 2] – Test First

Posted by – January 17, 2013

Post ก่อนเรามี test แล้ว ตอนนี้เราต้อง implement code เราให้เข้ากับ test นั่น

Model

พระเอกของเราจริงๆ ที่น่าจะพูดถึงก่อนน่าจะเป็น SQLAlchemy นอกจากการบันทึกที่มีรายละเอียดนิดหน่อยแล้ว ยังมีเรือ่ง transaction เข้ามาเกี่ยวด้วย คือมันเหมือนจะไม่มีอะไร แต่ถ้าเป็น code template ที่ generate มาจาก tool ของ Pyramid มันจะมี transaction ที่แปลกออกไป

เริ่มที่การประกาศ 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

ก็ง่ายๆ มี fields บอกรายละเอียด บันทึกเวลาตอนสร้าง เสร็จตอนไหน แล้วก็ระดับความสำคัญของ todo

อีกหนึ่งความมึนคือ transaction

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

code บรรทัดบน จะทำการสร้างการเชื่อมต่อสำหรับฐานข้อมูลซึ่งส่ง ZopeTransactionExtension ทำหน้าที่ commit transaction หลังจากที่ process งานตาม request ที่มีมา ถ้ามีการ raise เกิดขึ้น จะมีการ rollback สิ่งที่เราสั่งไป

ปัญหาคือ ถ้าเราทำตามตัวอย่างในเว็บมันเนี่ย มันจะให้เราสร้าง transaction เองมาใหม่ด้วย transaction.manager พอจบ scope ของ transaction.manager มันจะ commit ทีนี้ ตอนเรา tear down เราจะไม่มีอะไรให้ rollback ที่ผมต้องเพิ่ม DBSession.remove() มาก็เพื่อจะสั่งลบข้อมูลหลังจาก test เสร็จ ซึ่งตัวอย่างใน doc มันไม่ต้องเพราะมัน test ด้วย sqlite ที่เป็นแค่ฐานข้อมูลในหน่วยความจำชั่วคราว จบการ test เสร็จมันก็เคลียร์ไปเลย

ทีนี้ มันมีกระบวณท่าเพิ่มเข้ามาอีกเพื่อทำให้การ test ราบรื่น ต้องมาเริ่มที่ ORM method ของ SQLAlchemy ไม่ว่าจะสั่งให้เพิ่มหรือลบอะไรผ่าน session ของ SQLAlchemy มันจะเหมือนเราสั่งเก็บไว้ใน stack ก่อน แล้วค่อย execute ทีเดียว แต่สิ่งที่ผมคาดหวังไว้ คือ หลังจาก save แล้ว ผมควรจะได้ผลการ save ณ ตอนนั้นเลย วิธีการง่ายๆ คือใช้คำสั่ง

DBSession.flush()

พอ flush แล้ว todo instance ก็จะมีการเซ็ตค่าหลังจากมีการบันทึก เช่น ถ้าหาก new todo ยังไม่มี id จะถือว่าเป็น create พอ flush แล้ว instance นี้จะมี id กำกับตาม id ในฐานข้อมูล หากไม่ flush ก็ไม่มี id ก็จะไม่รู้เลยว่าบันทึกผ่านหรือเปล่า ต้องมาคอยจับ exception เอา ซึ่งตอนนี้ยังไม่รู้เลยว่าจะจับยังไง เพราะมันจะ execute ตามที่อธิบายไปก่อนหน้านี้และอยู่นอก scope ของ code ด้วย

กลับมาเรื่อง code ต่อ มันคงไม่สนุก ถ้าหากจะต้องมา flush ทุกครั้งหลังจากการบันทีกหรือลบ ผมเลยห่อมันไว้เป็น class แม่ซะ พร้อมทั้งเพิ่มส่วนของ validation helper ไปด้วย ที่ SQLAlchemy ไม่มี

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

และส่วนของ validation ที่เพิ่มเข้ามาเอง

    @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))

SQLAlchemy มี hook ไว้ที่นึงชื่อ validates เป้น decorator จะถูกเรียกตอนที่ assign ค่า

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

โดย task ก็คือ field ที่เราระบุไว้ใน Todo class โดย method นี้จะอยู่ใน Todo class เช่นกัน โดยเชื่อมระหว่าง class แม่นี้ กับ Todo ด้วย

Base = declarative_base(cls=AppBase)

และ code ใน 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

โดยตัว validation ที่ผมเลือกเข้ามาช่วยคือ formencode ที่ validate ได้ค่อนข้างครอบคลุม อย่างในตัวอย่าง ผมต้องการแค่ validators.String(not_empty=True) คือ ห้ามเป็นค่าว่าง กับ validators.Int() คือต้องเป็น Int เท่านั้น

Controller

Framework ฝั่ง Python เนี่ย มักจะใช้คำว่า views แทน controllers ซึ่งผมไม่ค่อยชอบเท่าไหร่ มันแปลกๆ กับ MVC เลยเปลีย่นเป็น controller(s) ซะเลย โดยสร้าง package เป็น controllers แล้วแยกเป็น file ไป โดยผมตั้งชื่อให้เป็น todos.py ตามใน test ที่มี

from todolist.controllers.todos import create

ใส่ไว้

มีท่ายากมานิดหน่อยสำหรับ template ที่จะใช้ เนื่องจาก template engine ตัว builtin มันเลยคือ ZPT(Chameleon) แต่มันไปติด bug นี้ ปัญหาคือ ใช้ macro แล้วใส่ doctype ไม่ได้ ผมเลยเปลี่ยนมาใช้ mako แทน

Read

ในส่วนนี้จะเป็นหน้า list ของ todo

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)

ผมห่อ get_todo_set แยกไว้ เพราะจะมีการเรียกใช้ที่อื่นด้วย ถ้าเปิดตัว test เทียบดูก็จะเห็น

        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")

จะเห็นว่ามีการเช็คด่ากับ response ซึ่งเป็นสิ่งที่ method ส่งกลับไป
ใน method ที่เหลือ ก็ลองเปิด test เทียบเอานะครับ

code นี้จะมี @view_config เป็น decorator ทำหน้าที่ config หน้าที่หลายๆ อย่างสำหรับ controller(หรือ view) เบื้องต้นทีใช้ในตัวอย่างก็มี

  • route_name ไว้เชื่อมกับการกำหนด path ของ uri ซึ่งต้องกำหนดก่อน
  • renderer ไว้กำหนดไฟล์ที่จะ render หรือถ้ากำหนดเป็น xml หรือ json มันจะแปลงข้อมูลจะส่งไปเป็น xml หรือ json ให้เลย
  • request_method ไว้กำหนด request method
  • xhr ไว้กำหนดให้รับ request ที่เป็น ajax

หรือดู list ทั้งหมดได้ที่ doc

สำหรับการ test ในสวนของ @view_config มีความจำเป็นต้องไป test ด้วย functional test ครับ เช่น test ว่า status เป็น 200 หรือเปล่า หรือคืน content-type เป็น 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 }

จะเห็นว่าต้องการรับ request มาเป้น ajax เท่านั้น
และการยืนยันว่า todo ที่สร้าง ถูกสร้างอย่างแน่นอนคือถ้าสร้างเสร็จแล้วจะต้องมีการเซ็ทค่า id ให้กับ todo ตามที่ผมอธิบายไปข้างบน คือ ถ้าไม่ flush() ก็จะไม่รู้เลยว่ามันบันทึกจริงๆ หรือเปล่า จะจับ exception ก็ยังคิดท่าไม่ออก แต่ข้อดีของการเช็คแบบนี้คือ มันเช็คได้ละเอียดแยกเป็นขั้นตอนได้ดีกว่า

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}

เบี้องหลังของการ update จริงๆ คือ merge() โดยการใส่ instance ลงไปใน DBSession ถ้ามี id อยู่ จะทำการเปลี่ยนค่า แต่ถ้าไม่มี id มันจะสร้างใหม่ให้เลย

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}

และอันสุดท้ายก็เป็นการลบ พร้อมกับข้อควมต่างๆ กรณีที่มีปัญหา

จริงๆ ผมตั้งใจจะทำฝั่ง UI ให้ดูด้วย เป็น ajax แต่พอทำไปสักพักแล้วรู้สึกนานไปละ และผมอยากให้ดูแค่เรื่องของการ test ก็เลยพักไว้แค่นี้

สรุป

การ test เบี้องต้นผมพอใจนะ สำหรับการ test แค่ integration test เครืองมืออย่าง nosetest ไว้ทำ code coverage เอามารวมปับ unittest ถือว่าสะดวกสบายพอสมควร ตัว unitest เองก็ test ได้ครอบคลุมดี อาจจะเพราะผมเองมือใหม่ บางท่าอาจจะยังไม่ถูกต้องนัก โดยเฉพาะกับ SQLAlchemy ถ้ามีโอกาสอยู่กับมันนานๆ คงจะมีท่าดีๆ กว่านี้

สามารถ Download source code มาดูทั้งหมดได้ครับ

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

หรือจะ folk กันก็เอาตามสะดวก

ไว้มีเวลาจะมาลอง functional test ให้ดูครับ ตอนนี้ขอสลับโหมดไป Rails ก่อน :) เลี้ยงปากเลี้ยงท้องหน่อยครับ

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

Respond

Respond

Comments

Comments