Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 31 additions & 14 deletions netplan_cli/cli/commands/try_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,10 +82,25 @@ def run(self): # pragma: nocover (requires user input)
self.parse_args()
self.run_command()

def _safe_revert(self, reason: str | None = None) -> int:
"""Attempt to revert configuration. Returns 0 on successful revert, 1 on revert failure."""
if reason:
print(f"\n{reason}")
print("\nReverting.")
try:
self.revert()
return 0
except Exception as e:
print(f"\nAn error occurred while reverting: {e}")
print("\nPlease check the configuration state")
return 1

def command_try(self): # pragma: nocover (requires user input)
if not self.is_revertable():
sys.exit(os.EX_CONFIG)

# 0 - user action success, 1 - failed
exit_code = 0
try:
fd = sys.stdin.fileno()
self.t = terminal.Terminal(fd)
Expand All @@ -105,19 +120,17 @@ def command_try(self): # pragma: nocover (requires user input)
self.touch_ready_stamp()
self.t.get_confirmation_input(timeout=self.timeout)
except terminal.InputRejected:
print("\nReverting.")
self.revert()
exit_code = self._safe_revert()
except terminal.InputAccepted:
print("\nConfiguration accepted.")
except Exception as e:
print("\nAn error occurred: %s" % e)
print("\nReverting.")
self.revert()
exit_code = self._safe_revert(f"An error occurred: {e}")
finally:
if self.t:
self.t.reset(self.t_settings)
self.cleanup()
self.clear_ready_stamp()
sys.exit(exit_code)

def backup(self): # pragma: nocover (requires user input)
backup_config_dir = False
Expand All @@ -135,16 +148,20 @@ def setup(self): # pragma: nocover (requires user input)
self.configuration_changed = True

def revert(self): # pragma: nocover (requires user input)
# backup the state we just tried to apply
tempdir = tempfile.mkdtemp()
confdir = os.path.join(tempdir, 'etc', 'netplan')
os.makedirs(confdir)
self.config_manager.copy_tree('/etc/netplan', confdir, dirs_exist_ok=True)
# restore previous state
self.config_manager.revert()
NetplanApply().command_apply(run_generate=False, sync=True, exit_on_error=False, state_dir=tempdir)
# clear the backup
shutil.rmtree(tempdir)
try:
# backup the state we just tried to apply
confdir = os.path.join(tempdir, 'etc', 'netplan')
os.makedirs(confdir)
self.config_manager.copy_tree('/etc/netplan', confdir, dirs_exist_ok=True)
# restore previous state
self.config_manager.revert()
NetplanApply().command_apply(run_generate=False, sync=True, exit_on_error=False, state_dir=tempdir)
finally:
try:
shutil.rmtree(tempdir)
except Exception as e:
logging.warning("Failed to remove temporary directory %s: %s", tempdir, e)
Comment thread
benhoyt marked this conversation as resolved.

def cleanup(self): # pragma: nocover (requires user input)
self.config_manager.cleanup()
Expand Down
16 changes: 16 additions & 0 deletions tests/cli/test_units.py
Original file line number Diff line number Diff line change
Expand Up @@ -240,3 +240,19 @@ def test_get_nm_interfaces_permission_error_raises(self, mock_nm):
NetplanApply._get_nm_interfaces([file_path],
['eth0'], exit_on_error=False)
self.assertTrue(any(os.strerror(errno.EACCES) in msg for msg in ctx.output))

def test_safe_revert_success(self):
"""_safe_revert returns 0 when revert() succeeds."""
cmd = NetplanTry()
with patch.object(cmd, 'revert') as mock_revert:
self.assertEqual(cmd._safe_revert("test reason"), 0)
mock_revert.assert_called_once()

def test_safe_revert_failure(self):
"""_safe_revert must catch exceptions from revert()
(e.g., insufficient privileges) and return 1 instead of letting
the unhandled exception propagate as an apport crash report)."""
cmd = NetplanTry()
with patch.object(cmd, 'revert',
side_effect=FileNotFoundError(2, 'No such file', '/etc/netplan')):
self.assertEqual(cmd._safe_revert("test reason"), 1)
Loading