test_odb.py
Go to the documentation of this file.
1 #! /usr/bin/env python
2 
3 """
4 usage: %(progname)s [args]
5 """
6 
7 import os, sys, string, time, getopt
8 from log import *
9 
10 import odb
11 import unittest
12 
13 ## -----------------------------------------------------------------------
14 ## T E S T S
15 ## -----------------------------------------------------------------------
16 
17 
18 def TEST_MYSQL(output=warn):
19  output("------ TESTING MySQLdb ---------")
20  import odb_mysql
21  conn = odb_mysql.Connection(host = 'localhost',user='root', passwd = '', db='test')
22  db = odb.Database(conn)
23  return db
24 
25 def TEST_SQLITE(output=warn):
26  output("------ TESTING sqlite2 ----------")
27  import odb_sqlite
28  conn = odb_sqlite.Connection("/tmp/test2.db", autocommit=1)
29  db = odb.Database(conn)
30  return db
31 
32 def TEST_SQLITE3(output=warn):
33  output("------ TESTING sqlite3 ----------")
34  import odb_sqlite3
35  conn = odb_sqlite3.Connection("/tmp/test3.db", autocommit=1)
36  db = odb.Database(conn)
37  return db
38 
39 def TEST_POSTGRES(output=warn):
40  output("------ TESTING postgres ----------")
41  import odb_postgres
42  conn = odb_postgres.Connection(database="test")
43  db = odb.Database(conn)
44  return db
45 
46 def setupDB(dbType="sqlite3"):
47  if dbType == "sqlite3":
48  db = TEST_SQLITE3()
49  elif dbType == "sqlite2":
50  db = TEST_SQLITE()
51  elif dbType == "mysql":
52  db = TEST_MYSQL()
53  elif dbType == "postgres":
54  db = TEST_POSTGRES()
55 
56  db.enabledCompression()
57 
58  db.addTable("agents", "agents", AgentsTable)
59  try: db.agents.dropTable()
60  except: pass
61 
62  db.addTable("roles", "roles", TestTable)
63  try: db.test.dropTable()
64  except: pass
65 
66  db.createTables()
67  db.createIndices()
68 
69  return db
70 
72  def _defineRows(self):
73  self.d_addColumn("agent_id",odb.kInteger,None,primarykey = 1,autoincrement = 1)
74  self.d_addColumn("login",odb.kVarString,200,notnull=1)
75  self.d_addColumn("ext_email",odb.kVarString,200,notnull=1, indexed=1)
76  self.d_addColumn("hashed_pw",odb.kVarString,20,notnull=1)
77  self.d_addColumn("name",odb.kBigString,compress_ok=1)
78  self.d_addColumn("auth_level",odb.kInteger,None)
79  self.d_addColumn("ticket_count",odb.kIncInteger,None)
80  self.d_addColumn("data",odb.kBlob,None, compress_ok=1)
81  self.d_addValueColumn()
82  self.d_addVColumn("columnA",odb.kInteger,None)
83  self.d_addVColumn("columnB",odb.kInteger,None)
84 
85  self.d_hasMany("roles")
86 
88  def _defineRows(self):
89  self.d_addColumn("oid",odb.kInteger,None,primarykey = 1,autoincrement = 1)
90  self.d_addColumn("agent_id",odb.kInteger)
91  self.d_addColumn("a",odb.kInteger)
92  self.d_addColumn("b",odb.kInteger)
93 
94  self.d_belongsTo("agents", foreign_key="agent_id")
95 
96 
97 
98 class Test_Database(unittest.TestCase):
99  def __init__(self, methodName='runTest', db=None):
100  unittest.TestCase.__init__(self, methodName=methodName)
101  self.db = db
102  if self.db is None: self.db = gDB
103 
105 
106  def setUp(self):
107  self.db.agents.deleteAllRows()
108 
109  def test_fetchRow(self):
110 
111  # ---------------------------------------------------------------
112  # make sure we can catch a missing row
113 
114  try:
115  a_row = self.db.agents.fetchRow( ("agent_id", 1000) )
116  self.fail("fetchRow eNoMatchingRows")
117  except odb.eNoMatchingRows:
118  pass
119 
120  def test_rowCreation(self):
121  # --------------------------------------------------------------
122  # create new rows and insert them
123 
124 
125  for n in range(self.TEST_INSERT_COUNT):
126  new_id = n + 1
127 
128  newrow = self.db.agents.newRow()
129  newrow.name = "name #%d" % new_id
130  newrow.login = "name%d" % new_id
131  newrow.ext_email = "%d@name" % new_id
132  newrow.hashed_pw = "hashedpw"
133  newrow.save()
134  msg = "new insert id (%d) does not match expected value (%d)" % (newrow.agent_id,new_id)
135  self.assertEqual(newrow.agent_id,new_id, msg=msg)
136 
137 
138  def test_fetchRow_One(self):
139  # --------------------------------------------------------------
140  # fetch one row
141  row = self.db.agents.newRow()
142  row.name = "name #1"
143  row.login = "name #2"
144  row.ext_email = "name #2"
145  row.hashed_pw = "lsjdf"
146  row.save()
147  a_row = self.db.agents.fetchRow( ("agent_id", 1) )
148 
149  self.assertEqual(a_row.name, "name #1")
150 
151  return a_row
152 
153  def test_save(self):
154  # ---------------------------------------------------------------
155  # don't change and save it
156  # (i.e. the "dummy cursor" string should never be called!)
157  #
158  a_row = self.test_fetchRow_One()
159  try:
160  a_row.save(cursor = "dummy cursor")
161  except AttributeError, reason:
162  self.fail("row tried to access cursor on save() when no changes were made!")
163 
164  def test_save(self):
165  # ---------------------------------------------------------------
166  # change, save, load, test
167 
168  a_row = self.test_fetchRow_One()
169 
170  a_row.auth_level = 10
171  a_row.save()
172  b_row = self.db.agents.fetchRow( ("agent_id", 1) )
173  self.assertEqual(b_row.auth_level, 10, "save and load failed")
174 
175  def misc(self):
176  # ---------------------------------------------------------------
177  # replace
178 
179  if 0:
180  repl_row = self.db.agents.newRow(replace=1)
181  repl_row.agent_id = a_row.agent_id
182  repl_row.login = a_row.login + "-" + a_row.login
183  repl_row.ext_email = "foo"
184  repl_row.hashed_pw = "hashed_pw"
185  repl_row.save()
186 
187  b_row = self.db.agents.fetchRow( ("agent_id", a_row.agent_id) )
188  if b_row.login != repl_row.login:
189  raise "replace failed"
190 
191  # --------------------------------------------------------------
192  # access unknown attribute
194  a_row = self.test_fetchRow_One()
195  try:
196  a = a_row.UNKNOWN_ATTRIBUTE
197  self.fail("unknown attribute")
198  except AttributeError, reason:
199  pass
200  except odb.eNoSuchColumn, reason:
201  pass
202 
204  a_row = self.test_fetchRow_One()
205  try:
206  a_row.UNKNOWN_ATTRIBUTE = 1
207  self.fail("unknown attribute")
208  except AttributeError, reason:
209  pass
210  except odb.eNoSuchColumn, reason:
211  pass
212 
213  # --------------------------------------------------------------
214  # access unknown dict item
215 
217  a_row = self.test_fetchRow_One()
218  try:
219  a = a_row["UNKNOWN_ATTRIBUTE"]
220  self.fail("unknown attribute")
221  except KeyError, reason:
222  pass
223  except odb.eNoSuchColumn, reason:
224  pass
225 
227  a_row = self.test_fetchRow_One()
228  try:
229  a_row["UNKNOWN_ATTRIBUTE"] = 1
230  self.fail("unknown attribute")
231  except KeyError, reason:
232  pass
233  except odb.eNoSuchColumn, reason:
234  pass
235 
236  def misc2(self):
237  # --------------------------------------------------------------
238  # use wrong data for column type
239 
240  if 0:
241  try:
242  a_row.agent_id = "this is a string"
243  raise "test error"
244  except eInvalidData, reason:
245  pass
246 
247  output("PASSED! invalid data for column type")
248 
249  # --------------------------------------------------------------
250  # fetch 1 rows
251 
252  def test_fetchRows(self):
253  arow = self.test_fetchRow_One()
254  rows = self.db.agents.fetchRows( ('agent_id', 1) )
255  self.assertEqual(len(rows), 1)
256 
257  # --------------------------------------------------------------
258  # fetch All rows
259 
260  def test_fetchRows(self):
261  arow = self.test_rowCreation()
262 
263  rows = self.db.agents.fetchAllRows()
264  self.assertEqual(len(rows), self.TEST_INSERT_COUNT)
265 
266 
267  def test_deleteObject(self):
268  # --------------------------------------------------------------
269  # delete row object
270 
271  arow = self.test_fetchRow_One()
272 
273  row = self.db.agents.fetchRow( ('agent_id', 1) )
274  row.delete()
275  try:
276  row = self.db.agents.fetchRow( ('agent_id', 1) )
277  self.fail()
278  except odb.eNoMatchingRows:
279  pass
280 
281  # --------------------------------------------------------------
282  # table deleteRow() call
283  def test_deleteRow(self):
284  self.test_rowCreation()
285 
286  row = self.db.agents.fetchRow( ('agent_id',2) )
287  self.db.agents.deleteRow( ('agent_id', 2) )
288  try:
289  row = self.db.agents.fetchRow( ('agent_id',2) )
290  self.fail()
291  except odb.eNoMatchingRows:
292  pass
293 
294  # --------------------------------------------------------------
295  def test_datasize(self):
296  self.test_rowCreation()
297  row = self.db.agents.fetchRow( ('agent_id',3) )
298  if row.databaseSizeForColumn('name') != len(row.name):
299  self.fail()
300 
301  # --------------------------------------------------------------
302  # table fetchRowUsingPrimaryKey
303  def test_primarykey(self):
304  self.test_rowCreation()
305 
306  row = self.db.agents.fetchRowUsingPrimaryKey(3)
307  if row.agent_id != 3:
308  self.fail()
309 
310  def test_lookup(self):
311  self.test_rowCreation()
312  row = self.db.agents.lookup(agent_id=3)
313  if row.agent_id != 3:
314  self.fail()
315 
316  def test_lookupCreate(self):
317  row = self.db.agents.lookupCreate(agent_id=2)
318  if row.isClean():
319  self.fail()
320  row.login = 1
321  row.ext_email = "hassan@dotfunk.com"
322  row.hashed_pw = "sdfj"
323  row.save()
324 
325  def misc3(self):
326  if 0:
327  # --------------------------------------------------------------
328  # test inc fields
329  row = self.db.agents.newRow()
330  new_id = 1092
331  row.name = "name #%d" % new_id
332  row.login = "name%d" % new_id
333  row.ext_email = "%d@name" % new_id
334  row.inc('ticket_count')
335  row.hashed_pw = "hashed_pw"
336  row.save()
337  new_id = row.agent_id
338 
339  trow = self.db.agents.fetchRow( ('agent_id',new_id) )
340  if trow.ticket_count != 1:
341  raise "ticket_count didn't inc!", repr(trow.ticket_count)
342 
343  row.inc('ticket_count', count=2)
344  row.save()
345  trow = self.db.agents.fetchRow( ('agent_id',new_id) )
346  if trow.ticket_count != 3:
347  raise "ticket_count wrong, expected 3, got %d" % trow.ticket_count
348 
349  trow.inc('ticket_count')
350  trow.save()
351  if trow.ticket_count != 4:
352  raise "ticket_count wrong, expected 4, got %d" % trow.ticket_count
353 
354 
356  self.test_rowCreation()
357 
358  try:
359  b_row = self.db.agents.newRow(replace=1)
360  except odb.eNoMatchingRows:
361  self.fail()
362 
363  b_row.login = "scott"
364  b_row.ext_email = "scott@"
365  b_row.hashed_pw = "lsdjf"
366  b_row.columnA = "hello1"
367  b_row.columnB = "hello2"
368  b_row.save()
369 
370  c_row = self.db.agents.fetchRow( ("agent_id", b_row.agent_id) )
371 
372  self.assertEqual(c_row.columnA, "hello1")
373  self.assertEqual(c_row.columnB, "hello2")
374 
375  def test_compression(self):
376  self.test_rowCreation()
377 
378  try:
379  b_row = self.db.agents.fetchRow( ("agent_id", 5) )
380  except odb.eNoMatchingRows:
381  self.fail()
382 
383  d = "hello\0hello" * 100
384  b_row.data = d
385  b_row.save()
386 
387  b_row = self.db.agents.fetchRow( ("agent_id", 5) )
388  self.assertEqual(b_row.data, d)
389 
390  def test_relations(self):
391  self.test_rowCreation()
392 
393  row = self.db.roles.newRow()
394  row.agent_id = 2
395  row.a = "1"
396  row.b = "2"
397  row.save()
398 
399  row2 = self.db.roles.lookup(oid=row.oid)
400 
401 
402 
403 
404 
405 
406 
407 def test():
408  pass
409 
410 def usage(progname):
411  print __doc__ % vars()
412 
413 def main(argv, stdout, environ):
414  progname = argv[0]
415  optlist, args = getopt.getopt(argv[1:], "", ["help", "test", "debug"])
416 
417  testflag = 0
418 
419  for (field, val) in optlist:
420  if field == "--help":
421  usage(progname)
422  return
423  elif field == "--debug":
424  debugfull()
425  elif field == "--test":
426  testflag = 1
427 
428  if testflag:
429  test()
430  return
431 
432  global gDB
433  gDB = setupDB("sqlite3")
434  unittest.main()
435  return
436 
437  for arg in args:
438  if arg == "sqlite2":
439  TEST_SQLITE()
440  elif arg == "sqlite3":
441  TEST_SQLITE3()
442  elif arg == "mysql":
443  TEST_MYSQL()
444  elif arg == "postgres":
445  TEST_POSTGRES()
446 
447 if __name__ == "__main__":
448  main(sys.argv, sys.stdout, os.environ)
def d_addValueColumn(self)
Definition: odb.py:1160
def TEST_POSTGRES(output=warn)
Definition: test_odb.py:39
def d_hasMany(self, tblname, col_name, foreign_key=None, order=None)
Definition: odb.py:1189
def d_addVColumn(self, col_name, type, size=None, default=None)
Definition: odb.py:1164
def TEST_SQLITE3(output=warn)
Definition: test_odb.py:32
def usage(progname)
Definition: test_odb.py:410
def d_addColumn(self, col_name, ctype, size=None, primarykey=0, notnull=0, indexed=0, default=None, unique=0, autoincrement=0, autoguid=0, safeupdate=0, enum_values=None, no_export=0, relations=None, foreign_key=None, compress_ok=0, int_date=0)
Definition: odb.py:1074
def TEST_SQLITE(output=warn)
Definition: test_odb.py:25
def TEST_MYSQL(output=warn)
T E S T S
Definition: test_odb.py:18
def debugfull()
Definition: log.py:120
if sql[:6] != "select": warn(repr(sql))
Definition: odb_sqlite.py:25
def main(argv, stdout, environ)
Definition: test_odb.py:413
def __init__(self, methodName='runTest', db=None)
Definition: test_odb.py:99
def d_belongsTo(self, col_name, tblNameStr=None, foreign_key=None, order=None)
Definition: odb.py:1184
def setupDB(dbType="sqlite3")
Definition: test_odb.py:46


pyclearsilver
Author(s): Scott Noob Hassan
autogenerated on Mon Jun 10 2019 15:51:13