# HG changeset patch # User Sylvain Thénault # Date 1397057338 -7200 # Node ID 5ee3d16b0df07c0f10e0509ec38eac0a92871c94 # Parent 10108d9f502a7b3580efc490e5dbb9b6c98ad27d [test] update sobjects/test/unittest_cwxmlparser to cw 3.19 api diff -r 10108d9f502a -r 5ee3d16b0df0 sobjects/test/unittest_cwxmlparser.py --- a/sobjects/test/unittest_cwxmlparser.py Wed Apr 09 17:15:25 2014 +0200 +++ b/sobjects/test/unittest_cwxmlparser.py Wed Apr 09 17:28:58 2014 +0200 @@ -1,4 +1,4 @@ -# copyright 2011-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved. +# copyright 2011-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved. # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr # # This file is part of CubicWeb. @@ -157,17 +157,18 @@ def test_complete_url(self): dfsource = self.repo.sources_by_uri['myfeed'] - parser = dfsource._get_parser(self.session) - self.assertEqual(parser.complete_url('http://www.cubicweb.org/CWUser'), - 'http://www.cubicweb.org/CWUser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject') - self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser'), - 'http://www.cubicweb.org/cwuser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject') - self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser?vid=rdf&relation=hop'), - 'http://www.cubicweb.org/cwuser?relation=hop&relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject&vid=rdf') - self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&vid=rdf&relation=hop'), - 'http://www.cubicweb.org/?rql=cwuser&relation=hop&vid=rdf') - self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&relation=hop'), - 'http://www.cubicweb.org/?rql=cwuser&relation=hop') + with self.admin_access.repo_cnx() as cnx: + parser = dfsource._get_parser(cnx) + self.assertEqual(parser.complete_url('http://www.cubicweb.org/CWUser'), + 'http://www.cubicweb.org/CWUser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject') + self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser'), + 'http://www.cubicweb.org/cwuser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject') + self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser?vid=rdf&relation=hop'), + 'http://www.cubicweb.org/cwuser?relation=hop&relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject&vid=rdf') + self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&vid=rdf&relation=hop'), + 'http://www.cubicweb.org/?rql=cwuser&relation=hop&vid=rdf') + self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&relation=hop'), + 'http://www.cubicweb.org/?rql=cwuser&relation=hop') def test_actions(self): @@ -192,113 +193,105 @@ (u'Tag', {u'linkattr': u'name'})], }, }) - session = self.repo.internal_session(safe=True) - stats = dfsource.pull_data(session, force=True, raise_on_error=True) - self.assertEqual(sorted(stats), ['checked', 'created', 'updated']) - self.assertEqual(len(stats['created']), 2) - self.assertEqual(stats['updated'], set()) + with self.repo.internal_cnx() as cnx: + stats = dfsource.pull_data(cnx, force=True, raise_on_error=True) + self.assertEqual(sorted(stats), ['checked', 'created', 'updated']) + self.assertEqual(len(stats['created']), 2) + self.assertEqual(stats['updated'], set()) - user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) - self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) - self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06)) - self.assertEqual(user.cwuri, 'http://pouet.org/5') - self.assertEqual(user.cw_source[0].name, 'myfeed') - self.assertEqual(user.absolute_url(), 'http://pouet.org/5') - self.assertEqual(len(user.use_email), 1) - # copy action - email = user.use_email[0] - self.assertEqual(email.address, 'syt@logilab.fr') - self.assertEqual(email.cwuri, 'http://pouet.org/6') - self.assertEqual(email.absolute_url(), 'http://pouet.org/6') - self.assertEqual(email.cw_source[0].name, 'myfeed') - self.assertEqual(len(email.reverse_tags), 1) - self.assertEqual(email.reverse_tags[0].name, 'hop') - # link action - self.assertFalse(self.execute('CWGroup X WHERE X name "unknown"')) - groups = sorted([g.name for g in user.in_group]) - self.assertEqual(groups, ['users']) - group = user.in_group[0] - self.assertEqual(len(group.reverse_tags), 1) - self.assertEqual(group.reverse_tags[0].name, 'hop') - # link or create action - tags = set([(t.name, t.cwuri.replace(str(t.eid), ''), t.cw_source[0].name) - for t in user.reverse_tags]) - self.assertEqual(tags, set((('hop', 'http://testing.fr/cubicweb/', 'system'), - ('unknown', 'http://testing.fr/cubicweb/', 'system'))) - ) - session.set_cnxset() - with session.security_enabled(read=False): # avoid Unauthorized due to password selection - stats = dfsource.pull_data(session, force=True, raise_on_error=True) - self.assertEqual(stats['created'], set()) - self.assertEqual(len(stats['updated']), 0) - self.assertEqual(len(stats['checked']), 2) - self.repo._type_source_cache.clear() - self.repo._extid_cache.clear() - session.set_cnxset() - with session.security_enabled(read=False): # avoid Unauthorized due to password selection - stats = dfsource.pull_data(session, force=True, raise_on_error=True) - self.assertEqual(stats['created'], set()) - self.assertEqual(len(stats['updated']), 0) - self.assertEqual(len(stats['checked']), 2) - session.commit() + with self.admin_access.web_request() as req: + user = req.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) + self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) + self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06)) + self.assertEqual(user.cwuri, 'http://pouet.org/5') + self.assertEqual(user.cw_source[0].name, 'myfeed') + self.assertEqual(user.absolute_url(), 'http://pouet.org/5') + self.assertEqual(len(user.use_email), 1) + # copy action + email = user.use_email[0] + self.assertEqual(email.address, 'syt@logilab.fr') + self.assertEqual(email.cwuri, 'http://pouet.org/6') + self.assertEqual(email.absolute_url(), 'http://pouet.org/6') + self.assertEqual(email.cw_source[0].name, 'myfeed') + self.assertEqual(len(email.reverse_tags), 1) + self.assertEqual(email.reverse_tags[0].name, 'hop') + # link action + self.assertFalse(req.execute('CWGroup X WHERE X name "unknown"')) + groups = sorted([g.name for g in user.in_group]) + self.assertEqual(groups, ['users']) + group = user.in_group[0] + self.assertEqual(len(group.reverse_tags), 1) + self.assertEqual(group.reverse_tags[0].name, 'hop') + # link or create action + tags = set([(t.name, t.cwuri.replace(str(t.eid), ''), t.cw_source[0].name) + for t in user.reverse_tags]) + self.assertEqual(tags, set((('hop', 'http://testing.fr/cubicweb/', 'system'), + ('unknown', 'http://testing.fr/cubicweb/', 'system'))) + ) + with self.repo.internal_cnx() as cnx: + stats = dfsource.pull_data(cnx, force=True, raise_on_error=True) + self.assertEqual(stats['created'], set()) + self.assertEqual(len(stats['updated']), 0) + self.assertEqual(len(stats['checked']), 2) + self.repo._type_source_cache.clear() + self.repo._extid_cache.clear() + stats = dfsource.pull_data(cnx, force=True, raise_on_error=True) + self.assertEqual(stats['created'], set()) + self.assertEqual(len(stats['updated']), 0) + self.assertEqual(len(stats['checked']), 2) - # test move to system source - self.sexecute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid}) - self.commit() - rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"') - self.assertEqual(len(rset), 1) - e = rset.get_entity(0, 0) - self.assertEqual(e.eid, email.eid) - self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system', - 'use-cwuri-as-url': False}, - 'type': 'EmailAddress', - 'extid': None}) - self.assertEqual(e.cw_source[0].name, 'system') - self.assertEqual(e.reverse_use_email[0].login, 'sthenault') - self.commit() - # test everything is still fine after source synchronization - session.set_cnxset() - with session.security_enabled(read=False): # avoid Unauthorized due to password selection - stats = dfsource.pull_data(session, force=True, raise_on_error=True) - rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"') - self.assertEqual(len(rset), 1) - e = rset.get_entity(0, 0) - self.assertEqual(e.eid, email.eid) - self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system', - 'use-cwuri-as-url': False}, - 'type': 'EmailAddress', - 'extid': None}) - self.assertEqual(e.cw_source[0].name, 'system') - self.assertEqual(e.reverse_use_email[0].login, 'sthenault') - session.commit() + # test move to system source + cnx.execute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid}) + cnx.commit() + rset = cnx.execute('EmailAddress X WHERE X address "syt@logilab.fr"') + self.assertEqual(len(rset), 1) + e = rset.get_entity(0, 0) + self.assertEqual(e.eid, email.eid) + self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system', + 'use-cwuri-as-url': False}, + 'type': 'EmailAddress', + 'extid': None}) + self.assertEqual(e.cw_source[0].name, 'system') + self.assertEqual(e.reverse_use_email[0].login, 'sthenault') + # test everything is still fine after source synchronization + stats = dfsource.pull_data(cnx, force=True, raise_on_error=True) + rset = cnx.execute('EmailAddress X WHERE X address "syt@logilab.fr"') + self.assertEqual(len(rset), 1) + e = rset.get_entity(0, 0) + self.assertEqual(e.eid, email.eid) + self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system', + 'use-cwuri-as-url': False}, + 'type': 'EmailAddress', + 'extid': None}) + self.assertEqual(e.cw_source[0].name, 'system') + self.assertEqual(e.reverse_use_email[0].login, 'sthenault') + cnx.commit() - # test delete entity - e.cw_delete() - self.commit() - # test everything is still fine after source synchronization - session.set_cnxset() - with session.security_enabled(read=False): # avoid Unauthorized due to password selection - stats = dfsource.pull_data(session, force=True, raise_on_error=True) - rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"') - self.assertEqual(len(rset), 0) - rset = self.sexecute('Any X WHERE X use_email E, X login "sthenault"') - self.assertEqual(len(rset), 0) + # test delete entity + e.cw_delete() + cnx.commit() + # test everything is still fine after source synchronization + stats = dfsource.pull_data(cnx, force=True, raise_on_error=True) + rset = cnx.execute('EmailAddress X WHERE X address "syt@logilab.fr"') + self.assertEqual(len(rset), 0) + rset = cnx.execute('Any X WHERE X use_email E, X login "sthenault"') + self.assertEqual(len(rset), 0) def test_external_entity(self): dfsource = self.repo.sources_by_uri['myotherfeed'] - session = self.repo.internal_session(safe=True) - stats = dfsource.pull_data(session, force=True, raise_on_error=True) - user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) - self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) - self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06)) - self.assertEqual(user.cwuri, 'http://pouet.org/5') - self.assertEqual(user.cw_source[0].name, 'myfeed') + with self.repo.internal_cnx() as cnx: + stats = dfsource.pull_data(cnx, force=True, raise_on_error=True) + user = cnx.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) + self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) + self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06)) + self.assertEqual(user.cwuri, 'http://pouet.org/5') + self.assertEqual(user.cw_source[0].name, 'myfeed') def test_noerror_missing_fti_attribute(self): dfsource = self.repo.sources_by_uri['myfeed'] - session = self.repo.internal_session(safe=True) - parser = dfsource._get_parser(session) - dfsource.process_urls(parser, [''' + with self.repo.internal_cnx() as cnx: + parser = dfsource._get_parser(cnx) + dfsource.process_urls(parser, [''' how-to @@ -308,9 +301,9 @@ def test_noerror_unspecified_date(self): dfsource = self.repo.sources_by_uri['myfeed'] - session = self.repo.internal_session(safe=True) - parser = dfsource._get_parser(session) - dfsource.process_urls(parser, [''' + with self.repo.internal_cnx() as cnx: + parser = dfsource._get_parser(cnx) + dfsource.process_urls(parser, [''' how-to