Mercurial Mercurial > pub > hg > cubicweb / comparison
summary | shortlog | changelog | graph | tags | bookmarks | branches | files | changeset | file | latest | revisions | annotate | diff | comparison | raw | help
Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
sobjects/test/unittest_cwxmlparser.py
changeset 9748 5ee3d16b0df0
parent 8696 0bb18407c053
child 9879 21278eb03bbf
equal deleted inserted replaced
9747:10108d9f502a 9748:5ee3d16b0df0 
     1 # copyright 2011-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
 
     1 # copyright 2011-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
 
     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
 
     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
 
     3 #
 
     3 #
 
     4 # This file is part of CubicWeb.
 
     4 # This file is part of CubicWeb.
 
     5 #
 
     5 #
 
     6 # CubicWeb is free software: you can redistribute it and/or modify it under the
 
     6 # CubicWeb is free software: you can redistribute it and/or modify it under the
 
   155                                   ])
 
   155                                   ])
 
   156         session.create_entity('Tag', name=u'hop')
 
   156         session.create_entity('Tag', name=u'hop')
 
   157 
   157 
   158     def test_complete_url(self):
 
   158     def test_complete_url(self):
 
   159         dfsource = self.repo.sources_by_uri['myfeed']
 
   159         dfsource = self.repo.sources_by_uri['myfeed']
 
   160         parser = dfsource._get_parser(self.session)
 
   160         with self.admin_access.repo_cnx() as cnx:
 
   161         self.assertEqual(parser.complete_url('http://www.cubicweb.org/CWUser'),
 
   161             parser = dfsource._get_parser(cnx)
 
   162                          'http://www.cubicweb.org/CWUser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject')
 
   162             self.assertEqual(parser.complete_url('http://www.cubicweb.org/CWUser'),
 
   163         self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser'),
 
   163                              'http://www.cubicweb.org/CWUser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject')
 
   164                          'http://www.cubicweb.org/cwuser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject')
 
   164             self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser'),
 
   165         self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser?vid=rdf&relation=hop'),
 
   165                              'http://www.cubicweb.org/cwuser?relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject')
 
   166                          'http://www.cubicweb.org/cwuser?relation=hop&relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject&vid=rdf')
 
   166             self.assertEqual(parser.complete_url('http://www.cubicweb.org/cwuser?vid=rdf&relation=hop'),
 
   167         self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&vid=rdf&relation=hop'),
 
   167                              'http://www.cubicweb.org/cwuser?relation=hop&relation=tags-object&relation=in_group-subject&relation=in_state-subject&relation=use_email-subject&vid=rdf')
 
   168                          'http://www.cubicweb.org/?rql=cwuser&relation=hop&vid=rdf')
 
   168             self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&vid=rdf&relation=hop'),
 
   169         self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&relation=hop'),
 
   169                              'http://www.cubicweb.org/?rql=cwuser&relation=hop&vid=rdf')
 
   170                          'http://www.cubicweb.org/?rql=cwuser&relation=hop')
 
   170             self.assertEqual(parser.complete_url('http://www.cubicweb.org/?rql=cwuser&relation=hop'),
 
       
   171                              'http://www.cubicweb.org/?rql=cwuser&relation=hop')
 
   171 
   172 
   172 
   173 
   173     def test_actions(self):
 
   174     def test_actions(self):
 
   174         dfsource = self.repo.sources_by_uri['myfeed']
 
   175         dfsource = self.repo.sources_by_uri['myfeed']
 
   175         self.assertEqual(dfsource.mapping,
 
   176         self.assertEqual(dfsource.mapping,
 
   190                           u'EmailAddress': {
 
   191                           u'EmailAddress': {
 
   191                              (u'tags', u'object', u'link-or-create'): [
 
   192                              (u'tags', u'object', u'link-or-create'): [
 
   192                                  (u'Tag', {u'linkattr': u'name'})],
 
   193                                  (u'Tag', {u'linkattr': u'name'})],
 
   193                              },
 
   194                              },
 
   194                           })
 
   195                           })
 
   195         session = self.repo.internal_session(safe=True)
 
   196         with self.repo.internal_cnx() as cnx:
 
   196         stats = dfsource.pull_data(session, force=True, raise_on_error=True)
 
   197             stats = dfsource.pull_data(cnx, force=True, raise_on_error=True)
 
   197         self.assertEqual(sorted(stats), ['checked', 'created', 'updated'])
 
   198             self.assertEqual(sorted(stats), ['checked', 'created', 'updated'])
 
   198         self.assertEqual(len(stats['created']), 2)
 
   199             self.assertEqual(len(stats['created']), 2)
 
   199         self.assertEqual(stats['updated'], set())
 
   200             self.assertEqual(stats['updated'], set())
 
   200 
   201 
   201         user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0)
 
   202         with self.admin_access.web_request() as req:
 
   202         self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59))
 
   203             user = req.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0)
 
   203         self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06))
 
   204             self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59))
 
   204         self.assertEqual(user.cwuri, 'http://pouet.org/5')
 
   205             self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06))
 
   205         self.assertEqual(user.cw_source[0].name, 'myfeed')
 
   206             self.assertEqual(user.cwuri, 'http://pouet.org/5')
 
   206         self.assertEqual(user.absolute_url(), 'http://pouet.org/5')
 
   207             self.assertEqual(user.cw_source[0].name, 'myfeed')
 
   207         self.assertEqual(len(user.use_email), 1)
 
   208             self.assertEqual(user.absolute_url(), 'http://pouet.org/5')
 
   208         # copy action
 
   209             self.assertEqual(len(user.use_email), 1)
 
   209         email = user.use_email[0]
 
   210             # copy action
 
   210         self.assertEqual(email.address, 'syt@logilab.fr')
 
   211             email = user.use_email[0]
 
   211         self.assertEqual(email.cwuri, 'http://pouet.org/6')
 
   212             self.assertEqual(email.address, 'syt@logilab.fr')
 
   212         self.assertEqual(email.absolute_url(), 'http://pouet.org/6')
 
   213             self.assertEqual(email.cwuri, 'http://pouet.org/6')
 
   213         self.assertEqual(email.cw_source[0].name, 'myfeed')
 
   214             self.assertEqual(email.absolute_url(), 'http://pouet.org/6')
 
   214         self.assertEqual(len(email.reverse_tags), 1)
 
   215             self.assertEqual(email.cw_source[0].name, 'myfeed')
 
   215         self.assertEqual(email.reverse_tags[0].name, 'hop')
 
   216             self.assertEqual(len(email.reverse_tags), 1)
 
   216         # link action
 
   217             self.assertEqual(email.reverse_tags[0].name, 'hop')
 
   217         self.assertFalse(self.execute('CWGroup X WHERE X name "unknown"'))
 
   218             # link action
 
   218         groups = sorted([g.name for g in user.in_group])
 
   219             self.assertFalse(req.execute('CWGroup X WHERE X name "unknown"'))
 
   219         self.assertEqual(groups, ['users'])
 
   220             groups = sorted([g.name for g in user.in_group])
 
   220         group = user.in_group[0]
 
   221             self.assertEqual(groups, ['users'])
 
   221         self.assertEqual(len(group.reverse_tags), 1)
 
   222             group = user.in_group[0]
 
   222         self.assertEqual(group.reverse_tags[0].name, 'hop')
 
   223             self.assertEqual(len(group.reverse_tags), 1)
 
   223         # link or create action
 
   224             self.assertEqual(group.reverse_tags[0].name, 'hop')
 
   224         tags = set([(t.name, t.cwuri.replace(str(t.eid), ''), t.cw_source[0].name)
 
   225             # link or create action
 
   225                     for t in user.reverse_tags])
 
   226             tags = set([(t.name, t.cwuri.replace(str(t.eid), ''), t.cw_source[0].name)
 
   226         self.assertEqual(tags, set((('hop', 'http://testing.fr/cubicweb/', 'system'),
 
   227                         for t in user.reverse_tags])
 
   227                                     ('unknown', 'http://testing.fr/cubicweb/', 'system')))
 
   228             self.assertEqual(tags, set((('hop', 'http://testing.fr/cubicweb/', 'system'),
 
   228                          )
 
   229                                         ('unknown', 'http://testing.fr/cubicweb/', 'system')))
 
   229         session.set_cnxset()
 
   230                              )
 
   230         with session.security_enabled(read=False): # avoid Unauthorized due to password selection
 
   231         with self.repo.internal_cnx() as cnx:
 
   231             stats = dfsource.pull_data(session, force=True, raise_on_error=True)
 
   232             stats = dfsource.pull_data(cnx, force=True, raise_on_error=True)
 
   232         self.assertEqual(stats['created'], set())
 
   233             self.assertEqual(stats['created'], set())
 
   233         self.assertEqual(len(stats['updated']), 0)
 
   234             self.assertEqual(len(stats['updated']), 0)
 
   234         self.assertEqual(len(stats['checked']), 2)
 
   235             self.assertEqual(len(stats['checked']), 2)
 
   235         self.repo._type_source_cache.clear()
 
   236             self.repo._type_source_cache.clear()
 
   236         self.repo._extid_cache.clear()
 
   237             self.repo._extid_cache.clear()
 
   237         session.set_cnxset()
 
   238             stats = dfsource.pull_data(cnx, force=True, raise_on_error=True)
 
   238         with session.security_enabled(read=False): # avoid Unauthorized due to password selection
 
   239             self.assertEqual(stats['created'], set())
 
   239             stats = dfsource.pull_data(session, force=True, raise_on_error=True)
 
   240             self.assertEqual(len(stats['updated']), 0)
 
   240         self.assertEqual(stats['created'], set())
 
   241             self.assertEqual(len(stats['checked']), 2)
 
   241         self.assertEqual(len(stats['updated']), 0)
 
   242 
   242         self.assertEqual(len(stats['checked']), 2)
 
   243             # test move to system source
 
   243         session.commit()
 
   244             cnx.execute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid})
 
   244 
   245             cnx.commit()
 
   245         # test move to system source
 
   246             rset = cnx.execute('EmailAddress X WHERE X address "syt@logilab.fr"')
 
   246         self.sexecute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid})
 
   247             self.assertEqual(len(rset), 1)
 
   247         self.commit()
 
   248             e = rset.get_entity(0, 0)
 
   248         rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"')
 
   249             self.assertEqual(e.eid, email.eid)
 
   249         self.assertEqual(len(rset), 1)
 
   250             self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system',
 
   250         e = rset.get_entity(0, 0)
 
   251                                                                  'use-cwuri-as-url': False},
 
   251         self.assertEqual(e.eid, email.eid)
 
   252                                                       'type': 'EmailAddress',
 
   252         self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system',
 
   253                                                       'extid': None})
 
   253                                                              'use-cwuri-as-url': False},
 
   254             self.assertEqual(e.cw_source[0].name, 'system')
 
   254                                                   'type': 'EmailAddress',
 
   255             self.assertEqual(e.reverse_use_email[0].login, 'sthenault')
 
   255                                                   'extid': None})
 
   256             # test everything is still fine after source synchronization
 
   256         self.assertEqual(e.cw_source[0].name, 'system')
 
   257             stats = dfsource.pull_data(cnx, force=True, raise_on_error=True)
 
   257         self.assertEqual(e.reverse_use_email[0].login, 'sthenault')
 
   258             rset = cnx.execute('EmailAddress X WHERE X address "syt@logilab.fr"')
 
   258         self.commit()
 
   259             self.assertEqual(len(rset), 1)
 
   259         # test everything is still fine after source synchronization
 
   260             e = rset.get_entity(0, 0)
 
   260         session.set_cnxset()
 
   261             self.assertEqual(e.eid, email.eid)
 
   261         with session.security_enabled(read=False): # avoid Unauthorized due to password selection
 
   262             self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system',
 
   262             stats = dfsource.pull_data(session, force=True, raise_on_error=True)
 
   263                                                                  'use-cwuri-as-url': False},
 
   263         rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"')
 
   264                                                       'type': 'EmailAddress',
 
   264         self.assertEqual(len(rset), 1)
 
   265                                                       'extid': None})
 
   265         e = rset.get_entity(0, 0)
 
   266             self.assertEqual(e.cw_source[0].name, 'system')
 
   266         self.assertEqual(e.eid, email.eid)
 
   267             self.assertEqual(e.reverse_use_email[0].login, 'sthenault')
 
   267         self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system',
 
   268             cnx.commit()
 
   268                                                              'use-cwuri-as-url': False},
 
   269 
   269                                                   'type': 'EmailAddress',
 
   270             # test delete entity
 
   270                                                   'extid': None})
 
   271             e.cw_delete()
 
   271         self.assertEqual(e.cw_source[0].name, 'system')
 
   272             cnx.commit()
 
   272         self.assertEqual(e.reverse_use_email[0].login, 'sthenault')
 
   273             # test everything is still fine after source synchronization
 
   273         session.commit()
 
   274             stats = dfsource.pull_data(cnx, force=True, raise_on_error=True)
 
   274 
   275             rset = cnx.execute('EmailAddress X WHERE X address "syt@logilab.fr"')
 
   275         # test delete entity
 
   276             self.assertEqual(len(rset), 0)
 
   276         e.cw_delete()
 
   277             rset = cnx.execute('Any X WHERE X use_email E, X login "sthenault"')
 
   277         self.commit()
 
   278             self.assertEqual(len(rset), 0)
 
   278         # test everything is still fine after source synchronization
 
       
   279         session.set_cnxset()
 
       
   280         with session.security_enabled(read=False): # avoid Unauthorized due to password selection
 
       
   281             stats = dfsource.pull_data(session, force=True, raise_on_error=True)
 
       
   282         rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"')
 
       
   283         self.assertEqual(len(rset), 0)
 
       
   284         rset = self.sexecute('Any X WHERE X use_email E, X login "sthenault"')
 
       
   285         self.assertEqual(len(rset), 0)
 
       
   286 
   279 
   287     def test_external_entity(self):
 
   280     def test_external_entity(self):
 
   288         dfsource = self.repo.sources_by_uri['myotherfeed']
 
   281         dfsource = self.repo.sources_by_uri['myotherfeed']
 
   289         session = self.repo.internal_session(safe=True)
 
   282         with self.repo.internal_cnx() as cnx:
 
   290         stats = dfsource.pull_data(session, force=True, raise_on_error=True)
 
   283             stats = dfsource.pull_data(cnx, force=True, raise_on_error=True)
 
   291         user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0)
 
   284             user = cnx.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0)
 
   292         self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59))
 
   285             self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59))
 
   293         self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06))
 
   286             self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06))
 
   294         self.assertEqual(user.cwuri, 'http://pouet.org/5')
 
   287             self.assertEqual(user.cwuri, 'http://pouet.org/5')
 
   295         self.assertEqual(user.cw_source[0].name, 'myfeed')
 
   288             self.assertEqual(user.cw_source[0].name, 'myfeed')
 
   296 
   289 
   297     def test_noerror_missing_fti_attribute(self):
 
   290     def test_noerror_missing_fti_attribute(self):
 
   298         dfsource = self.repo.sources_by_uri['myfeed']
 
   291         dfsource = self.repo.sources_by_uri['myfeed']
 
   299         session = self.repo.internal_session(safe=True)
 
   292         with self.repo.internal_cnx() as cnx:
 
   300         parser = dfsource._get_parser(session)
 
   293             parser = dfsource._get_parser(cnx)
 
   301         dfsource.process_urls(parser, ['''
 
   294             dfsource.process_urls(parser, ['''
 
   302 <rset size="1">
 
   295 <rset size="1">
 
   303  <Card eid="50" cwuri="http://pouet.org/50" cwsource="system">
 
   296  <Card eid="50" cwuri="http://pouet.org/50" cwsource="system">
 
   304   <title>how-to</title>
 
   297   <title>how-to</title>
 
   305  </Card>
 
   298  </Card>
 
   306 </rset>
 
   299 </rset>
 
   307 '''], raise_on_error=True)
 
   300 '''], raise_on_error=True)
 
   308 
   301 
   309     def test_noerror_unspecified_date(self):
 
   302     def test_noerror_unspecified_date(self):
 
   310         dfsource = self.repo.sources_by_uri['myfeed']
 
   303         dfsource = self.repo.sources_by_uri['myfeed']
 
   311         session = self.repo.internal_session(safe=True)
 
   304         with self.repo.internal_cnx() as cnx:
 
   312         parser = dfsource._get_parser(session)
 
   305             parser = dfsource._get_parser(cnx)
 
   313         dfsource.process_urls(parser, ['''
 
   306             dfsource.process_urls(parser, ['''
 
   314 <rset size="1">
 
   307 <rset size="1">
 
   315  <Card eid="50" cwuri="http://pouet.org/50" cwsource="system">
 
   308  <Card eid="50" cwuri="http://pouet.org/50" cwsource="system">
 
   316   <title>how-to</title>
 
   309   <title>how-to</title>
 
   317   <content>how-to</content>
 
   310   <content>how-to</content>
 
   318   <synopsis>how-to</synopsis>
 
   311   <synopsis>how-to</synopsis>
cubicweb