odb_test.py

在 linux平台上的网页编程的模板

#!/usr/bin/env python

from odb import *

## -----------------------------------------------------------------------
##                            T  E  S  T S
## -----------------------------------------------------------------------

import MySQLdb
import odb_mysql
import sqlite
import odb_sqlite
        
def TEST(output=log):
    LOGGING_STATUS[DEV_SELECT] = 1
    LOGGING_STATUS[DEV_UPDATE] = 1

    print "------ TESTING MySQLdb ---------"
    rdb = MySQLdb.connect(host = 'localhost',user='root', passwd = '', db='testdb')
    ndb = MySQLdb.connect(host = 'localhost',user='trakken', passwd = 'trakpas', db='testdb')
    cursor = rdb.cursor()
    
    output("drop table agents")
    try:
        cursor.execute("drop table agents")   # clean out the table
    except:
        pass
    output("creating table")

    SQL = """

    create table agents (
       agent_id integer not null primary key auto_increment,
       login varchar(200) not null,
       unique (login),
       ext_email varchar(200) not null,
       hashed_pw varchar(20) not null,
       name varchar(200),
       auth_level integer default 0,
       ticket_count integer default 0)
       """

    cursor.execute(SQL)
    db = odb_mysql.Database(ndb)
    TEST_DATABASE(rdb,db,output=output)

    print "------ TESTING sqlite ----------"
    rdb = sqlite.connect("/tmp/test.db",autocommit=1)
    cursor = rdb.cursor()
    try:
        cursor.execute("drop table agents")
    except:
        pass
    SQL = """
    create table agents (
       agent_id integer primary key,
       login varchar(200) not null,
       ext_email varchar(200) not null,
       hashed_pw varchar(20),
       name varchar(200),
       auth_level integer default 0,
       ticket_count integer default 0)"""
    cursor.execute(SQL)
    rdb = sqlite.connect("/tmp/test.db",autocommit=1)
    ndb = sqlite.connect("/tmp/test.db",autocommit=1)
        
    db = odb_sqlite.Database(ndb)
    TEST_DATABASE(rdb,db,output=output,is_mysql=0)
    
    


def TEST_DATABASE(rdb,db,output=log,is_mysql=1):

    cursor = rdb.cursor()
    
    class AgentsTable(Table):
        def _defineRows(self):
            self.d_addColumn("agent_id",kInteger,None,primarykey = 1,autoincrement = 1)
            self.d_addColumn("login",kVarString,200,notnull=1)
            self.d_addColumn("ext_email",kVarString,200,notnull=1)
            self.d_addColumn("hashed_pw",kVarString,20,notnull=1)
            self.d_addColumn("name",kBigString,compress_ok=1)
            self.d_addColumn("auth_level",kInteger,None)
            self.d_addColumn("ticket_count",kIncInteger,None)

    tbl = AgentsTable(db,"agents")




    TEST_INSERT_COUNT = 5

    # ---------------------------------------------------------------
    # make sure we can catch a missing row

    try:
        a_row = tbl.fetchRow( ("agent_id", 1000) )
        raise "test error"
    except eNoMatchingRows:
        pass

    output("PASSED! fetch missing row test")

    # --------------------------------------------------------------
    # create new rows and insert them

    for n in range(TEST_INSERT_COUNT):
        new_id = n + 1
        
        newrow = tbl.newRow()
        newrow.name = "name #%d" % new_id
        newrow.login = "name%d" % new_id
        newrow.ext_email = "%d@name" % new_id
        newrow.save()
        if newrow.agent_id != new_id:
            raise "new insert id (%s) does not match expected value (%d)" % (newrow.agent_id,new_id)

    output("PASSED! autoinsert test")

    # --------------------------------------------------------------
    # fetch one row
    a_row = tbl.fetchRow( ("agent_id", 1) )

    if a_row.name != "name #1":
        raise "row data incorrect"

    output("PASSED! fetch one row test")

    # ---------------------------------------------------------------
    # don't change and save it
    # (i.e. the "dummy cursor" string should never be called!)
    #
    try:
        a_row.save(cursor = "dummy cursor")
    except AttributeError, reason:
        raise "row tried to access cursor on save() when no changes were made!"

    output("PASSED! don't save when there are no changed")

    # ---------------------------------------------------------------
    # change, save, load, test
    
    a_row.auth_level = 10
    a_row.save()
    b_row = tbl.fetchRow( ("agent_id", 1) )
    if b_row.auth_level != 10:
        log(repr(b_row))
        raise "save and load failed"
    

    output("PASSED! change, save, load")

    # ---------------------------------------------------------------
    # replace


    repl_row = tbl.newRow(replace=1)
    repl_row.agent_id = a_row.agent_id
    repl_row.login = a_row.login + "-" + a_row.login
    repl_row.ext_email = "foo"
    repl_row.save()

    b_row = tbl.fetchRow( ("agent_id", a_row.agent_id) )
    if b_row.login != repl_row.login:
        raise "replace failed"
    output("PASSED! replace")

    # --------------------------------------------------------------
    # access unknown dict item

    try:
        a = a_row["UNKNOWN_ATTRIBUTE"]
        raise "test error"
    except KeyError, reason:
        pass

    try:
        a_row["UNKNOWN_ATTRIBUTE"] = 1
        raise "test error"
    except KeyError, reason:
        pass

    output("PASSED! unknown dict item exception")

    # --------------------------------------------------------------
    # access unknown attribute
    try:
        a = a_row.UNKNOWN_ATTRIBUTE
        raise "test error"
    except AttributeError, reason:
        pass

    try:
        a_row.UNKNOWN_ATTRIBUTE = 1
        raise "test error"
    except AttributeError, reason:
        pass

    output("PASSED! unknown attribute exception")

 
    # --------------------------------------------------------------
    # use wrong data for column type

    try:
        a_row.agent_id = "this is a string"
        raise "test error"
    except eInvalidData, reason:
        pass

    output("PASSED! invalid data for column type")

    # --------------------------------------------------------------
    # fetch 1 rows

    rows = tbl.fetchRows( ('agent_id', 1) )
    if len(rows) != 1:
        raise "fetchRows() did not return 1 row!" % (TEST_INSERT_COUNT)

    output("PASSED! fetch one row")


    # --------------------------------------------------------------
    # fetch All rows
    
    rows = tbl.fetchAllRows()
    if len(rows) != TEST_INSERT_COUNT:
        for a_row in rows:
            output(repr(a_row))
        raise "fetchAllRows() did not return TEST_INSERT_COUNT(%d) rows!" % (TEST_INSERT_COUNT)

    output("PASSED! fetchall rows")

  
    # --------------------------------------------------------------
    # delete row object

    row = tbl.fetchRow( ('agent_id', 1) )
    row.delete()
    try:
        row = tbl.fetchRow( ('agent_id', 1) )
        raise "delete failed to delete row!"
    except eNoMatchingRows:
        pass

    # --------------------------------------------------------------
    # table deleteRow() call

    row = tbl.fetchRow( ('agent_id',2) )
    tbl.deleteRow( ('agent_id', 2) )
    try:
        row = tbl.fetchRow( ('agent_id',2) )
        raise "table delete failed"
    except eNoMatchingRows:
        pass

    # --------------------------------------------------------------
    # table deleteRow() call

    row = tbl.fetchRow( ('agent_id',3) )
    if row.databaseSizeForColumn('name') != len(row.name):
        raise "databaseSizeForColumn('name') failed"
    
    # --------------------------------------------------------------
    # test inc fields
    row = tbl.newRow()
    new_id = 1092
    row.name = "name #%d" % new_id
    row.login = "name%d" % new_id
    row.ext_email = "%d@name" % new_id
    row.inc('ticket_count')
    row.save()
    new_id = row.agent_id

    trow = tbl.fetchRow( ('agent_id',new_id) )
    if trow.ticket_count != 1:
        raise "ticket_count didn't inc!"

    row.inc('ticket_count', count=2)
    row.save()
    trow = tbl.fetchRow( ('agent_id',new_id) )
    if trow.ticket_count != 3:
        raise "ticket_count wrong, expected 3, got %d" % trow.ticket_count

    trow.inc('ticket_count')
    trow.save()
    if trow.ticket_count != 4:
        raise "ticket_count wrong, expected 4, got %d" % trow.ticket_count

    output("\n==== ALL TESTS PASSED ====")
    

if __name__ == "__main__":
    TEST()