diff --git a/script/config/config_file.py b/script/config/config_file.py index 1885c96..78f9baf 100644 --- a/script/config/config_file.py +++ b/script/config/config_file.py @@ -26,27 +26,27 @@ _logger = logging.getLogger(__name__) class ConfigFile: def get_config(self, key_param: str): # Open file and update dct_data - dct_data_init = {} - dct_data_second = {} - dct_data_final = {} + config_base = {} + config_override = {} + config_private = {} if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE) as cfg: - dct_data_init = json.load(cfg) + config_base = json.load(cfg) if os.path.exists(CONFIG_OVERRIDE_FILE): with open(CONFIG_OVERRIDE_FILE) as cfg: - dct_data_second = json.load(cfg) + config_override = json.load(cfg) if os.path.exists(CONFIG_OVERRIDE_PRIVATE_FILE): with open(CONFIG_OVERRIDE_PRIVATE_FILE) as cfg: - dct_data_final = json.load(cfg) + config_private = json.load(cfg) - dct_data_first_merge = self.deep_merge_with_lists( - dct_data_init, dct_data_final, list_strategy="extend" + merged_base_private = self.deep_merge_with_lists( + config_base, config_private, list_strategy="extend" ) dct_data = self.deep_merge_with_lists( - dct_data_first_merge, dct_data_second, list_strategy="extend" + merged_base_private, config_override, list_strategy="extend" ) return dct_data.get(key_param) @@ -55,7 +55,6 @@ class ConfigFile: dct_data = self.get_config(lst_params[0]) for param in lst_params[1:]: if param in dct_data.keys(): - find_in_private = True dct_data = dct_data.get(param) return dct_data diff --git a/script/execute/execute.py b/script/execute/execute.py index 8ea3355..8fa9f8e 100644 --- a/script/execute/execute.py +++ b/script/execute/execute.py @@ -15,7 +15,7 @@ try: except ModuleNotFoundError as e: humanize = None -cst_venv_erplibre = ".venv.erplibre" +VENV_ERPLIBRE = ".venv.erplibre" new_path = os.path.normpath( os.path.join(os.path.dirname(__file__), "..", "..") @@ -42,7 +42,7 @@ class Execute: if exec_path_gnome_terminal: self.cmd_source_erplibre = ( f"gnome-terminal -- bash -c 'source" - f" ./{cst_venv_erplibre}/bin/activate;%s'" + f" ./{VENV_ERPLIBRE}/bin/activate;%s'" ) self.cmd_source_default = "gnome-terminal -- bash -c '" f"%s'" else: @@ -52,16 +52,16 @@ class Execute: "osascript -e 'tell application \"Terminal\"'" ) self.cmd_source_erplibre += " -e 'tell application \"System Events\" to keystroke \"t\" using {command down}' -e 'delay 0.1' -e 'do script \"" - self.cmd_source_erplibre += f"cd {os.getcwd()}; source ./{cst_venv_erplibre}/bin/activate; %s\" in front window'" + self.cmd_source_erplibre += f"cd {os.getcwd()}; source ./{VENV_ERPLIBRE}/bin/activate; %s\" in front window'" self.cmd_source_erplibre += " -e 'end tell'" else: self.cmd_source_erplibre = ( - f"source ./{cst_venv_erplibre}/bin/activate;%s" + f"source ./{VENV_ERPLIBRE}/bin/activate;%s" ) def exec_command_live( self, - commande, + command, source_erplibre=True, quiet=False, single_source_erplibre=False, @@ -74,10 +74,10 @@ class Execute: return_status_and_output_and_command=False, ): """ - Exécute une commande et affiche la sortie en direct. + Exécute une command et affiche la sortie en direct. Args: - commande (str): La commande à exécuter (sous forme de chaîne de caractères). + command (str): La command à exécuter (sous forme de chaîne de caractères). """ my_env = os.environ.copy() @@ -85,18 +85,18 @@ class Execute: my_env.update(new_env) process_start_time = time.time() - return_status = None + exit_code = None if source_erplibre: - # commande = f"source ./{cst_venv_erplibre}/bin/activate && " + commande + # command = f"source ./{VENV_ERPLIBRE}/bin/activate && " + command # cmd = ( # f"gnome-terminal --tab -- bash -c 'source" - # f" ./{cst_venv_erplibre}/bin/activate;{commande}'" + # f" ./{VENV_ERPLIBRE}/bin/activate;{command}'" # ) - commande = self.cmd_source_erplibre % commande - # os.system(f"./script/terminal/open_terminal.sh {commande}") + command = self.cmd_source_erplibre % command + # os.system(f"./script/terminal/open_terminal.sh {command}") elif single_source_erplibre: - commande = ( - f"source ./{cst_venv_erplibre}/bin/activate && %s" % commande + command = ( + f"source ./{VENV_ERPLIBRE}/bin/activate && %s" % command ) elif single_source_odoo: if not source_odoo and os.path.exists("./.erplibre-version"): @@ -104,68 +104,68 @@ class Execute: source_odoo = f.read() if not source_odoo: _logger.error( - f"You cannot execute Odoo command if no version is installed. Command : {commande}" + f"You cannot execute Odoo command if no version is installed. Command : {command}" ) return -1 - commande = ( - f"source ./.venv.{source_odoo}/bin/activate && {commande}" + command = ( + f"source ./.venv.{source_odoo}/bin/activate && {command}" ) if new_window and self.cmd_source_default: - commande = self.cmd_source_default % commande + command = self.cmd_source_default % command if not quiet: print("🏠 ⬇ Execute command :") - print(commande) - lst_output = [] + print(command) + output_lines = [] try: process = subprocess.Popen( - commande, + command, shell=True, executable="/bin/bash", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, # Désactive la mise en tampon pour la sortie en direct - universal_newlines=True, # Pour traiter les sauts de lignes correctement + universal_newlines=True, # Pour traiter les sauts de lines correctement env=my_env, ) while True: - ligne = process.stdout.readline() - if not ligne: + line = process.stdout.readline() + if not line: break if not quiet: - print(ligne, end="") + print(line, end="") if ( return_status_and_output or return_status_and_output_and_command ): # Remove last \n char - lst_output.append( - ligne.removesuffix("\r\n") + output_lines.append( + line.removesuffix("\r\n") .removesuffix("\n") .removesuffix("\r") ) process.wait() # Attendre la fin du process - return_status = process.returncode + exit_code = process.returncode if process.returncode != 0 and not quiet: print( - "La commande a retourné un code d'erreur :" + "La command a retourné un code d'erreur :" f" {process.returncode}" ) except FileNotFoundError: if not quiet: - if "password" in commande: + if "password" in command: print( - f"Erreur : La commande '{commande.split(' ')[0]}'[...] n'a" + f"Erreur : La command '{command.split(' ')[0]}'[...] n'a" " pas été trouvée." ) else: print( - f"Erreur : La commande '{commande}' n'a pas été trouvée." + f"Erreur : La command '{command}' n'a pas été trouvée." ) except Exception as e: if not quiet: @@ -181,12 +181,12 @@ class Execute: if not quiet: print(f"🏠 ⬆ Executed ({duration_sec:.2f} sec.) :") if not quiet: - print(commande) + print(command) print() if return_status_and_output_and_command: - return return_status, commande, lst_output + return exit_code, command, output_lines if return_status_and_command: - return return_status, commande + return exit_code, command if return_status_and_output: - return return_status, lst_output - return return_status + return exit_code, output_lines + return exit_code diff --git a/script/git/git_tool.py b/script/git/git_tool.py index 73d360a..9ff7b7d 100644 --- a/script/git/git_tool.py +++ b/script/git/git_tool.py @@ -16,14 +16,14 @@ from git import Repo from giturlparse import parse # pip install giturlparse from retrying import retry # pip install retrying -CST_FILE_SOURCE_REPO_ADDONS = "source_repo_addons.csv" -CST_EL_GITHUB_TOKEN = "EL_GITHUB_TOKEN" +SOURCE_REPO_ADDONS_FILE = "source_repo_addons.csv" +EL_GITHUB_TOKEN = "EL_GITHUB_TOKEN" DEFAULT_PROJECT_NAME = "ERPLibre" DEFAULT_WEBSITE = "erplibre.ca" DEFAULT_REMOTE_URL = "https://github.com/ERPLibre/ERPLibre.git" -class Struct(object): +class RepoAttrs(object): def __init__(self, **entries): self.__dict__.update(entries) @@ -146,7 +146,7 @@ class GitTool: "sub_path": sub_path, } if get_obj: - return Struct(**d) + return RepoAttrs(**d) return d def get_repo_info( @@ -190,10 +190,10 @@ class GitTool: name = "" url = "" - no_line = 0 + line_number = 0 first_execution = True for line in txt: - no_line += 1 + line_number += 1 if line[:12] == '[submodule "': if not first_execution: data = { @@ -395,7 +395,7 @@ class GitTool: :param repo_path: path of repo to get information env_var.sh :return: { - CST_EL_GITHUB_TOKEN: TOKEN, + EL_GITHUB_TOKEN: TOKEN, } """ filename = os.path.join(repo_path, "env_var.sh") @@ -403,7 +403,7 @@ class GitTool: txt = file.readlines() txt = [a[:-1] for a in txt if "=" in a] - lst_filter = [CST_EL_GITHUB_TOKEN] + lst_filter = [EL_GITHUB_TOKEN] dct_config = {} # Take filtered value and get bash string values for f in lst_filter: @@ -411,7 +411,7 @@ class GitTool: if f in v: lst_v = v.split("=") if len(lst_v) > 1: - dct_config[CST_EL_GITHUB_TOKEN] = v.split("=")[1][1:-1] + dct_config[EL_GITHUB_TOKEN] = v.split("=")[1][1:-1] return dct_config @staticmethod @@ -531,7 +531,7 @@ class GitTool: def generate_repo_manifest( self, - lst_repo: List[Struct] = [], + lst_repo: List[RepoAttrs] = [], output: str = "", dct_remote={}, dct_project={}, @@ -730,7 +730,7 @@ class GitTool: print(str_xml_text + "\n") def generate_git_modules( - self, lst_repo: List[Struct], repo_path: str = "." + self, lst_repo: List[RepoAttrs], repo_path: str = "." ): lst_modules = [] for repo in lst_repo: @@ -747,8 +747,8 @@ class GitTool: def get_source_repo_addons(self, repo_path=".", add_repo_root=False): """ - Read file CST_FILE_SOURCE_REPO_ADDONS and return structure of data - :param repo_path: path to find file CST_FILE_SOURCE_REPO_ADDONS + Read file SOURCE_REPO_ADDONS_FILE and return structure of data + :param repo_path: path to find file SOURCE_REPO_ADDONS_FILE :param add_repo_root: force adding repo root in the list :return: [{ @@ -760,7 +760,7 @@ class GitTool: "name": name of the submodule }] """ - file_name = os.path.join(repo_path, CST_FILE_SOURCE_REPO_ADDONS) + file_name = os.path.join(repo_path, SOURCE_REPO_ADDONS_FILE) lst_result = [] if add_repo_root: # TODO what to do if origin not exist? @@ -959,7 +959,7 @@ class GitTool: @staticmethod def add_and_fetch_remote( - repo_info: Struct, root_repo: Repo = None, branch_name: str = "" + repo_info: RepoAttrs, root_repo: Repo = None, branch_name: str = "" ): """ Deprecated function, not use anymore git submodule diff --git a/script/process/kill_process_by_port.py b/script/process/kill_process_by_port.py index 0610320..c34289e 100755 --- a/script/process/kill_process_by_port.py +++ b/script/process/kill_process_by_port.py @@ -69,7 +69,7 @@ def get_ancestry(pid: int): return chain -def choose_target(chain, nb_parent): +def choose_target(chain, parent_depth): """ Decide which process to kill. Default: kill the highest ancestor before PID 1 (so not systemd). @@ -77,13 +77,13 @@ def choose_target(chain, nb_parent): if not chain: return None - lst_chain_parent = [] + parent_chain = [] for pid in chain: - lst_chain_parent.append(pid) + parent_chain.append(pid) for str_to_stop in STOP_PARENT_KILL: if str_to_stop in pid.cmdline(): - return pid, lst_chain_parent - return chain[nb_parent - 1], [chain[nb_parent - 1]] + return pid, parent_chain + return chain[parent_depth - 1], [chain[parent_depth - 1]] def kill_process(p: psutil.Process, force: bool): @@ -151,6 +151,7 @@ def main(): ) ap.add_argument( "--nb_parent", + dest="parent_depth", type=int, default=1, help="Kill parent too", @@ -176,7 +177,7 @@ def main(): if not chain: continue - target, lst_target = choose_target(chain, args.nb_parent) + target, lst_target = choose_target(chain, args.parent_depth) print(f"\nListener PID {pid} ancestry:") for i, p in enumerate(chain): @@ -207,7 +208,7 @@ def main(): while not has_response and not ignore_kill: confirm = ( input( - f"Tuer ce processus index {args.nb_parent} (enter) ou mettre " + f"Tuer ce processus index {args.parent_depth} (enter) ou mettre " f"l'index [0 to {len(chain) - 1}] du " f"process à tuer, (c/C) pour annuler : \n" ) diff --git a/script/todo/todo.py b/script/todo/todo.py index 87b8de0..dcdda93 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -24,8 +24,8 @@ from script.config import config_file from script.execute import execute from script.todo.todo_i18n import get_lang, lang_is_configured, set_lang, t -file_error_path = ".erplibre.error.txt" -cst_venv_erplibre = ".venv.erplibre" +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" @@ -137,11 +137,11 @@ class TODO: status = click.prompt(help_info) except NameError: print("Do") - print(f"source ./{cst_venv_erplibre}/bin/activate && make") + print(f"source ./{VENV_ERPLIBRE}/bin/activate && make") sys.exit(1) except ImportError: print("Do") - print(f"source ./{cst_venv_erplibre}/bin/activate && make") + print(f"source ./{VENV_ERPLIBRE}/bin/activate && make") sys.exit(1) except click.exceptions.Abort: sys.exit(0) @@ -157,7 +157,7 @@ class TODO: elif status == "4": # cmd = ( # f"gnome-terminal --tab -- bash -c 'source" - # f" ./{cst_venv_erplibre}/bin/activate;make todo'" + # f" ./{VENV_ERPLIBRE}/bin/activate;make todo'" # ) cmd = "make todo" self.execute.exec_command_live(cmd, source_erplibre=True) @@ -1461,15 +1461,15 @@ class TODO: self.execute.exec_command_live(new_cmd) def crash_diagnostic(self, e): - # TODO show message at start if os.path.exists(file_error_path) - if os.path.exists(file_error_path) and not os.path.exists( - cst_venv_erplibre + # TODO show message at start if os.path.exists(ERROR_LOG_PATH) + if os.path.exists(ERROR_LOG_PATH) and not os.path.exists( + VENV_ERPLIBRE ): print("Got error : ") print(e) - print("Got error at first execution.", file_error_path) + print("Got error at first execution.", ERROR_LOG_PATH) try: - file = open(file_error_path, "r") + file = open(ERROR_LOG_PATH, "r") content = file.read() # TODO si vide, ajouter notre erreur print(content) @@ -1485,7 +1485,7 @@ class TODO: # self.restart_script(e) self.execute.exec_command_live(cmd, source_erplibre=True) sys.exit(1) - if os.path.exists(cst_venv_erplibre): + if os.path.exists(VENV_ERPLIBRE): print("Import error : ") print(e) # TODO auto-detect gnome-terminal, or choose another. Is it done already? @@ -1493,13 +1493,13 @@ class TODO: # self.prompt_install() # print( - # f"You forgot to activate source \nsource ./{cst_venv_erplibre}/bin/activate" + # f"You forgot to activate source \nsource ./{VENV_ERPLIBRE}/bin/activate" # ) # time.sleep(0.5) # cmd = "./script/todo/source_todo.sh" print("Re-execute TODO 🤖 or execute :") print() - print(f"source {cst_venv_erplibre}/bin/activate;make") + print(f"source {VENV_ERPLIBRE}/bin/activate;make") print() cmd = "./script/todo/todo.py" # # self.restart_script(e) @@ -1763,16 +1763,16 @@ class TODO: print(f"🤖 {t('reboot_todo')}") # os.execv(sys.executable, ['python'] + sys.argv) # TODO mettre check que le répertoire est créé, s'il existe, auto-loop à corriger - if os.path.exists(cst_venv_erplibre) and not os.path.exists( - file_error_path + if os.path.exists(VENV_ERPLIBRE) and not os.path.exists( + ERROR_LOG_PATH ): # TODO mettre check import suivant ne vont pas planter try: - with open(file_error_path, "w") as f_file: + with open(ERROR_LOG_PATH, "w") as f_file: f_file.write(str(last_error)) pass # The file is created and closed here, no content is written print( - f"Try to reopen process with before :\nsource ./{cst_venv_erplibre}/bin/activate && exec python " + f"Try to reopen process with before :\nsource ./{VENV_ERPLIBRE}/bin/activate && exec python " + " ".join(sys.argv) ) os.execv( @@ -1780,7 +1780,7 @@ class TODO: [ "/bin/bash", "-c", - f"source ./{cst_venv_erplibre}/bin/activate && exec python " + f"source ./{VENV_ERPLIBRE}/bin/activate && exec python " + " ".join(sys.argv), ], ) diff --git a/test/test_git_tool.py b/test/test_git_tool.py index 113514b..5580006 100644 --- a/test/test_git_tool.py +++ b/test/test_git_tool.py @@ -9,28 +9,28 @@ from collections import OrderedDict from unittest.mock import MagicMock, mock_open, patch from script.git.git_tool import ( - CST_EL_GITHUB_TOKEN, - CST_FILE_SOURCE_REPO_ADDONS, DEFAULT_PROJECT_NAME, DEFAULT_REMOTE_URL, DEFAULT_WEBSITE, + EL_GITHUB_TOKEN, GitTool, - Struct, + RepoAttrs, + SOURCE_REPO_ADDONS_FILE, ) -class TestStruct(unittest.TestCase): +class TestRepoAttrs(unittest.TestCase): def test_basic_attributes(self): - s = Struct(a=1, b="hello") + s = RepoAttrs(a=1, b="hello") self.assertEqual(s.a, 1) self.assertEqual(s.b, "hello") def test_empty_struct(self): - s = Struct() + s = RepoAttrs() self.assertEqual(s.__dict__, {}) def test_override_existing(self): - s = Struct(x=10) + s = RepoAttrs(x=10) self.assertEqual(s.x, 10) @@ -98,7 +98,7 @@ class TestGetTransformedRepoInfo(unittest.TestCase): "https://github.com/OCA/web.git", get_obj=True, ) - self.assertIsInstance(result, Struct) + self.assertIsInstance(result, RepoAttrs) self.assertEqual(result.organization, "OCA") def test_organization_force(self): @@ -216,7 +216,7 @@ class TestGetProjectConfig(unittest.TestCase): env_var_path = os.path.join(tmpdir, "env_var.sh") os.rename(f.name, env_var_path) result = GitTool.get_project_config(repo_path=tmpdir) - self.assertEqual(result[CST_EL_GITHUB_TOKEN], "my_token_123") + self.assertEqual(result[EL_GITHUB_TOKEN], "my_token_123") finally: if os.path.exists(env_var_path): os.unlink(env_var_path) @@ -314,7 +314,7 @@ class TestGetManifestXmlInfo(unittest.TestCase): class TestConstants(unittest.TestCase): def test_file_source_repo_addons(self): - self.assertEqual(CST_FILE_SOURCE_REPO_ADDONS, "source_repo_addons.csv") + self.assertEqual(SOURCE_REPO_ADDONS_FILE, "source_repo_addons.csv") def test_default_project_name(self): self.assertEqual(DEFAULT_PROJECT_NAME, "ERPLibre") diff --git a/test/test_todo.py b/test/test_todo.py index b2ef961..859b820 100644 --- a/test/test_todo.py +++ b/test/test_todo.py @@ -13,6 +13,7 @@ from script.todo.todo import ( CONFIG_FILE, CONFIG_OVERRIDE_FILE, ENABLE_CRASH, + ERROR_LOG_PATH, GRADLE_FILE, INSTALLED_ODOO_VERSION_FILE, LOGO_ASCII_FILE, @@ -20,9 +21,8 @@ from script.todo.todo import ( ODOO_VERSION_FILE, STRINGS_FILE, TODO, + VENV_ERPLIBRE, VERSION_DATA_FILE, - cst_venv_erplibre, - file_error_path, ) @@ -237,10 +237,10 @@ class TestConstants(unittest.TestCase): self.assertEqual(LOGO_ASCII_FILE, "./script/todo/logo_ascii.txt") def test_venv_erplibre(self): - self.assertEqual(cst_venv_erplibre, ".venv.erplibre") + self.assertEqual(VENV_ERPLIBRE, ".venv.erplibre") def test_file_error_path(self): - self.assertEqual(file_error_path, ".erplibre.error.txt") + self.assertEqual(ERROR_LOG_PATH, ".erplibre.error.txt") def test_version_data_file(self): self.assertEqual(