1 import os.path as osp |
1 import os.path as osp |
2 import shutil |
2 import shutil |
3 |
3 |
|
4 from cubicweb import ExecutionError |
4 from cubicweb.devtools import testlib, ApptestConfiguration |
5 from cubicweb.devtools import testlib, ApptestConfiguration |
5 from cubicweb.server.serverctl import _local_dump, DBDumpCommand |
6 from cubicweb.server.serverctl import _local_dump, DBDumpCommand, SynchronizeSourceCommand |
6 from cubicweb.server.serverconfig import ServerConfiguration |
7 from cubicweb.server.serverconfig import ServerConfiguration |
7 |
8 |
8 class ServerCTLTC(testlib.CubicWebTC): |
9 class ServerCTLTC(testlib.CubicWebTC): |
9 def setUp(self): |
10 def setUp(self): |
10 super(ServerCTLTC, self).setUp() |
11 super(ServerCTLTC, self).setUp() |
18 |
19 |
19 def test_dump(self): |
20 def test_dump(self): |
20 DBDumpCommand(None).run([self.appid]) |
21 DBDumpCommand(None).run([self.appid]) |
21 shutil.rmtree(osp.join(self.config.apphome, 'backup')) |
22 shutil.rmtree(osp.join(self.config.apphome, 'backup')) |
22 |
23 |
|
24 def test_source_sync(self): |
|
25 with self.admin_access.repo_cnx() as cnx: |
|
26 cnx.create_entity('CWSource', name=u'success_feed', type=u'datafeed', |
|
27 parser=u'test_source_parser_success', |
|
28 url=u'ignored') |
|
29 cnx.create_entity('CWSource', name=u'fail_feed', type=u'datafeed', |
|
30 parser=u'test_source_parser_fail', |
|
31 url=u'ignored') |
|
32 cnx.commit() |
|
33 |
|
34 cmd = SynchronizeSourceCommand(None) |
|
35 cmd.config.force = 1 |
|
36 |
|
37 # Should sync all sources even if one failed |
|
38 with self.assertRaises(ExecutionError) as exc: |
|
39 cmd.run([self.appid]) |
|
40 self.assertEqual(len(cnx.find('Card', title=u'success')), 1) |
|
41 self.assertEqual(len(cnx.find('Card', title=u'fail')), 0) |
|
42 self.assertEqual(str(exc.exception), 'All sources where not synced') |
|
43 |
|
44 # call with named sources |
|
45 cmd.run([self.appid, u'success_feed']) |
|
46 self.assertEqual(len(cnx.find('Card', title=u'success')), 2) |
|
47 |
|
48 with self.assertRaises(ExecutionError) as exc: |
|
49 cmd.run([self.appid, u'fail_feed']) |
|
50 self.assertEqual(str(exc.exception), 'All sources where not synced') |
|
51 self.assertEqual(len(cnx.find('Card', title=u'fail')), 0) |
|
52 |
23 |
53 |
24 if __name__ == '__main__': |
54 if __name__ == '__main__': |
25 from unittest import main |
55 from unittest import main |
26 main() |
56 main() |