|
| 1 | +# Copyright 2026 Sébastien Alix |
| 2 | +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) |
| 3 | + |
| 4 | +import io |
| 5 | +import shutil |
| 6 | +import tempfile |
| 7 | +import threading |
| 8 | +import time |
| 9 | +import unittest |
| 10 | +import zipfile |
| 11 | +from pathlib import Path |
| 12 | + |
| 13 | +import git |
| 14 | + |
| 15 | +cache = threading.local() |
| 16 | + |
| 17 | + |
| 18 | +class OdooRepoMixin(unittest.TestCase): |
| 19 | + @classmethod |
| 20 | + def setUpClass(cls): |
| 21 | + super().setUpClass() |
| 22 | + cls.upstream_org = "ORG" |
| 23 | + cls.fork_org = "FORK" |
| 24 | + cls.repo_name = "test" |
| 25 | + cls.source1 = "origin/15.0" |
| 26 | + cls.source2 = "origin/16.0" |
| 27 | + cls.target1 = "origin/16.0" |
| 28 | + cls.target2 = "origin/17.0" |
| 29 | + cls.target3 = "origin/18.0" |
| 30 | + cls.addon = "my_module" |
| 31 | + cls.target_addon = "my_module_renamed" |
| 32 | + |
| 33 | + def setUp(self): |
| 34 | + super().setUp() |
| 35 | + # Create a temporary Git repository |
| 36 | + self.repo_upstream_path = self._get_upstream_repository_path() |
| 37 | + self.addon_path = Path(self.repo_upstream_path) / self.addon |
| 38 | + self.manifest_path = self.addon_path / "__manifest__.py" |
| 39 | + # By cloning the first repository this will set an 'origin' remote |
| 40 | + self.repo_path = self._clone_tmp_git_repository(self.repo_upstream_path) |
| 41 | + self._add_fork_remote(self.repo_path) |
| 42 | + |
| 43 | + def _get_upstream_repository_path(self) -> Path: |
| 44 | + """Returns the path of upstream repository. |
| 45 | +
|
| 46 | + Generate the upstream git repository or re-use the one put in cache if any. |
| 47 | + """ |
| 48 | + if hasattr(cache, "archive_data") and cache.archive_data: |
| 49 | + # Unarchive the repository from memory |
| 50 | + repo_path = self._unarchive_upstream_repository(cache.archive_data) |
| 51 | + else: |
| 52 | + # Prepare and archive the repository in memory |
| 53 | + repo_path = self._create_tmp_git_repository() |
| 54 | + addon_path = repo_path / self.addon |
| 55 | + self._fill_git_repository(repo_path, addon_path) |
| 56 | + cache.archive_data = self._archive_upstream_repository(repo_path) |
| 57 | + return repo_path |
| 58 | + |
| 59 | + def _archive_upstream_repository(self, repo_path: Path) -> bytes: |
| 60 | + """Archive the repository located at `repo_path`. |
| 61 | +
|
| 62 | + Returns binary value of the archive. |
| 63 | + """ |
| 64 | + # Create in-memory zip archive |
| 65 | + zip_buffer = io.BytesIO() |
| 66 | + with zipfile.ZipFile(zip_buffer, "w") as zipf: |
| 67 | + for file_path in repo_path.rglob("*"): |
| 68 | + if file_path.is_file(): |
| 69 | + arcname = file_path.relative_to(repo_path) |
| 70 | + zipf.write(file_path, arcname) |
| 71 | + return zip_buffer.getvalue() |
| 72 | + |
| 73 | + def _unarchive_upstream_repository(self, archive_data: bytes) -> Path: |
| 74 | + """Unarchive the repository contained in `archive_data`. |
| 75 | +
|
| 76 | + Returns path of repository. |
| 77 | + """ |
| 78 | + temp_dir = tempfile.mkdtemp() |
| 79 | + with zipfile.ZipFile(io.BytesIO(archive_data), "r") as zip_ref: |
| 80 | + zip_ref.extractall(temp_dir) |
| 81 | + # Look for the repo directory and return its path |
| 82 | + for path in Path(temp_dir).rglob("*"): |
| 83 | + if path.is_dir() and ".git" in path.name: |
| 84 | + return path.parent |
| 85 | + |
| 86 | + def _create_tmp_git_repository(self) -> Path: |
| 87 | + """Create a temporary Git repository to run tests.""" |
| 88 | + repo_path = tempfile.mkdtemp() |
| 89 | + git.Repo.init(repo_path) |
| 90 | + return Path(repo_path) |
| 91 | + |
| 92 | + def _clone_tmp_git_repository(self, upstream_path: Path) -> Path: |
| 93 | + repo_path = tempfile.mkdtemp() |
| 94 | + git.Repo.clone_from(upstream_path, repo_path) |
| 95 | + return Path(repo_path) |
| 96 | + |
| 97 | + def _fill_git_repository(self, repo_path: Path, addon_path: Path): |
| 98 | + """Create branches with some content in the Git repository.""" |
| 99 | + repo = git.Repo(repo_path) |
| 100 | + # Commit a file in '15.0' |
| 101 | + branch1 = self.source1.split("/")[1] |
| 102 | + repo.git.checkout("--orphan", branch1) |
| 103 | + self._create_module(addon_path) |
| 104 | + repo.index.add(addon_path) |
| 105 | + commit = repo.index.commit(f"[ADD] {self.addon}") |
| 106 | + # Port the commit from 15.0 to 16.0 |
| 107 | + branch2 = self.source2.split("/")[1] |
| 108 | + repo.git.checkout("--orphan", branch2) |
| 109 | + repo.git.reset("--hard") |
| 110 | + # Some git operations do not appear to be atomic, so a delay is added |
| 111 | + # to allow them to complete |
| 112 | + time.sleep(1) |
| 113 | + repo.git.cherry_pick(commit.hexsha) |
| 114 | + # Create an empty branch 17.0 |
| 115 | + branch3 = self.target2.split("/")[1] |
| 116 | + repo.git.checkout("--orphan", branch3) |
| 117 | + repo.git.reset("--hard") |
| 118 | + repo.git.commit("-m", "Init", "--allow-empty") |
| 119 | + # Port the commit from 15.0 to 18.0 |
| 120 | + branch4 = self.target3.split("/")[1] |
| 121 | + repo.git.checkout("--orphan", branch4) |
| 122 | + repo.git.reset("--hard") |
| 123 | + time.sleep(1) |
| 124 | + repo.git.cherry_pick(commit.hexsha) |
| 125 | + # Rename the module on 18.0 |
| 126 | + repo.git.mv(self.addon, self.target_addon) |
| 127 | + repo.git.commit("-m", f"Rename {self.addon} to {self.target_addon}") |
| 128 | + |
| 129 | + def _create_module(self, module_path: Path): |
| 130 | + manifest_lines = [ |
| 131 | + "# Copyright 2026 Sébastien Alix\n", |
| 132 | + "# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)\n", |
| 133 | + "{\n", |
| 134 | + ' "name": "Test",\n', |
| 135 | + ' "version": "1.0.0",\n', |
| 136 | + ' "category": "Test Module",\n', |
| 137 | + ' "author": "Odoo Community Association (OCA)",\n', |
| 138 | + ' "website": "https://github.com/OCA/module-composition-analysis",\n', |
| 139 | + ' "license": "AGPL-3",\n', |
| 140 | + ' "depends": ["base"],\n', |
| 141 | + ' "data": [],\n', |
| 142 | + ' "demo": [],\n', |
| 143 | + ' "installable": True,\n', |
| 144 | + "}\n", |
| 145 | + ] |
| 146 | + module_path.mkdir(parents=True, exist_ok=True) |
| 147 | + manifest_path = module_path / "__manifest__.py" |
| 148 | + with open(manifest_path, "w") as manifest: |
| 149 | + manifest.writelines(manifest_lines) |
| 150 | + |
| 151 | + def _add_fork_remote(self, repo_path: Path): |
| 152 | + repo = git.Repo(repo_path) |
| 153 | + # We do not really care about the remote URL here, re-use origin one |
| 154 | + repo.create_remote(self.fork_org, repo.remotes.origin.url) |
| 155 | + |
| 156 | + def tearDown(self): |
| 157 | + super().tearDown() |
| 158 | + # Clean up the Git repository |
| 159 | + shutil.rmtree(self.repo_upstream_path) |
| 160 | + shutil.rmtree(self.repo_path) |
0 commit comments