Commit 61817ff2 authored by Laurent Peuch's avatar Laurent Peuch
Browse files

[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
parent 19aef4729d45
......@@ -200,7 +200,7 @@ class MigrationHelper(object):
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 @@ class MigrationHelper(object):
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):
......
......@@ -262,9 +262,10 @@ class ServerMigrationHelper(MigrationHelper):
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 @@ class ServerMigrationHelper(MigrationHelper):
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 @@ class ServerMigrationHelper(MigrationHelper):
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 @@ class ForRqlIterator:
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
......
......@@ -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 @@ def test_split_constraint():
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()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment