From ec134d929995223a2e695eaeb4a0c9b7a183b4e4 Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Tue, 10 Mar 2026 03:59:35 -0400 Subject: [PATCH] [REF] script: extract managers from todo into modules Reduce complexity of todo.py by extracting database, kdbx, and version logic into dedicated manager classes. Update tests to reference new module paths. Generated by Claude Code 2.1.72 model claude-sonnet-4-6 Co-Authored-By: Mathieu Benoit --- script/todo/database_manager.py | 231 ++++++++++++++++++++++++++++++++ script/todo/kdbx_manager.py | 100 ++++++++++++++ script/todo/todo.py | 131 +++--------------- script/todo/version_manager.py | 47 +++++++ test/test_todo.py | 106 ++++++++++++--- 5 files changed, 484 insertions(+), 131 deletions(-) create mode 100644 script/todo/database_manager.py create mode 100644 script/todo/kdbx_manager.py create mode 100644 script/todo/version_manager.py diff --git a/script/todo/database_manager.py b/script/todo/database_manager.py new file mode 100644 index 0000000..4159d07 --- /dev/null +++ b/script/todo/database_manager.py @@ -0,0 +1,231 @@ +#!/usr/bin/env python3 +# © 2021-2025 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +import datetime +import getpass +import logging +import os +import zipfile + +import click + +from script.todo.todo_i18n import t + +_logger = logging.getLogger(__name__) + +try: + from script.todo import todo_file_browser +except Exception: + todo_file_browser = None + + +class DatabaseManager: + def __init__(self, execute, fill_help_info): + self._execute = execute + self._fill_help_info = fill_help_info + self._dir_path = None + + def _on_dir_selected(self, path): + self._dir_path = path + + def select_database(self): + cmd_server = "./odoo_bin.sh db --list" + status, databases = self._execute.exec_command_live( + cmd_server, + return_status_and_output=True, + source_erplibre=False, + single_source_erplibre=True, + ) + choices = [{"prompt_description": a.strip()} for a in databases] + help_info = self._fill_help_info(choices) + valid_choices = [ + str(a) for a in range(len(choices) + 1) if a + ] + + while True: + status = click.prompt(help_info) + print() + if status == "0": + return False + elif status in valid_choices: + database_name = databases[int(status) - 1].strip() + print(database_name) + return database_name + else: + print(t("cmd_not_found")) + + def restore_from_database(self, show_remote_list=True): + path_image_db = os.path.join(os.getcwd(), "image_db") + print("[1] By filename from image_db") + print(f"[] Browser image_db {path_image_db}") + status = input("\U0001f4ac Select : ") + if status == "1": + file_name = status + else: + file_name = self.open_file_image_db() + + default_database_name = file_name.replace(" ", "_") + if default_database_name.endswith(".zip"): + default_database_name = default_database_name[:-4] + + database_name = input( + f"\U0001f4ac Database name (default={default_database_name}) : " + ) + if not database_name: + database_name = default_database_name + + status = ( + input( + "\U0001f4ac Would you like to neutralize database (n/N)? " + ) + .strip() + .lower() + ) + is_neutralize = False + more_arg = "" + if status != "n": + more_arg = "--neutralize " + is_neutralize = True + database_name += "_neutralize" + status, output_lines = self._execute.exec_command_live( + f"python3 ./script/database/db_restore.py -d {database_name} " + f"{more_arg}--ignore_cache --image {file_name}", + return_status_and_output=True, + single_source_erplibre=True, + source_erplibre=False, + ) + if is_neutralize: + status, output_lines = self._execute.exec_command_live( + f"./script/addons/update_prod_to_dev.sh {database_name}", + return_status_and_output=True, + single_source_erplibre=True, + source_erplibre=False, + ) + status = ( + input( + "\U0001f4ac Would you like to update all addons (y/Y)? " + ) + .strip() + .lower() + ) + if status == "y": + status, output_lines = self._execute.exec_command_live( + f"./script/addons/update_addons_all.sh {database_name}", + return_status_and_output=True, + single_source_erplibre=True, + source_erplibre=False, + ) + + def create_backup_from_database(self, show_remote_list=True): + database_name = self.select_database() + backup_name = input( + "\U0001f4ac Backup name (default = name+date.zip) : " + ) + if not backup_name: + backup_name = ( + database_name + + "_" + + datetime.datetime.now().strftime( + "%Y-%m-%d_%Hh%Mm%Ss" + ) + + ".zip" + ) + + if not backup_name.endswith(".zip"): + backup_name = backup_name + ".zip" + + print(backup_name) + + cmd = ( + f"./odoo_bin.sh db --backup --database {database_name}" + f" --restore_image {backup_name}" + ) + status, output_lines = self._execute.exec_command_live( + cmd, + return_status_and_output=True, + single_source_erplibre=True, + source_erplibre=False, + ) + + def open_file_image_db(self): + self._dir_path = "" + path_image_db = os.path.join(os.getcwd(), "image_db") + + file_browser = todo_file_browser.FileBrowser( + path_image_db, self._on_dir_selected + ) + file_browser.run_main_frame() + file_name = os.path.basename(self._dir_path) + print(file_name) + return file_name + + def download_database_backup_cli(self, show_remote_list=True): + database_domain = input( + "Domain Odoo (ex. https://mondomain.com) : " + ) + if show_remote_list: + status, output_lines = self._execute.exec_command_live( + f"python3 ./script/database/list_remote.py --raw" + f" --odoo-url {database_domain}", + return_status_and_output=True, + single_source_erplibre=True, + source_erplibre=False, + ) + if len(output_lines) > 1: + for index, output in enumerate(output_lines): + print(f"{index + 1} - {output}") + database_name = input( + "Select id of database :" + ).strip() + elif len(output_lines) == 1: + database_name = output_lines[0].strip() + else: + database_name = input( + "Cannot read remote database, Database name :\n" + ) + else: + database_name = input("Database name :\n") + + timestamp = datetime.datetime.now().strftime( + "%Y-%m-%d_%Hh%Mm%Ss" + ) + default_output_path = ( + f"./image_db/{database_name}_{timestamp}.zip" + ) + output_path = input( + f"Output path (default: {default_output_path}) : " + ).strip() + if not output_path: + output_path = default_output_path + + master_password = getpass.getpass(prompt="Master password : ") + + cmd = "script/database/download_remote.sh --quiet" + my_env = os.environ.copy() + my_env["MASTER_PWD"] = master_password + my_env["DATABASE_NAME"] = database_name + my_env["OUTPUT_FILE_PATH"] = output_path + my_env["ODOO_URL"] = database_domain + status, cmd_executed = self._execute.exec_command_live( + cmd, + source_erplibre=False, + return_status_and_command=True, + new_env=my_env, + ) + try: + with zipfile.ZipFile( + default_output_path, "r" + ) as zip_ref: + manifest_file_1 = zip_ref.open("manifest.json") + _logger.info( + f"Log file '{default_output_path}' is complete" + " and validated." + ) + except Exception as e: + _logger.error(e) + _logger.error( + "Failed to read manifest.json from backup file" + f" '{default_output_path}'." + ) + return status, output_path, database_name diff --git a/script/todo/kdbx_manager.py b/script/todo/kdbx_manager.py new file mode 100644 index 0000000..21b9035 --- /dev/null +++ b/script/todo/kdbx_manager.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +# © 2021-2025 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +import getpass +import logging + +from script.todo.todo_i18n import t + +_logger = logging.getLogger(__name__) + +try: + import tkinter as tk + from tkinter import filedialog + + from pykeepass import PyKeePass +except ModuleNotFoundError: + PyKeePass = None + tk = None + filedialog = None + + +class KdbxManager: + def __init__(self, config_file): + self._config_file = config_file + self._kdbx = None + + def get_kdbx(self): + if self._kdbx: + return self._kdbx + + kdbx_file_path = self._config_file.get_config_value( + ["kdbx", "path"] + ) + if not kdbx_file_path: + if tk is None: + _logger.error("tkinter is not available") + return None + root = tk.Tk() + root.withdraw() + kdbx_file_path = filedialog.askopenfilename( + title="Select a File", + filetypes=(("KeepassX files", "*.kdbx"),), + ) + if not kdbx_file_path: + _logger.error( + "KDBX is not configured, please fill" + f" {self._config_file.CONFIG_FILE}" + ) + return None + + kdbx_password = self._config_file.get_config_value( + ["kdbx", "password"] + ) + if not kdbx_password: + kdbx_password = getpass.getpass(prompt=t("enter_password")) + + if PyKeePass is None: + _logger.error("pykeepass is not installed") + return None + + kp = PyKeePass(kdbx_file_path, password=kdbx_password) + if kp: + self._kdbx = kp + return kp + + def get_extra_command_user(self, kdbx_key): + values = [] + if kdbx_key: + kp = self.get_kdbx() + if not kp: + return "" + if type(kdbx_key) is not list: + kdbx_keys = [kdbx_key] + else: + kdbx_keys = kdbx_key + for key in kdbx_keys: + entry = kp.find_entries_by_title(key, first=True) + try: + odoo_user = entry.username + except AttributeError: + _logger.error( + f"Cannot find username from keys {key}" + ) + try: + odoo_password = entry.password + except AttributeError: + _logger.error( + f"Cannot find password from keys {key}" + ) + values.append( + " --default_email_auth" + f" {odoo_user} --default_password_auth" + f" '{odoo_password}'" + ) + if len(values) == 0: + return "" + elif len(values) == 1: + return values[0] + return values diff --git a/script/todo/todo.py b/script/todo/todo.py index 28339c0..79067eb 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -22,15 +22,13 @@ sys.path.append(new_path) from script.config import config_file from script.execute import execute +from script.todo.database_manager import DatabaseManager +from script.todo.kdbx_manager import KdbxManager from script.todo.todo_i18n import get_lang, lang_is_configured, set_lang, t +from script.todo.version_manager import get_odoo_version ERROR_LOG_PATH = ".erplibre.error.txt" VENV_ERPLIBRE = ".venv.erplibre" -VERSION_DATA_FILE = os.path.join("conf", "supported_version_erplibre.json") -INSTALLED_ODOO_VERSION_FILE = os.path.join( - ".repo", "installed_odoo_version.txt" -) -ODOO_VERSION_FILE = ".odoo-version" ENABLE_CRASH = False CRASH_E = None # Support mobile ERPLibre @@ -83,10 +81,13 @@ LOGO_ASCII_FILE = "./script/todo/logo_ascii.txt" class TODO: def __init__(self): self.dir_path = None - self.kdbx = None self.selected_file_path = None self.config_file = config_file.ConfigFile() self.execute = execute.Execute() + self.kdbx_manager = KdbxManager(self.config_file) + self.db_manager = DatabaseManager( + self.execute, self.fill_help_info + ) def _ask_language(self): if not lang_is_configured(): @@ -169,38 +170,6 @@ class TODO: print(status) # manipuler() - def get_kdbx(self): - if self.kdbx: - return self.kdbx - # Open file - kdbx_file_path = self.config_file.get_config_value( - ["kdbx", "path"] - ) - if not kdbx_file_path: - root = tk.Tk() - root.withdraw() # Hide the main window - kdbx_file_path = filedialog.askopenfilename( - title="Select a File", - filetypes=(("KeepassX files", "*.kdbx"),), - ) - if not kdbx_file_path: - _logger.error( - f"KDBX is not configured, please fill {self.config_file.CONFIG_FILE}" - ) - return - - kdbx_password = self.config_file.get_config_value( - ["kdbx", "password"] - ) - if not kdbx_password: - kdbx_password = getpass.getpass(prompt=t("enter_password")) - - kp = PyKeePass(kdbx_file_path, password=kdbx_password) - - if kp: - self.kdbx = kp - return kp - def execute_prompt_ia(self): while True: help_info = f"""{t("command")} @@ -210,7 +179,7 @@ class TODO: print() if status == "0": return - kp = self.get_kdbx() + kp = self.kdbx_manager.get_kdbx() if not kp: return config_name = self.config_file.get_config_value( @@ -393,7 +362,7 @@ class TODO: } commands_end = {} versions, installed_versions, odoo_installed_version = ( - self.get_odoo_version() + get_odoo_version() ) for version_info in versions[::-1]: @@ -462,7 +431,7 @@ class TODO: odoo_password = instance.get("password") if kdbx_key: - extra_cmd_web_login = self.kdbx_get_extra_command_user(kdbx_key) + extra_cmd_web_login = self.kdbx_manager.get_extra_command_user(kdbx_key) elif odoo_user and odoo_password: extra_cmd_web_login = ( f" --default_email_auth {odoo_user} --default_password_auth" @@ -925,11 +894,11 @@ class TODO: if status == "0": return False elif status == "1": - self.download_database_backup_cli() + self.db_manager.download_database_backup_cli() elif status == "2": - self.restore_from_database() + self.db_manager.restore_from_database() elif status == "3": - self.create_backup_from_database() + self.db_manager.create_backup_from_database() else: print(t("cmd_not_found")) @@ -1181,7 +1150,7 @@ class TODO: def execute_pip_audit(self): versions, installed_versions, odoo_installed_version = ( - self.get_odoo_version() + get_odoo_version() ) # Build list of installed environments @@ -1322,12 +1291,12 @@ class TODO: print(t("cmd_not_found")) def generate_config_from_backup(self): - file_name = self.open_file_image_db() + file_name = self.db_manager.open_file_image_db() add_arg = f"--from_backup_name {file_name} --add_repo odoo18.0/addons/MathBenTech_development" self.generate_config(add_arg=add_arg) def generate_config_from_database(self): - database_name = self.select_database() + database_name = self.db_manager.select_database() str_arg = f"--database {database_name}" self.generate_config(add_arg=str_arg) return False @@ -1366,66 +1335,6 @@ class TODO: else: print(t("cmd_not_found")) - def get_odoo_version(self): - with open(VERSION_DATA_FILE) as txt: - data_version = json.load(txt) - - if not data_version: - raise Exception( - f"Internal error, no Odoo version is supported, please valide file '{VERSION_DATA_FILE}'" - ) - version_entries = [] - for key, value in data_version.items(): - version_entries.append(value) - value["erplibre_version"] = key - - installed_versions = [] - if os.path.exists(INSTALLED_ODOO_VERSION_FILE): - with open(INSTALLED_ODOO_VERSION_FILE) as txt: - installed_versions = sorted(txt.read().splitlines()) - - odoo_installed_version = None - if os.path.exists(ODOO_VERSION_FILE): - with open(ODOO_VERSION_FILE) as txt: - odoo_installed_version = f"odoo{txt.read().strip()}" - - # Add odoo version installation on command - versions = sorted( - version_entries, key=lambda k: k.get("erplibre_version") - ) - - return versions, installed_versions, odoo_installed_version - - def kdbx_get_extra_command_user(self, kdbx_key): - values = [] - if kdbx_key: - kp = self.get_kdbx() - if not kp: - return "" - if type(kdbx_key) is not list: - kdbx_keys = [kdbx_key] - else: - kdbx_keys = kdbx_key - for key in kdbx_keys: - entry = kp.find_entries_by_title(key, first=True) - try: - odoo_user = entry.username - except AttributeError: - _logger.error(f"Cannot find username from keys {key}") - try: - odoo_password = entry.password - except AttributeError: - _logger.error(f"Cannot find password from keys {key}") - values.append( - " --default_email_auth" - f" {odoo_user} --default_password_auth '{odoo_password}'" - ) - if len(values) == 0: - return "" - elif len(values) == 1: - return values[0] - return values - def prompt_execute_selenium_and_run_db(self, db_name, extra_cmd_web_login=""): cmd_server = f"./run.sh -d {db_name};bash" self.execute.exec_command_live(cmd_server) @@ -1517,7 +1426,7 @@ class TODO: self.prompt_install() def open_shell_on_database(self): - database = self.select_database() + database = self.db_manager.select_database() if database: cmd_server = f"./odoo_bin.sh shell -d {database}" status, databases = self.execute.exec_command_live( @@ -1588,7 +1497,7 @@ class TODO: shutil.copy2(poetry_lock, path_file_odoo_lock) def callback_execute_custom_database(self, config): - database_name = self.select_database() + database_name = self.db_manager.select_database() self.prompt_execute_selenium_and_run_db(database_name) def restore_from_database(self, show_remote_list=True): @@ -1599,7 +1508,7 @@ class TODO: if status == "1": file_name = status else: - file_name = self.open_file_image_db() + file_name = self.db_manager.open_file_image_db() default_database_name = file_name.replace(" ", "_") if default_database_name.endswith(".zip"): @@ -1649,7 +1558,7 @@ class TODO: ) def create_backup_from_database(self, show_remote_list=True): - database_name = self.select_database() + database_name = self.db_manager.select_database() str_arg = f"--database {database_name}" backup_name = input("💬 Backup name (default = name+date.zip) : ") diff --git a/script/todo/version_manager.py b/script/todo/version_manager.py new file mode 100644 index 0000000..0640e7e --- /dev/null +++ b/script/todo/version_manager.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +# © 2021-2025 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +import json +import os + +VERSION_DATA_FILE = os.path.join("conf", "supported_version_erplibre.json") +INSTALLED_ODOO_VERSION_FILE = os.path.join( + ".repo", "installed_odoo_version.txt" +) +ODOO_VERSION_FILE = ".odoo-version" + + +def get_odoo_version(): + """ + Read version configuration and return sorted versions, + installed versions, and current version. + """ + with open(VERSION_DATA_FILE) as txt: + data_version = json.load(txt) + + if not data_version: + raise Exception( + "Internal error, no Odoo version is supported," + f" please validate file '{VERSION_DATA_FILE}'" + ) + version_entries = [] + for key, value in data_version.items(): + version_entries.append(value) + value["erplibre_version"] = key + + installed_versions = [] + if os.path.exists(INSTALLED_ODOO_VERSION_FILE): + with open(INSTALLED_ODOO_VERSION_FILE) as txt: + installed_versions = sorted(txt.read().splitlines()) + + odoo_installed_version = None + if os.path.exists(ODOO_VERSION_FILE): + with open(ODOO_VERSION_FILE) as txt: + odoo_installed_version = f"odoo{txt.read().strip()}" + + versions = sorted( + version_entries, key=lambda k: k.get("erplibre_version") + ) + + return versions, installed_versions, odoo_installed_version diff --git a/test/test_todo.py b/test/test_todo.py index 92a2186..71374eb 100644 --- a/test/test_todo.py +++ b/test/test_todo.py @@ -15,14 +15,17 @@ from script.todo.todo import ( ENABLE_CRASH, ERROR_LOG_PATH, GRADLE_FILE, - INSTALLED_ODOO_VERSION_FILE, LOGO_ASCII_FILE, MOBILE_HOME_PATH, - ODOO_VERSION_FILE, STRINGS_FILE, TODO, VENV_ERPLIBRE, +) +from script.todo.version_manager import ( + INSTALLED_ODOO_VERSION_FILE, + ODOO_VERSION_FILE, VERSION_DATA_FILE, + get_odoo_version, ) @@ -30,10 +33,10 @@ class TestTODOInit(unittest.TestCase): def test_initial_attributes(self): todo = TODO() self.assertIsNone(todo.dir_path) - self.assertIsNone(todo.kdbx) self.assertIsNone(todo.selected_file_path) self.assertIsNotNone(todo.config_file) self.assertIsNotNone(todo.execute) + self.assertIsNotNone(todo.kdbx_manager) class TestFillHelpInfo(unittest.TestCase): @@ -99,7 +102,6 @@ class TestGetOdooVersion(unittest.TestCase): "is_deprecated": False, }, } - todo = TODO() with tempfile.TemporaryDirectory() as tmpdir: version_file = os.path.join(tmpdir, "version.json") with open(version_file, "w") as f: @@ -110,16 +112,16 @@ class TestGetOdooVersion(unittest.TestCase): f.write("18.0") with patch( - "script.todo.todo.VERSION_DATA_FILE", version_file + "script.todo.version_manager.VERSION_DATA_FILE", version_file ), patch( - "script.todo.todo.INSTALLED_ODOO_VERSION_FILE", + "script.todo.version_manager.INSTALLED_ODOO_VERSION_FILE", os.path.join(tmpdir, "nonexistent.txt"), ), patch( - "script.todo.todo.ODOO_VERSION_FILE", + "script.todo.version_manager.ODOO_VERSION_FILE", odoo_version_file, ): lst_version, lst_installed, odoo_current = ( - todo.get_odoo_version() + get_odoo_version() ) self.assertEqual(len(lst_version), 2) @@ -138,7 +140,6 @@ class TestGetOdooVersion(unittest.TestCase): "is_deprecated": False, }, } - todo = TODO() with tempfile.TemporaryDirectory() as tmpdir: version_file = os.path.join(tmpdir, "version.json") with open(version_file, "w") as f: @@ -149,31 +150,30 @@ class TestGetOdooVersion(unittest.TestCase): f.write("odoo18.0\nodoo16.0\n") with patch( - "script.todo.todo.VERSION_DATA_FILE", version_file + "script.todo.version_manager.VERSION_DATA_FILE", version_file ), patch( - "script.todo.todo.INSTALLED_ODOO_VERSION_FILE", + "script.todo.version_manager.INSTALLED_ODOO_VERSION_FILE", installed_file, ), patch( - "script.todo.todo.ODOO_VERSION_FILE", + "script.todo.version_manager.ODOO_VERSION_FILE", os.path.join(tmpdir, "nonexistent"), ): lst_version, lst_installed, odoo_current = ( - todo.get_odoo_version() + get_odoo_version() ) self.assertEqual(lst_installed, ["odoo16.0", "odoo18.0"]) self.assertIsNone(odoo_current) def test_no_version_data_raises(self): - todo = TODO() with tempfile.TemporaryDirectory() as tmpdir: version_file = os.path.join(tmpdir, "empty.json") with open(version_file, "w") as f: json.dump({}, f) - with patch("script.todo.todo.VERSION_DATA_FILE", version_file): + with patch("script.todo.version_manager.VERSION_DATA_FILE", version_file): with self.assertRaises(Exception): - todo.get_odoo_version() + get_odoo_version() class TestOnDirSelected(unittest.TestCase): @@ -303,18 +303,18 @@ class TestExecuteUnitTests(unittest.TestCase): class TestKdbxGetExtraCommandUser(unittest.TestCase): def test_empty_kdbx_key(self): todo = TODO() - result = todo.kdbx_get_extra_command_user("") + result = todo.kdbx_manager.get_extra_command_user("") self.assertEqual(result, "") def test_none_kdbx_key(self): todo = TODO() - result = todo.kdbx_get_extra_command_user(None) + result = todo.kdbx_manager.get_extra_command_user(None) self.assertEqual(result, "") def test_kdbx_not_available(self): todo = TODO() - todo.get_kdbx = MagicMock(return_value=None) - result = todo.kdbx_get_extra_command_user("some_key") + todo.kdbx_manager.get_kdbx = MagicMock(return_value=None) + result = todo.kdbx_manager.get_extra_command_user("some_key") self.assertEqual(result, "") @@ -328,5 +328,71 @@ class TestSetupClaudeCommit(unittest.TestCase): # Should print exists message without asking for input +class TestSelectDatabase(unittest.TestCase): + @patch("script.todo.todo.click") + def test_select_database_returns_name(self, mock_click): + todo = TODO() + todo.execute = MagicMock() + todo.execute.exec_command_live.return_value = ( + 0, + ["db_test", "db_prod"], + ) + mock_click.prompt.return_value = "1" + result = todo.select_database() + self.assertEqual(result, "db_test") + + @patch("script.todo.todo.click") + def test_select_database_returns_false_on_zero(self, mock_click): + todo = TODO() + todo.execute = MagicMock() + todo.execute.exec_command_live.return_value = (0, ["db_test"]) + mock_click.prompt.return_value = "0" + result = todo.select_database() + self.assertFalse(result) + + +class TestRestoreFromDatabase(unittest.TestCase): + @patch("builtins.input") + def test_restore_by_filename(self, mock_input): + todo = TODO() + todo.execute = MagicMock() + todo.execute.exec_command_live.return_value = (0, []) + # status="1" (by filename), db name default, no neutralize + mock_input.side_effect = ["1", "", "n", "n"] + todo.restore_from_database() + cmd = todo.execute.exec_command_live.call_args_list[0][0][0] + self.assertIn("db_restore.py", cmd) + + @patch("builtins.input") + def test_restore_with_neutralize(self, mock_input): + todo = TODO() + todo.execute = MagicMock() + todo.execute.exec_command_live.return_value = (0, []) + mock_input.side_effect = ["1", "mydb", "y", "n"] + todo.restore_from_database() + cmd = todo.execute.exec_command_live.call_args_list[0][0][0] + self.assertIn("--neutralize", cmd) + self.assertIn("mydb_neutralize", cmd) + + +class TestCreateBackupFromDatabase(unittest.TestCase): + @patch("script.todo.todo.click") + @patch("builtins.input") + def test_creates_backup_command(self, mock_input, mock_click): + todo = TODO() + todo.execute = MagicMock() + todo.execute.exec_command_live.return_value = ( + 0, + ["test_db"], + ) + mock_click.prompt.return_value = "1" + # backup name input + mock_input.return_value = "backup.zip" + todo.create_backup_from_database() + cmd = todo.execute.exec_command_live.call_args_list[-1][0][0] + self.assertIn("--backup", cmd) + self.assertIn("test_db", cmd) + + if __name__ == "__main__": unittest.main()