[migration/pdb] add option to use pdb.post_mortem if traceback is provided
Post mortem is a mode where the pdb shell is opened **where** the exception as
occured instead at the breakpoint for set_trace. This is way more useful for
debugging for the user because is will have the full context of the error.
Closes #17219827
--- a/cubicweb/migration.py Wed May 22 17:08:09 2019 +0200
+++ b/cubicweb/migration.py Wed May 22 17:10:06 2019 +0200
@@ -200,7 +200,7 @@
return meth(*args, **kwargs)
def confirm(self, question, # pylint: disable=E0202
- shell=True, abort=True, retry=False, pdb=False, default='y'):
+ shell=True, abort=True, retry=False, pdb=False, default='y', traceback=None):
"""ask for confirmation and return true on positive answer
if `retry` is true the r[etry] answer may return 2
@@ -226,11 +226,14 @@
raise SystemExit(1)
if answer == 'shell':
self.interactive_shell()
- return self.confirm(question, shell, abort, retry, pdb, default)
+ return self.confirm(question, shell, abort, retry, pdb, default, traceback)
if answer == 'pdb':
pdb = utils.get_pdb()
- pdb.set_trace()
- return self.confirm(question, shell, abort, retry, pdb, default)
+ if traceback:
+ pdb.post_mortem(traceback)
+ else:
+ pdb.set_trace()
+ return self.confirm(question, shell, abort, retry, pdb, default, traceback)
return True
def interactive_shell(self):
--- a/cubicweb/server/migractions.py Wed May 22 17:08:09 2019 +0200
+++ b/cubicweb/server/migractions.py Wed May 22 17:10:06 2019 +0200
@@ -262,9 +262,10 @@
source = repo.system_source
try:
source.restore(osp.join(tmpdir, source.uri), self.confirm, drop, format)
- except Exception as exc:
+ except Exception:
+ _, exc, traceback_ = sys.exc_info()
print('-> error trying to restore %s [%s]' % (source.uri, exc))
- if not self.confirm('Continue anyway?', default='n'):
+ if not self.confirm('Continue anyway?', default='n', pdb=True, traceback=traceback_):
raise SystemExit(1)
finally:
shutil.rmtree(tmpdir)
@@ -1454,8 +1455,9 @@
try:
cu = self.cnx.system_sql(sql, args)
except Exception:
- ex = sys.exc_info()[1]
- if self.confirm('Error: %s\nabort?' % ex, pdb=True):
+ _, ex, traceback_ = sys.exc_info()
+ if self.confirm('Error: %s\nabort?' % ex,
+ pdb=True, traceback=traceback_):
raise
return
try:
@@ -1479,8 +1481,10 @@
if not ask_confirm or self.confirm('Execute rql: %s ?' % msg):
try:
res = execute(rql, kwargs, build_descr=build_descr)
- except Exception as ex:
- if self.confirm('Error: %s\nabort?' % ex, pdb=True):
+ except Exception:
+ _, ex, traceback_ = sys.exc_info()
+ if self.confirm('Error: %s\nabort?' % ex,
+ pdb=True, traceback=traceback_):
raise
return res
@@ -1568,8 +1572,10 @@
raise StopIteration
try:
return self._h._cw.execute(rql, kwargs)
- except Exception as ex:
- if self._h.confirm('Error: %s\nabort?' % ex):
+ except Exception:
+ _, ex, traceback_ = sys.exc_info()
+ if self._h.confirm('Error: %s\nabort?' % ex,
+ pdb=True, traceback=traceback_):
raise
else:
raise StopIteration
--- a/cubicweb/test/unittest_migration.py Wed May 22 17:08:09 2019 +0200
+++ b/cubicweb/test/unittest_migration.py Wed May 22 17:10:06 2019 +0200
@@ -18,14 +18,18 @@
"""cubicweb.migration unit tests"""
from os.path import dirname, join
+from unittest.mock import patch
+
from logilab.common.testlib import TestCase, unittest_main
-from cubicweb import devtools
+from cubicweb import devtools, utils
+from logilab.common.shellutils import ASK
from cubicweb.cwconfig import CubicWebConfiguration
from cubicweb.migration import (
filter_scripts,
split_constraint,
version_strictly_lower,
+ MigrationHelper,
)
@@ -128,5 +132,54 @@
assert split_constraint("<= 42.1.0") == ("<=", "42.1.0")
+class WontColideWithOtherExceptionsException(Exception):
+ pass
+
+
+class MigrationHelperTC(TestCase):
+ @patch.object(utils, 'get_pdb')
+ @patch.object(ASK, 'ask', return_value="pdb")
+ def test_confirm_no_traceback(self, ask, get_pdb):
+ post_mortem = get_pdb.return_value.post_mortem
+ set_trace = get_pdb.return_value.set_trace
+
+ # we need to break after post_mortem is called otherwise we get
+ # infinite recursion
+ set_trace.side_effect = WontColideWithOtherExceptionsException
+
+ mh = MigrationHelper(config=None)
+
+ with self.assertRaises(WontColideWithOtherExceptionsException):
+ mh.confirm("some question")
+
+ get_pdb.assert_called_once()
+ set_trace.assert_called_once()
+ post_mortem.assert_not_called()
+
+ @patch.object(utils, 'get_pdb')
+ @patch.object(ASK, 'ask', return_value="pdb")
+ def test_confirm_got_traceback(self, ask, get_pdb):
+ post_mortem = get_pdb.return_value.post_mortem
+ set_trace = get_pdb.return_value.set_trace
+
+ # we need to break after post_mortem is called otherwise we get
+ # infinite recursion
+ post_mortem.side_effect = WontColideWithOtherExceptionsException
+
+ mh = MigrationHelper(config=None)
+
+ fake_traceback = object()
+
+ with self.assertRaises(WontColideWithOtherExceptionsException):
+ mh.confirm("some question", traceback=fake_traceback)
+
+ get_pdb.assert_called_once()
+ set_trace.assert_not_called()
+ post_mortem.assert_called_once()
+
+ # we want post_mortem to actually receive the traceback
+ self.assertEqual(post_mortem.call_args, ((fake_traceback,),))
+
+
if __name__ == '__main__':
unittest_main()