1#!/usr/bin/env python 2 3from odb import * 4 5## ----------------------------------------------------------------------- 6## T E S T S 7## ----------------------------------------------------------------------- 8 9import MySQLdb 10import odb_mysql 11import sqlite 12import odb_sqlite 13 14def TEST(output=log): 15 LOGGING_STATUS[DEV_SELECT] = 1 16 LOGGING_STATUS[DEV_UPDATE] = 1 17 18 print "------ TESTING MySQLdb ---------" 19 rdb = MySQLdb.connect(host = 'localhost',user='root', passwd = '', db='testdb') 20 ndb = MySQLdb.connect(host = 'localhost',user='trakken', passwd = 'trakpas', db='testdb') 21 cursor = rdb.cursor() 22 23 output("drop table agents") 24 try: 25 cursor.execute("drop table agents") # clean out the table 26 except: 27 pass 28 output("creating table") 29 30 SQL = """ 31 32 create table agents ( 33 agent_id integer not null primary key auto_increment, 34 login varchar(200) not null, 35 unique (login), 36 ext_email varchar(200) not null, 37 hashed_pw varchar(20) not null, 38 name varchar(200), 39 auth_level integer default 0, 40 ticket_count integer default 0) 41 """ 42 43 cursor.execute(SQL) 44 db = odb_mysql.Database(ndb) 45 TEST_DATABASE(rdb,db,output=output) 46 47 print "------ TESTING sqlite ----------" 48 rdb = sqlite.connect("/tmp/test.db",autocommit=1) 49 cursor = rdb.cursor() 50 try: 51 cursor.execute("drop table agents") 52 except: 53 pass 54 SQL = """ 55 create table agents ( 56 agent_id integer primary key, 57 login varchar(200) not null, 58 ext_email varchar(200) not null, 59 hashed_pw varchar(20), 60 name varchar(200), 61 auth_level integer default 0, 62 ticket_count integer default 0)""" 63 cursor.execute(SQL) 64 rdb = sqlite.connect("/tmp/test.db",autocommit=1) 65 ndb = sqlite.connect("/tmp/test.db",autocommit=1) 66 67 db = odb_sqlite.Database(ndb) 68 TEST_DATABASE(rdb,db,output=output,is_mysql=0) 69 70 71 72 73def TEST_DATABASE(rdb,db,output=log,is_mysql=1): 74 75 cursor = rdb.cursor() 76 77 class AgentsTable(Table): 78 def _defineRows(self): 79 self.d_addColumn("agent_id",kInteger,None,primarykey = 1,autoincrement = 1) 80 self.d_addColumn("login",kVarString,200,notnull=1) 81 self.d_addColumn("ext_email",kVarString,200,notnull=1) 82 self.d_addColumn("hashed_pw",kVarString,20,notnull=1) 83 self.d_addColumn("name",kBigString,compress_ok=1) 84 self.d_addColumn("auth_level",kInteger,None) 85 self.d_addColumn("ticket_count",kIncInteger,None) 86 87 tbl = AgentsTable(db,"agents") 88 89 90 91 92 TEST_INSERT_COUNT = 5 93 94 # --------------------------------------------------------------- 95 # make sure we can catch a missing row 96 97 try: 98 a_row = tbl.fetchRow( ("agent_id", 1000) ) 99 raise "test error" 100 except eNoMatchingRows: 101 pass 102 103 output("PASSED! fetch missing row test") 104 105 # -------------------------------------------------------------- 106 # create new rows and insert them 107 108 for n in range(TEST_INSERT_COUNT): 109 new_id = n + 1 110 111 newrow = tbl.newRow() 112 newrow.name = "name #%d" % new_id 113 newrow.login = "name%d" % new_id 114 newrow.ext_email = "%d@name" % new_id 115 newrow.save() 116 if newrow.agent_id != new_id: 117 raise "new insert id (%s) does not match expected value (%d)" % (newrow.agent_id,new_id) 118 119 output("PASSED! autoinsert test") 120 121 # -------------------------------------------------------------- 122 # fetch one row 123 a_row = tbl.fetchRow( ("agent_id", 1) ) 124 125 if a_row.name != "name #1": 126 raise "row data incorrect" 127 128 output("PASSED! fetch one row test") 129 130 # --------------------------------------------------------------- 131 # don't change and save it 132 # (i.e. the "dummy cursor" string should never be called!) 133 # 134 try: 135 a_row.save(cursor = "dummy cursor") 136 except AttributeError, reason: 137 raise "row tried to access cursor on save() when no changes were made!" 138 139 output("PASSED! don't save when there are no changed") 140 141 # --------------------------------------------------------------- 142 # change, save, load, test 143 144 a_row.auth_level = 10 145 a_row.save() 146 b_row = tbl.fetchRow( ("agent_id", 1) ) 147 if b_row.auth_level != 10: 148 log(repr(b_row)) 149 raise "save and load failed" 150 151 152 output("PASSED! change, save, load") 153 154 # --------------------------------------------------------------- 155 # replace 156 157 158 repl_row = tbl.newRow(replace=1) 159 repl_row.agent_id = a_row.agent_id 160 repl_row.login = a_row.login + "-" + a_row.login 161 repl_row.ext_email = "foo" 162 repl_row.save() 163 164 b_row = tbl.fetchRow( ("agent_id", a_row.agent_id) ) 165 if b_row.login != repl_row.login: 166 raise "replace failed" 167 output("PASSED! replace") 168 169 # -------------------------------------------------------------- 170 # access unknown dict item 171 172 try: 173 a = a_row["UNKNOWN_ATTRIBUTE"] 174 raise "test error" 175 except KeyError, reason: 176 pass 177 178 try: 179 a_row["UNKNOWN_ATTRIBUTE"] = 1 180 raise "test error" 181 except KeyError, reason: 182 pass 183 184 output("PASSED! unknown dict item exception") 185 186 # -------------------------------------------------------------- 187 # access unknown attribute 188 try: 189 a = a_row.UNKNOWN_ATTRIBUTE 190 raise "test error" 191 except AttributeError, reason: 192 pass 193 194 try: 195 a_row.UNKNOWN_ATTRIBUTE = 1 196 raise "test error" 197 except AttributeError, reason: 198 pass 199 200 output("PASSED! unknown attribute exception") 201 202 203 # -------------------------------------------------------------- 204 # use wrong data for column type 205 206 try: 207 a_row.agent_id = "this is a string" 208 raise "test error" 209 except eInvalidData, reason: 210 pass 211 212 output("PASSED! invalid data for column type") 213 214 # -------------------------------------------------------------- 215 # fetch 1 rows 216 217 rows = tbl.fetchRows( ('agent_id', 1) ) 218 if len(rows) != 1: 219 raise "fetchRows() did not return 1 row!" % (TEST_INSERT_COUNT) 220 221 output("PASSED! fetch one row") 222 223 224 # -------------------------------------------------------------- 225 # fetch All rows 226 227 rows = tbl.fetchAllRows() 228 if len(rows) != TEST_INSERT_COUNT: 229 for a_row in rows: 230 output(repr(a_row)) 231 raise "fetchAllRows() did not return TEST_INSERT_COUNT(%d) rows!" % (TEST_INSERT_COUNT) 232 233 output("PASSED! fetchall rows") 234 235 236 # -------------------------------------------------------------- 237 # delete row object 238 239 row = tbl.fetchRow( ('agent_id', 1) ) 240 row.delete() 241 try: 242 row = tbl.fetchRow( ('agent_id', 1) ) 243 raise "delete failed to delete row!" 244 except eNoMatchingRows: 245 pass 246 247 # -------------------------------------------------------------- 248 # table deleteRow() call 249 250 row = tbl.fetchRow( ('agent_id',2) ) 251 tbl.deleteRow( ('agent_id', 2) ) 252 try: 253 row = tbl.fetchRow( ('agent_id',2) ) 254 raise "table delete failed" 255 except eNoMatchingRows: 256 pass 257 258 # -------------------------------------------------------------- 259 # table deleteRow() call 260 261 row = tbl.fetchRow( ('agent_id',3) ) 262 if row.databaseSizeForColumn('name') != len(row.name): 263 raise "databaseSizeForColumn('name') failed" 264 265 # -------------------------------------------------------------- 266 # test inc fields 267 row = tbl.newRow() 268 new_id = 1092 269 row.name = "name #%d" % new_id 270 row.login = "name%d" % new_id 271 row.ext_email = "%d@name" % new_id 272 row.inc('ticket_count') 273 row.save() 274 new_id = row.agent_id 275 276 trow = tbl.fetchRow( ('agent_id',new_id) ) 277 if trow.ticket_count != 1: 278 raise "ticket_count didn't inc!" 279 280 row.inc('ticket_count', count=2) 281 row.save() 282 trow = tbl.fetchRow( ('agent_id',new_id) ) 283 if trow.ticket_count != 3: 284 raise "ticket_count wrong, expected 3, got %d" % trow.ticket_count 285 286 trow.inc('ticket_count') 287 trow.save() 288 if trow.ticket_count != 4: 289 raise "ticket_count wrong, expected 4, got %d" % trow.ticket_count 290 291 output("\n==== ALL TESTS PASSED ====") 292 293 294if __name__ == "__main__": 295 TEST() 296 297