[REF] script: add type hints and remove duplicate methods

Complete the manager extraction by removing methods from todo.py
that were already moved to database_manager.py (select_database,
restore_from_database, create_backup_from_database, open_file_image_db,
download_database_backup_cli). Add type annotations to manager
classes for better code clarity. Update tests to target the
manager instances directly.

Generated by Claude Code 2.1.72 model claude-sonnet-4-6

Co-Authored-By: Mathieu Benoit <mathben@technolibre.ca>
This commit is contained in:
Mathieu Benoit 2026-03-10 04:05:04 -04:00
parent ec134d9299
commit 2dde5467db
6 changed files with 86 additions and 270 deletions

View file

@ -24,10 +24,10 @@ _logger = logging.getLogger(__name__)
class ConfigFile:
def get_config(self, key_param: str):
config_base = {}
config_override = {}
config_private = {}
def get_config(self, key_param: str) -> Any:
config_base: dict = {}
config_override: dict = {}
config_private: dict = {}
if os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE) as cfg:
@ -50,14 +50,14 @@ class ConfigFile:
return merged_config.get(key_param)
def get_config_value(self, params: list):
def get_config_value(self, params: list[str]) -> Any:
config_data = self.get_config(params[0])
for param in params[1:]:
if param in config_data:
config_data = config_data.get(param)
return config_data
def get_logo_ascii_file_path(self):
def get_logo_ascii_file_path(self) -> str:
return LOGO_ASCII_FILE
def deep_merge_with_lists(

View file

@ -21,15 +21,15 @@ except Exception:
class DatabaseManager:
def __init__(self, execute, fill_help_info):
def __init__(self, execute, fill_help_info) -> None:
self._execute = execute
self._fill_help_info = fill_help_info
self._dir_path = None
self._dir_path: str | None = None
def _on_dir_selected(self, path):
def _on_dir_selected(self, path: str) -> None:
self._dir_path = path
def select_database(self):
def select_database(self) -> str | bool:
cmd_server = "./odoo_bin.sh db --list"
status, databases = self._execute.exec_command_live(
cmd_server,
@ -39,9 +39,7 @@ class DatabaseManager:
)
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
]
valid_choices = [str(a) for a in range(len(choices) + 1) if a]
while True:
status = click.prompt(help_info)
@ -55,7 +53,7 @@ class DatabaseManager:
else:
print(t("cmd_not_found"))
def restore_from_database(self, show_remote_list=True):
def restore_from_database(self, show_remote_list: bool = True) -> None:
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}")
@ -76,9 +74,7 @@ class DatabaseManager:
database_name = default_database_name
status = (
input(
"\U0001f4ac Would you like to neutralize database (n/N)? "
)
input("\U0001f4ac Would you like to neutralize database (n/N)? ")
.strip()
.lower()
)
@ -103,9 +99,7 @@ class DatabaseManager:
source_erplibre=False,
)
status = (
input(
"\U0001f4ac Would you like to update all addons (y/Y)? "
)
input("\U0001f4ac Would you like to update all addons (y/Y)? ")
.strip()
.lower()
)
@ -117,7 +111,9 @@ class DatabaseManager:
source_erplibre=False,
)
def create_backup_from_database(self, show_remote_list=True):
def create_backup_from_database(
self, show_remote_list: bool = True
) -> None:
database_name = self.select_database()
backup_name = input(
"\U0001f4ac Backup name (default = name+date.zip) : "
@ -126,9 +122,7 @@ class DatabaseManager:
backup_name = (
database_name
+ "_"
+ datetime.datetime.now().strftime(
"%Y-%m-%d_%Hh%Mm%Ss"
)
+ datetime.datetime.now().strftime("%Y-%m-%d_%Hh%Mm%Ss")
+ ".zip"
)
@ -148,7 +142,7 @@ class DatabaseManager:
source_erplibre=False,
)
def open_file_image_db(self):
def open_file_image_db(self) -> str:
self._dir_path = ""
path_image_db = os.path.join(os.getcwd(), "image_db")
@ -160,10 +154,10 @@ class DatabaseManager:
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) : "
)
def download_database_backup_cli(
self, show_remote_list: bool = True
) -> tuple[int, str, str]:
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"
@ -175,9 +169,7 @@ class DatabaseManager:
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()
database_name = input("Select id of database :").strip()
elif len(output_lines) == 1:
database_name = output_lines[0].strip()
else:
@ -187,12 +179,8 @@ class DatabaseManager:
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"
)
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()
@ -214,9 +202,7 @@ class DatabaseManager:
new_env=my_env,
)
try:
with zipfile.ZipFile(
default_output_path, "r"
) as zip_ref:
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"

View file

@ -21,7 +21,7 @@ except ModuleNotFoundError:
class KdbxManager:
def __init__(self, config_file):
def __init__(self, config_file) -> None:
self._config_file = config_file
self._kdbx = None
@ -29,9 +29,7 @@ class KdbxManager:
if self._kdbx:
return self._kdbx
kdbx_file_path = self._config_file.get_config_value(
["kdbx", "path"]
)
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")
@ -64,7 +62,9 @@ class KdbxManager:
self._kdbx = kp
return kp
def get_extra_command_user(self, kdbx_key):
def get_extra_command_user(
self, kdbx_key: str | list | None
) -> str | list:
values = []
if kdbx_key:
kp = self.get_kdbx()
@ -79,15 +79,11 @@ class KdbxManager:
try:
odoo_user = entry.username
except AttributeError:
_logger.error(
f"Cannot find username from keys {key}"
)
_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}"
)
_logger.error(f"Cannot find password from keys {key}")
values.append(
" --default_email_auth"
f" {odoo_user} --default_password_auth"

View file

@ -85,9 +85,7 @@ class TODO:
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
)
self.db_manager = DatabaseManager(self.execute, self.fill_help_info)
def _ask_language(self):
if not lang_is_configured():
@ -431,7 +429,9 @@ class TODO:
odoo_password = instance.get("password")
if kdbx_key:
extra_cmd_web_login = self.kdbx_manager.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"
@ -818,8 +818,8 @@ class TODO:
f"{name} <{email}>",
)
content = content.replace(
'Your Name ',
f'{name} ',
"Your Name ",
f"{name} ",
)
os.makedirs(dest_dir, exist_ok=True)
@ -1309,33 +1309,9 @@ class TODO:
single_source_erplibre=True,
)
def select_database(self):
cmd_server = f"./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 prompt_execute_selenium_and_run_db(self, db_name, extra_cmd_web_login=""):
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)
cmd_client = (
@ -1500,102 +1476,6 @@ class TODO:
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):
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("💬 Select : ")
if status == "1":
file_name = status
else:
file_name = self.db_manager.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"💬 Database name (default={default_database_name}) : "
)
if not database_name:
database_name = default_database_name
status = (
input("💬 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} {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("💬 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.db_manager.select_database()
str_arg = f"--database {database_name}"
backup_name = input("💬 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} --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")
# self.dir_path is over-write into on_dir_selected
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 process_kill_from_port(self):
cfg = configparser.ConfigParser()
cfg.read("./config.conf")
@ -1606,63 +1486,6 @@ class TODO:
source_erplibre=False,
)
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 --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(
f"Failed to read manifest.json from backup file '{default_output_path}'."
)
return status, output_path, database_name
def restart_script(self, last_error):
print(f"🤖 {t('reboot_todo')}")
# os.execv(sys.executable, ['python'] + sys.argv)

View file

@ -12,7 +12,7 @@ INSTALLED_ODOO_VERSION_FILE = os.path.join(
ODOO_VERSION_FILE = ".odoo-version"
def get_odoo_version():
def get_odoo_version() -> tuple[list[dict], list[str], str | None]:
"""
Read version configuration and return sorted versions,
installed versions, and current version.
@ -40,8 +40,6 @@ def get_odoo_version():
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")
)
versions = sorted(version_entries, key=lambda k: k.get("erplibre_version"))
return versions, installed_versions, odoo_installed_version

View file

@ -120,9 +120,7 @@ class TestGetOdooVersion(unittest.TestCase):
"script.todo.version_manager.ODOO_VERSION_FILE",
odoo_version_file,
):
lst_version, lst_installed, odoo_current = (
get_odoo_version()
)
lst_version, lst_installed, odoo_current = get_odoo_version()
self.assertEqual(len(lst_version), 2)
self.assertEqual(odoo_current, "odoo18.0")
@ -158,9 +156,7 @@ class TestGetOdooVersion(unittest.TestCase):
"script.todo.version_manager.ODOO_VERSION_FILE",
os.path.join(tmpdir, "nonexistent"),
):
lst_version, lst_installed, odoo_current = (
get_odoo_version()
)
lst_version, lst_installed, odoo_current = get_odoo_version()
self.assertEqual(lst_installed, ["odoo16.0", "odoo18.0"])
self.assertIsNone(odoo_current)
@ -171,7 +167,9 @@ class TestGetOdooVersion(unittest.TestCase):
with open(version_file, "w") as f:
json.dump({}, f)
with patch("script.todo.version_manager.VERSION_DATA_FILE", version_file):
with patch(
"script.todo.version_manager.VERSION_DATA_FILE", version_file
):
with self.assertRaises(Exception):
get_odoo_version()
@ -329,25 +327,28 @@ class TestSetupClaudeCommit(unittest.TestCase):
class TestSelectDatabase(unittest.TestCase):
@patch("script.todo.todo.click")
@patch("script.todo.database_manager.click")
def test_select_database_returns_name(self, mock_click):
todo = TODO()
todo.execute = MagicMock()
todo.execute.exec_command_live.return_value = (
todo.db_manager._execute = MagicMock()
todo.db_manager._execute.exec_command_live.return_value = (
0,
["db_test", "db_prod"],
)
mock_click.prompt.return_value = "1"
result = todo.select_database()
result = todo.db_manager.select_database()
self.assertEqual(result, "db_test")
@patch("script.todo.todo.click")
@patch("script.todo.database_manager.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"])
todo.db_manager._execute = MagicMock()
todo.db_manager._execute.exec_command_live.return_value = (
0,
["db_test"],
)
mock_click.prompt.return_value = "0"
result = todo.select_database()
result = todo.db_manager.select_database()
self.assertFalse(result)
@ -355,41 +356,53 @@ 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, [])
todo.db_manager._execute = MagicMock()
todo.db_manager._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]
todo.db_manager.restore_from_database()
cmd = todo.db_manager._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, [])
todo.db_manager._execute = MagicMock()
todo.db_manager._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]
todo.db_manager.restore_from_database()
cmd = todo.db_manager._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("script.todo.database_manager.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 = (
todo.db_manager._execute = MagicMock()
todo.db_manager._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]
todo.db_manager.create_backup_from_database()
cmd = todo.db_manager._execute.exec_command_live.call_args_list[-1][0][
0
]
self.assertIn("--backup", cmd)
self.assertIn("test_db", cmd)