#614630040

Committed 06 Dec 2023 12:35PM UTC coverage: 86.675%. First build

Build # #614630040

Build Type

Pull #87

travis-ci

Pull Request Pull Request #87: Master with controller adhoc

Run Details

20 of 34 new or added lines in 2 files covered. (58.82%)

683 of 788 relevant lines covered (86.68%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.87

/odoo_module_migrate/base_migration_script.py

import os
from .config import _ALLOWED_EXTENSIONS
from .tools import _execute_shell
from .log import logger
from . import tools
import re
import pathlib
import traceback
import inspect
import glob
import yaml
import importlib


class BaseMigrationScript(object):
    _TEXT_REPLACES = {}
    _TEXT_ERRORS = {}
    _TEXT_WARNINGS = {}
    _DEPRECATED_MODULES = []
    _FILE_RENAMES = {}
    _REMOVED_FIELDS = []
    _RENAMED_FIELDS = []
    _RENAMED_MODELS = []
    _REMOVED_MODELS = []
    _GLOBAL_FUNCTIONS = []  # [function_object]
    _module_path = ""

    def parse_rules(self):
        script_parts = inspect.getfile(self.__class__).split("/")
        migrate_from_to = script_parts[-1].split(".")[0]
        migration_scripts_dir = "/".join(script_parts[:-1])

        TYPE_ARRAY = "TYPE_ARRAY"
        TYPE_DICT = "TYPE_DICT"
        TYPE_DICT_OF_DICT = "TYPE_DICT_OF_DICT"
        rules = {
            # {filetype: {regex: replacement}}
            "_TEXT_REPLACES": {
                "type": TYPE_DICT_OF_DICT,
                "doc": {},
            },
            # {filetype: {regex: message}}
            "_TEXT_ERRORS": {
                "type": TYPE_DICT_OF_DICT,
                "doc": {},
            },
            # {filetype: {regex: message}}
            "_TEXT_WARNINGS": {
                "type": TYPE_DICT_OF_DICT,
                "doc": {},
            },
            # [(module, why, ...)]
            "_DEPRECATED_MODULES": {
                "type": TYPE_ARRAY,
                "doc": [],
            },
            # {old_name: new_name}
            "_FILE_RENAMES": {
                "type": TYPE_DICT,
                "doc": {},
            },
            # [(model_name, field_name, more_info), ...)]
            "_REMOVED_FIELDS": {
                "type": TYPE_ARRAY,
                "doc": [],
            },
            # [(model_name, old_field_name, new_field_name, more_info), ...)]
            "_RENAMED_FIELDS": {
                "type": TYPE_ARRAY,
                "doc": [],
            },
            # [(old.model.name, new.model.name, more_info)]
            "_RENAMED_MODELS": {
                "type": TYPE_ARRAY,
                "doc": [],
            },
            # [(old.model.name, more_info)]
            "_REMOVED_MODELS": {
                "type": TYPE_ARRAY,
                "doc": [],
            },
        }
        # read
        for rule in rules.keys():
            rule_folder = rule[1:].lower()
            file_pattern = "%s/%s/%s/*.yaml" % (
                migration_scripts_dir,
                rule_folder,
                migrate_from_to,
            )
            for filename in glob.glob(file_pattern):
                with open(filename) as f:
                    new_rules = yaml.safe_load(f)
                    if rules[rule]["type"] == TYPE_DICT_OF_DICT:
                        for f_type, data in new_rules.items():
                            if f_type not in rules[rule]["doc"]:
                                rules[rule]["doc"][f_type] = {}
                            rules[rule]["doc"][f_type].update(data)
                    elif rules[rule]["type"] == TYPE_DICT:
                        rules[rule]["doc"].update(new_rules)
                    elif rules[rule]["type"] == TYPE_ARRAY:
                        rules[rule]["doc"].extend(new_rules)
        # extend
        for rule, data in rules.items():
            rtype = data["type"]
            doc = data.get("doc")
            if not doc:
                continue

            rvalues = getattr(self, rule)
            if rtype == TYPE_ARRAY:
                rvalues.extend(doc)
            elif rtype == TYPE_DICT:
                rvalues.update(doc)
            else:
                # TYPE_DICT_OF_DICT
                for filetype, values in doc.items():
                    rvalues.setdefault(filetype, {})
                    rvalues[filetype].update(values or {})

        file_pattern = "%s/python_scripts/%s/*.py" % (
            migration_scripts_dir,
            migrate_from_to,
        )
        for path in glob.glob(file_pattern):
            module_name = path.split("/")[-1].split(".")[0]
            module_name = ".".join(
                [
                    "odoo_module_migrate.migration_scripts.python_scripts",
                    migrate_from_to,
                    module_name,
                ]
            )
            module = importlib.import_module(module_name)
            for name, value in inspect.getmembers(module, inspect.isfunction):
                if not name.startswith("_"):
                    self._GLOBAL_FUNCTIONS.append(value)

    def run(
        self,
        module_path,
        manifest_path,
        module_name,
        migration_steps,
        directory_path,
        commit_enabled,
    ):
        logger.debug(
            "Running %s script" % inspect.getfile(self.__class__).split("/")[-1]
        )
        self.parse_rules()
        manifest_path = self._get_correct_manifest_path(
            manifest_path, self._FILE_RENAMES
        )
        for root, directories, filenames in os.walk(module_path.resolve()):
            for filename in filenames:
                extension = os.path.splitext(filename)[1]
                if extension not in _ALLOWED_EXTENSIONS:
                    continue
                self.process_file(
                    root,
                    filename,
                    extension,
                    self._FILE_RENAMES,
                    directory_path,
                    commit_enabled,
                )

        self.handle_deprecated_modules(manifest_path, self._DEPRECATED_MODULES)

        if self._GLOBAL_FUNCTIONS:
            for function in self._GLOBAL_FUNCTIONS:
                function(
                    logger=logger,
                    module_path=module_path,
                    module_name=module_name,
                    manifest_path=manifest_path,
                    migration_steps=migration_steps,
                    tools=tools,
                )

    def process_file(
        self, root, filename, extension, file_renames, directory_path, commit_enabled
    ):
        # Skip useless file
        # TODO, skip files present in some folders. (for exemple 'lib')
        absolute_file_path = os.path.join(root, filename)
        logger.debug("Migrate '%s' file" % absolute_file_path)

        # Rename file, if required
        new_name = file_renames.get(filename)
        if new_name:
            self._rename_file(
                directory_path,
                absolute_file_path,
                os.path.join(root, new_name),
                commit_enabled,
            )
            absolute_file_path = os.path.join(root, new_name)

        removed_fields = self.handle_removed_fields(self._REMOVED_FIELDS)
        renamed_fields = self.handle_renamed_fields(self._RENAMED_FIELDS)
        renamed_models = self.handle_renamed_models(self._RENAMED_MODELS)
        removed_models = self.handle_removed_models(self._REMOVED_MODELS)

        # Operate changes in the file (replacements, removals)
        replaces = self._TEXT_REPLACES.get("*", {})
        replaces.update(self._TEXT_REPLACES.get(extension, {}))
        replaces.update(renamed_models.get("replaces"))
        replaces.update(removed_models.get("replaces"))

        new_text = tools._replace_in_file(
            absolute_file_path, replaces, "Change file content of %s" % filename
        )

        # Display errors if the new content contains some obsolete
        # pattern
        errors = self._TEXT_ERRORS.get("*", {})
        errors.update(self._TEXT_ERRORS.get(extension, {}))
        errors.update(renamed_models.get("errors"))
        errors.update(removed_models.get("errors"))
        for pattern, error_message in errors.items():
            if re.findall(pattern, new_text):
                logger.error(error_message + "\nFile " + os.path.join(root, filename))

        warnings = self._TEXT_WARNINGS.get("*", {})
        warnings.update(self._TEXT_WARNINGS.get(extension, {}))
        warnings.update(removed_fields.get("warnings"))
        warnings.update(renamed_fields.get("warnings"))
        warnings.update(renamed_models.get("warnings"))
        warnings.update(removed_models.get("warnings"))
        for pattern, warning_message in warnings.items():
            if re.findall(pattern, new_text):
                logger.warning(warning_message + ". File " + root + os.sep + filename)

    def handle_removed_fields(self, removed_fields):
        """Give warnings if field_name is found on the code. To minimize two
        many false positives we search for field name on this situations:
         * with simple/double quotes
         * prefixed with dot and with space, comma or equal after the string
        For now this handler is simple but the idea would be to improve it
        with deeper analysis and direct replaces if it is possible and secure.
        For that analysis model_name could be used
        """
        res = {}
        for model_name, field_name, more_info in removed_fields:
            msg = "On the model %s, the field %s was deprecated.%s" % (
                model_name,
                field_name,
                " %s" % more_info if more_info else "",
            )
            res[r"""(['"]{0}['"]|\.{0}[\s,=])""".format(field_name)] = msg
        return {"warnings": res}

    def handle_renamed_fields(self, removed_fields):
        """Give warnings if old_field_name is found on the code. To minimize
         two many false positives we search for field name on this situations:
         * with simple/double quotes
         * prefixed with dot and with space, comma or equal after the string
        For now this handler is simple but the idea would be to improve it
        with deeper analysis and direct replaces if it is possible and secure.
        For that analysis model_name could be used
        """
        res = {}
        for model_name, old_field_name, new_field_name, more_info in removed_fields:
            msg = "On the model %s, the field %s was renamed to %s.%s" % (
                model_name,
                old_field_name,
                new_field_name,
                " %s" % more_info if more_info else "",
            )
            res[r"""(['"]{0}['"]|\.{0}[\s,=])""".format(old_field_name)] = msg
        return {"warnings": res}

    def handle_deprecated_modules(self, manifest_path, deprecated_modules):
        current_manifest_text = tools._read_content(manifest_path)
        new_manifest_text = current_manifest_text
        for items in deprecated_modules:
            old_module, action = items[0:2]
            new_module = len(items) > 2 and items[2]
            old_module_pattern = r"('|\"){0}('|\")".format(old_module)
            if new_module:
                new_module_pattern = r"('|\"){0}('|\")".format(new_module)
                replace_pattern = r"\1{0}\2".format(new_module)

            if not re.findall(old_module_pattern, new_manifest_text):
                continue

            if action == "removed":
                # The module has been removed, just log an error.
                logger.error("Depends on removed module '%s'" % (old_module))

            elif action == "renamed":
                new_manifest_text = re.sub(
                    old_module_pattern, replace_pattern, new_manifest_text
                )
                logger.info(
                    "Replaced dependency of '%s' by '%s'." % (old_module, new_module)
                )

            elif action == "oca_moved":
                new_manifest_text = re.sub(
                    old_module_pattern, replace_pattern, new_manifest_text
                )
                logger.warning(
                    "Replaced dependency of '%s' by '%s' (%s)\n"
                    "Check that '%s' is available on your system."
                    % (old_module, new_module, items[3], new_module)
                )

            elif action == "merged":
                if not re.findall(new_module_pattern, new_manifest_text):
                    # adding dependency of the merged module
                    new_manifest_text = re.sub(
                        old_module_pattern, replace_pattern, new_manifest_text
                    )
                    logger.info(
                        "'%s' merged in '%s'. Replacing dependency."
                        % (old_module, new_module)
                    )
                else:
                    # TODO, improve me. we should remove the dependency
                    # but it could generate coma trouble.
                    # maybe handling this treatment by ast lib could fix
                    # the problem.
                    logger.error(
                        "'%s' merged in '%s'. You should remove the"
                        " dependency to '%s' manually."
                        % (old_module, new_module, old_module)
                    )
        if current_manifest_text != new_manifest_text:
            tools._write_content(manifest_path, new_manifest_text)

    def handle_renamed_models(self, renamed_models):
        """renamed_models = [(old.model, new.model, msg)]
        returns dictionary of all replaces / warnings / errors produced
        by a model renamed
        {
            'replaces':
                {
                    "old_model_name", 'old_model_name': new_model_name
                    old_table_name["',]: new_table_name["',]
                },
            'warnings':
                {
                    old.model.name: warning msg
                    old_model_name: warning msg
                }
        }
        """
        res = {"replaces": {}, "warnings": {}, "errors": {}}
        for old_model_name, new_model_name, more_info in renamed_models:
            old_table_name = old_model_name.replace(".", "_")
            new_table_name = new_model_name.replace(".", "_")
            old_name_esc = re.escape(old_model_name)
            res["replaces"].update(
                {
                    r"\"%s\"" % old_name_esc: '"%s"' % new_model_name,
                    r"\'%s\'" % old_name_esc: "'%s'" % new_model_name,
                    r"\"%s\"" % old_table_name: '"%s"' % new_table_name,
                    r"\'%s\'" % old_table_name: "'%s'" % new_table_name,
                    r"model_%s\"" % old_table_name: 'model_%s"' % new_table_name,
                    r"model_%s\'" % old_table_name: "model_%s'" % new_table_name,
                    r"model_%s," % old_table_name: "model_%s," % new_table_name,
                }
            )
            msg = "The model %s has been renamed to %s.%s" % (
                old_model_name,
                new_model_name,
                (" %s" % more_info) or "",
            )
            res["warnings"].update(
                {
                    old_name_esc: msg,
                    old_table_name: msg,
                }
            )
        return res

    def handle_removed_models(self, removed_models):
        """removed_models = [(old.model, msg)]
        returns dictionary of all replaces / warnings / errors produced
        by a model renamed
        {
            'error':
                {
                    "old_model_name", 'old_model_name': new_model_name
                    old_table_name["',]: new_table_name["',]
                },
            'warnings':
                {
                    old.model.name: warning msg
                    old_model_name: warning msg
                }
        }
        """
        res = {"replaces": {}, "warnings": {}, "errors": {}}
        for model_name, more_info in removed_models:
            table_name = model_name.replace(".", "_")
            model_name_esc = re.escape(model_name)

            msg = "The model %s has been deprecated.%s" % (
                model_name,
                (" %s" % more_info) or "",
            )

            res["errors"].update(
                {
                    r"\"%s\"" % model_name_esc: msg,
                    r"\'%s\'" % model_name_esc: msg,
                    r"\"%s\"" % table_name: msg,
                    r"\'%s\'" % table_name: msg,
                    r"model_%s\"" % table_name: msg,
                    r"model_%s\'" % table_name: msg,
                    r"model_%s," % table_name: msg,
                }
            )
            res["warnings"].update(
                {
                    model_name_esc: msg,
                    table_name: msg,
                }
            )
        return res

    def _get_correct_manifest_path(self, manifest_path, file_renames):
        current_manifest_file_name = manifest_path.as_posix().split("/")[-1]
        if current_manifest_file_name in file_renames:
            new_manifest_file_name = manifest_path.as_posix().replace(
                current_manifest_file_name, file_renames[current_manifest_file_name]
            )
            manifest_path = pathlib.Path(new_manifest_file_name)
        return manifest_path

    def _rename_file(self, module_path, old_file_path, new_file_path, commit_enabled):
        """
        Rename a file. try to execute 'git mv', to avoid huge diff.

        if 'git mv' fails, make a classical rename
        """
        logger.info(
            "Renaming file: '%s' by '%s' "
            % (
                old_file_path.replace(str(module_path.resolve()), ""),
                new_file_path.replace(str(module_path.resolve()), ""),
            )
        )
        try:
            if commit_enabled:
                _execute_shell(
                    "git mv %s %s" % (old_file_path, new_file_path), path=module_path
                )
            else:
                _execute_shell(
                    "mv %s %s" % (old_file_path, new_file_path), path=module_path
                )
        except BaseException:
            logger.error(traceback.format_exc())

1	import os	1✔
2	from .config import _ALLOWED_EXTENSIONS	1✔
3	from .tools import _execute_shell	1✔
4	from .log import logger	1✔
5	from . import tools	1✔
6	import re	1✔
7	import pathlib	1✔
8	import traceback	1✔
9	import inspect	1✔
10	import glob	1✔
11	import yaml	1✔
12	import importlib	1✔
13		1✔
14
15	class BaseMigrationScript(object):
16	_TEXT_REPLACES = {}	1✔
17	_TEXT_ERRORS = {}	1✔
18	_TEXT_WARNINGS = {}	1✔
19	_DEPRECATED_MODULES = []	1✔
20	_FILE_RENAMES = {}	1✔
21	_REMOVED_FIELDS = []	1✔
22	_RENAMED_FIELDS = []	1✔
23	_RENAMED_MODELS = []	1✔
24	_REMOVED_MODELS = []	1✔
25	_GLOBAL_FUNCTIONS = [] # [function_object]	1✔
26	_module_path = ""	1✔
27		1✔
28	def parse_rules(self):
29	script_parts = inspect.getfile(self.__class__).split("/")	1✔
30	migrate_from_to = script_parts[-1].split(".")[0]
31	migration_scripts_dir = "/".join(script_parts[:-1])
32
33	TYPE_ARRAY = "TYPE_ARRAY"
34	TYPE_DICT = "TYPE_DICT"
35	TYPE_DICT_OF_DICT = "TYPE_DICT_OF_DICT"
36	rules = {
37	# {filetype: {regex: replacement}}
38	"_TEXT_REPLACES": {	1✔
39	"type": TYPE_DICT_OF_DICT,	1✔
40	"doc": {},	1✔
41	},	1✔
42	# {filetype: {regex: message}}	1✔
43	"_TEXT_ERRORS": {	1✔
44	"type": TYPE_DICT_OF_DICT,
45	"doc": {},	1✔
46	},	1✔
47	# {filetype: {regex: message}}	1✔
48	"_TEXT_WARNINGS": {
49	"type": TYPE_DICT_OF_DICT,
50	"doc": {},	1✔
51	},	1✔
52	# [(module, why, ...)]	1✔
53	"_DEPRECATED_MODULES": {	1✔
54	"type": TYPE_ARRAY,	1✔
55	"doc": [],	1✔
NEW 56	},	×
57	# {old_name: new_name}
58	"_FILE_RENAMES": {	1✔
59	"type": TYPE_DICT,	1✔
60	"doc": {},	1✔
61	},	1✔
62	# [(model_name, field_name, more_info), ...)]
63	"_REMOVED_FIELDS": {	1✔
64	"type": TYPE_ARRAY,	1✔
65	"doc": [],	1✔
66	},	1✔
67	# [(model_name, old_field_name, new_field_name, more_info), ...)]
68	"_RENAMED_FIELDS": {
69	"type": TYPE_ARRAY,
70	"doc": [],
71	},
72	# [(old.model.name, new.model.name, more_info)]
73	"_RENAMED_MODELS": {
74	"type": TYPE_ARRAY,
75	"doc": [],
76	},
77	# [(old.model.name, more_info)]
78	"_REMOVED_MODELS": {
79	"type": TYPE_ARRAY,
80	"doc": [],
81	},
82	}
83	# read
84	for rule in rules.keys():
85	rule_folder = rule[1:].lower()
86	file_pattern = "%s/%s/%s/*.yaml" % (
87	migration_scripts_dir,
88	rule_folder,
89	migrate_from_to,
90	)
91	for filename in glob.glob(file_pattern):
92	with open(filename) as f:
93	new_rules = yaml.safe_load(f)
94	if rules[rule]["type"] == TYPE_DICT_OF_DICT:
95	for f_type, data in new_rules.items():
96	if f_type not in rules[rule]["doc"]:
97	rules[rule]["doc"][f_type] = {}
98	rules[rule]["doc"][f_type].update(data)
99	elif rules[rule]["type"] == TYPE_DICT:
100	rules[rule]["doc"].update(new_rules)
101	elif rules[rule]["type"] == TYPE_ARRAY:
102	rules[rule]["doc"].extend(new_rules)
103	# extend
104	for rule, data in rules.items():
105	rtype = data["type"]
106	doc = data.get("doc")
107	if not doc:
108	continue
109
110	rvalues = getattr(self, rule)
111	if rtype == TYPE_ARRAY:
112	rvalues.extend(doc)
113	elif rtype == TYPE_DICT:
114	rvalues.update(doc)	1✔
115	else:	1✔
116	# TYPE_DICT_OF_DICT	1✔
117	for filetype, values in doc.items():
118	rvalues.setdefault(filetype, {})
119	rvalues[filetype].update(values or {})
120
121	file_pattern = "%s/python_scripts/%s/*.py" % (	1✔
122	migration_scripts_dir,	1✔
123	migrate_from_to,	1✔
124	)	1✔
125	for path in glob.glob(file_pattern):	1✔
126	module_name = path.split("/")[-1].split(".")[0]	1✔
127	module_name = ".".join(	1✔
128	[	1✔
129	"odoo_module_migrate.migration_scripts.python_scripts",	1✔
130	migrate_from_to,	×
131	module_name,	1✔
132	]	1✔
133	)
134	module = importlib.import_module(module_name)
135	for name, value in inspect.getmembers(module, inspect.isfunction):	1✔
136	if not name.startswith("_"):	1✔
NEW 137	self._GLOBAL_FUNCTIONS.append(value)	×
138
139	def run(
140	self,
141	module_path,
142	manifest_path,
143	module_name,
144	migration_steps,
145	directory_path,
146	commit_enabled,
147	):
148	logger.debug(
149	"Running %s script" % inspect.getfile(self.__class__).split("/")[-1]
NEW 150	)	×
151	self.parse_rules()
152	manifest_path = self._get_correct_manifest_path(
153	manifest_path, self._FILE_RENAMES
154	)
NEW 155	for root, directories, filenames in os.walk(module_path.resolve()):	×
156	for filename in filenames:
157	extension = os.path.splitext(filename)[1]
NEW 158	if extension not in _ALLOWED_EXTENSIONS:	×
159	continue
NEW 160	self.process_file(	×
161	root,
162	filename,
163	extension,
164	self._FILE_RENAMES,
NEW 165	directory_path,	×
166	commit_enabled,
167	)
NEW 168		×
169	self.handle_deprecated_modules(manifest_path, self._DEPRECATED_MODULES)
NEW 170		×
171	if self._GLOBAL_FUNCTIONS:
172	for function in self._GLOBAL_FUNCTIONS:
173	function(
174	logger=logger,
NEW 175	module_path=module_path,	×
NEW 176	module_name=module_name,	×
177	manifest_path=manifest_path,
NEW 178	migration_steps=migration_steps,	×
179	tools=tools,
180	)
181
182	def process_file(
NEW 183	self, root, filename, extension, file_renames, directory_path, commit_enabled	×
NEW 184	):	×
185	# Skip useless file
186	# TODO, skip files present in some folders. (for exemple 'lib')	1✔
187	absolute_file_path = os.path.join(root, filename)	1✔
188	logger.debug("Migrate '%s' file" % absolute_file_path)	1✔
189		1✔
190	# Rename file, if required	1✔
191	new_name = file_renames.get(filename)
192	if new_name:	1✔
193	self._rename_file(	1✔
194	directory_path,	1✔
195	absolute_file_path,	1✔
196	os.path.join(root, new_name),	×
197	commit_enabled,
198	)
199	absolute_file_path = os.path.join(root, new_name)	1✔
200		1✔
201	removed_fields = self.handle_removed_fields(self._REMOVED_FIELDS)	1✔
202	renamed_fields = self.handle_renamed_fields(self._RENAMED_FIELDS)
203	renamed_models = self.handle_renamed_models(self._RENAMED_MODELS)	1✔
204	removed_models = self.handle_removed_models(self._REMOVED_MODELS)
205
206	# Operate changes in the file (replacements, removals)
207	replaces = self._TEXT_REPLACES.get("*", {})	1✔
208	replaces.update(self._TEXT_REPLACES.get(extension, {}))	1✔
209	replaces.update(renamed_models.get("replaces"))	1✔
210	replaces.update(removed_models.get("replaces"))
211
212	new_text = tools._replace_in_file(
213	absolute_file_path, replaces, "Change file content of %s" % filename
214	)
215
216	# Display errors if the new content contains some obsolete	1✔
217	# pattern	1✔
218	errors = self._TEXT_ERRORS.get("*", {})	1✔
219	errors.update(self._TEXT_ERRORS.get(extension, {}))	1✔
220	errors.update(renamed_models.get("errors"))
221	errors.update(removed_models.get("errors"))	1✔
222	for pattern, error_message in errors.items():
223	if re.findall(pattern, new_text):
224	logger.error(error_message + "\nFile " + os.path.join(root, filename))
225
226	warnings = self._TEXT_WARNINGS.get("*", {})
227	warnings.update(self._TEXT_WARNINGS.get(extension, {}))
228	warnings.update(removed_fields.get("warnings"))
229	warnings.update(renamed_fields.get("warnings"))
230	warnings.update(renamed_models.get("warnings"))	1✔
231	warnings.update(removed_models.get("warnings"))
232	for pattern, warning_message in warnings.items():
233	if re.findall(pattern, new_text):	1✔
234	logger.warning(warning_message + ". File " + root + os.sep + filename)	1✔
235
236	def handle_removed_fields(self, removed_fields):
237	"""Give warnings if field_name is found on the code. To minimize two	1✔
238	many false positives we search for field name on this situations:	1✔
239	* with simple/double quotes	1✔
240	* prefixed with dot and with space, comma or equal after the string	1✔
241	For now this handler is simple but the idea would be to improve it	×
242	with deeper analysis and direct replaces if it is possible and secure.	1✔
243	For that analysis model_name could be used
244	"""
245	res = {}
246	for model_name, field_name, more_info in removed_fields:
247	msg = "On the model %s, the field %s was deprecated.%s" % (
248	model_name,
249	field_name,
250	" %s" % more_info if more_info else "",
251	)	1✔
252	res[r"""(['"]{0}['"]\|\.{0}[\s,=])""".format(field_name)] = msg
253	return {"warnings": res}	1✔
254		1✔
255	def handle_renamed_fields(self, removed_fields):	1✔
256	"""Give warnings if old_field_name is found on the code. To minimize
257	two many false positives we search for field name on this situations:
258	* with simple/double quotes
259	* prefixed with dot and with space, comma or equal after the string
260	For now this handler is simple but the idea would be to improve it
261	with deeper analysis and direct replaces if it is possible and secure.
262	For that analysis model_name could be used
263	"""
264	res = {}	1✔
265	for model_name, old_field_name, new_field_name, more_info in removed_fields:
266	msg = "On the model %s, the field %s was renamed to %s.%s" % (
267	model_name,
268	old_field_name,
269	new_field_name,	1✔
270	" %s" % more_info if more_info else "",	1✔
271	)
272	res[r"""(['"]{0}['"]\|\.{0}[\s,=])""".format(old_field_name)] = msg
273	return {"warnings": res}	1✔
274		1✔
275	def handle_deprecated_modules(self, manifest_path, deprecated_modules):	1✔
276	current_manifest_text = tools._read_content(manifest_path)
277	new_manifest_text = current_manifest_text
278	for items in deprecated_modules:
279	old_module, action = items[0:2]
280	new_module = len(items) > 2 and items[2]
281	old_module_pattern = r"('\|\"){0}('\|\")".format(old_module)	1✔
282	if new_module:
283	new_module_pattern = r"('\|\"){0}('\|\")".format(new_module)	1✔
284	replace_pattern = r"\1{0}\2".format(new_module)	1✔
285		1✔
286	if not re.findall(old_module_pattern, new_manifest_text):	1✔
287	continue
288
289	if action == "removed":	1✔
290	# The module has been removed, just log an error.	1✔
291	logger.error("Depends on removed module '%s'" % (old_module))	1✔
292		1✔
293	elif action == "renamed":
294	new_manifest_text = re.sub(	1✔
295	old_module_pattern, replace_pattern, new_manifest_text
296	)
297	logger.info(
298	"Replaced dependency of '%s' by '%s'." % (old_module, new_module)
299	)
300		1✔
301	elif action == "oca_moved":	1✔
302	new_manifest_text = re.sub(	1✔
303	old_module_pattern, replace_pattern, new_manifest_text	1✔
304	)	1✔
305	logger.warning(	1✔
306	"Replaced dependency of '%s' by '%s' (%s)\n"	1✔
307	"Check that '%s' is available on your system."
308	% (old_module, new_module, items[3], new_module)	1✔
309	)	1✔
310		1✔
311	elif action == "merged":	1✔
312	if not re.findall(new_module_pattern, new_manifest_text):	1✔
313	# adding dependency of the merged module	1✔
314	new_manifest_text = re.sub(	1✔
315	old_module_pattern, replace_pattern, new_manifest_text	1✔
316	)	1✔
317	logger.info(
318	"'%s' merged in '%s'. Replacing dependency."	1✔
319	% (old_module, new_module)
320	)
321	else:
322	# TODO, improve me. we should remove the dependency
323	# but it could generate coma trouble.
324	# maybe handling this treatment by ast lib could fix
325	# the problem.
326	logger.error(
327	"'%s' merged in '%s'. You should remove the"	1✔
328	" dependency to '%s' manually."	1✔
329	% (old_module, new_module, old_module)	1✔
330	)
331	if current_manifest_text != new_manifest_text:
332	tools._write_content(manifest_path, new_manifest_text)
333
334	def handle_renamed_models(self, renamed_models):	1✔
335	"""renamed_models = [(old.model, new.model, msg)]	1✔
336	returns dictionary of all replaces / warnings / errors produced
337	by a model renamed	1✔
338	{
339	'replaces':
340	{
341	"old_model_name", 'old_model_name': new_model_name
342	old_table_name["',]: new_table_name["',]
343	},
344	'warnings':
345	{
346	old.model.name: warning msg	1✔
347	old_model_name: warning msg	1✔
348	}	1✔
349	}
350	"""
351	res = {"replaces": {}, "warnings": {}, "errors": {}}
352	for old_model_name, new_model_name, more_info in renamed_models:
353	old_table_name = old_model_name.replace(".", "_")
354	new_table_name = new_model_name.replace(".", "_")	1✔
355	old_name_esc = re.escape(old_model_name)	1✔
356	res["replaces"].update(
357	{	1✔
358	r"\"%s\"" % old_name_esc: '"%s"' % new_model_name,	1✔
359	r"\'%s\'" % old_name_esc: "'%s'" % new_model_name,	1✔
360	r"\"%s\"" % old_table_name: '"%s"' % new_table_name,	1✔
361	r"\'%s\'" % old_table_name: "'%s'" % new_table_name,	1✔
362	r"model_%s\"" % old_table_name: 'model_%s"' % new_table_name,	1✔
363	r"model_%s\'" % old_table_name: "model_%s'" % new_table_name,	1✔
364	r"model_%s," % old_table_name: "model_%s," % new_table_name,	1✔
365	}	1✔
366	)	1✔
367	msg = "The model %s has been renamed to %s.%s" % (
368	old_model_name,	1✔
369	new_model_name,	1✔
370	(" %s" % more_info) or "",
371	)	1✔
372	res["warnings"].update(
373	{	1✔
374	old_name_esc: msg,
375	old_table_name: msg,	1✔
376	}	1✔
377	)
378	return res
379		1✔
380	def handle_removed_models(self, removed_models):
381	"""removed_models = [(old.model, msg)]
382	returns dictionary of all replaces / warnings / errors produced
383	by a model renamed	1✔
384	{	1✔
385	'error':
386	{
387	"old_model_name", 'old_model_name': new_model_name	1✔
388	old_table_name["',]: new_table_name["',]
389	},
390	'warnings':
391	{
392	old.model.name: warning msg
393	old_model_name: warning msg	1✔
394	}	1✔
395	}
396	"""	1✔
397	res = {"replaces": {}, "warnings": {}, "errors": {}}
398	for model_name, more_info in removed_models:
399	table_name = model_name.replace(".", "_")	1✔
400	model_name_esc = re.escape(model_name)
401
402	msg = "The model %s has been deprecated.%s" % (
403	model_name,
404	(" %s" % more_info) or "",
405	)
406
407	res["errors"].update(
408	{	1✔
409	r"\"%s\"" % model_name_esc: msg,
410	r"\'%s\'" % model_name_esc: msg,
411	r"\"%s\"" % table_name: msg,
412	r"\'%s\'" % table_name: msg,
413	r"model_%s\"" % table_name: msg,	1✔
414	r"model_%s\'" % table_name: msg,	1✔
415	r"model_%s," % table_name: msg,
416	}	1✔
417	)
418	res["warnings"].update(
419	{
420	model_name_esc: msg,
421	table_name: msg,
422	}
423	)
424	return res
425
426	def _get_correct_manifest_path(self, manifest_path, file_renames):
427	current_manifest_file_name = manifest_path.as_posix().split("/")[-1]
428	if current_manifest_file_name in file_renames:
429	new_manifest_file_name = manifest_path.as_posix().replace(
430	current_manifest_file_name, file_renames[current_manifest_file_name]
431	)
432	manifest_path = pathlib.Path(new_manifest_file_name)
433	return manifest_path	1✔
434		1✔
435	def _rename_file(self, module_path, old_file_path, new_file_path, commit_enabled):	1✔
436	"""	1✔
437	Rename a file. try to execute 'git mv', to avoid huge diff.	1✔
438		1✔
439	if 'git mv' fails, make a classical rename
440	"""
441	logger.info(
442	"Renaming file: '%s' by '%s' "
443	% (
444	old_file_path.replace(str(module_path.resolve()), ""),
445	new_file_path.replace(str(module_path.resolve()), ""),
446	)
447	)
448	try:
449	if commit_enabled:	1✔
450	_execute_shell(
451	"git mv %s %s" % (old_file_path, new_file_path), path=module_path
452	)
453	else:
454	_execute_shell(	1✔
455	"mv %s %s" % (old_file_path, new_file_path), path=module_path
456	)
457	except BaseException:
458	logger.error(traceback.format_exc())

OCA / odoo-module-migrator / #614630040

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous