mathben/erplibre
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions
erplibre/script/todo/todo_upgrade.py
Mathieu Benoit 045ccc1c85 [ADD] migration: predict the website COW views a version bump will break 
A copy-on-write view freezes the structure of the module view it was copied
from. When that module view changes mode between two versions, the copy keeps
an arch written for the old mode and the upgrade dies on
« Element ... cannot be located in parent view », hours into the migration.

Measured on a real 12.0 database: portal.frontend_layout is declared primary in
12.0 (full QWeb template, so a full-template arch is legitimate) and becomes an
extension in 13.0 (inherit_id + xpath). The 2021 COW copy follows the module,
turns into an extension, and keeps its 12.0 arch. The customization is genuine;
the breakage is produced by the migration.

check_cow_views.py compares the mode stored in database with the mode declared
in the target version sources and sorts the views in three buckets: at risk
(mode change), module absent from the target version, and pages made from the
editor (not at risk). Read-only, and called from step 2 so the list is known
before the long version loop rather than during it.

On the reference database it reports exactly the view that broke the upgrade,
plus 7 views whose module is gone in 13.0 (custom theme included).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-31 23:28:54 -04:00

2244 lines
95 KiB
Python
Executable file
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
import datetime
import json
import logging
import os
import shutil
import sys
import zipfile
from uuid import uuid4
import click
import todo_file_browser
from script.todo.version_manager import get_odoo_version
new_path = os.path.normpath(
os.path.join(os.path.dirname(__file__), "..", "..")
)
sys.path.append(new_path)
from script.execute import execute
from script.git.git_tool import GitTool
_logger = logging.getLogger(__name__)
PYTHON_BIN = ".venv.erplibre/bin/python3"
UPGRADE_DATABASE_CONFIG_LOG = ".venv.erplibre/odoo_database_migration_log.json"
# UPGRADE_MODULE_CONFIG_LOG = ".venv.erplibre/odoo_module_migration_log.json"
VENV_NAME_MODULE_MIGRATOR = ".venv"
LST_PATH_OCA_ODOO_MODULE_MIGRATOR = ["script", "OCA_odoo-module-migrator"]
PATH_OCA_ODOO_MODULE_MIGRATOR = "./" + "/".join(
LST_PATH_OCA_ODOO_MODULE_MIGRATOR
)
PATH_VENV_MODULE_MIGRATOR = os.path.join(
PATH_OCA_ODOO_MODULE_MIGRATOR, VENV_NAME_MODULE_MIGRATOR
)
PATH_SOURCE_VENV_MODULE_MIGRATOR = os.path.join(
PATH_VENV_MODULE_MIGRATOR, "bin", "activate"
)
FILENAME_ODOO_VERSION = ".odoo-version"
LOCAL_MANIFEST = os.path.join(
".repo", "local_manifests", "erplibre_manifest.xml"
)
# Module lists for a version bump. The shared, versioned defaults live under
# script/; the per-database lists live under private/ (mirroring script/, like
# script/todo/todo.json -> private/todo/todo_override.json). Which modules must
# be dropped depends on the database, so that choice is never versioned.
PATH_MIGRATION_GLOBAL = os.path.join("script", "odoo", "migration")
PATH_MIGRATION_PRIVATE = os.path.join("private", "odoo", "migration")
class TodoUpgrade:
def __init__(self, todo):
self.file_path = None
self.dir_path = None
self.todo = todo
self.dct_progression = {}
self.lst_command_executed = []
self.dct_module_per_version = {}
self.dct_module_per_dct_version_path = {}
self.execute = execute.Execute()
def write_config(self):
if "date_create" not in self.dct_progression.keys():
self.dct_progression["date_create"] = str(datetime.datetime.now())
self.dct_progression["date_update"] = str(datetime.datetime.now())
# Always put command_executed at the end
if "command_executed" in self.dct_progression.keys():
value = self.dct_progression["command_executed"]
del self.dct_progression["command_executed"]
self.dct_progression["command_executed"] = value
with open(UPGRADE_DATABASE_CONFIG_LOG, "w") as f:
json.dump(self.dct_progression, f, indent=4)
def on_file_selected(self, file_path):
self.file_path = file_path
todo_file_browser.exit_program()
def on_dir_selected(self, dir_path):
self.dir_path = dir_path
todo_file_browser.exit_program()
def execute_module_upgrade(self):
print("Welcome to Odoo module upgrade processus with ERPLibre 🤖")
if os.path.exists(UPGRADE_DATABASE_CONFIG_LOG):
with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f:
try:
old_dct_progression = json.load(f)
self.dct_progression = old_dct_progression
self.lst_command_executed = old_dct_progression.get(
"command_executed"
)
except json.decoder.JSONDecodeError:
print(
f'⚠️ The config file "{UPGRADE_DATABASE_CONFIG_LOG}" is invalid, ignore it.'
)
print(
"Migrate a directory repo to migrate all module, or select a directory module."
)
print("[m] Migrate one module")
print("[r] Migrate one repo")
print("[] Directory browser")
lst_dir_path = []
from_version = 0
to_version = 0
status = input("💬 What do you choose : ").strip().lower()
if status == "m":
print("[p] Path (default)")
print("[n] Name")
status = input("💬 What do you choose : ").strip().lower()
if status == "n":
module_name = input("💬 Module name : ").strip().lower()
from_version = int(
input("💬 From odoo version (12 to 18) : ").strip()
)
to_version = int(
input("💬 To odoo version (12 to 18) : ").strip()
)
self.switch_odoo(from_version)
# TODO detect path
(
lst_module_missing,
lst_module_duplicate,
lst_module_exist,
lst_module_error,
) = self.check_addons_exist([module_name], get_all_info=True)
if (
lst_module_missing
or lst_module_duplicate
or lst_module_error
):
if lst_module_missing:
print(f"Missing list : {lst_module_missing}")
if lst_module_duplicate:
print(f"Duplicate list : {lst_module_duplicate}")
if lst_module_error:
print(f"Error list : {lst_module_error}")
return
lst_dir_path.extend(lst_module_exist)
else:
self.dir_path = input("💬 Path : ").strip()
elif status == "r":
self.dir_path = input("💬 Path : ").strip()
else:
self.dir_path = None
if not self.dir_path and not lst_dir_path:
initial_dir = os.getcwd()
file_browser = todo_file_browser.FileBrowser(
initial_dir, self.on_dir_selected, open_dir=True
)
file_browser.run_main_frame()
if not lst_dir_path and not os.path.exists(self.dir_path):
_logger.error(f"Path '{self.dir_path}' not exists.")
return
if not from_version:
from_version = int(
input("💬 From odoo version (12 to 18) : ").strip()
)
if not to_version:
to_version = int(input("💬 To odoo version (12 to 18) : ").strip())
# TODO ask option direct migration
lst_version_to_update = [
(a, a + 1) for a in range(from_version, to_version)
]
# lst_version_to_update = [(from_version, to_version)]
# for actual_version in range(from_version, to_version):
# next_version = actual_version + 1
for actual_version, next_version in lst_version_to_update:
set_path_migrate_addons = set()
if not lst_dir_path:
if os.path.exists(
os.path.join(self.dir_path, "__manifest__.py")
):
lst_dir_path = [
[os.path.basename(self.dir_path), self.dir_path]
]
else:
lst_dir_path = [
[a, os.path.join(self.dir_path, a)]
for a in os.listdir(self.dir_path)
if os.path.exists(
os.path.join(self.dir_path, a, "__manifest__.py")
)
]
lst_path_git_clone_migrate = []
lst_module_to_migrate_all = []
for dir_path in lst_dir_path:
source_manifest_path = os.path.join(
dir_path[1], "__manifest__.py"
).replace(f"odoo{from_version}.0", f"odoo{actual_version}.0")
is_dir_module = os.path.exists(source_manifest_path)
if not is_dir_module:
continue
source_module_path = dir_path[1].replace(
f"odoo{from_version}.0", f"odoo{actual_version}.0"
)
source_addons_path = os.path.dirname(dir_path[1]).replace(
f"odoo{from_version}.0", f"odoo{actual_version}.0"
)
target_module_path = source_module_path.replace(
f"odoo{actual_version}.0", f"odoo{next_version}.0"
)
target_manifest_path = source_manifest_path.replace(
f"odoo{actual_version}.0", f"odoo{next_version}.0"
)
target_addons_path = source_addons_path.replace(
f"odoo{actual_version}.0", f"odoo{next_version}.0"
)
dct_module = {
"source_module_path": source_module_path,
"source_manifest_path": source_manifest_path,
"source_addons_path": source_addons_path,
"target_module_path": target_module_path,
"target_manifest_path": target_manifest_path,
"target_addons_path": target_addons_path,
"module_name": dir_path[0],
"source_version_odoo": actual_version,
"target_version_odoo": next_version,
}
lst_module_to_migrate_all.append(dct_module)
set_path_migrate_addons.add(target_addons_path)
# lst_path_git_clone_migrate.append(target_addons_path)
# TODO auto detect version from
# TODO if detect from directory, check from repo list
# TODO detect from manifest version
self.switch_odoo(next_version)
self.install_OCA_odoo_module_migrator()
self.internal_module_upgrade(
next_version,
lst_module_to_migrate_all,
lst_path_git_clone_migrate,
)
for commit_path in set_path_migrate_addons:
cmd = f"./script/code/git_commit_migration_addons_path.py --path {commit_path} --odoo_version {next_version}.0"
self.todo_upgrade_execute(cmd)
print(set_path_migrate_addons)
status = input(
f"💬 Please validate git commit on repos, press to continue : "
).strip()
def internal_module_upgrade(
self,
next_version,
lst_module_to_migrate_all,
lst_path_git_clone_migrate,
):
has_cmd = False
cmd_parallel = "parallel :::"
for dct_module in lst_module_to_migrate_all:
target_addons_path = dct_module.get("target_addons_path")
source_addons_path = dct_module.get("source_addons_path")
module_name = dct_module.get("module_name")
source_version_odoo = f'{dct_module.get("source_version_odoo")}.0'
target_version_odoo = f'{dct_module.get("target_version_odoo")}.0'
source_module_path_to_copy = dct_module.get("source_module_path")
# Prepare git environment for target
if target_addons_path not in lst_path_git_clone_migrate:
lst_path_git_clone_migrate.append(target_addons_path)
self.check_and_clone_source_to_target_migration_code(
next_version,
source_addons_path,
target_addons_path,
)
cmd_migration = (
f"echo 'odoo_module_migrate {module_name}' && "
f"cp -r {source_module_path_to_copy} {target_addons_path} && "
f"cd {PATH_OCA_ODOO_MODULE_MIGRATOR} && "
f"source {VENV_NAME_MODULE_MIGRATOR}/bin/activate && "
f"python -m odoo_module_migrate --directory {target_addons_path} --modules {module_name} "
f"--init-version-name {source_version_odoo} --target-version-name {target_version_odoo} "
f"--no-commit && "
f"cd ~- "
# f"cp -r {source_module_path_to_copy} {target_module_path_to_copy} && "
# f"cd {target_module_path_to_copy} && git commit -am '[MIG] {module_name}: Migration to {target_version_odoo}' && cd ~-"
)
cmd_parallel += f' "{cmd_migration}"'
has_cmd = True
if lst_module_to_migrate_all:
if has_cmd:
self.todo_upgrade_execute(cmd_parallel)
print("List of path with migrate code :")
print(lst_path_git_clone_migrate)
print("ℹ To show repo status :\nmake repo_show_status")
input("💬 Check migration code, press to continue : ")
# source_module_path = dct_module_result.get(
# "source_module_path"
# )
# if not source_module_path:
# _logger.error(
# f"Missing source module path '{source_module_path}'"
# )
# else:
# if os.path.exists(
# os.path.join(source_module_path, ".git")
# ):
# self.todo_upgrade_execute(
# f"cd '{source_module_path}' && git stash && cd ~-",
# )
#
# target_module_path = dct_module_result.get(
# "target_module_path"
# )
# if not target_module_path:
# _logger.error(
# f"Missing target module path '{target_module_path}'"
# )
# else:
# # TODO check if has file to commit
# self.todo_upgrade_execute(
# f"cd '{target_module_path}' && git commit -am '[MIG] {len(lst_module_to_migrate_all)} modules: Migration to {next_version}' && cd ~-",
# )
# TODO copie to next odoo version
# do commit and continue
# continue migration to loop
if next_version in [18]:
# TODO need odoo 18, validate python version without switch
status = input(
f"💬 Please validate repo is ready to run upgrade views_migration_18, press to continue : "
).strip()
# Apply modification with views_migration_18
has_cmd = False
# cmd_serial = ""
cmd_parallel = "parallel :::"
for path_git_clone_migrate in lst_path_git_clone_migrate:
cmd_migration = (
f"echo 'views_migration_18 {path_git_clone_migrate}' && "
f"./.venv.odoo18.0_python3.12.10/bin/python ./script/code/odoo_upgrade_code_with_dir_module.py --path {path_git_clone_migrate}"
)
cmd_parallel += f' "{cmd_migration}"'
# cmd_serial += f"{cmd_migration};"
has_cmd = True
if has_cmd:
# self.todo_upgrade_execute(
# cmd_serial
# )
self.todo_upgrade_execute(cmd_parallel)
print("List of module with migration 18 :")
print(lst_module_to_migrate_all)
print("ℹ To show repo status :\nmake repo_show_status")
input("💬 Check migration 18 code, press to continue : ")
if next_version == 17:
status = input(
f"💬 Please validate repo is ready to run upgrade views_migration_17, press to continue : "
).strip()
# Apply modification with views_migration_17
has_cmd = False
cmd_serial = ""
cmd_parallel = "parallel :::"
for dct_module in lst_module_to_migrate_all:
database_migration_17_name = (
f"migration_odoo_{next_version}_{str(uuid4())[:6]}"
)
module_name = dct_module.get("module_name")
cmd_migration = (
f"echo 'views_migration_17 {module_name}' && "
f"./run.sh -d {database_migration_17_name} -i {module_name} --load=base,web,views_migration_17 --dev upgrade --no-http --stop-after-init"
)
cmd_parallel += f' "{cmd_migration}"'
cmd_serial += f"{cmd_migration};"
has_cmd = True
if has_cmd:
# self.todo_upgrade_execute(
# cmd_serial
# )
self.todo_upgrade_execute(cmd_parallel)
print("List of module with migration 17 :")
print(lst_module_to_migrate_all)
print("ℹ To show repo status :\nmake repo_show_status")
input("💬 Check migration 17 code, press to continue : ")
def execute_odoo_upgrade(self):
# TODO update dev environment for git project
# TODO Redeploy new production after upgrade
# 2 upgrades version = 5 environnement. 0-prod init, 1-dev init, 2-dev01, 3-dev02, 4-prod final
print("Welcome to Odoo database upgrade processus with ERPLibre 🤖")
self.lst_command_executed = []
self.dct_module_per_version = {}
self.dct_module_per_dct_version_path = {}
default_database_name = "test"
if os.path.exists(UPGRADE_DATABASE_CONFIG_LOG):
with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f:
try:
old_dct_progression = json.load(f)
self.dct_progression = {
"migration_file": old_dct_progression.get(
"migration_file"
),
# More useful to ask this question each time
"target_odoo_version": old_dct_progression.get(
"target_odoo_version"
),
"date_create": old_dct_progression.get("date_create"),
}
except json.decoder.JSONDecodeError:
print(
f'⚠️ The config file "{UPGRADE_DATABASE_CONFIG_LOG}" is invalid, ignore it.'
)
print(
f"✨ Detected migration \"{self.dct_progression.get('migration_file')}\" "
f"to version \"{self.dct_progression.get('target_odoo_version')}\", "
f"please select an option."
)
print("[1] Erase progression for a new migration")
print("[2] Reuse database with new process")
print(
"[3] Reuse database without state_4, before looping on next version"
)
print("[4] Reuse database without configuration")
erase_progression_input = (
input("💬 Select an option or press to continue : ")
.strip()
.lower()
)
self.dct_progression = {}
if erase_progression_input in ["2", "3", "4"]:
with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f:
try:
old_dct_progression = json.load(f)
self.dct_progression = {
"migration_file": old_dct_progression.get(
"migration_file"
),
# More useful to ask this question each time
# "target_odoo_version": old_dct_progression.get(
# "target_odoo_version"
# ),
"date_create": old_dct_progression.get(
"date_create"
),
}
for key, value in old_dct_progression.items():
if erase_progression_input == "3":
if (
key.startswith("state_0")
or key.startswith("state_1")
or key.startswith("state_2")
or key.startswith("state_3")
):
self.dct_progression[key] = value
if key.startswith(f"config_state") and not (
key.startswith(f"config_state_0")
or key.startswith(f"config_state_1")
or key.startswith(f"config_state_2")
or key.startswith(f"config_state_3")
):
continue
if (
key.startswith("config_")
and erase_progression_input != "4"
):
self.dct_progression[key] = value
# Force to search missing module to fill dict
self.dct_progression[
"state_0_search_missing_module"
] = False
except json.decoder.JSONDecodeError:
print(
f"⚠️ The config file '{UPGRADE_DATABASE_CONFIG_LOG}' is invalid, ignore it."
)
self.write_config()
elif erase_progression_input not in ["1"]:
with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f:
try:
self.dct_progression = json.load(f)
except json.decoder.JSONDecodeError:
print(
f"⚠️ The config file '{UPGRADE_DATABASE_CONFIG_LOG}' is invalid, ignore it."
)
if "migration_file" in self.dct_progression:
self.file_path = self.dct_progression["migration_file"]
else:
print("")
print("Select the zip file of you database backup.")
self.file_path = input(
"💬 Give the path of file, or empty to use a File Browser, or type 'remote' to download from production : "
)
if not self.file_path.strip():
self.file_path = None
if not self.file_path:
initial_dir = os.path.join(os.getcwd(), "image_db")
file_browser = todo_file_browser.FileBrowser(
initial_dir, self.on_file_selected
)
file_browser.run_main_frame()
elif self.file_path == "remote":
status, self.file_path, default_database_name = (
self.todo.db_manager.download_database_backup_cli()
)
if status:
_logger.error(
"Cannot retrieve database from remote, please retry migration."
)
return
self.dct_progression["migration_file"] = self.file_path
self.write_config()
print(f"✅ Open file {self.file_path}")
with zipfile.ZipFile(self.file_path, "r") as zip_ref:
manifest_file_1 = zip_ref.open("manifest.json")
json_manifest_file_1 = json.load(manifest_file_1)
odoo_actual_version = json_manifest_file_1.get("version")
print(f"✅ Detect version Odoo CE '{odoo_actual_version}'.")
# print("What is your actual Odoo version?")
lst_version, lst_version_installed, odoo_installed_version = (
get_odoo_version()
)
lst_odoo_version = [
{"prompt_description": a.get("odoo_version")}
for a in lst_version
if float(a.get("odoo_version")) > float(odoo_actual_version)
]
help_info = self.todo.fill_help_info(lst_odoo_version)
if "target_odoo_version" in self.dct_progression:
odoo_target_version = self.dct_progression["target_odoo_version"]
else:
print("💬 Which version do you want to upgrade to?")
odoo_target_version = None
cmd_no_found = True
while cmd_no_found:
status = click.prompt(help_info)
try:
int_cmd = int(status)
if 0 < int_cmd <= len(lst_odoo_version):
cmd_no_found = False
odoo_target_version = lst_odoo_version[
int_cmd - 1
].get("prompt_description")
except ValueError:
pass
if cmd_no_found:
print("Commande non trouvée 🤖!")
self.dct_progression["target_odoo_version"] = odoo_target_version
self.write_config()
# Search nb diff to use range
start_version = int(float(odoo_actual_version))
end_version = int(float(odoo_target_version))
range_version = range(start_version, end_version)
lst_module = sorted(
list(set(json_manifest_file_1.get("modules").keys()))
)
self.dct_module_per_version[start_version] = lst_module
self.dct_progression["dct_module_per_version"] = (
self.dct_module_per_version
)
self.dct_progression["lst_module_per_version_origin"] = lst_module
# TODO need support minor version, example 18.2, the .2 (no need for OCE OCB)
print("✨ Show documentation version :")
# TODO Generate it locally and show it if asked
for next_version in range_version:
print(
f"https://oca.github.io/OpenUpgrade/coverage_analysis/modules{next_version*10}-{(next_version+1)*10}.html"
)
# ⚠️ ℹ 💬 ❗ 🔷 ✨ 🟦 🔹 🔵 ⟳ ⧖ ⚙ ✔ ✅ ❌ ⏵ ⏸ ⏹ ◆ ◇ … ➤ ⚑ ★ ☆ ☰ ⬍ ⍟ ⊗ ⌘ ⏻ ⍰
msg = "0 - Inspect zip"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
print("✅ -> Search odoo version")
print("✅ -> Find good environment, read the .zip file")
is_state_4_reach_open_upgrade = self.dct_progression.get(
"state_4_reach_open_upgrade"
)
if not is_state_4_reach_open_upgrade and not self.dct_progression.get(
"state_0_install_odoo"
):
lst_diff_version = sorted(
list(
set([f"odoo{a}.0" for a in range_version]).difference(
set(lst_version_installed)
)
)
)
for odoo_version_to_install in lst_diff_version:
iter_range_version = odoo_version_to_install.replace(
"odoo", ""
).replace(".0", "")
want_continue = input(
f"💬 Would you like to install '{odoo_version_to_install}' (y/Y) : "
)
if want_continue.strip().lower() != "y":
return
self.todo_upgrade_execute(
f"make install_odoo_{iter_range_version}"
)
if not os.path.isfile(FILENAME_ODOO_VERSION):
print(
"⚠️ You need an installed system before continue, check your Odoo installation."
)
return
self.dct_progression["state_0_install_odoo"] = True
self.write_config()
# not self.dct_progression.get("state_0_switch_odoo")
if not is_state_4_reach_open_upgrade:
self.switch_odoo(odoo_actual_version)
# self.dct_progression["state_0_switch_odoo"] = True
# self.write_config()
print("✅ -> Install environment if missing")
if not self.dct_progression.get("state_0_search_missing_module"):
self.switch_odoo(odoo_actual_version)
dct_bd_modules = json_manifest_file_1.get("modules")
lst_module_to_check = [a for a in dct_bd_modules.keys()]
(
lst_module_missing,
lst_module_duplicate,
lst_module_exist,
lst_module_error,
) = self.check_addons_exist(lst_module_to_check, get_all_info=True)
if not lst_module_missing:
lst_module_missing = []
dct_module_exist = {}
if not lst_module_exist:
lst_module_exist = []
else:
for item_lst_module_exist in lst_module_exist:
dct_module_exist[item_lst_module_exist[0]] = (
item_lst_module_exist[1].replace(os.getcwd(), ".")
)
if not lst_module_duplicate:
lst_module_duplicate = []
lst_module_missing = sorted(list(set(lst_module_missing)))
self.dct_progression["len_lst_module_missing"] = len(
lst_module_missing
)
self.dct_progression["lst_module_missing"] = lst_module_missing
self.dct_progression["len_dct_module_exist"] = len(
lst_module_exist
)
self.dct_progression["dct_module_exist"] = dct_module_exist
self.dct_progression["len_lst_module_duplicate"] = len(
lst_module_duplicate
)
self.dct_progression["lst_module_duplicate"] = lst_module_duplicate
self.write_config()
if lst_module_missing or lst_module_duplicate:
print("Cannot setup environment to begin.")
if lst_module_missing:
print("Missing module :")
print(lst_module_missing)
if lst_module_duplicate:
print("Duplicate module :")
print(lst_module_duplicate)
want_continue = input(
"💬 Detect error missing/duplicate module init, do you want to continue? (Y/N): "
)
if want_continue.strip().lower() != "y":
return
self.dct_progression["state_0_search_missing_module"] = True
self.write_config()
else:
# TODO fill from config
lst_module_missing = []
print("✅ -> Search missing module")
print(
"❌ -> Install missing module, do a research or ask to uninstall it (can break data)"
)
msg = "1 - Import database from zip"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
database_name = self.dct_progression.get("config_database_name")
if not database_name:
database_name = (
input(
f"💬 Witch database name do you want to work with? Default ({default_database_name}) : "
).strip()
or default_database_name
)
self.dct_progression["config_database_name"] = database_name
self.write_config()
do_neutralize = False
if not self.dct_progression.get("state_1_neutralize_database"):
print("[1] Ignore neutralize database")
wait_continue = (
input("💬 Neutralize database, press to continue : ")
.strip()
.lower()
)
if wait_continue != "1":
do_neutralize = True
database_name += "_neutralize"
self.dct_progression["config_database_name"] = database_name
self.write_config()
print(f"★ Work with database '{database_name}'")
if not self.dct_progression.get("state_1_restore_database"):
file_name = os.path.basename(self.file_path)
image_db_file_path = os.path.join("image_db", file_name)
str_will_copy = (
f"🤖 will copy '{self.file_path}' to '{image_db_file_path}'"
)
if not shutil._samefile(self.file_path, image_db_file_path):
do_copy = False
if os.path.exists(image_db_file_path):
status_overwrite_image_db = input(
f"{str_will_copy}, "
f"a file already exist, do you want to continue (y/Y) : "
).strip()
if status_overwrite_image_db.lower() == "y":
do_copy = True
os.remove(image_db_file_path)
else:
print(str_will_copy)
do_copy = True
if do_copy:
shutil.copy(self.file_path, image_db_file_path)
neutralize_arg = ""
if start_version >= 16 and do_neutralize:
neutralize_arg = " --neutralize"
status, cmd_executed = self.todo_upgrade_execute(
f"./script/database/db_restore.py --database {database_name} --image {file_name} --ignore_cache{neutralize_arg}",
single_source_odoo=True,
)
if not status:
self.dct_progression["state_1_restore_database"] = True
self.write_config()
print("✅ -> Restore database")
already_update_state_1 = False
if not self.dct_progression.get("state_1_update_all"):
print(
"[1] Update all addons before neutralize (already neutralize by odoo if supported)"
)
wait_continue = (
input(
"💬 Do you need to upgrade before a database neutralization, press to ignore : "
)
.strip()
.lower()
)
if wait_continue == "1":
status, cmd_executed = self.todo_upgrade_execute(
f"./script/addons/update_addons_all.sh {database_name}",
single_source_odoo=True,
)
if not status:
already_update_state_1 = True
self.dct_progression["state_1_update_all"] = True
self.write_config()
print("✅ -> Update database before neutralize by module")
print("✅ -> Neutralize database")
if do_neutralize:
status, cmd_executed = self.todo_upgrade_execute(
f"./script/addons/update_prod_to_dev.sh {database_name}",
single_source_odoo=True,
)
# TODO exécuter next line si status != 0 et log contient
# psycopg2.errors.UndefinedTable: relation "discuss_channel" does not exist
# LIGNE 1 : SELECT "discuss_channel"."id" FROM "discuss_channel" WHERE (...
# source ./.venv.odoo18.0_python3.12.10/bin/activate && cat script/postgresql/migration/fix_migration_postgresql_17_to_postgresql_18_module_mail_nov_2025.py | ./odoo18.0/odoo/odoo-bin shell -d ripbylop_stage_prod_17_nov_2025
# psycopg2.errors.ForeignKeyViolation: insert or update on table "discuss_channel_member" violates foreign key constraint "discuss_channel_member_channel_id_fkey"
# DÉTAIL : Key (channel_id)=(20) is not present in table "discuss_channel".
# ./script/database/migrate/process_backup_file.py --path_backup_zip image_db/db.zip --path_output_zip image_db/dbFIX.zip --word_to_delete discuss_channel_channel_type_not_null
# Puis faire un retry de la commande, sinon rien
# Only record the step when it actually succeeded. The previous
# unconditional assignment made the test below dead code: a failed
# neutralization was remembered as done and skipped on resume.
if not status:
self.dct_progression["state_1_neutralize_database"] = True
self.write_config()
else:
self.dct_progression["state_1_neutralize_database"] = True
self.write_config()
config_state_1_uninstall_module = self.dct_progression.get(
"config_state_1_uninstall_module"
)
config_state_1_install_module = self.dct_progression.get(
"config_state_1_install_module"
)
if not is_state_4_reach_open_upgrade:
lst_module_to_uninstall, lst_uninstall_reason = (
self.read_uninstall_module_list(start_version, database_name)
)
if lst_uninstall_reason:
# Show WHY each module goes away: which modules must be dropped
# depends on the database, and a removal without a stated reason
# is a decision nobody can review later.
print("✨ Modules to uninstall before migration :")
for name, reason, origin in lst_uninstall_reason:
print(
f" - {name}"
+ (
f" — {reason}"
if reason
else " — ⚠️ no reason given"
)
+ f" [{origin}]"
)
if config_state_1_uninstall_module:
lst_module_to_uninstall = (
lst_module_to_uninstall + config_state_1_uninstall_module
)
if lst_module_to_uninstall:
self.uninstall_from_database(
lst_module_to_uninstall, database_name, start_version
)
self.dct_progression["state_1_uninstall_module"] = True
self.write_config()
self.dct_progression["config_state_1_uninstall_module"] = (
config_state_1_uninstall_module
)
self.write_config()
print("✅ -> Uninstall module")
print("✅ -> install module")
if not is_state_4_reach_open_upgrade:
lst_module_to_install = []
if config_state_1_install_module:
lst_module_to_install = (
lst_module_to_install + config_state_1_install_module
)
if lst_module_to_install:
self.install_from_database(
lst_module_to_install, database_name, start_version
)
self.dct_progression["state_1_install_module"] = True
self.write_config()
self.dct_progression["config_state_1_install_module"] = (
config_state_1_install_module
)
self.write_config()
msg = "2 - Succeed update all addons"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
if (
not self.dct_progression.get("state_2_update_all")
and not already_update_state_1
):
status, cmd_executed = self.todo_upgrade_execute(
f"./script/addons/update_addons_all.sh {database_name}",
single_source_odoo=True,
)
if not status:
self.dct_progression["state_2_update_all"] = True
self.write_config()
# Predict the website COW views that the NEXT version bump will break.
# A copy-on-write view freezes the structure of the module view it was
# copied from; when that module view changes mode between two versions,
# the copy keeps an arch written for the old mode and the upgrade dies
# on « Element ... cannot be located in parent view ». Reporting it here
# -- before hours of migration -- leaves time to arbitrate.
# Only the next bump can be predicted: the modes in database describe
# the current version.
self.todo_upgrade_execute(
f"{PYTHON_BIN} ./script/odoo/migration/check_cow_views.py"
f" -d {database_name} -t odoo{start_version + 1}.0",
wait_at_error=False,
)
msg = "3 - Clean up database before data migration"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
if not self.dct_progression.get("state_3_install_clean_database"):
status, cmd_executed = self.todo_upgrade_execute(
f"./script/addons/install_addons.sh {database_name} database_cleanup",
single_source_odoo=True,
)
if not status:
self.dct_progression["state_3_install_clean_database"] = True
self.write_config()
if not self.dct_progression.get("state_3_clean_database"):
print(
"✨ Aller dans «configuration/Technique/Nettoyage.../Purger» les modules obsolètes"
)
status = input(
"💬 Did you finish to clean database? Press y/Y to open server with selenium, else ignore it : "
).strip()
if status.lower().strip() == "y":
self.todo.prompt_execute_selenium_and_run_db(database_name)
status = input("💬 Press to continue state.3 : ").strip()
self.dct_progression["state_3_clean_database"] = True
self.write_config()
self.install_OCA_odoo_module_migrator()
msg = "4 - Upgrade version with OpenUpgrade"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
self.dct_progression["state_4_reach_open_upgrade"] = True
self.write_config()
lst_next_version = [
a for a in range(start_version + 1, end_version + 1)
]
lst_database_name_upgrade = [
f"{database_name}_upgrade_{str(a)}" for a in lst_next_version
]
# Setup lst_switch_odoo
lst_clone_odoo = self.dct_progression.get(
"state_4_clone_odoo_lst", [False] * len(lst_next_version)
)
lst_switch_odoo = self.dct_progression.get(
"state_4_switch_odoo_lst", [False] * len(lst_next_version)
)
lst_module_migrate_odoo = self.dct_progression.get(
"state_4_module_migrate_odoo_lst", [False] * len(lst_next_version)
)
lst_module_uninstall_module = self.dct_progression.get(
"state_4_uninstall_module", [False] * len(lst_next_version)
)
lst_module_install_module = self.dct_progression.get(
"state_4_install_module", [False] * len(lst_next_version)
)
lst_module_search_missing_module = self.dct_progression.get(
"state_4_search_missing_module", [False] * len(lst_next_version)
)
nb_missing_value_switch_odoo = abs(
len(lst_switch_odoo) - len(lst_next_version)
)
if nb_missing_value_switch_odoo:
lst_switch_odoo += [False] * nb_missing_value_switch_odoo
# Setup lst_upgrade_odoo
lst_upgrade_odoo = self.dct_progression.get(
"state_4_upgrade_odoo_lst", [[]] * len(lst_next_version)
)
lst_fix_migration_odoo = self.dct_progression.get(
"state_4_fix_migration_odoo_lst", [[]] * len(lst_next_version)
)
nb_missing_value_upgrade_odoo = abs(
len(lst_upgrade_odoo) - len(lst_next_version)
)
if nb_missing_value_upgrade_odoo:
lst_upgrade_odoo += [[]] * nb_missing_value_upgrade_odoo
database_name_upgrade = None
lst_module_missing_next_version = []
lst_module_to_delete = []
lst_module_to_delete_last_version = []
for index, next_version in enumerate(lst_next_version):
# Reinit the list
lst_module_missing_last_version = lst_module_missing_next_version[
:
]
lst_module_to_delete_last_version.extend(lst_module_to_delete)
lst_module_to_delete = []
msg = f"4.{index} - Ready to work with version {next_version}"
self.add_comment_progression(msg)
option_comment = 0
msg = f"4.{index}.{chr(option_comment + 65)} - Search updated module list to next version"
self.add_comment_progression(msg)
if not database_name_upgrade:
last_database_name = database_name
else:
last_database_name = database_name_upgrade
database_name_upgrade = lst_database_name_upgrade[index]
lst_module_to_uninstall = []
lst_module_to_install = []
lst_module_to_analyse = self.get_rename_module(
self.dct_module_per_version[next_version - 1],
next_version,
)
self.dct_module_per_version[next_version] = sorted(
list(set(lst_module_to_analyse))
)
self.dct_progression["dct_module_per_version"] = (
self.dct_module_per_version
)
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Clone Odoo"
self.add_comment_progression(msg)
if not os.path.exists(f"odoo{next_version}.0/addons/addons"):
os.makedirs(f"odoo{next_version}.0/addons/addons")
if not lst_clone_odoo[index]:
self.switch_odoo(next_version - 1)
print(
f"⧖ -> Clone to odoo.'{next_version}', from '{database_name}' to '{database_name_upgrade}'."
)
# Delete if exist database
self.todo_upgrade_execute(
f"./script/database/db_restore.py -d {database_name_upgrade} --only_drop",
)
# Duplicate database
cmd_clone_database = f"./odoo_bin.sh db --clone --from_database {last_database_name} --database {database_name_upgrade}"
status, cmd_executed = self.todo_upgrade_execute(
cmd_clone_database
)
# Everything downstream runs against this clone: if it failed,
# do not mark it done (a rerun would skip the clone and migrate
# a missing or truncated database).
if status:
print(
f"❌ -> Clone to Odoo{next_version} FAILED (status"
f" {status}). Stopping: '{database_name_upgrade}' is"
" not usable."
)
return
lst_clone_odoo[index] = True
self.dct_progression["state_4_clone_odoo_lst"] = lst_clone_odoo
self.write_config()
print(f"✅ -> Clone Odoo{next_version} done")
else:
print(f"✅ -> Clone Odoo{next_version} - nothing")
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Uninstall module"
self.add_comment_progression(msg)
config_state_4_uninstall_module = self.dct_progression.get(
"config_state_4_uninstall_module",
[False] * len(lst_next_version),
)
if not lst_module_uninstall_module[index]:
lst_module_to_uninstall = config_state_4_uninstall_module[
index
]
if lst_module_to_uninstall:
self.uninstall_from_database(
lst_module_to_uninstall,
database_name_upgrade,
next_version - 1,
)
lst_module_uninstall_module[index] = True
self.dct_progression["state_4_module_migrate_odoo_lst"] = (
lst_module_uninstall_module
)
self.write_config()
self.dct_progression["config_state_4_uninstall_module"] = (
config_state_4_uninstall_module
)
self.dct_progression["state_4_uninstall_module"] = (
lst_module_uninstall_module
)
self.write_config()
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Install module"
self.add_comment_progression(msg)
config_state_4_install_module = self.dct_progression.get(
"config_state_4_install_module",
[False] * len(lst_next_version),
)
# Special case to install module to fix migration
if (
next_version == 13
and "dms" in lst_module_to_analyse
and "muk_dms"
not in self.dct_progression.get("dct_module_exist", {}).keys()
):
# Force install dms into odoo 12
if not config_state_4_install_module[index]:
config_state_4_install_module[index] = ["dms"]
elif "dms" not in config_state_4_install_module[index]:
config_state_4_install_module[index].append("dms")
if not lst_module_install_module[index]:
lst_module_to_install = config_state_4_install_module[index]
if not lst_module_to_install:
lst_module_to_install = []
if lst_module_to_install:
self.install_from_database(
lst_module_to_install,
database_name_upgrade,
next_version - 1,
)
lst_module_install_module[index] = True
self.dct_progression["state_4_module_migrate_odoo_lst"] = (
lst_module_install_module
)
self.write_config()
self.dct_progression["config_state_4_install_module"] = (
config_state_4_install_module
)
self.dct_progression["state_4_install_module"] = (
lst_module_install_module
)
self.write_config()
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Switch Odoo"
self.add_comment_progression(msg)
if not lst_switch_odoo[index]:
self.switch_odoo(next_version)
lst_switch_odoo[index] = True
self.dct_progression["state_4_switch_odoo_lst"] = (
lst_switch_odoo
)
self.write_config()
print(f"✅ -> Switch Odoo{next_version} done with update")
else:
print(f"✅ -> Switch Odoo{next_version} - nothing")
lst_state_4_module_migrate_code = self.dct_progression.get(
"config_state_4_module_to_migrate_code",
[[]] * len(lst_next_version),
)
if (
"config_state_4_module_to_migrate_code"
not in self.dct_progression.keys()
):
self.dct_progression[
"config_state_4_module_to_migrate_code"
] = lst_state_4_module_migrate_code
lst_module_to_migrate = lst_state_4_module_migrate_code[index]
option_comment += 1
msg = (
f"4.{index}.{chr(option_comment + 65)} - Search missing module"
)
self.add_comment_progression(msg)
if not lst_module_search_missing_module[index]:
lst_module_to_analyse_updated = []
for bd_module in lst_module_to_analyse:
if (
lst_module_to_uninstall
and bd_module in lst_module_to_uninstall
):
# Ignore check if uninstall before
continue
lst_module_to_analyse_updated.append(bd_module)
# TODO remove from list past module deleted
lst_module_to_check = [
a
for a in lst_module_to_analyse_updated
if a not in lst_module_to_delete_last_version
]
(
lst_module_missing_next_version,
lst_module_duplicate_next_version,
) = self.check_addons_exist(lst_module_to_check)
lst_module_missing_next_version = sorted(
list(set(lst_module_missing_next_version))
)
self.dct_progression["state_4_len_lst_module_missing"] = len(
lst_module_missing_next_version
)
self.dct_progression["state_4_lst_module_missing"] = (
lst_module_missing_next_version
)
lst_module_duplicate = sorted(
list(set(lst_module_duplicate_next_version))
)
self.dct_progression["state_4_len_lst_module_duplicate"] = len(
lst_module_duplicate
)
self.dct_progression["state_4_lst_module_duplicate"] = (
lst_module_duplicate
)
if lst_module_duplicate:
print(f"Duplicate module into odoo{next_version} : ")
print(lst_module_duplicate)
input(
f"💬 Detect error duplicate module, manage this problem manually and press to continue."
)
# if lst_module_missing_next_version and not lst_module_to_migrate:
if lst_module_missing_next_version:
# TODO support when lst_module_to_migrate is fill
lst_module_to_migrate = []
print(
f"👹 Detect error missing module, missing module into odoo{next_version} :"
)
for index_missing_module, module_missing in enumerate(
lst_module_missing_next_version
):
old_path = self.dct_progression.get(
"dct_module_exist", {}
).get(module_missing)
print(
f"[{index_missing_module}] {module_missing} - {old_path}"
)
print("[a] All list above")
print("[e] Add extra custom")
want_continue = (
input(
f"💬 Enumerate missing module separate by coma to delete it"
f". The others will be migrate : "
)
.strip()
.lower()
)
is_delete_all = False
if want_continue:
lst_want_continue = [
a.strip() for a in want_continue.split(",")
]
if "a" in lst_want_continue:
is_delete_all = True
lst_module_to_delete = [
lst_module_missing_next_version[a]
for a in range(
len(lst_module_missing_next_version)
)
]
else:
# TODO show error if the index is wrong
lst_want_continue_number = [
int(a)
for a in lst_want_continue
if a.isdigit()
]
lst_module_to_delete = [
lst_module_missing_next_version[a]
for a in lst_want_continue_number
if 0
<= a
< len(lst_module_missing_next_version)
]
if len(lst_module_to_delete) == len(
lst_module_missing_next_version
):
is_delete_all = True
if next_version == 15:
lst_module_to_delete.append("users_default_groups")
lst_module_to_delete.append(
"web_editor_backend_context"
)
lst_module_to_delete.append(
"website_google_analytics_fixed"
)
elif next_version == 17:
lst_module_to_delete.append(
"export_delete_login_log"
)
elif next_version == 18:
lst_module_to_delete.append(
"base_attachment_object_storage"
)
lst_module_to_delete.append(
"user_password_strength"
)
if "e" in lst_want_continue:
want_continue = (
input(
f"💬 Enumerate module name to delete, separate by coma : "
)
.strip()
.lower()
)
lst_to_extend = [
a.strip() for a in want_continue.split(",")
]
lst_module_to_delete.extend(lst_to_extend)
if lst_module_to_delete:
msg = f"4.{index}.{chr(option_comment + 65)}.option - Choose delete missing module"
self.add_comment_progression(msg)
self.switch_odoo(next_version - 1)
if lst_module_to_delete:
# Delete if exist database
self.todo_upgrade_execute(
f"./script/database/db_restore.py -d {database_name_upgrade} --only_drop",
)
# Duplicate database
cmd_clone_database = f"./odoo_bin.sh db --clone --from_database {last_database_name} --database {database_name_upgrade}"
self.todo_upgrade_execute(cmd_clone_database)
self.uninstall_from_database(
lst_module_to_delete,
database_name_upgrade,
next_version,
)
self.install_from_database(
lst_module_to_install,
database_name_upgrade,
next_version - 1,
)
if not is_delete_all:
msg = f"4.{index}.{chr(option_comment + 65)}.option - Choose auto-fix (not implemented yet)"
self.add_comment_progression(msg)
lst_module_to_migrate_code = set(
lst_module_missing_next_version
) - set(lst_module_to_delete)
(
lst_module_missing_last,
lst_module_duplicate_last,
lst_module_exist_last,
lst_module_error_last,
) = self.check_addons_exist(
lst_module_to_migrate_code, get_all_info=True
)
if lst_module_missing_last:
print(
f"Error missing module : {lst_module_missing_last}"
)
if lst_module_duplicate_last:
print(
f"Error duplicate module : {lst_module_duplicate_last}"
)
if lst_module_error_last:
print(
f"Error error module : {lst_module_error_last}"
)
if lst_module_exist_last:
odoo_name_last_version = (
f"odoo{next_version - 1}.0"
)
odoo_name_actual_version = f"odoo{next_version}.0"
for (
module_name,
module_path,
) in lst_module_exist_last:
module_dir_path = os.path.dirname(module_path)
module_dir_path_new_version = (
module_dir_path.replace(
odoo_name_last_version,
odoo_name_actual_version,
)
)
module_dir_path_manifest = os.path.join(
module_path, "__manifest__.py"
)
module_dir_new_version = os.path.join(
module_dir_path_new_version, module_name
)
module_dir_new_version_manifest = os.path.join(
module_dir_new_version, "__manifest__.py"
)
dct_module_to_migrate_module = {
"source_module_path": module_path,
"source_manifest_path": module_dir_path_manifest,
"source_addons_path": module_dir_path,
"target_module_path": module_dir_new_version,
"target_manifest_path": module_dir_new_version_manifest,
"target_addons_path": module_dir_path_new_version,
"module_name": module_name,
"source_version_odoo": next_version - 1,
"target_version_odoo": next_version,
}
# TODO move this into config
lst_module_to_migrate.append(
dct_module_to_migrate_module
)
self.dct_progression[
"config_state_4_module_to_migrate_code"
][index] = lst_module_to_migrate
self.write_config()
self.switch_odoo(next_version)
# TODO auto-fix
# TODO try to migrate module, find in previous version, application la migration vers une nouvelle version
# TODO ajouté menu todo qui permet de faire une migration d'un module et migrer le générateur de code.
# TODO when check module, reminder provenance
# TODO implement asyncio instead of parallel
# TODO detect when duplicate path module ou module manquant, prendre décision qui ont efface si dupliqué
# TODO pourquoi web_ir_actions_act_multi est doublé dans odoo 13
lst_module_search_missing_module[index] = True
self.dct_progression["state_4_search_missing_module"] = (
lst_module_search_missing_module
)
self.write_config()
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Migrate module"
self.add_comment_progression(msg)
lst_path_git_clone_migrate = []
if not lst_module_migrate_odoo[index]:
# TODO Searching module
# search addons/addons
# search read manifest and detect branch difference, manifest into private?
# Extract module name and run migration to another list
# Maybe check if already exist and show list or continue with overwrite
# Expliquer pourquoi on ne fait pas le oca-port, c'est
config_migrate_repo = self.dct_progression.get(
"config_migrate_repo", False
)
self.dct_progression["config_migrate_repo"] = (
config_migrate_repo
)
if config_migrate_repo:
dct_module_result = self.search_module_to_move(
next_version - 1, next_version
)
# TODO code migration
# git stash
# call odoo-module-migrate, without commit
source_module_path = dct_module_result.get(
"source_module_path"
)
if not source_module_path:
_logger.error(
f"Missing source module path '{source_module_path}'"
)
else:
if os.path.exists(
os.path.join(source_module_path, ".git")
):
self.todo_upgrade_execute(
f"cd '{source_module_path}' && git stash && cd ~-"
)
target_module_path = dct_module_result.get(
"target_module_path"
)
if not target_module_path:
_logger.error(
f"Missing target module path '{target_module_path}'"
)
else:
if os.path.exists(
os.path.join(target_module_path, ".git")
):
self.todo_upgrade_execute(
f"cd '{target_module_path}' && git stash && cd ~-"
)
lst_module_to_migrate_all = dct_module_result.get(
"lst_module", []
)
else:
lst_module_to_migrate_all = []
# TODO remove duplicate au lieu d'extend
lst_module_to_migrate_all.extend(lst_module_to_migrate)
self.internal_module_upgrade(
next_version,
lst_module_to_migrate_all,
lst_path_git_clone_migrate,
)
lst_module_migrate_odoo[index] = True
self.dct_progression["state_4_module_migrate_odoo_lst"] = (
lst_module_migrate_odoo
)
self.write_config()
print(f"✅ -> Module upgrade Odoo{next_version} done")
else:
print(f"✅ -> Module upgrade Odoo{next_version} - nothing")
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Fix migrate code"
self.add_comment_progression(msg)
if not lst_fix_migration_odoo[index]:
print("")
file_path_fix_migration = os.path.join(
"script",
"odoo",
"migration",
f"fix_migration_odoo{(next_version-1)*10}_to_odoo{next_version*10}.py",
)
if os.path.exists(file_path_fix_migration):
status, cmd_executed = self.todo_upgrade_execute(
f"cat ./{file_path_fix_migration} | ./odoo{next_version}.0/odoo/odoo-bin shell -d {database_name_upgrade}",
single_source_odoo=True,
)
# A fix that did not run must not be recorded as applied,
# otherwise the rerun skips it and OpenUpgrade hits the very
# problem the fix exists to prevent.
if status:
print(
f"❌ -> Fix migration Odoo{next_version} FAILED"
f" (status {status}):"
f" {file_path_fix_migration}"
)
return
lst_fix_migration_odoo[index] = file_path_fix_migration
self.dct_progression["state_4_fix_migration_odoo_lst"] = (
lst_fix_migration_odoo
)
self.write_config()
print(f"✅ -> Fix migration Odoo{next_version} done")
else:
print(
f"✅ -> Fix migration Odoo{next_version} - no fix to execute"
)
else:
print(f"✅ -> Fix migration Odoo{next_version} - nothing")
for path_git_clone_migrate in lst_path_git_clone_migrate:
cmd = f"./script/code/git_commit_migration_addons_path.py --path {path_git_clone_migrate} --odoo_version {next_version}.0"
self.todo_upgrade_execute(cmd)
option_comment += 1
msg = f"4.{index}.{chr(option_comment + 65)} - Migrate database"
self.add_comment_progression(msg)
if not lst_upgrade_odoo[index]:
path_addons_openupgrade = os.path.join(
os.getcwd(), f"odoo{next_version}.0", "OCA_OpenUpgrade"
)
# Update config with OCA_OpenUpgrade
ignore_path = (
"--ignore-odoo-path " if next_version <= 13 else ""
)
extra_addons_path_extra = (
f",{path_addons_openupgrade}/addons,{path_addons_openupgrade}/odoo/addons"
if next_version <= 13
else ""
)
cmd_update_config = (
f"./script/git/git_repo_update_group.py {ignore_path}"
f"--extra-addons-path {path_addons_openupgrade}{extra_addons_path_extra} "
f"&& ./script/generate_config.sh"
)
self.todo_upgrade_execute(cmd_update_config)
print("🚸 Please, validate commits after code migration.")
print("ℹ To show repo status :\nmake repo_show_status")
print(
f"🚸 Please, validate this path into config.conf : '{path_addons_openupgrade}'."
)
status = input(f"💬 Press to continue {msg} : ").strip()
# The technique change at version 14
if next_version <= 13:
erplibre_version = self.install_OCA_openupgrade(
next_version
)
cmd_upgrade = f".venv.{erplibre_version}/bin/python ./odoo{next_version}.0/OCA_OpenUpgrade/odoo-bin -c ./config.conf --update all --no-http --stop-after-init -d {database_name_upgrade}"
else:
cmd_upgrade = f"./run.sh --upgrade-path=./odoo{next_version}.0/OCA_OpenUpgrade/openupgrade_scripts/scripts --update all -c config.conf --stop-after-init --no-http --load=base,web,openupgrade_framework -d {database_name_upgrade}"
lst_upgrade_odoo[index] = cmd_upgrade
status, cmd_executed = self.todo_upgrade_execute(
cmd_upgrade,
new_env={
"OPENUPGRADE_TARGET_VERSION": f"{next_version}.0"
},
)
# This is THE data migration. Recording it as done when it
# failed used to send the loop to the next version on top of a
# half-migrated database. Stop here instead: the state stays
# unset, so a rerun replays this version.
if status:
print(
f"❌ -> Database migration to Odoo{next_version} FAILED"
f" (status {status}). Stopping before version"
f" {next_version + 1} to avoid migrating a broken"
" database. Fix the cause, then relaunch: this version"
" will be replayed."
)
return
self.dct_progression["state_4_upgrade_odoo_lst"] = (
lst_upgrade_odoo
)
self.write_config()
str_wait_next_version = (
" (or wait next version 🤖)"
if next_version != lst_next_version[-1]
else ""
)
status = (
input(
f"💬 Do you want to upgrade all{str_wait_next_version}? Press y/Y to upgrade all addons database : "
)
.strip()
.lower()
)
if status == "y":
self.todo_upgrade_execute(
f"./script/addons/update_addons_all.sh {database_name_upgrade}",
)
print(f"✅ -> Database upgrade Odoo{next_version} done")
# Update config without OCA_OpenUpgrade
cmd_update_config = f"./script/git/git_repo_update_group.py && ./script/generate_config.sh"
self.todo_upgrade_execute(cmd_update_config)
print("[y] Open server with Selenium")
status = (
input(
"💬 Do you want to test this upgrade? Choose or press to ignore it : "
)
.strip()
.lower()
)
"make repo_show_status"
if status == "y":
self.todo.prompt_execute_selenium_and_run_db(
database_name_upgrade
)
status = input(
f"💬 Press to continue 4.{index} : "
).strip()
else:
print(f"✅ -> Database upgrade Odoo{next_version} - nothing")
#
# waiting_input = input("💬 Press any keyboard key to continue...")
print("")
msg = "5 - Cleaning up database after upgrade"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
print(
"✨ Re-update i18n, purger data, tables (except mail_test and mail_test_full)"
)
# waiting_input = input("💬print Press any keyboard key to continue...")
msg = "6 - Migration finished"
print(f"🔷 {msg}")
self.add_comment_progression(msg)
cmd_backup_template = f"./odoo_bin.sh db --backup --database {database_name_upgrade} --restore_image"
cmd_backup = f"{cmd_backup_template} {database_name_upgrade}_finish_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"
print(f"✨ Can execute backup creation :\n{cmd_backup}")
status = input(
"💬 Press y/Y or write filename.zip to export or enter to continue : "
).strip()
if status.lower():
if status.lower() != "y":
cmd_backup = f"{cmd_backup_template} {status}"
self.todo_upgrade_execute(cmd_backup)
status = input("💬 Test the migration, press y/Y : ")
if status.lower().strip() == "y":
self.todo.prompt_execute_selenium_and_run_db(database_name_upgrade)
def get_rename_module(self, lst_module, next_version):
path_search = f"odoo{next_version}.0/OCA_OpenUpgrade/"
status, cmd_executed, lst_output = self.todo_upgrade_execute(
f"find {path_search} -name apriori.py",
get_output=True,
)
if not lst_output:
_logger.error(
f"Cannot find renamed module script apriori.py into path '{path_search}'"
)
return lst_module
apriory_py = lst_output[0].strip()
with open(apriory_py, "r") as f:
file_content = f.read()
data_vars = {}
exec(file_content, data_vars)
renamed_modules = data_vars.get("renamed_modules", {})
merged_modules = data_vars.get("merged_modules", {})
deleted_modules = data_vars.get("deleted_modules", [])
lst_index_to_delete = []
for index, module in enumerate(lst_module):
renamed_module = renamed_modules.get(module)
merged_module = merged_modules.get(module)
if renamed_module:
lst_module[index] = renamed_module
if merged_module:
lst_module[index] = merged_module
if module in deleted_modules:
lst_index_to_delete.append(index)
for index_to_delete in lst_index_to_delete[::-1]:
lst_module.pop(index_to_delete)
return list(set(lst_module))
def search_module_to_move(self, source_version_odoo, target_version_odoo):
lst_module = []
# lst_path_to_check = [
# os.path.join(f"odoo{actual_version_odoo}", "addons"),
# os.path.join(f"odoo{target_version_odoo}", "addons"),
# os.path.join(f"private", "addons"),
# ]
source_path_to_check = os.path.join(
f"odoo{source_version_odoo}.0", "addons", "addons"
)
target_path_to_check = os.path.join(
f"odoo{target_version_odoo}.0", "addons", "addons"
)
# Search
is_moving_git = False
if os.path.exists(source_path_to_check):
if os.path.exists(os.path.join(source_path_to_check, ".git")):
is_moving_git = True
if not os.path.exists(target_path_to_check):
shutil.copytree(source_path_to_check, target_path_to_check)
# if not is_moving_git:
# os.mkdir(path_to_check_target)
# else:
# # TODO clone
# pass
if not is_moving_git:
# TODO do something
# Time to compare
os.listdir(source_path_to_check)
if os.path.exists(target_path_to_check):
for dir_name in os.listdir(target_path_to_check):
source_module_path = os.path.join(
source_path_to_check, dir_name
)
source_manifest_path = os.path.join(
source_module_path, "__manifest__.py"
)
target_module_path = os.path.join(
target_path_to_check, dir_name
)
target_manifest_path = os.path.join(
target_module_path, "__manifest__.py"
)
if os.path.exists(target_manifest_path):
# TODO remove from list when module already exist in version 15
dct_module = {
"source_module_path": source_module_path,
"source_manifest_path": source_manifest_path,
"source_addons_path": source_path_to_check,
"target_module_path": target_module_path,
"target_manifest_path": target_manifest_path,
"target_addons_path": target_path_to_check,
"module_name": dir_name,
"source_version_odoo": source_version_odoo,
"target_version_odoo": target_version_odoo,
}
lst_module.append(dct_module)
dct_module = {
"lst_module": lst_module,
"source_module_path": source_path_to_check,
"target_module_path": target_path_to_check,
}
return dct_module
@staticmethod
def parse_module_list_file(file_path):
"""Read a module list file, return [(module, reason), ...].
Accepted syntax, one module per line with an optional justification:
queue_job # blocks 12->13, trigger queue_job_notify
mgmtsystem_hazard # not ported to 13.0
Commas and several names per line are also accepted, so a list copied
from a command line works as-is. Blank lines and full-line comments are
ignored.
The previous parser was « f.readline().split() »: it kept only the FIRST
line, so a multi-line list was silently truncated, and a comma-separated
list collapsed into one bogus module name.
"""
lst_module = []
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
content, _, reason = line.partition("#")
for module_name in content.replace(",", " ").split():
lst_module.append((module_name, reason.strip()))
return lst_module
def read_uninstall_module_list(self, start_version, database_name):
"""Modules to uninstall before migrating start_version -> next.
Reads the per-database private list first, then the shared versioned
defaults; duplicates are dropped, keeping the first occurrence.
Returns (lst_module, lst_detail) where lst_detail carries
(module, reason, origin_file) so the caller can justify each removal.
"""
file_name = (
f"uninstall_module_list_odoo{start_version * 10}"
f"_to_odoo{(start_version + 1) * 10}.txt"
)
lst_path = [
os.path.join(PATH_MIGRATION_PRIVATE, database_name, file_name),
os.path.join(PATH_MIGRATION_GLOBAL, file_name),
]
lst_module = []
lst_detail = []
for file_path in lst_path:
if not os.path.exists(file_path):
continue
for module_name, reason in self.parse_module_list_file(file_path):
if module_name in lst_module:
continue
lst_module.append(module_name)
lst_detail.append((module_name, reason, file_path))
return lst_module, lst_detail
def uninstall_from_database(
self, lst_module_to_uninstall, database_name, actual_version
):
if not lst_module_to_uninstall:
return
uninstall_module = ",".join(lst_module_to_uninstall)
self.todo_upgrade_execute(
f"./script/addons/uninstall_addons.sh {database_name} {uninstall_module}",
single_source_odoo=True,
)
# Update list installed module
self.dct_module_per_version[actual_version] = sorted(
list(
set(self.dct_module_per_version[actual_version])
- set(lst_module_to_uninstall)
)
)
self.dct_progression["dct_module_per_version"] = (
self.dct_module_per_version
)
self.write_config()
def install_from_database(
self, lst_module_to_install, database_name, actual_version
):
if not lst_module_to_install:
return
install_module = ",".join(lst_module_to_install)
self.todo_upgrade_execute(
f"./script/addons/install_addons.sh {database_name} {install_module}",
single_source_odoo=True,
)
# Update list installed module
self.dct_module_per_version[actual_version] = sorted(
list(
set(
self.dct_module_per_version[actual_version]
+ lst_module_to_install
)
)
)
self.dct_progression["dct_module_per_version"] = (
self.dct_module_per_version
)
self.write_config()
def check_addons_exist(
self, lst_module_to_check, ignore_error=True, get_all_info=False
):
str_module_to_check = ",".join(sorted(lst_module_to_check))
status, cmd_executed, dct_output = self.todo_upgrade_execute(
f"{PYTHON_BIN} ./script/addons/check_addons_exist.py --format_json --output_json -m {str_module_to_check}",
get_output=True,
output_is_json=True,
wait_at_error=not ignore_error,
)
lst_module_missing = dct_output.get("missing")
lst_module_duplicate = dct_output.get("duplicate")
if get_all_info:
lst_module_error = dct_output.get("error")
lst_module_exist = dct_output.get("exist")
return (
lst_module_missing,
lst_module_duplicate,
lst_module_exist,
lst_module_error,
)
return lst_module_missing, lst_module_duplicate
def switch_odoo(self, odoo_version):
int_odoo_version = int(float(odoo_version))
# Expect odoo_version like 12.0
lst_version, lst_version_installed, odoo_installed_version = (
get_odoo_version()
)
if odoo_installed_version != f"odoo{int_odoo_version}.0":
print(
f"⧖ -> Was '{odoo_installed_version}', Switch to odoo{int_odoo_version}.0"
)
self.todo_upgrade_execute(f"make switch_odoo_{int_odoo_version}")
self.todo_upgrade_execute("make config_gen_all")
def install_OCA_odoo_module_migrator(self):
if not os.path.exists(PATH_VENV_MODULE_MIGRATOR):
self.todo_upgrade_execute(
f"cd {PATH_OCA_ODOO_MODULE_MIGRATOR} && python -m venv {VENV_NAME_MODULE_MIGRATOR} && source {VENV_NAME_MODULE_MIGRATOR}/bin/activate && pip3 install -r requirements.txt"
)
def install_OCA_openupgrade(self, next_version):
# TODO install odoorpc==0.7.0
# openupgradelib
# openupgrade_path = f"odoo{next_version}.0/OCA_OpenUpgrade"
# venv_oca_path = f"{openupgrade_path}/.venv"
# if os.path.exists(venv_oca_path):
# return
lst_version, lst_version_installed, odoo_installed_version = (
get_odoo_version()
)
extract_version = f"{next_version}.0"
dct_erplibre_info = [
a for a in lst_version if a.get("odoo_version") == extract_version
]
if not dct_erplibre_info:
raise Exception(f"Cannot extract {extract_version}")
dct_erplibre_info = dct_erplibre_info[0]
erplibre_version = dct_erplibre_info.get("erplibre_version")
# self.todo_upgrade_execute(
# f".venv.{erplibre_version}/bin/python -m venv {venv_oca_path} && {venv_oca_path}/bin/pip3 install -r {openupgrade_path}/requirements.txt"
# )
self.todo_upgrade_execute(
f".venv.{erplibre_version}/bin/pip install odoorpc==0.7.0"
)
self.todo_upgrade_execute(
f".venv.{erplibre_version}/bin/pip install openupgradelib"
)
return erplibre_version
def todo_upgrade_execute(
self,
cmd,
single_source_odoo=False,
new_env=None,
quiet=False,
get_output=False,
output_is_json=False,
wait_at_error=True,
):
if output_is_json and not get_output:
get_output = True
output = None
if get_output:
status, cmd_executed, output = self.execute.exec_command_live(
cmd,
source_erplibre=False,
single_source_odoo=single_source_odoo,
new_env=new_env,
return_status_and_output_and_command=True,
quiet=quiet,
)
else:
status, cmd_executed = self.execute.exec_command_live(
cmd,
source_erplibre=False,
single_source_odoo=single_source_odoo,
new_env=new_env,
return_status_and_command=True,
quiet=quiet,
)
self.lst_command_executed.append(cmd_executed)
self.dct_progression["command_executed"] = self.lst_command_executed
self.write_config()
# None means « the command never reported a status » -> treat it as a
# failure, never as a success (defence in depth: exec_command_live now
# always sets one, but a silent None must not skip this prompt).
if (status is None or status) and wait_at_error:
print("[1] to redo the command")
wait_status = (
input(
"💬 Error detected, press to continue or ctrl+c to stop : "
)
.strip()
.lower()
)
# psycopg2.errors.UndefinedTable: relation "discuss_channel" does not exist
# LIGNE 1 : SELECT "discuss_channel"."id" FROM "discuss_channel" WHERE (...
if wait_status == "1":
return self.todo_upgrade_execute(
cmd,
single_source_odoo=single_source_odoo,
new_env=new_env,
quiet=quiet,
get_output=get_output,
output_is_json=output_is_json,
wait_at_error=wait_at_error,
)
if get_output:
if output_is_json:
str_output = json.loads("".join(output))
return status, cmd_executed, str_output
return status, cmd_executed, output
return status, cmd_executed
def check_and_clone_source_to_target_migration_code(
self, next_version, source_addons_path, target_addons_path
):
if not os.path.exists(os.path.join(source_addons_path, ".git")):
return
if os.path.exists(os.path.join(target_addons_path, ".git")):
return
# if not os.path.exists(target_addons_path):
# cmd_mkdir = f"mkdir -p {target_addons_path}"
# status, cmd_executed, lst_output = self.todo_upgrade_execute(
# cmd_mkdir,
# get_output=True,
# )
# return
source_dir_name = os.path.basename(source_addons_path)
# Clone a project for next version
# Get actual branch
cmd_git_clone_migrate_source = (
f"cd {source_addons_path} && "
f"git branch --show-current && "
f"cd ~-"
)
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_git_clone_migrate_source,
get_output=True,
)
branch_source = lst_output[0].strip() if len(lst_output) else ""
if not branch_source:
# Get branch from repo
branch_source = self.get_branch_name_from_local_manifest(
source_addons_path, f"{next_version}.0"
)
branch_target = branch_source.replace(
str(next_version - 1), str(next_version)
)
# Get remote branch for actual version
cmd_git_clone_migrate_source_same_target = (
f"cd {source_addons_path} "
f"&& git fetch --all "
f'&& git branch -vv | grep "{branch_source}" '
f"&& cd ~-"
)
cmd_git_clone_migrate_source_same_target_remote_only = (
f"cd {source_addons_path} && git remote && cd ~-"
)
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_git_clone_migrate_source_same_target,
get_output=True,
wait_at_error=False,
)
if status == 1:
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_git_clone_migrate_source_same_target_remote_only,
get_output=True,
)
if lst_output:
remote = lst_output[0].strip()
# TODO write this modification into repo manifest
else:
remote = input(
"👹 BUG, you need to push last branch. Please write the remote/ :"
)
else:
local_branch, remote, remote_branch = (
self.get_local_branch_remote_actual_branch_git(
lst_output,
)
)
# Get remote branch for next version
remote_branch_target = f"{remote}/{branch_target}"
cmd_git_clone_migrate_source_same_target = (
f"cd {source_addons_path} "
f"&& git fetch --all "
f'&& git branch --remotes -vv | grep "{remote_branch_target} " '
f"&& cd ~-"
)
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_git_clone_migrate_source_same_target,
get_output=True,
wait_at_error=False,
)
has_existing_target_branch = any([a.strip() for a in lst_output])
# TODO check config if path is added
# Get remote branch address
cmd_remote_address = (
f"cd {source_addons_path} "
f"&& git remote get-url {remote} "
f"&& cd ~-"
)
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_remote_address,
get_output=True,
)
remote_address = lst_output[0].strip()
# TODO some time, the clone has error, need to repeat
branch_to_clone = (
branch_target if has_existing_target_branch else branch_source
)
cmd_git_clone = (
f"cd {os.path.dirname(target_addons_path)} "
f"&& git clone {remote_address} {source_dir_name} -b {branch_to_clone} && cd ~-"
)
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_git_clone,
get_output=True,
)
if not has_existing_target_branch:
cmd_git_clone = (
f"cd {target_addons_path} "
f"&& if git rev-parse --verify --quiet refs/heads/{branch_target}; then "
f"git checkout {branch_target}; else git checkout -b {branch_target}; fi "
f"&& cd ~-"
)
status, cmd_executed, lst_output = self.todo_upgrade_execute(
cmd_git_clone,
get_output=True,
)
def get_branch_name_from_local_manifest(self, addons_path, default_branch):
relative_addons_path = addons_path.replace(os.getcwd() + "/", "")
git_tool = GitTool()
dct_remote, dct_project, default_remote = (
git_tool.get_manifest_xml_info(filename=LOCAL_MANIFEST)
)
for dct_repo in dct_project.values():
path = dct_repo.get("@path")
if path == relative_addons_path:
revision = dct_repo.get("@revision")
return revision
return default_branch
def get_local_branch_remote_actual_branch_git(self, lst_output):
for line in lst_output:
# The current branch is marked with an asterisk (*) at the start of the line
if not line.strip().startswith("*"):
continue
# Split the line to isolate the remote and remote branch name
parts = line.split()
if len(parts) >= 4:
# The remote and remote branch are in the 4th part, e.g., '[origin/main]'
remote_info = parts[3].strip("[]")
# Split 'origin/main' into 'origin' and 'main'
# remote, remote_branch = remote_info.split("/", 1)
try:
remote, remote_branch = remote_info.split("/", 1)
except ValueError as e:
# # TODO this means no remote, take default one? or last one?
# # TODO supporter la migration 17 vers 18
# print("👹 BUG, you need to push ")
# TODO search remote and associate branch
value = input(
"👹 BUG, you need to push last branch. Please write the remote/branch_name :"
)
remote, remote_branch = value.split("/", 1)
return parts[1], remote, remote_branch
return None, None, None
def add_comment_progression(self, comment):
comment_to_add = f"# {comment}"
self.lst_command_executed.append(comment_to_add)
self.dct_progression["command_executed"] = self.lst_command_executed
self.write_config()
Reference in a new issue View git blame Copy permalink