54 pool = self.repo._get_pool() |
54 pool = self.repo._get_pool() |
55 table = SQL_PREFIX + 'CWEType' |
55 table = SQL_PREFIX + 'CWEType' |
56 namecol = SQL_PREFIX + 'name' |
56 namecol = SQL_PREFIX + 'name' |
57 finalcol = SQL_PREFIX + 'final' |
57 finalcol = SQL_PREFIX + 'final' |
58 try: |
58 try: |
59 sqlcursor = pool['system'] |
59 cu = self.session.system_sql('SELECT %s FROM %s WHERE %s is NULL' % ( |
60 sqlcursor.execute('SELECT %s FROM %s WHERE %s is NULL' % ( |
|
61 namecol, table, finalcol)) |
60 namecol, table, finalcol)) |
62 self.assertEquals(sqlcursor.fetchall(), []) |
61 self.assertEquals(cu.fetchall(), []) |
63 sqlcursor.execute('SELECT %s FROM %s WHERE %s=%%(final)s ORDER BY %s' |
62 cu = self.session.system_sql('SELECT %s FROM %s WHERE %s=%%(final)s ORDER BY %s' |
64 % (namecol, table, finalcol, namecol), {'final': 'TRUE'}) |
63 % (namecol, table, finalcol, namecol), {'final': 'TRUE'}) |
65 self.assertEquals(sqlcursor.fetchall(), [(u'Boolean',), (u'Bytes',), |
64 self.assertEquals(cu.fetchall(), [(u'Boolean',), (u'Bytes',), |
66 (u'Date',), (u'Datetime',), |
65 (u'Date',), (u'Datetime',), |
67 (u'Decimal',),(u'Float',), |
66 (u'Decimal',),(u'Float',), |
68 (u'Int',), |
67 (u'Int',), |
69 (u'Interval',), (u'Password',), |
68 (u'Interval',), (u'Password',), |
70 (u'String',), (u'Time',)]) |
69 (u'String',), (u'Time',)]) |
356 def test_add_delete_info(self): |
355 def test_add_delete_info(self): |
357 entity = self.repo.vreg.etype_class('Personne')(self.session, None, None) |
356 entity = self.repo.vreg.etype_class('Personne')(self.session, None, None) |
358 entity.eid = -1 |
357 entity.eid = -1 |
359 entity.complete = lambda x: None |
358 entity.complete = lambda x: None |
360 self.repo.add_info(self.session, entity, self.repo.sources_by_uri['system']) |
359 self.repo.add_info(self.session, entity, self.repo.sources_by_uri['system']) |
361 cursor = self.session.pool['system'] |
360 cu = self.session.system_sql('SELECT * FROM entities WHERE eid = -1') |
362 cursor.execute('SELECT * FROM entities WHERE eid = -1') |
361 data = cu.fetchall() |
363 data = cursor.fetchall() |
|
364 self.assertIsInstance(data[0][3], datetime) |
362 self.assertIsInstance(data[0][3], datetime) |
365 data[0] = list(data[0]) |
363 data[0] = list(data[0]) |
366 data[0][3] = None |
364 data[0][3] = None |
367 self.assertEquals(tuplify(data), [(-1, 'Personne', 'system', None, None)]) |
365 self.assertEquals(tuplify(data), [(-1, 'Personne', 'system', None, None)]) |
368 self.repo.delete_info(self.session, -1) |
366 self.repo.delete_info(self.session, -1) |
369 #self.repo.commit() |
367 #self.repo.commit() |
370 cursor.execute('SELECT * FROM entities WHERE eid = -1') |
368 cu = self.session.system_sql('SELECT * FROM entities WHERE eid = -1') |
371 data = cursor.fetchall() |
369 data = cu.fetchall() |
372 self.assertEquals(data, []) |
370 self.assertEquals(data, []) |
373 |
371 |
374 |
372 |
375 class FTITC(RepositoryBasedTC): |
373 class FTITC(RepositoryBasedTC): |
376 |
374 |
377 def test_reindex_and_modified_since(self): |
375 def test_reindex_and_modified_since(self): |
378 cursor = self.session.pool['system'] |
|
379 eidp = self.execute('INSERT Personne X: X nom "toto", X prenom "tutu"')[0][0] |
376 eidp = self.execute('INSERT Personne X: X nom "toto", X prenom "tutu"')[0][0] |
380 self.commit() |
377 self.commit() |
381 ts = datetime.now() |
378 ts = datetime.now() |
382 self.assertEquals(len(self.execute('Personne X WHERE X has_text "tutu"')), 1) |
379 self.assertEquals(len(self.execute('Personne X WHERE X has_text "tutu"')), 1) |
383 cursor.execute('SELECT mtime, eid FROM entities WHERE eid = %s' % eidp) |
380 cu = self.session.system_sql('SELECT mtime, eid FROM entities WHERE eid = %s' % eidp) |
384 omtime = cursor.fetchone()[0] |
381 omtime = cu.fetchone()[0] |
385 # our sqlite datetime adapter is ignore seconds fraction, so we have to |
382 # our sqlite datetime adapter is ignore seconds fraction, so we have to |
386 # ensure update is done the next seconds |
383 # ensure update is done the next seconds |
387 time.sleep(1 - (ts.second - int(ts.second))) |
384 time.sleep(1 - (ts.second - int(ts.second))) |
388 self.execute('SET X nom "tata" WHERE X eid %(x)s', {'x': eidp}, 'x') |
385 self.execute('SET X nom "tata" WHERE X eid %(x)s', {'x': eidp}, 'x') |
389 self.commit() |
386 self.commit() |
390 self.assertEquals(len(self.execute('Personne X WHERE X has_text "tutu"')), 1) |
387 self.assertEquals(len(self.execute('Personne X WHERE X has_text "tutu"')), 1) |
391 cursor.execute('SELECT mtime FROM entities WHERE eid = %s' % eidp) |
388 cu = self.session.system_sql('SELECT mtime FROM entities WHERE eid = %s' % eidp) |
392 mtime = cursor.fetchone()[0] |
389 mtime = cu.fetchone()[0] |
393 self.failUnless(omtime < mtime) |
390 self.failUnless(omtime < mtime) |
394 self.commit() |
391 self.commit() |
395 date, modified, deleted = self.repo.entities_modified_since(('Personne',), omtime) |
392 date, modified, deleted = self.repo.entities_modified_since(('Personne',), omtime) |
396 self.assertEquals(modified, [('Personne', eidp)]) |
393 self.assertEquals(modified, [('Personne', eidp)]) |
397 self.assertEquals(deleted, []) |
394 self.assertEquals(deleted, []) |