201 def test_internal_api(self): |
201 def test_internal_api(self): |
202 repo = self.repo |
202 repo = self.repo |
203 cnxid = repo.connect(self.admlogin, password=self.admpassword) |
203 cnxid = repo.connect(self.admlogin, password=self.admpassword) |
204 session = repo._get_session(cnxid) |
204 session = repo._get_session(cnxid) |
205 with session.new_cnx() as cnx: |
205 with session.new_cnx() as cnx: |
206 with cnx.ensure_cnx_set: |
206 self.assertEqual(repo.type_and_source_from_eid(2, cnx), |
207 self.assertEqual(repo.type_and_source_from_eid(2, cnx), |
207 ('CWGroup', None, 'system')) |
208 ('CWGroup', None, 'system')) |
208 self.assertEqual(repo.type_from_eid(2, cnx), 'CWGroup') |
209 self.assertEqual(repo.type_from_eid(2, cnx), 'CWGroup') |
|
210 repo.close(cnxid) |
209 repo.close(cnxid) |
211 |
210 |
212 def test_public_api(self): |
211 def test_public_api(self): |
213 self.assertEqual(self.repo.get_schema(), self.repo.schema) |
212 self.assertEqual(self.repo.get_schema(), self.repo.schema) |
214 self.assertEqual(self.repo.source_defs(), {'system': {'type': 'native', |
213 self.assertEqual(self.repo.source_defs(), {'system': {'type': 'native', |
383 self.repo.set_schema(self.repo.deserialize_schema()) |
382 self.repo.set_schema(self.repo.deserialize_schema()) |
384 table = SQL_PREFIX + 'CWEType' |
383 table = SQL_PREFIX + 'CWEType' |
385 namecol = SQL_PREFIX + 'name' |
384 namecol = SQL_PREFIX + 'name' |
386 finalcol = SQL_PREFIX + 'final' |
385 finalcol = SQL_PREFIX + 'final' |
387 with self.admin_access.repo_cnx() as cnx: |
386 with self.admin_access.repo_cnx() as cnx: |
388 with cnx.ensure_cnx_set: |
387 cu = cnx.system_sql('SELECT %s FROM %s WHERE %s is NULL' |
389 cu = cnx.system_sql('SELECT %s FROM %s WHERE %s is NULL' |
388 % (namecol, table, finalcol)) |
390 % (namecol, table, finalcol)) |
389 self.assertEqual(cu.fetchall(), []) |
391 self.assertEqual(cu.fetchall(), []) |
390 cu = cnx.system_sql('SELECT %s FROM %s ' |
392 cu = cnx.system_sql('SELECT %s FROM %s ' |
391 'WHERE %s=%%(final)s ORDER BY %s' |
393 'WHERE %s=%%(final)s ORDER BY %s' |
392 % (namecol, table, finalcol, namecol), |
394 % (namecol, table, finalcol, namecol), |
393 {'final': True}) |
395 {'final': True}) |
394 self.assertEqual(cu.fetchall(), |
396 self.assertEqual(cu.fetchall(), |
395 [(u'BabarTestType',), |
397 [(u'BabarTestType',), |
396 (u'BigInt',), (u'Boolean',), (u'Bytes',), |
398 (u'BigInt',), (u'Boolean',), (u'Bytes',), |
397 (u'Date',), (u'Datetime',), |
399 (u'Date',), (u'Datetime',), |
398 (u'Decimal',),(u'Float',), |
400 (u'Decimal',),(u'Float',), |
399 (u'Int',), |
401 (u'Int',), |
400 (u'Interval',), (u'Password',), |
402 (u'Interval',), (u'Password',), |
401 (u'String',), |
403 (u'String',), |
402 (u'TZDatetime',), (u'TZTime',), (u'Time',)]) |
404 (u'TZDatetime',), (u'TZTime',), (u'Time',)]) |
403 sql = ("SELECT etype.cw_eid, etype.cw_name, cstr.cw_eid, rel.eid_to " |
405 sql = ("SELECT etype.cw_eid, etype.cw_name, cstr.cw_eid, rel.eid_to " |
404 "FROM cw_CWUniqueTogetherConstraint as cstr, " |
406 "FROM cw_CWUniqueTogetherConstraint as cstr, " |
405 " relations_relation as rel, " |
407 " relations_relation as rel, " |
406 " cw_CWEType as etype " |
408 " cw_CWEType as etype " |
407 "WHERE cstr.cw_eid = rel.eid_from " |
409 "WHERE cstr.cw_eid = rel.eid_from " |
408 " AND cstr.cw_constraint_of = etype.cw_eid " |
410 " AND cstr.cw_constraint_of = etype.cw_eid " |
409 " AND etype.cw_name = 'Personne' " |
411 " AND etype.cw_name = 'Personne' " |
410 ";") |
412 ";") |
411 cu = cnx.system_sql(sql) |
413 cu = cnx.system_sql(sql) |
412 rows = cu.fetchall() |
414 rows = cu.fetchall() |
413 self.assertEqual(len(rows), 3) |
415 self.assertEqual(len(rows), 3) |
414 person = self.repo.schema.eschema('Personne') |
416 person = self.repo.schema.eschema('Personne') |
415 self.assertEqual(len(person._unique_together), 1) |
417 self.assertEqual(len(person._unique_together), 1) |
416 self.assertItemsEqual(person._unique_together[0], |
418 self.assertItemsEqual(person._unique_together[0], |
417 ('nom', 'prenom', 'inline2')) |
419 ('nom', 'prenom', 'inline2')) |
|
420 |
418 |
421 finally: |
419 finally: |
422 self.repo.set_schema(origshema) |
420 self.repo.set_schema(origshema) |
423 |
421 |
424 def test_custom_attribute_param(self): |
422 def test_custom_attribute_param(self): |
437 |
435 |
438 class DataHelpersTC(CubicWebTC): |
436 class DataHelpersTC(CubicWebTC): |
439 |
437 |
440 def test_type_from_eid(self): |
438 def test_type_from_eid(self): |
441 with self.admin_access.repo_cnx() as cnx: |
439 with self.admin_access.repo_cnx() as cnx: |
442 with cnx.ensure_cnx_set: |
440 self.assertEqual(self.repo.type_from_eid(2, cnx), 'CWGroup') |
443 self.assertEqual(self.repo.type_from_eid(2, cnx), 'CWGroup') |
|
444 |
441 |
445 def test_type_from_eid_raise(self): |
442 def test_type_from_eid_raise(self): |
446 with self.admin_access.repo_cnx() as cnx: |
443 with self.admin_access.repo_cnx() as cnx: |
447 with cnx.ensure_cnx_set: |
444 self.assertRaises(UnknownEid, self.repo.type_from_eid, -2, cnx) |
448 self.assertRaises(UnknownEid, self.repo.type_from_eid, -2, cnx) |
|
449 |
445 |
450 def test_add_delete_info(self): |
446 def test_add_delete_info(self): |
451 with self.admin_access.repo_cnx() as cnx: |
447 with self.admin_access.repo_cnx() as cnx: |
452 with cnx.ensure_cnx_set: |
448 cnx.mode = 'write' |
453 cnx.mode = 'write' |
449 entity = self.repo.vreg['etypes'].etype_class('Personne')(cnx) |
454 entity = self.repo.vreg['etypes'].etype_class('Personne')(cnx) |
450 entity.eid = -1 |
455 entity.eid = -1 |
451 entity.complete = lambda x: None |
456 entity.complete = lambda x: None |
452 self.repo.add_info(cnx, entity, self.repo.system_source) |
457 self.repo.add_info(cnx, entity, self.repo.system_source) |
453 cu = cnx.system_sql('SELECT * FROM entities WHERE eid = -1') |
458 cu = cnx.system_sql('SELECT * FROM entities WHERE eid = -1') |
454 data = cu.fetchall() |
459 data = cu.fetchall() |
455 self.assertEqual(tuplify(data), [(-1, 'Personne', 'system', None)]) |
460 self.assertEqual(tuplify(data), [(-1, 'Personne', 'system', None)]) |
456 self.repo._delete_cascade_multi(cnx, [entity]) |
461 self.repo._delete_cascade_multi(cnx, [entity]) |
457 self.repo.system_source.delete_info_multi(cnx, [entity]) |
462 self.repo.system_source.delete_info_multi(cnx, [entity]) |
458 cu = cnx.system_sql('SELECT * FROM entities WHERE eid = -1') |
463 cu = cnx.system_sql('SELECT * FROM entities WHERE eid = -1') |
459 data = cu.fetchall() |
464 data = cu.fetchall() |
460 self.assertEqual(data, []) |
465 self.assertEqual(data, []) |
|
466 |
461 |
467 |
462 |
468 class FTITC(CubicWebTC): |
463 class FTITC(CubicWebTC): |
469 |
464 |
470 def test_fulltext_container_entity(self): |
465 def test_fulltext_container_entity(self): |