124 REMOVE THE DATABASE TEMPLATE else it won't be considered |
127 REMOVE THE DATABASE TEMPLATE else it won't be considered |
125 """ |
128 """ |
126 test_db_id = 'xmlparser' |
129 test_db_id = 'xmlparser' |
127 @classmethod |
130 @classmethod |
128 def pre_setup_database(cls, session, config): |
131 def pre_setup_database(cls, session, config): |
129 source = session.create_entity('CWSource', name=u'myfeed', type=u'datafeed', |
132 myfeed = session.create_entity('CWSource', name=u'myfeed', type=u'datafeed', |
130 parser=u'cw.entityxml', url=BASEXML) |
133 parser=u'cw.entityxml', url=BASEXML) |
131 session.create_entity('CWSource', name=u'myotherfeed', type=u'datafeed', |
134 myotherfeed = session.create_entity('CWSource', name=u'myotherfeed', type=u'datafeed', |
132 parser=u'cw.entityxml', url=OTHERXML) |
135 parser=u'cw.entityxml', url=OTHERXML) |
133 session.commit() |
136 session.commit() |
134 source.init_mapping([(('CWUser', 'use_email', '*'), |
137 myfeed.init_mapping([(('CWUser', 'use_email', '*'), |
135 u'role=subject\naction=copy'), |
138 u'role=subject\naction=copy'), |
136 (('CWUser', 'in_group', '*'), |
139 (('CWUser', 'in_group', '*'), |
137 u'role=subject\naction=link\nlinkattr=name'), |
140 u'role=subject\naction=link\nlinkattr=name'), |
138 (('CWUser', 'in_state', '*'), |
141 (('CWUser', 'in_state', '*'), |
139 u'role=subject\naction=link\nlinkattr=name'), |
142 u'role=subject\naction=link\nlinkattr=name'), |
140 (('*', 'tags', 'CWUser'), |
143 (('*', 'tags', 'CWUser'), |
141 u'role=object\naction=link-or-create\nlinkattr=name'), |
144 u'role=object\naction=link-or-create\nlinkattr=name'), |
142 ]) |
145 ]) |
|
146 myotherfeed.init_mapping([(('CWUser', 'in_group', '*'), |
|
147 u'role=subject\naction=link\nlinkattr=name'), |
|
148 (('CWUser', 'in_state', '*'), |
|
149 u'role=subject\naction=link\nlinkattr=name'), |
|
150 ]) |
143 session.create_entity('Tag', name=u'hop') |
151 session.create_entity('Tag', name=u'hop') |
144 |
152 |
145 def test_complete_url(self): |
153 def test_complete_url(self): |
146 dfsource = self.repo.sources_by_uri['myfeed'] |
154 dfsource = self.repo.sources_by_uri['myfeed'] |
147 parser = dfsource._get_parser(self.session) |
155 parser = dfsource._get_parser(self.session) |
169 (u'Tag', {u'linkattr': u'name'})], |
177 (u'Tag', {u'linkattr': u'name'})], |
170 (u'use_email', u'subject', u'copy'): [ |
178 (u'use_email', u'subject', u'copy'): [ |
171 (u'EmailAddress', {})] |
179 (u'EmailAddress', {})] |
172 } |
180 } |
173 }) |
181 }) |
174 session = self.repo.internal_session() |
182 session = self.repo.internal_session(safe=True) |
175 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
183 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
176 self.assertEqual(sorted(stats.keys()), ['created', 'updated']) |
184 self.assertEqual(sorted(stats.keys()), ['created', 'updated']) |
177 self.assertEqual(len(stats['created']), 2) |
185 self.assertEqual(len(stats['created']), 2) |
178 self.assertEqual(stats['updated'], set()) |
186 self.assertEqual(stats['updated'], set()) |
179 |
187 |
253 rset = self.sexecute('Any X WHERE X use_email E, X login "sthenault"') |
261 rset = self.sexecute('Any X WHERE X use_email E, X login "sthenault"') |
254 self.assertEqual(len(rset), 0) |
262 self.assertEqual(len(rset), 0) |
255 |
263 |
256 def test_external_entity(self): |
264 def test_external_entity(self): |
257 dfsource = self.repo.sources_by_uri['myotherfeed'] |
265 dfsource = self.repo.sources_by_uri['myotherfeed'] |
258 session = self.repo.internal_session() |
266 session = self.repo.internal_session(safe=True) |
259 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
267 stats = dfsource.pull_data(session, force=True, raise_on_error=True) |
260 user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) |
268 user = self.execute('CWUser X WHERE X login "sthenault"').get_entity(0, 0) |
261 self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) |
269 self.assertEqual(user.creation_date, datetime(2010, 01, 22, 10, 27, 59)) |
262 self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06)) |
270 self.assertEqual(user.modification_date, datetime(2011, 01, 25, 14, 14, 06)) |
263 self.assertEqual(user.cwuri, 'http://pouet.org/5') |
271 self.assertEqual(user.cwuri, 'http://pouet.org/5') |
264 self.assertEqual(user.cw_source[0].name, 'myfeed') |
272 self.assertEqual(user.cw_source[0].name, 'myfeed') |
265 |
273 |
266 def test_noerror_missing_fti_attribute(self): |
274 def test_noerror_missing_fti_attribute(self): |
267 dfsource = self.repo.sources_by_uri['myfeed'] |
275 dfsource = self.repo.sources_by_uri['myfeed'] |
268 session = self.repo.internal_session() |
276 session = self.repo.internal_session(safe=True) |
269 parser = dfsource._get_parser(session) |
277 parser = dfsource._get_parser(session) |
270 dfsource.process_urls(parser, [''' |
278 dfsource.process_urls(parser, [''' |
271 <rset size="1"> |
279 <rset size="1"> |
272 <Card eid="50" cwuri="http://pouet.org/50" cwsource="system"> |
280 <Card eid="50" cwuri="http://pouet.org/50" cwsource="system"> |
273 <title>how-to</title> |
281 <title>how-to</title> |
275 </rset> |
283 </rset> |
276 '''], raise_on_error=True) |
284 '''], raise_on_error=True) |
277 |
285 |
278 def test_noerror_unspecified_date(self): |
286 def test_noerror_unspecified_date(self): |
279 dfsource = self.repo.sources_by_uri['myfeed'] |
287 dfsource = self.repo.sources_by_uri['myfeed'] |
280 session = self.repo.internal_session() |
288 session = self.repo.internal_session(safe=True) |
281 parser = dfsource._get_parser(session) |
289 parser = dfsource._get_parser(session) |
282 dfsource.process_urls(parser, [''' |
290 dfsource.process_urls(parser, [''' |
283 <rset size="1"> |
291 <rset size="1"> |
284 <Card eid="50" cwuri="http://pouet.org/50" cwsource="system"> |
292 <Card eid="50" cwuri="http://pouet.org/50" cwsource="system"> |
285 <title>how-to</title> |
293 <title>how-to</title> |