--- a/server/test/unittest_datafeed.py Fri May 30 17:46:40 2014 +0200
+++ b/server/test/unittest_datafeed.py Fri May 30 17:52:15 2014 +0200
@@ -1,4 +1,4 @@
-# copyright 2011-2013 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.
@@ -24,9 +24,11 @@
class DataFeedTC(CubicWebTC):
def setup_database(self):
- self.request().create_entity('CWSource', name=u'myfeed', type=u'datafeed',
- parser=u'testparser', url=u'ignored',
- config=u'synchronization-interval=1min')
+ with self.admin_access.repo_cnx() as cnx:
+ cnx.create_entity('CWSource', name=u'myfeed', type=u'datafeed',
+ parser=u'testparser', url=u'ignored',
+ config=u'synchronization-interval=1min')
+ cnx.commit()
def test(self):
self.assertIn('myfeed', self.repo.sources_by_uri)
@@ -48,47 +50,45 @@
entity.cw_edited.update(sourceparams['item'])
with self.temporary_appobjects(AParser):
- session = self.repo.internal_session()
- stats = dfsource.pull_data(session, force=True)
- self.commit()
- # test import stats
- self.assertEqual(sorted(stats), ['checked', 'created', 'updated'])
- self.assertEqual(len(stats['created']), 1)
- entity = self.execute('Card X').get_entity(0, 0)
- self.assertIn(entity.eid, stats['created'])
- self.assertEqual(stats['updated'], set())
- # test imported entities
- self.assertEqual(entity.title, 'cubicweb.org')
- self.assertEqual(entity.content, 'the cw web site')
- self.assertEqual(entity.cwuri, 'http://www.cubicweb.org/')
- self.assertEqual(entity.cw_source[0].name, 'myfeed')
- self.assertEqual(entity.cw_metainformation(),
- {'type': 'Card',
- 'source': {'uri': 'myfeed', 'type': 'datafeed', 'use-cwuri-as-url': True},
- 'extid': 'http://www.cubicweb.org/'}
- )
- self.assertEqual(entity.absolute_url(), 'http://www.cubicweb.org/')
- # test repo cache keys
- self.assertEqual(self.repo._type_source_cache[entity.eid],
- ('Card', 'http://www.cubicweb.org/', 'myfeed'))
- self.assertEqual(self.repo._extid_cache['http://www.cubicweb.org/'],
- entity.eid)
- # test repull
- session.set_cnxset()
- stats = dfsource.pull_data(session, force=True)
- self.assertEqual(stats['created'], set())
- self.assertEqual(stats['updated'], set((entity.eid,)))
- # test repull with caches reseted
- self.repo._type_source_cache.clear()
- self.repo._extid_cache.clear()
- session.set_cnxset()
- stats = dfsource.pull_data(session, force=True)
- self.assertEqual(stats['created'], set())
- self.assertEqual(stats['updated'], set((entity.eid,)))
- self.assertEqual(self.repo._type_source_cache[entity.eid],
- ('Card', 'http://www.cubicweb.org/', 'myfeed'))
- self.assertEqual(self.repo._extid_cache['http://www.cubicweb.org/'],
- entity.eid)
+ with self.repo.internal_cnx() as cnx:
+ stats = dfsource.pull_data(cnx, force=True)
+ cnx.commit()
+ # test import stats
+ self.assertEqual(sorted(stats), ['checked', 'created', 'updated'])
+ self.assertEqual(len(stats['created']), 1)
+ entity = cnx.execute('Card X').get_entity(0, 0)
+ self.assertIn(entity.eid, stats['created'])
+ self.assertEqual(stats['updated'], set())
+ # test imported entities
+ self.assertEqual(entity.title, 'cubicweb.org')
+ self.assertEqual(entity.content, 'the cw web site')
+ self.assertEqual(entity.cwuri, 'http://www.cubicweb.org/')
+ self.assertEqual(entity.cw_source[0].name, 'myfeed')
+ self.assertEqual(entity.cw_metainformation(),
+ {'type': 'Card',
+ 'source': {'uri': 'myfeed', 'type': 'datafeed', 'use-cwuri-as-url': True},
+ 'extid': 'http://www.cubicweb.org/'}
+ )
+ self.assertEqual(entity.absolute_url(), 'http://www.cubicweb.org/')
+ # test repo cache keys
+ self.assertEqual(self.repo._type_source_cache[entity.eid],
+ ('Card', 'http://www.cubicweb.org/', 'myfeed'))
+ self.assertEqual(self.repo._extid_cache['http://www.cubicweb.org/'],
+ entity.eid)
+ # test repull
+ stats = dfsource.pull_data(cnx, force=True)
+ self.assertEqual(stats['created'], set())
+ self.assertEqual(stats['updated'], set((entity.eid,)))
+ # test repull with caches reseted
+ self.repo._type_source_cache.clear()
+ self.repo._extid_cache.clear()
+ stats = dfsource.pull_data(cnx, force=True)
+ self.assertEqual(stats['created'], set())
+ self.assertEqual(stats['updated'], set((entity.eid,)))
+ self.assertEqual(self.repo._type_source_cache[entity.eid],
+ ('Card', 'http://www.cubicweb.org/', 'myfeed'))
+ self.assertEqual(self.repo._extid_cache['http://www.cubicweb.org/'],
+ entity.eid)
self.assertEqual(dfsource.source_cwuris(self.session),
{'http://www.cubicweb.org/': (entity.eid, 'Card')}
@@ -97,28 +97,27 @@
self.assertTrue(dfsource.fresh())
# test_rename_source
- req = self.request()
- req.execute('SET S name "myrenamedfeed" WHERE S is CWSource, S name "myfeed"')
- self.commit()
- entity = self.execute('Card X').get_entity(0, 0)
- self.assertEqual(entity.cwuri, 'http://www.cubicweb.org/')
- self.assertEqual(entity.cw_source[0].name, 'myrenamedfeed')
- self.assertEqual(entity.cw_metainformation(),
- {'type': 'Card',
- 'source': {'uri': 'myrenamedfeed', 'type': 'datafeed', 'use-cwuri-as-url': True},
- 'extid': 'http://www.cubicweb.org/'}
- )
- self.assertEqual(self.repo._type_source_cache[entity.eid],
- ('Card', 'http://www.cubicweb.org/', 'myrenamedfeed'))
- self.assertEqual(self.repo._extid_cache['http://www.cubicweb.org/'],
- entity.eid)
+ with self.admin_access.repo_cnx() as cnx:
+ cnx.execute('SET S name "myrenamedfeed" WHERE S is CWSource, S name "myfeed"')
+ cnx.commit()
+ entity = cnx.execute('Card X').get_entity(0, 0)
+ self.assertEqual(entity.cwuri, 'http://www.cubicweb.org/')
+ self.assertEqual(entity.cw_source[0].name, 'myrenamedfeed')
+ self.assertEqual(entity.cw_metainformation(),
+ {'type': 'Card',
+ 'source': {'uri': 'myrenamedfeed', 'type': 'datafeed', 'use-cwuri-as-url': True},
+ 'extid': 'http://www.cubicweb.org/'}
+ )
+ self.assertEqual(self.repo._type_source_cache[entity.eid],
+ ('Card', 'http://www.cubicweb.org/', 'myrenamedfeed'))
+ self.assertEqual(self.repo._extid_cache['http://www.cubicweb.org/'],
+ entity.eid)
- # test_delete_source
- req = self.request()
- req.execute('DELETE CWSource S WHERE S name "myrenamedfeed"')
- self.commit()
- self.assertFalse(self.execute('Card X WHERE X title "cubicweb.org"'))
- self.assertFalse(self.execute('Any X WHERE X has_text "cubicweb.org"'))
+ # test_delete_source
+ cnx.execute('DELETE CWSource S WHERE S name "myrenamedfeed"')
+ cnx.commit()
+ self.assertFalse(cnx.execute('Card X WHERE X title "cubicweb.org"'))
+ self.assertFalse(cnx.execute('Any X WHERE X has_text "cubicweb.org"'))
if __name__ == '__main__':
from logilab.common.testlib import unittest_main