#!./.venv/bin/python import argparse import asyncio import configparser import datetime import logging import os import sys import tempfile import time import uuid from collections import deque from typing import Tuple import aioshutil import git from colorama import Fore logging.basicConfig(level=logging.DEBUG) _logger = logging.getLogger(__name__) LOG_FILE = "./.venv/make_test.log" def get_config(): parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, description="""\ Run code generator test in parallel (asyncio). """, epilog="""\ """, ) parser.add_argument( "--ignore_init_check_git", action="store_true", help="Will not stop or init check if contain git change.", ) parser.add_argument( "--no_parallel", action="store_true", help="Will run in serial.", ) parser.add_argument( "--coverage", action="store_true", help="Execute coverage file.", ) parser.add_argument( "--keep_cache", action="store_true", help=( "Will not delete the temporary directory, check in print and log." ), ) parser.add_argument( "-p", "--max_process", type=int, default=0, help="Max processor to use. If 0, use max.", ) parser.add_argument( "--debug", action="store_true", help="Enable asyncio debugging", ) args = parser.parse_args() return args lst_ignore_warning = [ "have the same label:", "odoo.addons.code_generator.extractor_module_file: Ignore next error about" " ALTER TABLE DROP CONSTRAINT.", ] lst_ignore_error = [ "fetchmail_notify_error_to_sender", 'odoo.sql_db: bad query: ALTER TABLE "db_backup" DROP CONSTRAINT' ' "db_backup_db_backup_name_unique"', 'ERROR: constraint "db_backup_db_backup_name_unique" of relation' ' "db_backup" does not exist', 'odoo.sql_db: bad query: ALTER TABLE "db_backup" DROP CONSTRAINT' ' "db_backup_db_backup_days_to_keep_positive"', 'ERROR: constraint "db_backup_db_backup_days_to_keep_positive" of relation' ' "db_backup" does not exist', "odoo.addons.code_generator.extractor_module_file: Ignore next error about" " ALTER TABLE DROP CONSTRAINT.", ] def extract_result(result, test_name, lst_error, lst_warning): lst_log = result[0].split("\n") for log_line in lst_log: is_ignore_error = False if "error" in log_line.lower(): # Remove ignore error for ignore_error in lst_ignore_error: if ignore_error in log_line: is_ignore_error = True break if not is_ignore_error: lst_error.append(log_line) is_ignore_warning = False if "warning" in log_line.lower(): # Remove ignore warning for ignore_warning in lst_ignore_warning: if ignore_warning in log_line: is_ignore_warning = True break if not is_ignore_warning: lst_warning.append(log_line) if result[1]: lst_error.append(f"Return status error for test {test_name}") def check_result(task_list, tpl_result): lst_error = [] lst_warning = [] for i, result in enumerate(tpl_result): extract_result( result, task_list[i].cr_code.co_name, lst_error, lst_warning ) if lst_warning: print(f"{Fore.YELLOW}{len(lst_warning)} WARNING{Fore.RESET}") i = 0 for warning in lst_warning: i += 1 print(f"[{i}]{warning}") if lst_error: print(f"{Fore.RED}{len(lst_error)} ERROR{Fore.RESET}") i = 0 for error in lst_error: i += 1 print(f"[{i}]{error}") if lst_error or lst_warning: str_result = ( f"{Fore.RED}{len(lst_error)} ERROR" f" {Fore.YELLOW}{len(lst_warning)} WARNING" ) else: str_result = f"{Fore.GREEN}SUCCESS 🍰" print(f"{Fore.BLUE}Summary TEST {str_result}{Fore.RESET}") def print_log(lst_task, tpl_result): if len(lst_task) != len(tpl_result): _logger.error("Different length for log... What happen?") return with open(LOG_FILE, "w") as f: for i, task in enumerate(lst_task): result = tpl_result[i] status_str = "PASS" if not result[1] else "FAIL" f.write( f"\nTest execution {i + 1} - {status_str} -" f" {task.cr_code.co_name}\n\n" ) if result[0]: f.write(result[0]) f.write("\n") print(f"Log file {LOG_FILE}") async def run_command_get_output(*args, cwd=None): if cwd is not None: process = await asyncio.create_subprocess_exec( *args, # stdout must a pipe to be accessible as process.stdout stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd, ) else: process = await asyncio.create_subprocess_exec( *args, # stdout must a pipe to be accessible as process.stdout stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) # Wait for the subprocess to finish stdout, stderr = await process.communicate() return stdout.decode() async def run_command(*args, test_name=None): # Create subprocess start_time = time.time() cmd_str = " ".join(args) process = await asyncio.create_subprocess_exec( *args, # stdout must a pipe to be accessible as process.stdout stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) # Wait for the subprocess to finish stdout, stderr = await process.communicate() end_time = time.time() diff_sec = end_time - start_time # Return stdout + stderr, returncode str_out = "\n" + stdout.decode().strip() + "\n" if stdout else "" str_err = "\n" + stderr.decode().strip() + "\n" if stderr else "" status_str = "FAIL" if process.returncode else "PASS" if test_name: str_output_init = ( f"\n\n{status_str} [{test_name}] [{diff_sec:.3f}s] Execute" f' "{cmd_str}"\n\n' ) else: str_output_init = ( f'\n\n{status_str} [{diff_sec:.3f}s] Execute "{cmd_str}"\n\n' ) all_output = str_out + str_err print(str_output_init) if process.returncode: lst_error = [] lst_warning = [] extract_result( (all_output, process.returncode), test_name, lst_error, lst_warning ) for error in lst_error: print(f"\t{error}") for warning in lst_warning: print(f"\t{warning}") return str_output_init + all_output, process.returncode def update_config( origin_config, new_path_config, lst_path_to_add_config=None, lst_path_to_remove_config=None, module_name=None, ): config_parser = configparser.ConfigParser() config_parser.read(origin_config) str_path = config_parser["options"]["addons_path"].strip(",") lst_path = str_path.split(",") # Clean path from test if lst_path_to_remove_config: for remove_key in lst_path_to_remove_config: if remove_key.startswith("./"): remove_key = remove_key[2:] if module_name and remove_key.endswith(module_name): remove_key = remove_key[: -(len(module_name) + 1)] for s_path in lst_path: if s_path.endswith(remove_key): lst_path.remove(s_path) break if lst_path_to_add_config: for add_path in lst_path_to_add_config: if module_name and add_path.endswith(module_name): add_path = add_path[: -(len(module_name) + 1)] lst_path.insert(0, add_path) s_new_path = ",".join(lst_path) config_parser["options"]["addons_path"] = s_new_path with open(new_path_config, "w") as configfile: config_parser.write(configfile) async def test_exec( config, path_module_check: str, generated_module=None, generate_path=None, tested_module=None, search_class_module=None, script_after_init_check=None, lst_init_module_name=None, test_name=None, install_path=None, run_in_sandbox=False, restore_db_image_name="erplibre_base", ) -> Tuple[str, int]: test_result = "" test_status = 0 new_destination_path = None if search_class_module: if install_path is not None: path_template_to_generate = os.path.join( install_path, tested_module ) elif generate_path: path_template_to_generate = os.path.join( generate_path, tested_module ) else: path_template_to_generate = os.path.join( path_module_check, tested_module ) if generate_path is not None: path_module_to_generate = os.path.join( generate_path, search_class_module ) else: path_module_to_generate = os.path.join( path_module_check, search_class_module ) else: path_template_to_generate = None path_module_to_generate = None use_test_path_generic = False destination_path = None temp_dir_name = None if run_in_sandbox: if config.keep_cache: temp_dir = tempfile.mkdtemp() temp_dir_name = temp_dir else: temp_dir = tempfile.TemporaryDirectory() temp_dir_name = temp_dir.name print(temp_dir_name) temp_dir_name = os.path.join(temp_dir_name, "workspace") os.mkdir(temp_dir_name) lst_path_to_add_config = [] lst_path_to_remove_config = [] if not os.path.exists(path_module_check): return ( f"Error var path_module_check '{path_module_check}' not" " exist.", -1, ) copy_path_module_check = path_module_check path_module_check = os.path.normpath( os.path.join(temp_dir_name, path_module_check) ) if generated_module and path_module_check.endswith( "/" + generated_module ): use_test_path_generic = True destination_path = path_module_check[ : -(len(generated_module) + 1) ] copy_path = copy_path_module_check[: -(len(generated_module) + 1)] elif search_class_module and path_module_check.endswith( "/" + search_class_module ): destination_path = path_module_check[ : -(len(search_class_module) + 1) ] copy_path = copy_path_module_check[ : -(len(search_class_module) + 1) ] else: destination_path = path_module_check copy_path = copy_path_module_check lst_path_to_add_config.append(destination_path) lst_path_to_remove_config.append(copy_path) ignore_tree = await aioshutil.ignore_patterns(".git", "setup") await aioshutil.copytree( copy_path, destination_path, ignore=ignore_tree ) # destination_path_with_git = os.path.join(destination_path, ".git") # if os.path.exists(destination_path_with_git): # await aioshutil.rmtree(destination_path_with_git) # destination_path_with_setup = os.path.join(destination_path, "setup") # if os.path.exists(destination_path_with_setup): # await aioshutil.rmtree(destination_path_with_setup) if tested_module: lst_module_to_test = tested_module.split(",") for module_name in lst_module_to_test: # Update path to change new emplacement s_lst_path_tested_module = await run_command_get_output( "find", ".", "-name", module_name ) if not s_lst_path_tested_module: return ( f"Error cannot find module '{path_module_check}' not" " exist.", -1, ) else: lst_path_tested_module = ( s_lst_path_tested_module.strip().split("\n") ) s_first_path = lst_path_tested_module[0] parent_dir = os.path.dirname(s_first_path) # Copy it if copy_path != parent_dir: os.path.basename(parent_dir) new_destination_path = os.path.join( os.path.dirname(destination_path), os.path.basename(parent_dir), ) ignore_tree = await aioshutil.ignore_patterns( ".git", "setup" ) await aioshutil.copytree( parent_dir, new_destination_path, ignore=ignore_tree, ) # destination_path_with_git = os.path.join( # new_destination_path, ".git" # ) # if os.path.exists(destination_path_with_git): # await aioshutil.rmtree(destination_path_with_git) lst_path_to_add_config.append(new_destination_path) lst_path_to_remove_config.append(parent_dir) new_s_first_path = os.path.normpath( os.path.join(temp_dir_name, s_first_path) ) s_lst_path_generated_module = await run_command_get_output( "find", ".", "-name", generated_module, cwd=temp_dir_name, ) if s_lst_path_generated_module: lst_path_generated_module = ( s_lst_path_generated_module.strip().split("\n") ) s_first_path = os.path.normpath( os.path.join( temp_dir_name, os.path.dirname(lst_path_generated_module[0]), ) ) else: # TODO This is wrong... bug in template if reach this case s_first_path = destination_path hook_file = os.path.join(new_s_first_path, "hooks.py") with open(hook_file) as hook: hook_line = hook.read() has_template = ( "template_dir = os.path.normpath" in hook_line ) # Goal, update path_module_generate, maybe commented # Goal, update template_dir, maybe not exist # TODO need to refactor this and use AST and not string research # try find nb space indentation lst_f_key = [ "# path_module_generate = ", "#path_module_generate = ", "path_module_generate = ", ] nb_space_indentation = None first_index = None for f_key in lst_f_key: if f_key in hook_line: first_index = hook_line.find(f_key) f_begin_index = ( hook_line.rfind("\n", 0, first_index) + 1 ) nb_space_indentation = ( first_index - f_begin_index ) break if nb_space_indentation is None: return ( f"Cannot find keys '{lst_f_key}' in" f" {hook_file}", -1, ) end_string = "\n\n" index_end_string = hook_line.find( end_string, first_index ) if has_template: new_hook_line = ( hook_line[: first_index + len(f_key)] + f'"{s_first_path}"\n' + f'{nb_space_indentation * " "}template_dir =' f' "{s_first_path}/" + MODULE_NAME\n\n' + hook_line[index_end_string:] ) else: new_hook_line = ( hook_line[: first_index + len(f_key)] + f'"{s_first_path}"\n' + hook_line[index_end_string:] ) new_hook_line = new_hook_line.replace( "# path_module_generate = ", "path_module_generate = ", ) new_hook_line = new_hook_line.replace( '# "path_sync_code": path_module_generate,', '"path_sync_code": path_module_generate,', ) with open(hook_file, "w") as hook: hook.write(new_hook_line) # Format editing code before commit await run_command_get_output( "./script/maintenance/black.sh", temp_dir_name ) # init repo with git for dir_to_git in lst_path_to_add_config: await run_command_get_output("git", "init", ".", cwd=dir_to_git) await run_command_get_output("git", "add", ".", cwd=dir_to_git) await run_command_get_output( "git", "commit", "-am", "'first commit'", cwd=dir_to_git ) new_config_path = os.path.join(temp_dir_name, "config.conf") update_config( "./config.conf", new_config_path, lst_path_to_add_config=lst_path_to_add_config, lst_path_to_remove_config=lst_path_to_remove_config, module_name=generated_module, ) else: new_config_path = None if install_path is None: install_path = path_module_check elif run_in_sandbox: install_path = os.path.normpath( os.path.join(temp_dir_name, install_path) ) # Check code, init module to install if lst_init_module_name: for module_name in lst_init_module_name: res, status = await run_command( "./script/code_generator/check_git_change_code_generator.sh", path_module_check, module_name, test_name=test_name, ) test_result += res test_status += status # Check code, module to generate if tested_module: res, status = await run_command( "./script/code_generator/check_git_change_code_generator.sh", path_module_check, tested_module, test_name=test_name, ) test_result += res test_status += status # Leave when detect anomaly in check before start if test_status: return test_result, test_status # Execute script before start if script_after_init_check and not test_status: res, status = await run_command(script_after_init_check) test_result += res test_status += status is_db_create = False unique_database_name = f"test_demo_{uuid.uuid4()}"[:63] if not test_status: # Create database res, status = await run_command( "./script/database/db_restore.py", "--database", unique_database_name, "--image", restore_db_image_name, test_name=test_name, ) test_result += res test_status += status is_db_create = not status if not test_status and lst_init_module_name: # Install required module str_test = ",".join(lst_init_module_name) if config.coverage: script_name = ( "./script/addons/coverage_install_addons_dev.sh" if tested_module else "./script/addons/coverage_install_addons.sh" ) else: script_name = ( "./script/addons/install_addons_dev.sh" if tested_module else "./script/addons/install_addons.sh" ) if new_config_path: res, status = await run_command( script_name, unique_database_name, str_test, new_config_path, test_name=test_name, ) else: res, status = await run_command( script_name, unique_database_name, str_test, test_name=test_name, ) test_result += res test_status += status if not test_status and search_class_module and generated_module: # Update template with class/model/inherit res, status = await run_command( "./script/code_generator/search_class_model.py", "--quiet", "-d", path_module_to_generate, "-t", path_template_to_generate, test_name=test_name, ) test_result += res test_status += status # test_generated_path = ( # install_path if destination_path is None else new_destination_path # ) test_generated_path = ( destination_path if use_test_path_generic else install_path ) # if destination_path is None: # destination_path = install_path if not test_status and tested_module and generated_module: cmd = ( "./script/code_generator/coverage_install_and_test_code_generator.sh" if config.coverage else "./script/code_generator/install_and_test_code_generator.sh" ) # Finally, the test if new_config_path: res, status = await run_command( cmd, unique_database_name, tested_module, test_generated_path, generated_module, new_config_path, test_name=test_name, ) else: res, status = await run_command( cmd, unique_database_name, tested_module, install_path, generated_module, test_name=test_name, ) test_result += res test_status += status if is_db_create: res, status = await run_command( "./.venv/bin/python3", "./odoo/odoo-bin", "db", "--drop", "--database", unique_database_name, test_name=test_name, ) test_result += res test_status += status return test_result, test_status def check_git_change(): """ return True if success """ loop = asyncio.get_event_loop() task_list = [ run_command( "./script/code_generator/check_git_change_code_generator.sh", "./addons/TechnoLibre_odoo-code-generator-template", test_name="Init check_git_change TechnoLibre_odoo-code-generator-template", ), run_command( "./script/code_generator/check_git_change_code_generator.sh", "./addons/OCA_server-tools", test_name="Init check_git_change OCA_server-tools", ) ] commands = asyncio.gather(*task_list) tpl_result = loop.run_until_complete(commands) status = any([a[1] for a in tpl_result]) loop.close() return not status def print_summary_task(task_list): for task in task_list: print(task.cr_code.co_name) # START TEST async def run_demo_test(config) -> Tuple[str, int]: lst_test_name = [ "demo_helpdesk_data", "demo_internal", "demo_internal_inherit", "demo_mariadb_sql_example_1", "demo_portal", "demo_website_data", "demo_website_leaflet", "demo_website_snippet", ] res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", lst_init_module_name=lst_test_name, test_name="demo_test", ) return res, status async def run_code_generator_migrator_demo_mariadb_sql_example_1_test( config, ) -> Tuple[str, int]: # Migrator res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module="demo_mariadb_sql_example_1", tested_module="code_generator_migrator_demo_mariadb_sql_example_1", script_after_init_check=( "./script/database/restore_mariadb_sql_example_1.sh" ), lst_init_module_name=[ "code_generator_portal", ], test_name="mariadb_test-migrator", run_in_sandbox=True, ) return res, status async def run_code_generator_template_demo_mariadb_sql_example_1_test( config, ) -> Tuple[str, int]: # Template res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module="code_generator_demo_mariadb_sql_example_1", tested_module="code_generator_template_demo_mariadb_sql_example_1", search_class_module="demo_mariadb_sql_example_1", lst_init_module_name=[ "code_generator_portal", "demo_mariadb_sql_example_1", ], test_name="mariadb_test-template", run_in_sandbox=True, ) return res, status async def run_code_generator_demo_mariadb_sql_example_1_test( config, ) -> Tuple[str, int]: # Code generator res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module="demo_mariadb_sql_example_1", lst_init_module_name=[ "code_generator_portal", ], tested_module="code_generator_demo_mariadb_sql_example_1", test_name="mariadb_test-code-generator", run_in_sandbox=True, ) return res, status async def run_code_generator_data_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "demo_helpdesk_data", # "demo_website_data", ] lst_tested_module = [ "code_generator_demo_export_helpdesk", # "code_generator_demo_export_website", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_data_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_data_test_part_2(config) -> Tuple[str, int]: # TODO merge this test into run_code_generator_data_test test_result = "" test_status = 0 lst_generated_module = [ "demo_website_data", ] lst_tested_module = [ "code_generator_demo_export_website", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_data_part_2_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_export_website_attachments_test( config, ) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "demo_website_attachments_data", ] lst_tested_module = [ "code_generator_demo_export_website_attachments", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_export_website_attachments_test", run_in_sandbox=True, restore_db_image_name="test_website_attachments", ) test_result += res test_status += status return test_result, test_status async def run_code_generator_theme_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "theme_website_demo_code_generator", ] lst_tested_module = [ "code_generator_demo_theme_website", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_theme_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_generic_all_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "demo_internal", "demo_portal", "demo_helpdesk_data", "demo_website_data", "demo_website_leaflet", "demo_website_snippet", "theme_website_demo_code_generator", ] lst_tested_module = [ "code_generator_demo_internal", "code_generator_demo_portal", "code_generator_demo_export_helpdesk", "code_generator_demo_export_website", "code_generator_demo_website_leaflet", "code_generator_demo_website_snippet", "code_generator_demo_theme_website", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_generic_all_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_website_snippet_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "demo_website_leaflet", "demo_website_snippet", "demo_website_multiple_snippet", ] lst_tested_module = [ "code_generator_demo_website_leaflet", "code_generator_demo_website_snippet", "code_generator_demo_website_multiple_snippet", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_website_snippet_test", run_in_sandbox=True, ) test_result += res test_status += status if "code_generator_demo_website_multiple_snippet" in lst_tested_module: # Because code_generator_demo_website_multiple_snippet depend on code_generator_demo_portal, it will # execute it and this delete file demo_portal/i18n/demo_portal.pot and demo_portal/i18n/fr_CA.po lst_file_is_delete = [ "demo_portal/i18n/demo_portal.pot", "demo_portal/i18n/fr_CA.po", ] path_to_check = os.path.join( "addons", "TechnoLibre_odoo-code-generator-template" ) repo_demo_portal = git.Repo(path_to_check) status_to_check = repo_demo_portal.git.status("-s") if all([f"D {a}" in status_to_check for a in lst_file_is_delete]): # Revert it for file_name in lst_file_is_delete: repo_demo_portal.git.checkout(file_name) else: test_status += 1 test_result += ( f"\n\nFAIL - inspect to delete file ${lst_file_is_delete}" ) return test_result, test_status async def run_code_generator_demo_generic_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "demo_internal", "demo_portal", ] lst_tested_module = [ "code_generator_demo_internal", "code_generator_demo_portal", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_demo_generic_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_demo_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "code_generator_demo", ] lst_tested_module = [ "code_generator_demo", ] # Multiple res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_demo_test", ) test_result += res test_status += status return test_result, test_status async def run_code_generator_inherit_test(config) -> Tuple[str, int]: # TODO can be merge into code_generator_multiple test_result = "" test_status = 0 lst_generated_module = [ "demo_internal_inherit", ] lst_tested_module = [ "code_generator_demo_internal_inherit", ] # Inherit res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_inherit_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_auto_backup_test(config) -> Tuple[str, int]: test_result = "" test_status = 0 lst_generated_module = [ "auto_backup", ] lst_tested_module = [ "code_generator_auto_backup", ] # Auto-backup res, status = await test_exec( config, "./addons/OCA_server-tools/auto_backup", generated_module=",".join(lst_generated_module), tested_module=",".join(lst_tested_module), test_name="code_generator_auto_backup_test", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_template_demo_portal_test( config, ) -> Tuple[str, int]: test_result = "" test_status = 0 # Template res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module="code_generator_demo_portal", tested_module="code_generator_template_demo_portal", search_class_module="demo_portal", lst_init_module_name=[ "demo_portal", ], test_name="code_generator_template_demo_portal", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_template_demo_internal_inherit_test( config, ) -> Tuple[str, int]: test_result = "" test_status = 0 # Template res, status = await test_exec( config, "./addons/TechnoLibre_odoo-code-generator-template", generated_module="code_generator_demo_internal_inherit", tested_module="code_generator_template_demo_internal_inherit", search_class_module="demo_internal_inherit", lst_init_module_name=[ "demo_internal_inherit", ], test_name="code_generator_template_demo_internal_inherit", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_code_generator_template_demo_sysadmin_cron_test( config, ) -> Tuple[str, int]: test_result = "" test_status = 0 # Template res, status = await test_exec( config, "./addons/OCA_server-tools/auto_backup", generated_module="code_generator_auto_backup", generate_path="./addons/OCA_server-tools/", tested_module="code_generator_template_demo_sysadmin_cron", search_class_module="auto_backup", lst_init_module_name=[ "auto_backup", ], test_name="code_generator_template_demo_sysadmin_cron", install_path="./addons/TechnoLibre_odoo-code-generator-template", run_in_sandbox=True, ) test_result += res test_status += status return test_result, test_status async def run_helloworld_test(config) -> Tuple[str, int]: res, status = await run_command( "./test/code_generator/hello_world.sh", test_name="helloworld_test" ) return res, status async def run_in_serial(task_list): q = asyncio.Queue() for task in task_list: await q.put(task) lst_result = [] for i in range(len(task_list)): co = await q.get() result = await co lst_result.append(result) return lst_result class AsyncioPool: # TODO check to replace this pool by https://github.com/dano/aioprocessing def __init__(self, concurrency, loop=None): """ @param loop: asyncio loop @param concurrency: Maximum number of concurrently running tasks """ self._loop = loop or asyncio.get_event_loop() self._concurrency = concurrency self._coros = deque([]) # All coroutines queued for execution self._futures = [] # All currently running coroutines self._lst_result = [] def close(self): self._loop.close() def add_coro(self, coro): """ @param coro: coroutine to add """ self._coros.append(coro) self.print_status() def run_until_complete(self): self._loop.run_until_complete(self._wait_for_futures()) return self._lst_result def print_status(self): print( " Status: coros:%s - futures:%s" % (len(self._coros), len(self._futures)) ) def _start_futures(self): num_to_start = self._concurrency - len(self._futures) num_to_start = min(num_to_start, len(self._coros)) for _ in range(num_to_start): coro = self._coros.popleft() future = asyncio.ensure_future(coro, loop=self._loop) self._futures.append(future) self.print_status() async def _wait_for_futures(self): while len(self._coros) > 0 or len(self._futures) > 0: self._start_futures() futures_completed, futures_pending = await asyncio.wait( self._futures, loop=self._loop, return_when=asyncio.FIRST_COMPLETED, ) for future in futures_completed: self._lst_result.append(future.result()) self._futures.remove(future) self._start_futures() def run_all_test(config) -> None: # low in time, at the end for more speed task_list = [ run_code_generator_migrator_demo_mariadb_sql_example_1_test(config), run_code_generator_template_demo_mariadb_sql_example_1_test(config), run_code_generator_demo_mariadb_sql_example_1_test(config), run_code_generator_auto_backup_test(config), run_code_generator_template_demo_portal_test(config), run_code_generator_template_demo_internal_inherit_test(config), run_code_generator_template_demo_sysadmin_cron_test(config), run_code_generator_demo_test(config), # Begin to run generic test # run_code_generator_generic_all_test(config), run_code_generator_data_test(config), run_code_generator_data_test_part_2(config), run_code_generator_export_website_attachments_test(config), run_code_generator_theme_test(config), run_code_generator_website_snippet_test(config), run_code_generator_demo_generic_test(config), # End run generic test # TODO Will cause conflict with the other because write in code_generator_demo/hooks.py # run_helloworld_test(config), run_code_generator_inherit_test(config), run_demo_test(config), ] print_summary_task(task_list) if asyncio.get_event_loop().is_closed(): asyncio.set_event_loop(asyncio.new_event_loop()) if config.no_parallel: tpl_result = asyncio.run(run_in_serial(task_list)) elif config.max_process: pool = AsyncioPool(config.max_process) for task in task_list: pool.add_coro(task) try: tpl_result = pool.run_until_complete() finally: pool.close() else: # Use maximal resource loop = asyncio.get_event_loop() if config.debug: loop.set_debug(True) try: commands = asyncio.gather(*task_list) tpl_result = loop.run_until_complete(commands) finally: loop.close() print_log(task_list, tpl_result) check_result(task_list, tpl_result) def main(): # TODO configure logger with thread # logging.basicConfig( # level=logging.INFO, # format='%(threadName)10s %(name)18s: %(message)s', # stream=sys.stderr, # ) config = get_config() start_time = time.time() if not config.ignore_init_check_git: success = check_git_change() else: success = True if success: run_all_test(config) end_time = time.time() diff_sec = end_time - start_time # print(f"Time execution {diff_sec:.3f}s") print(f"Time execution {datetime.timedelta(seconds=diff_sec)}") return 0 if __name__ == "__main__": sys.exit(main())