1 # copyright 2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

     3 #

     4 # This file is part of CubicWeb.

     5 #

     6 # CubicWeb is free software: you can redistribute it and/or modify it under the

     7 # terms of the GNU Lesser General Public License as published by the Free

     8 # Software Foundation, either version 2.1 of the License, or (at your option)

     9 # any later version.

    10 #

    11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT

    12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

    13 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

    14 # details.

    15 #

    16 # You should have received a copy of the GNU Lesser General Public License along

    17 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

    18 from __future__ import with_statement

    19 

    20 from datetime import timedelta

    21 

    22 from cubicweb.devtools.testlib import CubicWebTC

    23 from cubicweb.server.sources import datafeed

    24 

    25 

    26 class DataFeedTC(CubicWebTC):

    27     def setup_database(self):

    28         self.request().create_entity('CWSource', name=u'myfeed', type=u'datafeed',

    29                                     parser=u'testparser', url=u'ignored',

    30                                     config=u'synchronization-interval=1min')

    31 

    32     def test(self):

    33         self.assertIn('myfeed', self.repo.sources_by_uri)

    34         dfsource = self.repo.sources_by_uri['myfeed']

    35         self.assertNotIn(dfsource, self.repo.sources)

    36         self.assertEqual(dfsource.latest_retrieval, None)

    37         self.assertEqual(dfsource.synchro_interval, timedelta(seconds=60))

    38         self.assertFalse(dfsource.fresh())

    39 

    40         class AParser(datafeed.DataFeedParser):

    41             __regid__ = 'testparser'

    42             def process(self, url):

    43                 entity = self.extid2entity('http://www.cubicweb.org/', 'Card',

    44                                   item={'title': u'cubicweb.org',

    45                                         'content': u'the cw web site'})

    46                 if not self.created_during_pull(entity):

    47                     self.notify_updated(entity)

    48             def before_entity_copy(self, entity, sourceparams):

    49                 entity.cw_edited.update(sourceparams['item'])

    50 

    51         with self.temporary_appobjects(AParser):

    52             stats = dfsource.pull_data(self.session, force=True)

    53             self.commit()

    54             # test import stats

    55             self.assertEqual(sorted(stats.keys()), ['created', 'updated'])

    56             self.assertEqual(len(stats['created']), 1)

    57             entity = self.execute('Card X').get_entity(0, 0)

    58             self.assertIn(entity.eid, stats['created'])

    59             self.assertEqual(stats['updated'], set())

    60             # test imported entities

    61             self.assertEqual(entity.title, 'cubicweb.org')

    62             self.assertEqual(entity.content, 'the cw web site')

    63             self.assertEqual(entity.cwuri, 'http://www.cubicweb.org/')

    64             self.assertEqual(entity.cw_source[0].name, 'myfeed')

    65             self.assertEqual(entity.cw_metainformation(),

    66                              {'type': 'Card',

    67                               'source': {'uri': 'system', 'type': 'native'},

    68                               'extid': 'http://www.cubicweb.org/'}

    69                              )

    70             # test repo cache keys

    71             self.assertEqual(self.repo._type_source_cache[entity.eid],

    72                              ('Card', 'system', 'http://www.cubicweb.org/'))

    73             self.assertEqual(self.repo._extid_cache[('http://www.cubicweb.org/', 'system')],

    74                              entity.eid)

    75             # test repull

    76             stats = dfsource.pull_data(self.session, force=True)

    77             self.assertEqual(stats['created'], set())

    78             self.assertEqual(stats['updated'], set((entity.eid,)))

    79             # test repull with caches reseted

    80             self.repo._type_source_cache.clear()

    81             self.repo._extid_cache.clear()

    82             stats = dfsource.pull_data(self.session, force=True)

    83             self.assertEqual(stats['created'], set())

    84             self.assertEqual(stats['updated'], set((entity.eid,)))

    85             self.assertEqual(self.repo._type_source_cache[entity.eid],

    86                              ('Card', 'system', 'http://www.cubicweb.org/'))

    87             self.assertEqual(self.repo._extid_cache[('http://www.cubicweb.org/', 'system')],

    88                              entity.eid)

    89 

    90         self.assertEqual(dfsource.source_cwuris(self.session),

    91                          {'http://www.cubicweb.org/': (entity.eid, 'Card')}

    92                          )

    93         self.assertTrue(dfsource.latest_retrieval)

    94         self.assertTrue(dfsource.fresh())

    95 

    96 if __name__ == '__main__':

    97     from logilab.common.testlib import unittest_main

    98     unittest_main()