#!/usr/bin/env pythonfrom odb import *## -----------------------------------------------------------------------##                            T  E  S  T S## -----------------------------------------------------------------------import MySQLdbimport odb_mysqlimport sqliteimport 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()