root/trunk/radiomate/daotest.py @ 6

Revision 6, 9.4 KB (checked in by clauz, 4 years ago)

beginning to work on the JSON interface

Line 
1# Test of the dao module
2
3import json
4import mysqldao as dao
5from dao import *
6import sys
7
8# connection parameters
9DBHOST="127.0.0.1"
10DBUSER="mate"
11DBPASSWORD="radi0"
12DATABASE="radiomate0"
13
14def test0():
15        try:
16                conn = MySQLdb.connect( 
17                                host = DBHOST, 
18                                user = DBUSER,
19                                passwd = DBPASSWORD,
20                                db = DATABASE
21                                )
22                cursor = conn.cursor()
23                cursor.execute("""INSERT IGNORE INTO roles (
24                        rolename,
25                        canManageRoles,
26                        canManageUsers,
27                        canManageAllPlaylists,
28                        canRegisterFiles,
29                        canManageRegisteredFiles,
30                        canSearchRegisteredFiles,
31                        canManageTimetable,
32                        fixedSlotTime,
33                        changeTimeBeforeTransmission,
34                        canCreateTestMountpoint,
35                        canListNetcasts,
36                        fixedSlotTimesList
37                        ) VALUES (
38                        'admin',
39                        1, 1, 1, 1, 1, 1, 1, 1, 120, 1, 1, '60,120')""")
40                conn.commit()
41                print "Number of rows inserted: %d" % cursor.rowcount
42                cursor.close()
43                cursor = conn.cursor(MySQLdb.cursors.DictCursor)
44                cursor.execute("SELECT * from roles")
45                rows = cursor.fetchall()
46                print len(rows), "rows fetched"
47                for row in rows:
48                        print ">", row
49                        r = Role(row)
50                        print "<", r
51                        print "-->", r.rolename
52                cursor.close()
53                conn.close()
54                       
55        except MySQLdb.Error, e:
56                print "Error %d: %s" % (e.args[0], e.args[1])
57                sys.exit(1)
58
59        # JSON
60        jsonrequest = """
61        {
62            "request": "createrole",
63            "username": "foobar",
64            "password": "secret",
65            "role" : {
66                "rolename": "testrole",
67                "canManageRoles": false,
68                "canManageUsers": false,
69                "canManageAllPlaylists": false,
70                "canRegisterFiles": false,
71                "canManageRegisteredFiles": false,
72                "canSearchRegisteredFiles": false,
73                "canManageTimetable": false,
74                "fixedSlotTime": false,
75                "fixedSlotTimesList": "45,90",
76                "changeTimeBeforeTransmission": 60,
77                "canCreateTestMountpoint": false,
78                "canListNetcasts": false
79            }
80        }
81        """
82
83        try:
84                d = json.loads(jsonrequest)
85        except ValueError, e:
86                raise SystemExit("JSON Validation error: %s" % (e))
87
88        print "-----"
89        print d['role']
90        print "-----"
91        r = Role(d['role'])
92        print r
93        print "-----"
94        print "-->", r.rolename
95
96        print "------>"
97        print json.dumps(r.dictexport(), skipkeys=True)
98
99        print "-----"
100        r = Role()
101        print r
102        print "-->", r.rolename
103        r.rolename = "scemodelvillaggio3"
104        print "-->", r.rolename
105        print "-----"
106
107        #DAO
108        conn = MySQLdb.connect( 
109                        host = DBHOST, 
110                        user = DBUSER,
111                        passwd = DBPASSWORD,
112                        db = DATABASE
113                        )
114        roledao = RoleMysqlDAO(conn)
115        roledao.insert(r)
116        r.rolename = "scemodelvillaggio"
117        roledao.insert(r)
118        r.rolename = "scemodelvillaggio2"
119        roledao.insert(r)
120        r.rolename = "scemodelvillaggio3"
121        roledao.insert(r)
122
123        r = roledao.getByName("admin")
124        print r
125        r = roledao.getByName("scemodelvillaggio")
126        print r
127        r = roledao.getByName("scemodelvillaggio2")
128        print r
129        r = roledao.getByName("scemodelvillaggio3")
130        print r
131        print r.fixedSlotTimesList, r.fixedSlotTimesListList()
132        roledao.removeByName("scemodelvillaggio3")
133        rl = roledao.getAll()
134        for r in rl:
135                print r
136
137        print "+++++++++++++++++++++++++++++++++++++"
138        u = User()
139        u.name = "foobar3"
140        u.password = "segreta"
141        u.displayname = "Mr. Pippo 3"
142        u.rolename = "scemodelvillaggio2"
143        print u
144        udao = UserMysqlDAO(conn)
145        udao.insert(u)
146        ulist = udao.getAll()
147        for u in ulist:
148                print u
149        udao.removeByName(u.name)
150
151def roledaotest():
152        r = Role()
153        r.rolename = "testrole"
154        conn = MySQLdb.connect( 
155                        host = DBHOST, 
156                        user = DBUSER,
157                        passwd = DBPASSWORD,
158                        db = DATABASE
159                        )
160        roledao = RoleMysqlDAO(conn)
161        roledao.insert(r)
162        r1 = roledao.getByName("testrole")
163        print r1
164        rall = roledao.getAll()
165        print rall
166        r1.canManageRoles = True
167        roledao.update(r1)
168        r2 = roledao.getByName("testrole")
169        print r2
170        roledao.removeByName("testrole")
171        conn.close()
172
173def userdaotest():
174        r = Role()
175        r.rolename = "testrole"
176        conn = MySQLdb.connect( 
177                        host = DBHOST, 
178                        user = DBUSER,
179                        passwd = DBPASSWORD,
180                        db = DATABASE
181                        )
182        roledao = RoleMysqlDAO(conn)
183        roledao.insert(r)
184        u = User()
185        u.name = "testuser"
186        u.displayname = "Test User"
187        u.rolename = r.rolename
188        userdao = UserMysqlDAO(conn)
189        userdao.insert(u)
190        u1 = userdao.getByName(u.name)
191        print u1
192        u1.displayname = "Modified Test User"
193        userdao.update(u1)
194        u2 = userdao.getByName(u1.name)
195        print u2
196        res = userdao.getAll()
197        print res
198        userdao.removeByName(u.name)
199        roledao.removeByName(r.rolename)
200
201def mediafilesdaotest():
202        r = Role()
203        r.rolename = "testrole"
204        u = User()
205        u.name = "foobar"
206        u.rolename = r.rolename
207
208        mf = MediaFile()
209        mf.user = u.name
210        mf.path = "/tmp/ulululu.mp3"
211        mf.type = 'audio'
212        mf.title = "Ulululu"
213        mf.author = "Shakerobba"
214        mf.album = "Steam this album"
215        mf.genre = "steam pop"
216        mf.year = 1932
217        mf.comment = "very good music"
218        mf.license = "CC"
219        mf.tags = "shakerobba,steam,wolves"
220        conn = MySQLdb.connect( 
221                        host = DBHOST, 
222                        user = DBUSER,
223                        passwd = DBPASSWORD,
224                        db = DATABASE
225                        )
226        rdao = RoleMysqlDAO(conn)
227        rdao.insert(r)
228
229        udao = UserMysqlDAO(conn)
230        udao.insert(u)
231        mediafiledao = MediaFileMysqlDAO(conn)
232        id = mediafiledao.insert(mf)
233        mf2 = mediafiledao.getById(id)
234        print mf2
235        mf2.license = "AGPL"
236        mediafiledao.update(mf2)
237        mf3 = mediafiledao.getById(id)
238        print mf3
239        mediafiledao.removeById(id)
240        udao.removeByName(u.name)
241        rdao.removeByName(r.rolename)
242
243def playlistdaotest():
244        conn = MySQLdb.connect( 
245                        host = DBHOST, 
246                        user = DBUSER,
247                        passwd = DBPASSWORD,
248                        db = DATABASE
249                        )
250        r = Role()
251        r.rolename = "testrole"
252        rdao = RoleMysqlDAO(conn)
253        rdao.insert(r)
254        u1 = User()
255        u1.name = "foobar"
256        u1.rolename = r.rolename
257        udao = UserMysqlDAO(conn)
258        udao.insert(u1)
259        u2 = User()
260        u2.name = "foobar2"
261        u2.rolename = r.rolename
262        udao.insert(u2)
263
264        mf = MediaFile()
265        mf.user = "foobar2"
266        mf.path = "/tmp/ulululu.mp3"
267        mf.type = 'audio'
268        mf.title = "Ulululu"
269        mf.author = "Shakerobba"
270        mf.album = "Steam this album"
271        mf.genre = "steam pop"
272        mf.year = 1932
273        mf.comment = "very good music"
274        mf.license = "CC"
275        mf.tags = "shakerobba,steam,wolves"
276
277        mf2 = MediaFile()
278        mf2.user = "foobar2"
279        mf2.path = "/tmp/ulululu2.mp3"
280        mf2.type = 'audio'
281        mf2.title = "Ulululu2"
282        mf2.author = "Shakerobba"
283        mf2.album = "Steam this album"
284        mf2.genre = "steam pop"
285        mf2.year = 1932
286        mf2.comment = "very good music"
287        mf2.license = "CC"
288        mf2.tags = "shakerobba,steam,wolves,sequel"
289       
290        mediafiledao = MediaFileMysqlDAO(conn)
291        id1 = mediafiledao.insert(mf)
292        id2 = mediafiledao.insert(mf2)
293        mf = mediafiledao.getById(id1)
294        mf2 = mediafiledao.getById(id2)
295
296        pl = PlayList()
297        pl.title = "just a test"
298        pl.creator = "foobar2"
299        pl.fallback = False
300        pl.description = "testing dao"
301        pl.comment = ":)"
302        pl.tags = "test,prova,foo,bar"
303
304        pl.addMediaFile(mf)
305        pl.addMediaFile(mf2)
306       
307        pl.addOwner("foobar")
308        pl.addViewer("foobar2")
309
310        print pl
311
312        playlistdao = PlayListMysqlDAO(conn)
313        plid = playlistdao.insert(pl)
314        print plid
315        pl2 = playlistdao.getById(plid)
316        print pl2
317
318        pl2.description = "testing dao 2"
319        playlistdao.update(pl2)
320        pl3 = playlistdao.getById(pl2.id)
321        print pl3
322
323        res = playlistdao.getByUser("foobar")
324        print res
325
326        res = playlistdao.getByCreator("foobar2")
327        print res
328
329        mediafiledao.removeById(id1)
330        mediafiledao.removeById(id2)
331        playlistdao.removeById(plid)
332        udao.removeByName(u1.name)
333        udao.removeByName(u2.name)
334        rdao.removeByName(r.rolename)
335        conn.close()
336
337def timeslotdaotest():
338        conn = MySQLdb.connect( 
339                        host = DBHOST, 
340                        user = DBUSER,
341                        passwd = DBPASSWORD,
342                        db = DATABASE
343                        )
344        r = Role()
345        r.rolename = "testrole"
346        rdao = RoleMysqlDAO(conn)
347        rdao.insert(r)
348        udao = UserMysqlDAO(conn)
349        u1 = User()
350        u1.name = "foobar"
351        u1.rolename = "testrole"
352        udao.insert(u1)
353        u2 = User()
354        u2.name = "foobar2"
355        u2.rolename = "testrole"
356        udao.insert(u2)
357
358        ts = TimeSlot()
359        ts.creator = 'foobar'
360        ts.slottype = 'dummy'
361        ts.slotparams = {'par1': 0, 'par2': 'ciao'}
362        ts.beginningyear = 2010
363        ts.beginningmonth = 07
364        ts.beginningday = 1
365        ts.beginninghour = 10
366        ts.beginningminute = 0
367        ts.duration = 60
368        ts.title = "The Test Show"
369        ts.description = "the best test show ever"
370        ts.comment = "just a test"
371        ts.tags = "test, show, radio"
372       
373        pl = PlayList()
374        pl.title = "just a test"
375        pl.creator = "foobar2"
376        pl.fallback = False
377        pl.description = "testing dao"
378        pl.comment = ":)"
379        pl.tags = "test,prova,foo,bar"
380        playlistdao = PlayListMysqlDAO(conn)
381        plid = playlistdao.insert(pl)
382
383        ts.fallbackplaylist = plid
384
385        tsdao = TimeSlotMysqlDAO(conn)
386        tsid = tsdao.insert(ts)
387
388        ts1 = tsdao.getById(tsid)
389        ts1.description = "testing dao"
390        tsdao.update(ts1)
391
392        ts2 = tsdao.getById(tsid)
393        print ts2
394
395        res = tsdao.getFromTo("2010-07-01 9:00", "2010-07-01 12:00")
396        print res
397
398        tss = TimeSlot()
399        tss.title = "show"
400        res = tsdao.search(tss)
401        print res
402       
403        tsdao.removeById(tsid)
404        playlistdao.removeById(plid)
405        udao.removeByName(u1.name)
406        udao.removeByName(u2.name)
407        rdao.removeByName(r.rolename)
408        conn.close()
409
410def exceptiontest():
411        conn = MySQLdb.connect( 
412                        host = DBHOST, 
413                        user = DBUSER,
414                        passwd = DBPASSWORD,
415                        db = DATABASE
416                        )
417        try:
418                selstring = "SELECT dummyfield FROM roles"
419                cursor = conn.cursor(MySQLdb.cursors.DictCursor)
420                cursor.execute(selstring)
421                cursor.close()
422        except MySQLdb.Error, e:
423                raise RadioMateDAOException(e.args)
424       
425        conn.close()
426
427
428#execute
429print "---"
430#test0()
431#roledaotest()
432#userdaotest()
433#mediafilesdaotest()
434#playlistdaotest()
435timeslotdaotest()
436#exceptiontest()
437
Note: See TracBrowser for help on using the browser.