# HG changeset patch # User Sylvain Thénault # Date 1329130244 -3600 # Node ID fd02d0120ddc29bdcd9d51b0225b791a9f30c2ea # Parent e4ea285510e5f24f71d65dd92e3f0f7073031297 [test] fix test broken by renaming of module containing the cwxml parser and handling of vid in urls diff -r e4ea285510e5 -r fd02d0120ddc sobjects/test/unittest_cwxmlparser.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sobjects/test/unittest_cwxmlparser.py Mon Feb 13 11:50:44 2012 +0100 @@ -0,0 +1,327 @@ +# copyright 2011-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved. +# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr +# +# This file is part of CubicWeb. +# +# CubicWeb is free software: you can redistribute it and/or modify it under the +# terms of the GNU Lesser General Public License as published by the Free +# Software Foundation, either version 2.1 of the License, or (at your option) +# any later version. +# +# CubicWeb is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more +# details. +# +# You should have received a copy of the GNU Lesser General Public License along +# with CubicWeb. If not, see . + +from __future__ import with_statement + +from datetime import datetime + +from cubicweb.devtools.testlib import CubicWebTC + +from cubicweb.sobjects.cwxmlparser import CWEntityXMLParser + +orig_parse = CWEntityXMLParser.parse + +def parse(self, url): + try: + url = RELATEDXML[url.split('?')[0]] + except KeyError: + pass + return orig_parse(self, url) + +def setUpModule(): + CWEntityXMLParser.parse = parse + +def tearDownModule(): + CWEntityXMLParser.parse = orig_parse + + +BASEXML = ''.join(u''' + + + sthenault + toto + 2011-01-25 14:14:06 + 2010-01-22 10:27:59 + 2011-01-25 14:14:06 + + + + + + + + + + + + + + + + +'''.splitlines()) + +RELATEDXML = { + 'http://pouet.org/6': u''' + + +
syt@logilab.fr
+ 2010-04-13 14:35:56 + 2010-04-13 14:35:56 + + + +
+
+''', + 'http://pouet.org/7': u''' + + + users + + + + + +''', + 'http://pouet.org/8': u''' + + + unknown + + +''', + 'http://pouet.org/9': u''' + + + hop + + +''', + 'http://pouet.org/10': u''' + + + unknown + + +''', + } + + +OTHERXML = ''.join(u''' + + + sthenault + toto + 2011-01-25 14:14:06 + 2010-01-22 10:27:59 + 2011-01-25 14:14:06 + + + + + +'''.splitlines() +) + + +class CWEntityXMLParserTC(CubicWebTC): + """/!\ this test use a pre-setup database /!\, if you modify above xml, + REMOVE THE DATABASE TEMPLATE else it won't be considered + """ + test_db_id = 'xmlparser' + @classmethod + def pre_setup_database(cls, session, config): + myfeed = session.create_entity('CWSource', name=u'myfeed', type=u'datafeed', + parser=u'cw.entityxml', url=BASEXML) + myotherfeed = session.create_entity('CWSource', name=u'myotherfeed', type=u'datafeed', + parser=u'cw.entityxml', url=OTHERXML) + session.commit() + myfeed.init_mapping([(('CWUser', 'use_email', '*'), + u'role=subject\naction=copy'), + (('CWUser', 'in_group', '*'), + u'role=subject\naction=link\nlinkattr=name'), + (('CWUser', 'in_state', '*'), + u'role=subject\naction=link\nlinkattr=name'), + (('*', 'tags', '*'), + u'role=object\naction=link-or-create\nlinkattr=name'), + ]) + myotherfeed.init_mapping([(('CWUser', 'in_group', '*'), + u'role=subject\naction=link\nlinkattr=name'), + (('CWUser', 'in_state', '*'), + u'role=subject\naction=link\nlinkattr=name'), + ]) + session.create_entity('Tag', name=u'hop') + + 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') + + + def test_actions(self): + dfsource = self.repo.sources_by_uri['myfeed'] + self.assertEqual(dfsource.mapping, + {u'CWUser': { + (u'in_group', u'subject', u'link'): [ + (u'CWGroup', {u'linkattr': u'name'})], + (u'in_state', u'subject', u'link'): [ + (u'State', {u'linkattr': u'name'})], + (u'tags', u'object', u'link-or-create'): [ + (u'Tag', {u'linkattr': u'name'})], + (u'use_email', u'subject', u'copy'): [ + (u'EmailAddress', {})] + }, + u'CWGroup': { + (u'tags', u'object', u'link-or-create'): [ + (u'Tag', {u'linkattr': u'name'})], + }, + u'EmailAddress': { + (u'tags', u'object', u'link-or-create'): [ + (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.keys()), ['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']), 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']), 2) + session.commit() + + # 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 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) + + 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') + + 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, [''' + + + how-to + + +'''], raise_on_error=True) + + 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, [''' + + + how-to + how-to + how-to + + + +'''], raise_on_error=True) + +if __name__ == '__main__': + from logilab.common.testlib import unittest_main + unittest_main() diff -r e4ea285510e5 -r fd02d0120ddc sobjects/test/unittest_parsers.py --- a/sobjects/test/unittest_parsers.py Mon Feb 13 11:47:07 2012 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,327 +0,0 @@ -# copyright 2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved. -# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr -# -# This file is part of CubicWeb. -# -# CubicWeb is free software: you can redistribute it and/or modify it under the -# terms of the GNU Lesser General Public License as published by the Free -# Software Foundation, either version 2.1 of the License, or (at your option) -# any later version. -# -# CubicWeb is distributed in the hope that it will be useful, but WITHOUT -# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS -# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more -# details. -# -# You should have received a copy of the GNU Lesser General Public License along -# with CubicWeb. If not, see . - -from __future__ import with_statement - -from datetime import datetime - -from cubicweb.devtools.testlib import CubicWebTC - -from cubicweb.sobjects.parsers import CWEntityXMLParser - -orig_parse = CWEntityXMLParser.parse - -def parse(self, url): - try: - url = RELATEDXML[url.split('?')[0]] - except KeyError: - pass - return orig_parse(self, url) - -def setUpModule(): - CWEntityXMLParser.parse = parse - -def tearDownModule(): - CWEntityXMLParser.parse = orig_parse - - -BASEXML = ''.join(u''' - - - sthenault - toto - 2011-01-25 14:14:06 - 2010-01-22 10:27:59 - 2011-01-25 14:14:06 - - - - - - - - - - - - - - - - -'''.splitlines()) - -RELATEDXML = { - 'http://pouet.org/6': u''' - - -
syt@logilab.fr
- 2010-04-13 14:35:56 - 2010-04-13 14:35:56 - - - -
-
-''', - 'http://pouet.org/7': u''' - - - users - - - - - -''', - 'http://pouet.org/8': u''' - - - unknown - - -''', - 'http://pouet.org/9': u''' - - - hop - - -''', - 'http://pouet.org/10': u''' - - - unknown - - -''', - } - - -OTHERXML = ''.join(u''' - - - sthenault - toto - 2011-01-25 14:14:06 - 2010-01-22 10:27:59 - 2011-01-25 14:14:06 - - - - - -'''.splitlines() -) - - -class CWEntityXMLParserTC(CubicWebTC): - """/!\ this test use a pre-setup database /!\, if you modify above xml, - REMOVE THE DATABASE TEMPLATE else it won't be considered - """ - test_db_id = 'xmlparser' - @classmethod - def pre_setup_database(cls, session, config): - myfeed = session.create_entity('CWSource', name=u'myfeed', type=u'datafeed', - parser=u'cw.entityxml', url=BASEXML) - myotherfeed = session.create_entity('CWSource', name=u'myotherfeed', type=u'datafeed', - parser=u'cw.entityxml', url=OTHERXML) - session.commit() - myfeed.init_mapping([(('CWUser', 'use_email', '*'), - u'role=subject\naction=copy'), - (('CWUser', 'in_group', '*'), - u'role=subject\naction=link\nlinkattr=name'), - (('CWUser', 'in_state', '*'), - u'role=subject\naction=link\nlinkattr=name'), - (('*', 'tags', '*'), - u'role=object\naction=link-or-create\nlinkattr=name'), - ]) - myotherfeed.init_mapping([(('CWUser', 'in_group', '*'), - u'role=subject\naction=link\nlinkattr=name'), - (('CWUser', 'in_state', '*'), - u'role=subject\naction=link\nlinkattr=name'), - ]) - session.create_entity('Tag', name=u'hop') - - 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_tate-subject&relation=use_email-subject&vid=xml') - 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&vid=xml') - 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&vid=xml') - - - def test_actions(self): - dfsource = self.repo.sources_by_uri['myfeed'] - self.assertEqual(dfsource.mapping, - {u'CWUser': { - (u'in_group', u'subject', u'link'): [ - (u'CWGroup', {u'linkattr': u'name'})], - (u'in_state', u'subject', u'link'): [ - (u'State', {u'linkattr': u'name'})], - (u'tags', u'object', u'link-or-create'): [ - (u'Tag', {u'linkattr': u'name'})], - (u'use_email', u'subject', u'copy'): [ - (u'EmailAddress', {})] - }, - u'CWGroup': { - (u'tags', u'object', u'link-or-create'): [ - (u'Tag', {u'linkattr': u'name'})], - }, - u'EmailAddress': { - (u'tags', u'object', u'link-or-create'): [ - (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.keys()), ['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']), 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']), 2) - session.commit() - - # 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 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) - - 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') - - 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, [''' - - - how-to - - -'''], raise_on_error=True) - - 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, [''' - - - how-to - how-to - how-to - - - -'''], raise_on_error=True) - -if __name__ == '__main__': - from logilab.common.testlib import unittest_main - unittest_main()