195 (u'Tag', {u'linkattr': u'name'})], |
195 (u'Tag', {u'linkattr': u'name'})], |
196 }, |
196 }, |
197 }) |
197 }) |
198 session = self.repo.internal_session(safe=True) |
198 session = self.repo.internal_session(safe=True) |
199 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
199 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
200 self.assertEqual(sorted(stats.keys()), ['created', 'updated']) |
200 self.assertEqual(sorted(stats.keys()), ['checked', 'created', 'updated']) |
201 self.assertEqual(len(stats['created']), 2) |
201 self.assertEqual(len(stats['created']), 2) |
202 self.assertEqual(stats['updated'], set()) |
202 self.assertEqual(stats['updated'], set()) |
203 |
203 |
204 user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) |
204 user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) |
205 self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) |
205 self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) |
231 ) |
231 ) |
232 session.set_cnxset() |
232 session.set_cnxset() |
233 with session.security_enabled(read=False): # avoid Unauthorized due to password selection |
233 with session.security_enabled(read=False): # avoid Unauthorized due to password selection |
234 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
234 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
235 self.assertEqual(stats['created'], set()) |
235 self.assertEqual(stats['created'], set()) |
236 self.assertEqual(len(stats['updated']), 2) |
236 self.assertEqual(len(stats['updated']), 0) |
|
237 self.assertEqual(len(stats['checked']), 2) |
237 self.repo._type_source_cache.clear() |
238 self.repo._type_source_cache.clear() |
238 self.repo._extid_cache.clear() |
239 self.repo._extid_cache.clear() |
239 session.set_cnxset() |
240 session.set_cnxset() |
240 with session.security_enabled(read=False): # avoid Unauthorized due to password selection |
241 with session.security_enabled(read=False): # avoid Unauthorized due to password selection |
241 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
242 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
242 self.assertEqual(stats['created'], set()) |
243 self.assertEqual(stats['created'], set()) |
243 self.assertEqual(len(stats['updated']), 2) |
244 self.assertEqual(len(stats['updated']), 0) |
|
245 self.assertEqual(len(stats['checked']), 2) |
244 session.commit() |
246 session.commit() |
245 |
247 |
246 # test move to system source |
248 # test move to system source |
247 self.sexecute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid}) |
249 self.sexecute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid}) |
248 self.commit() |
250 self.commit() |