feat(cli): Warn before commands that desync a Frappe Cloud bench

bench/cli.py

@@ -28,6 +28,7 @@
 	get_cmd_from_sysargv,
 )
 from bench.utils.bench import get_env_cmd
+from bench.utils.prod_guard import confirm_if_managed
 from importlib.util import find_spec
 
 
@@ -118,6 +119,11 @@ def cli():
 			print(get_frappe_help())
 		return
 
+	# must run before both dispatch paths below: commands not registered with
+	# bench are handed to frappe via execv, after which there's nothing to ask
+	if is_cli_command:
+		confirm_if_managed(cmd_from_sys, bench_config)
+
 	_opts = [x.opts + x.secondary_opts for x in bench_command.params]
 	opts = {item for sublist in _opts for item in sublist}
 

bench/tests/test_prod_guard.py

@@ -0,0 +1,150 @@
+import unittest
+from unittest.mock import patch
+
+from bench.utils import prod_guard
+from bench.utils.prod_guard import (
+	MANAGED_CONFIG_KEY,
+	MANAGED_ENV_VAR,
+	SKIP_ENV_VAR,
+	confirm_if_managed,
+	get_guard,
+	is_managed_bench,
+)
+
+MANAGED = {MANAGED_CONFIG_KEY: True}
+
+
+class TestIsManagedBench(unittest.TestCase):
+	def setUp(self):
+		patcher = patch.dict("os.environ", {}, clear=True)
+		patcher.start()
+		self.addCleanup(patcher.stop)
+
+	def test_managed_via_config(self):
+		self.assertTrue(is_managed_bench(MANAGED))
+
+	def test_managed_via_env(self):
+		with patch.dict("os.environ", {MANAGED_ENV_VAR: "1"}):
+			self.assertTrue(is_managed_bench({}))
+
+	def test_unmanaged(self):
+		self.assertFalse(is_managed_bench({}))
+		self.assertFalse(is_managed_bench({"monitor": True}))
+		self.assertFalse(is_managed_bench({MANAGED_CONFIG_KEY: False}))
+
+
+class TestGuardReason(unittest.TestCase):
+	def test_guarded_commands(self):
+		for command in ("install-app", "drop-site", "set-config", "get-app", "update"):
+			with self.subTest(command=command):
+				self.assertIsNotNone(get_guard(command))
+
+	def test_unguarded_commands(self):
+		for command in ("backup", "clear-cache", "console", "mariadb", "doctor", "build"):
+			with self.subTest(command=command):
+				self.assertIsNone(get_guard(command))
+
+	def test_aliases_of_guarded_commands_are_guarded(self):
+		"""`bench get` and `bench rm` must not slip past `get-app`/`remove-app`."""
+		for alias in ("get", "remove", "rm"):
+			with self.subTest(alias=alias):
+				self.assertIsNotNone(get_guard(alias))
+
+	def test_no_alias_of_a_guarded_command_is_unguarded(self):
+		"""Catches a future alias added to a guarded command without a table entry."""
+		from collections import defaultdict
+
+		from bench.commands import bench_command
+
+		names_by_command = defaultdict(list)
+		for name, command in bench_command.commands.items():
+			names_by_command[id(command)].append(name)
+
+		for names in names_by_command.values():
+			if any(name in prod_guard.GUARDED_COMMANDS for name in names):
+				for name in names:
+					with self.subTest(name=name):
+						self.assertIsNotNone(get_guard(name))
+
+	def test_aliases_of_unguarded_commands_stay_unguarded(self):
+		self.assertIsNone(get_guard("src"))
+		self.assertIsNone(get_guard("find"))
+
+	def test_every_command_explains_its_desync(self):
+		for command, (reason, docs_url) in prod_guard.GUARDED_COMMANDS.items():
+			with self.subTest(command=command):
+				self.assertIsInstance(reason, str)
+				# implicit concatenation across wrapped lines drops spaces easily
+				self.assertEqual(reason, " ".join(reason.split()))
+				if docs_url is not None:
+					self.assertTrue(docs_url.startswith(f"{prod_guard.DOCS}/"))
+
+
+class TestConfirmIfManaged(unittest.TestCase):
+	"""The guard only prompts for a guarded command, on a managed bench, at a TTY."""
+
+	def setUp(self):
+		env = patch.dict("os.environ", {}, clear=True)
+		env.start()
+		self.addCleanup(env.stop)
+
+		argv = patch.object(prod_guard.sys, "argv", ["bench", "install-app", "erpnext"])
+		argv.start()
+		self.addCleanup(argv.stop)
+
+		tty = patch.object(prod_guard.sys.stdin, "isatty", return_value=True)
+		tty.start()
+		self.addCleanup(tty.stop)
+
+		self.confirm = patch.object(prod_guard.click, "confirm", return_value=False).start()
+		self.addCleanup(patch.stopall)
+
+	def assert_did_not_prompt(self, *args, **kwargs):
+		confirm_if_managed(*args, **kwargs)
+		self.confirm.assert_not_called()
+
+	def test_aborts_when_declined(self):
+		with self.assertRaises(SystemExit) as cm:
+			confirm_if_managed("install-app", MANAGED)
+
+		self.assertNotEqual(cm.exception.code, 0)
+		self.confirm.assert_called_once()
+
+	def test_proceeds_when_accepted(self):
+		self.confirm.return_value = True
+		confirm_if_managed("install-app", MANAGED)
+		self.confirm.assert_called_once()
+
+	def test_aborts_when_prompt_is_interrupted(self):
+		self.confirm.side_effect = prod_guard.click.Abort()
+		with self.assertRaises(SystemExit):
+			confirm_if_managed("install-app", MANAGED)
+
+	def test_silent_on_unmanaged_bench(self):
+		self.assert_did_not_prompt("install-app", {})
+
+	def test_silent_for_unguarded_command(self):
+		self.assert_did_not_prompt("backup", MANAGED)
+
+	def test_silent_without_a_command(self):
+		self.assert_did_not_prompt(None, MANAGED)
+
+	def test_silent_when_not_a_tty(self):
+		"""The agent drives benches over `docker exec` without a TTY."""
+		with patch.object(prod_guard.sys.stdin, "isatty", return_value=False):
+			self.assert_did_not_prompt("install-app", MANAGED)
+
+	def test_silent_when_bypass_is_set(self):
+		with patch.dict("os.environ", {SKIP_ENV_VAR: "1"}):
+			self.assert_did_not_prompt("install-app", MANAGED)
+
+	def test_silent_for_help(self):
+		for flag in ("--help", "-h"):
+			with self.subTest(flag=flag):
+				self.confirm.reset_mock()
+				with patch.object(prod_guard.sys, "argv", ["bench", "install-app", flag]):
+					self.assert_did_not_prompt("install-app", MANAGED)
+
+
+if __name__ == "__main__":
+	unittest.main()