cubicweb/server/test/unittest_serverctl.py
author Denis Laxalde <denis.laxalde@logilab.fr>
Mon, 20 Feb 2017 15:56:07 +0100
changeset 11962 36851c8b6763
parent 11355 47b0b08fbb4b
child 12002 26453d9467f6
permissions -rw-r--r--
[server] Remove pseudo-handling of exceptions in Repository._delete_cascade_multi() It seems to me that we never want to pass on an exception here be it in "test" mode or not, so remove the last `except Exception:` clause. For instance, the actual implementation would pass on an IntegrityError and thus possibly hide it in a log message. Concerning the `except Unauthorized:`, as said by the log message this should not happen because we are within a `security_enabled(write=False)` context. So remove it as well. As far as I can tell, this strange handling of exceptions dates from 37668bf302f5 and there is no clear justification to it.

import os.path as osp
import shutil

from cubicweb import ExecutionError
from cubicweb.devtools import testlib, ApptestConfiguration
from cubicweb.server.serverctl import _local_dump, DBDumpCommand, SynchronizeSourceCommand
from cubicweb.server.serverconfig import ServerConfiguration

class ServerCTLTC(testlib.CubicWebTC):
    def setUp(self):
        super(ServerCTLTC, self).setUp()
        self.orig_config_for = ServerConfiguration.config_for
        config_for = lambda appid: ApptestConfiguration(appid, __file__)
        ServerConfiguration.config_for = staticmethod(config_for)

    def tearDown(self):
        ServerConfiguration.config_for = self.orig_config_for
        super(ServerCTLTC, self).tearDown()

    def test_dump(self):
        DBDumpCommand(None).run([self.appid])
        shutil.rmtree(osp.join(self.config.apphome, 'backup'))

    def test_source_sync(self):
        with self.admin_access.repo_cnx() as cnx:
            cnx.create_entity('CWSource', name=u'success_feed', type=u'datafeed',
                              parser=u'test_source_parser_success',
                              url=u'ignored')
            cnx.create_entity('CWSource', name=u'fail_feed', type=u'datafeed',
                              parser=u'test_source_parser_fail',
                              url=u'ignored')
            cnx.commit()

            cmd = SynchronizeSourceCommand(None)
            cmd.config.force = 1

            # Should sync all sources even if one failed
            with self.assertRaises(ExecutionError) as exc:
                cmd.run([self.appid])
            self.assertEqual(len(cnx.find('Card', title=u'success')), 1)
            self.assertEqual(len(cnx.find('Card', title=u'fail')), 0)
            self.assertEqual(str(exc.exception), 'All sources where not synced')

            # call with named sources
            cmd.run([self.appid, u'success_feed'])
            self.assertEqual(len(cnx.find('Card', title=u'success')), 2)

            with self.assertRaises(ExecutionError) as exc:
                cmd.run([self.appid, u'fail_feed'])
            self.assertEqual(str(exc.exception), 'All sources where not synced')
            self.assertEqual(len(cnx.find('Card', title=u'fail')), 0)


if __name__ == '__main__':
    from unittest import main
    main()