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