1170 self.assertEqual(len(rset.rows), 0, rset.rows) |
1170 self.assertEqual(len(rset.rows), 0, rset.rows) |
1171 |
1171 |
1172 def test_delete_3(self): |
1172 def test_delete_3(self): |
1173 s = self.user_groups_session('users') |
1173 s = self.user_groups_session('users') |
1174 with s.new_cnx() as cnx: |
1174 with s.new_cnx() as cnx: |
1175 with cnx.ensure_cnx_set: |
1175 peid, = self.o.execute(cnx, "INSERT Personne P: P nom 'toto'")[0] |
1176 peid, = self.o.execute(cnx, "INSERT Personne P: P nom 'toto'")[0] |
1176 seid, = self.o.execute(cnx, "INSERT Societe S: S nom 'logilab'")[0] |
1177 seid, = self.o.execute(cnx, "INSERT Societe S: S nom 'logilab'")[0] |
1177 self.o.execute(cnx, "SET P travaille S") |
1178 self.o.execute(cnx, "SET P travaille S") |
1178 cnx.commit() |
1179 cnx.commit() |
|
1180 rset = self.qexecute('Personne P WHERE P travaille S') |
1179 rset = self.qexecute('Personne P WHERE P travaille S') |
1181 self.assertEqual(len(rset.rows), 1) |
1180 self.assertEqual(len(rset.rows), 1) |
1182 self.qexecute("DELETE X travaille Y WHERE X eid %s, Y eid %s" % (peid, seid)) |
1181 self.qexecute("DELETE X travaille Y WHERE X eid %s, Y eid %s" % (peid, seid)) |
1183 rset = self.qexecute('Personne P WHERE P travaille S') |
1182 rset = self.qexecute('Personne P WHERE P travaille S') |
1184 self.assertEqual(len(rset.rows), 0) |
1183 self.assertEqual(len(rset.rows), 0) |
1209 #'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y' |
1208 #'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y' |
1210 eeid, = self.qexecute('INSERT Email X: X messageid "<1234>", X subject "test", ' |
1209 eeid, = self.qexecute('INSERT Email X: X messageid "<1234>", X subject "test", ' |
1211 'X sender Y, X recipients Y WHERE Y is EmailAddress')[0] |
1210 'X sender Y, X recipients Y WHERE Y is EmailAddress')[0] |
1212 self.qexecute("DELETE Email X") |
1211 self.qexecute("DELETE Email X") |
1213 with self.session.new_cnx() as cnx: |
1212 with self.session.new_cnx() as cnx: |
1214 with cnx.ensure_cnx_set: |
1213 sqlc = cnx.cnxset.cu |
1215 sqlc = cnx.cnxset.cu |
1214 sqlc.execute('SELECT * FROM recipients_relation') |
1216 sqlc.execute('SELECT * FROM recipients_relation') |
1215 self.assertEqual(len(sqlc.fetchall()), 0) |
1217 self.assertEqual(len(sqlc.fetchall()), 0) |
1216 sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid) |
1218 sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid) |
1217 self.assertEqual(len(sqlc.fetchall()), 0) |
1219 self.assertEqual(len(sqlc.fetchall()), 0) |
|
1220 |
1218 |
1221 def test_nonregr_delete_cache2(self): |
1219 def test_nonregr_delete_cache2(self): |
1222 eid = self.qexecute("INSERT Folder T: T name 'toto'")[0][0] |
1220 eid = self.qexecute("INSERT Folder T: T name 'toto'")[0][0] |
1223 # fill the cache |
1221 # fill the cache |
1224 self.qexecute("Any X WHERE X eid %(x)s", {'x': eid}) |
1222 self.qexecute("Any X WHERE X eid %(x)s", {'x': eid}) |
1361 self.assertEqual(len(rset.rows), 1) |
1359 self.assertEqual(len(rset.rows), 1) |
1362 self.assertEqual(rset.description, [('CWUser',)]) |
1360 self.assertEqual(rset.description, [('CWUser',)]) |
1363 self.assertRaises(Unauthorized, |
1361 self.assertRaises(Unauthorized, |
1364 self.qexecute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1362 self.qexecute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1365 with self.session.new_cnx() as cnx: |
1363 with self.session.new_cnx() as cnx: |
1366 with cnx.ensure_cnx_set: |
1364 cursor = cnx.cnxset.cu |
1367 cursor = cnx.cnxset.cu |
1365 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1368 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1366 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1369 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1367 passwd = str(cursor.fetchone()[0]) |
1370 passwd = str(cursor.fetchone()[0]) |
1368 self.assertEqual(passwd, crypt_password('toto', passwd)) |
1371 self.assertEqual(passwd, crypt_password('toto', passwd)) |
|
1372 rset = self.qexecute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1369 rset = self.qexecute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1373 {'pwd': Binary(passwd)}) |
1370 {'pwd': Binary(passwd)}) |
1374 self.assertEqual(len(rset.rows), 1) |
1371 self.assertEqual(len(rset.rows), 1) |
1375 self.assertEqual(rset.description, [('CWUser',)]) |
1372 self.assertEqual(rset.description, [('CWUser',)]) |
1376 |
1373 |
1377 def test_update_upassword(self): |
1374 def test_update_upassword(self): |
1378 with self.session.new_cnx() as cnx: |
1375 with self.session.new_cnx() as cnx: |
1379 with cnx.ensure_cnx_set: |
1376 rset = cnx.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", |
1380 rset = cnx.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", |
1377 {'pwd': 'toto'}) |
1381 {'pwd': 'toto'}) |
1378 self.assertEqual(rset.description[0][0], 'CWUser') |
1382 self.assertEqual(rset.description[0][0], 'CWUser') |
1379 rset = cnx.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1383 rset = cnx.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1380 {'pwd': 'tutu'}) |
1384 {'pwd': 'tutu'}) |
1381 cursor = cnx.cnxset.cu |
1385 cursor = cnx.cnxset.cu |
1382 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1386 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1383 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1387 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1384 passwd = str(cursor.fetchone()[0]) |
1388 passwd = str(cursor.fetchone()[0]) |
1385 self.assertEqual(passwd, crypt_password('tutu', passwd)) |
1389 self.assertEqual(passwd, crypt_password('tutu', passwd)) |
1386 rset = cnx.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1390 rset = cnx.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1387 {'pwd': Binary(passwd)}) |
1391 {'pwd': Binary(passwd)}) |
1388 self.assertEqual(len(rset.rows), 1) |
1392 self.assertEqual(len(rset.rows), 1) |
1389 self.assertEqual(rset.description, [('CWUser',)]) |
1393 self.assertEqual(rset.description, [('CWUser',)]) |
|
1394 |
1390 |
1395 # ZT datetime tests ######################################################## |
1391 # ZT datetime tests ######################################################## |
1396 |
1392 |
1397 def test_tz_datetime(self): |
1393 def test_tz_datetime(self): |
1398 self.qexecute("INSERT Personne X: X nom 'bob', X tzdatenaiss %(date)s", |
1394 self.qexecute("INSERT Personne X: X nom 'bob', X tzdatenaiss %(date)s", |