1 import sys, os
2 import logging
3 import MySQLdb
4 import MySQLdb.cursors
5 import logging
6
8 """
9 Helper class for parsing external data for data driven testing
10 """
11
12
13 db = {'db':None,
14 'host':None,
15 'user':None,
16 'password':None,
17 'database':None,
18 'cursor':None,
19 'results':None,
20 'query':None}
21
24
25
27 try:
28
29 self.db['db'] = MySQLdb.connect(host=self.db['host'],
30 user=self.db['user'],
31 passwd=self.db['password'],
32 db=self.db['database'],
33 charset='utf8',
34 cursorclass = MySQLdb.cursors.Cursor)
35 self.db['cursor'] = self.db['db'].cursor()
36
37 except MySQLdb.Error, e:
38 print "Mysql Error %d: %s" % (e.args[0],e.args[1])
39
41 """
42 Closes the connetion
43 """
44 try:
45 self.db['db'].close()
46 except MySQLdb.Error, e:
47 print "Mysql Error %d: %s" % (e.args[0],e.args[1])
48
50 """
51 Retrieves all entries from db
52 """
53 if limits is not None:
54 sql = """SELECT * FROM """+str(table)+""" limit"""+str(limits)
55 else:
56 sql = """SELECT * FROM """+str(table)
57 records = []
58
59 try:
60 self.cursor.execute(sql)
61 except MySQLdb.Error, e:
62 print "Mysql Error %d: %s" % (e.args[0],e.args[1])
63
64 numrows = int(self.cursor.rowcount)
65
66 if numrows >= 1:
67 records = self.cursor.fetchall()
68 else:
69 print "No test data was returned. Maybe data for tests is empty?"
70
71
72 columnNames = self.getColumnNames(table)
73
74 testData = []
75
76 for record in records:
77 rowNamesAndData = {}
78 for i, column in enumerate(columnNames):
79 rowNamesAndData[column] = record[i]
80 testData.append(rowNamesAndData)
81
82 return testData
83
84
85
86
87
88
89
90
91 dbInj = DataInjector()
92
93 dbInj.db = {'db':None,
94 'host':'127.0.0.1',
95 'user':'yid_user',
96 'password':'y3ll0w1d',
97 'database':'yid',
98 'cursor':None,
99 'results':None,
100 'query':None}
101 dbInj.connect()
102