[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 <mathben@technolibre.ca>
This commit is contained in:
Mathieu Benoit 2026-03-10 03:59:35 -04:00
parent 666f285fca
commit ec134d9299
5 changed files with 484 additions and 131 deletions

View file

@ -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

100
script/todo/kdbx_manager.py Normal file
View file

@ -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

View file

@ -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) : ")

View file

@ -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

View file

@ -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()