diff -r 058bb3dc685f -r 0b59724cb3f2 sobjects/test/unittest_cwxmlparser.py --- a/sobjects/test/unittest_cwxmlparser.py Mon Jan 04 18:40:30 2016 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,338 +0,0 @@ -# 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. -# -# 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 datetime import datetime - -from six.moves.urllib.parse import urlsplit, parse_qsl - -import pytz -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' - - def assertURLEquiv(self, first, second): - # ignore ordering differences in query params - parsed_first = urlsplit(first) - parsed_second = urlsplit(second) - self.assertEqual(parsed_first.scheme, parsed_second.scheme) - self.assertEqual(parsed_first.netloc, parsed_second.netloc) - self.assertEqual(parsed_first.path, parsed_second.path) - self.assertEqual(parsed_first.fragment, parsed_second.fragment) - self.assertCountEqual(parse_qsl(parsed_first.query), parse_qsl(parsed_second.query)) - - @classmethod - def pre_setup_database(cls, cnx, config): - myfeed = cnx.create_entity('CWSource', name=u'myfeed', type=u'datafeed', - parser=u'cw.entityxml', url=BASEXML) - myotherfeed = cnx.create_entity('CWSource', name=u'myotherfeed', type=u'datafeed', - parser=u'cw.entityxml', url=OTHERXML) - cnx.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'), - ]) - cnx.create_entity('Tag', name=u'hop') - cnx.commit() - - def test_complete_url(self): - dfsource = self.repo.sources_by_uri['myfeed'] - with self.admin_access.repo_cnx() as cnx: - parser = dfsource._get_parser(cnx) - self.assertURLEquiv(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.assertURLEquiv(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.assertURLEquiv(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.assertURLEquiv(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&vid=rdf&relation=hop'), - 'http://www.cubicweb.org/?rql=cwuser&relation=hop&vid=rdf') - self.assertURLEquiv(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'})], - }, - }) - 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()) - - 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, 1, 22, 10, 27, 59, tzinfo=pytz.utc)) - self.assertEqual(user.modification_date, datetime(2011, 1, 25, 14, 14, 6, tzinfo=pytz.utc)) - 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 - 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 - # clear caches to make sure we look at the moved_entities table - 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['updated'], set((email.eid,))) - 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() - 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'] - 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, 1, 22, 10, 27, 59, tzinfo=pytz.utc)) - self.assertEqual(user.modification_date, datetime(2011, 1, 25, 14, 14, 6, tzinfo=pytz.utc)) - 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'] - with self.repo.internal_cnx() as cnx: - parser = dfsource._get_parser(cnx) - dfsource.process_urls(parser, [''' - - - how-to - - -'''], raise_on_error=True) - - def test_noerror_unspecified_date(self): - dfsource = self.repo.sources_by_uri['myfeed'] - with self.repo.internal_cnx() as cnx: - parser = dfsource._get_parser(cnx) - 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()