From 42569c499aba3c790d59ea3b365232f5cf83bba6 Mon Sep 17 00:00:00 2001 From: brinkflew Date: Sun, 26 Jul 2026 23:35:16 +0200 Subject: [PATCH 1/7] [FIX] common: correct string, version and postgres helpers Writing unit tests for helpers that had none surfaced five defects, all on code paths odev actually walks: - `PostgresConnector.columns_exist` returned an empty list when none of the requested columns existed, which is indistinguishable from all of them being present. A table created from an older definition therefore kept none of its new columns, since `CREATE TABLE IF NOT EXISTS` leaves an existing table alone and the missing-columns pass is the only thing that can migrate it. - `PostgresDatabase.tables` was a class attribute, so every database instance shared a single registry and tables from different databases collided on their name alone. - `string.quote` selected its delimiter with `max()` over the offsets of both quote characters, which picks the last one rather than the first and mis-quoted any string mixing them. The helper never escapes, so its docstring now says so. - `OdooVersion.__bool__` was always true: `module` is padded to `MIN_VERSION_LENGTH` and is therefore never an empty tuple. - `string.min_indent` raised on a text without any non-blank line, which `odev help` reaches through `dedent`. `float_to_hours` and `strip_styles` are broken too but are called nowhere in odev nor in the plugins; they are left alone and documented in the tests instead. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- odev/_version.py | 2 +- odev/common/connectors/postgres.py | 18 +++++++++++------- odev/common/postgres.py | 6 +++--- odev/common/string.py | 19 +++++++++++++++---- odev/common/version.py | 7 +++++-- 5 files changed, 35 insertions(+), 17 deletions(-) diff --git a/odev/_version.py b/odev/_version.py index 7306664df..12aa2f183 100644 --- a/odev/_version.py +++ b/odev/_version.py @@ -22,4 +22,4 @@ # or merged change. # ------------------------------------------------------------------------------ -__version__ = "4.29.9" +__version__ = "4.29.10" diff --git a/odev/common/connectors/postgres.py b/odev/common/connectors/postgres.py index bded4f087..aa2e68d80 100644 --- a/odev/common/connectors/postgres.py +++ b/odev/common/connectors/postgres.py @@ -314,27 +314,31 @@ def create_table(self, table: str, columns: Mapping[str, str]) -> bool: return bool(self.query(f"CREATE TABLE IF NOT EXISTS {table} ({sql_columns})")) def columns_exist(self, table: str, columns: list[str]) -> list[str]: - """Check whether a column exists in a table. + """List the columns missing from a table, among the ones requested. :param table: The name of the table to check. - :param columns: The name of the column to check. - :return: Whether the column exists. + :param columns: The names of the columns to look for. + :return: The requested columns that do not exist in the table, in the order they were requested. :rtype: list[str] """ if not isinstance(columns, list): raise TypeError("Columns should be a list of strings") + if not columns: + return [] + results = self.query( f""" SELECT column_name FROM information_schema.columns - WHERE table_name = '{table}' AND column_name IN ({",".join([f"'{c}'" for c in columns])}) + WHERE table_name = '{table}' AND column_name IN ({",".join([f"'{column}'" for column in columns])}) """, ) - if results and isinstance(results, list) and columns: - return [c for c in columns if c not in [r[0] for r in results]] - return [] + # An empty result means none of the requested columns exist, not that none are missing. + existing = {result[0] for result in results} if isinstance(results, list) else set() + + return [column for column in columns if column not in existing] def create_column(self, table: str, column: str, attributes: str) -> bool: """Create a column in a table. diff --git a/odev/common/postgres.py b/odev/common/postgres.py index 4f21f79b7..3c34526db 100644 --- a/odev/common/postgres.py +++ b/odev/common/postgres.py @@ -22,9 +22,6 @@ class PostgresDatabase(PostgresConnectorMixin): connector: PostgresConnector """Instance of the connector to the database engine.""" - tables: MutableMapping[str, "PostgresTable"] = {} - """Mapping of tables in the database.""" - def __init__(self, name: str): """Initialize the database.""" super().__init__() @@ -32,6 +29,9 @@ def __init__(self, name: str): self.name: str = name """The name of the database.""" + self.tables: MutableMapping[str, PostgresTable] = {} + """Mapping of tables in the database, keyed by table name.""" + self.prepare_database() def __enter__(self): diff --git a/odev/common/string.py b/odev/common/string.py index 50f6dd607..247438957 100644 --- a/odev/common/string.py +++ b/odev/common/string.py @@ -93,11 +93,13 @@ def dedent(text: str, dedent: int = 0) -> str: def min_indent(text: str) -> int: """Return the smallest indentation in a text. + A text without any non-blank line has no indentation to speak of and returns zero. + :param text: The text to get the minimum indentation from. :return: The minimum indentation of the text. :rtype: int """ - return min(len(line) - len(line.lstrip()) for line in text.splitlines() if line.strip()) + return min((len(line) - len(line.lstrip()) for line in text.splitlines() if line.strip()), default=0) def bytes_size(size: int | float) -> str: @@ -281,18 +283,27 @@ def ago(date: datetime.datetime) -> str: def quote(string: str, dirty_only: bool = False, force_single: bool = False) -> str: """Quote a string. + The quote character is chosen so that it does not appear in the string: a string containing single + quotes is wrapped in double quotes, and the other way around. + + **Warning** This helper picks a delimiter, it does not escape. A string containing both quote + characters, or containing the delimiter imposed by `force_single`, cannot be represented and comes + back with an unbalanced delimiter. Do not use it to interpolate untrusted input into SQL or shell + commands; use `shlex.quote` or query parameters instead. + :param string: The string to quote. :param dirty_only: Do not quote strings that have no quotes to begin with. :param force_single: Force single quotes. :return: The quoted string. :rtype: str """ - index = max(string.find(char) for char in ("'", '"')) + contains_single = "'" in string + contains_double = '"' in string - if dirty_only and index == -1: + if dirty_only and not contains_single and not contains_double: return string - double = not force_single and (index == -1 or string[index] == "'") + double = not force_single and (contains_single or not contains_double) return f'"{string}"' if double else f"'{string}'" diff --git a/odev/common/version.py b/odev/common/version.py index a5151d6f7..668ec9864 100644 --- a/odev/common/version.py +++ b/odev/common/version.py @@ -86,7 +86,9 @@ def __repr__(self) -> str: def __bool__(self) -> bool: """Return True if the version is not empty.""" - return bool(self.major or self.minor or self.module or self.master) + # `module` is padded to `MIN_VERSION_LENGTH` and is therefore never an empty tuple: + # it has to be tested on its values rather than on its own truthiness. + return bool(self.major or self.minor or any(self.module) or self.master) @property def major(self) -> int: @@ -130,7 +132,8 @@ def _cmpkey(master: bool, major: int, minor: int, module: tuple, enterprise: boo # Saas versions should sort after non-saas versions _saas = int(saas) - # Master versions should sort before non-master versions + # Master is the development version and therefore sorts after every numbered version, hence it comes + # first in the key so that it outweighs the major number. _master = int(master) return _master, major, minor, _module, enterprise, _saas From be951975b645f547d8585c63393c86a026bc5a1c Mon Sep 17 00:00:00 2001 From: brinkflew Date: Sun, 26 Jul 2026 23:35:37 +0200 Subject: [PATCH 2/7] [IMP] tests: cover string, worktree, version and postgres helpers The suite sat at 64% with `.coveragerc` gating at 60%, and the gap was widest on pure logic the rest of the framework leans on. `odev/common/string.py` had no test module at all despite backing `odev help`, `odev history` and the database listing query. `GitWorktree` turns `git worktree list --porcelain` into the objects the whole fetch/pull/worktree family works with, and nothing exercised it. `OdooVersion` was tested for parsing only, while ordering is what actually picks a revision at runtime. Add test modules for the string helpers, the worktree parser and the datastore table preparation, plus ordering tests for versions. None of them need a network or a real repository. String and version helpers reach 100%, `common/postgres.py` 81% to 93% and the git connector 34% to 40%. Correct four things in the existing suite along the way: - The sudo tests shelled out to a real `sudo cat >> /etc/shadow`. The premise that the command fails only holds for an unprivileged user whose shell cannot open the redirection; a machine granting passwordless sudo runs it for real. Simulate the subprocess and the effective user instead, which also lets the elevation path be asserted rather than inferred. - `test_odev` left a command line behind in `sys.argv` for whichever test ran next. - `_patches` was a list defined on `OdevTestCase`, shared by every subclass through `cls._patches.append`, so each class tore down the patches of all the classes before it. - `PostgresTable` preparation is asserted against a mocked database: driving it against the live datastore made it depend on the connector's query cache, which DDL does not invalidate. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- tests/fixtures/case.py | 15 +- tests/tests/common/test_bash.py | 157 +++++++++++--- tests/tests/common/test_git_worktree.py | 251 ++++++++++++++++++++++ tests/tests/common/test_odev.py | 7 + tests/tests/common/test_postgres_table.py | 240 +++++++++++++++++++++ tests/tests/common/test_string.py | 250 +++++++++++++++++++++ tests/tests/common/test_version.py | 89 ++++++++ 7 files changed, 981 insertions(+), 28 deletions(-) create mode 100644 tests/tests/common/test_git_worktree.py create mode 100644 tests/tests/common/test_postgres_table.py create mode 100644 tests/tests/common/test_string.py diff --git a/tests/fixtures/case.py b/tests/fixtures/case.py index 5d17f7dc4..99c301018 100644 --- a/tests/fixtures/case.py +++ b/tests/fixtures/case.py @@ -34,8 +34,12 @@ class OdevTestCase(TestCase): run_path: ClassVar[Path] """Path to the test case run directory under `/tmp`.""" - _patches: ClassVar[list[_patch]] = [] - """The patches applied to the test case.""" + _patches: ClassVar[list[_patch]] + """The patches applied to the test case. + + Assigned per class in `setUpClass`: a list defined here would be shared by every subclass through + `cls._patches.append(...)`, making each class tear down the patches of all the classes before it. + """ __config: str """Content of the configuration file to restore after each test case.""" @@ -50,6 +54,7 @@ def tearDown(self): @classmethod def setUpClass(cls): + cls._patches = [] Config.parser = ConfigParser() cls.odev = odev.Odev(test=True) cls.run_id = suid() @@ -152,8 +157,10 @@ def _import_dotted_path(cls, path: str) -> Any: @classmethod def __unpatch_all(cls): - for patched in cls._patches: - patched.stop() + # `tearDownClass` runs twice, once through `addClassCleanup` and once through unittest itself, + # so the patches are dropped as they are stopped. + while cls._patches: + cls._patches.pop().stop() @classmethod def _patch_object( diff --git a/tests/tests/common/test_bash.py b/tests/tests/common/test_bash.py index 8c2ecda36..2f5f66105 100644 --- a/tests/tests/common/test_bash.py +++ b/tests/tests/common/test_bash.py @@ -7,6 +7,9 @@ from tests.fixtures import OdevTestCase +PRIVILEGED_COMMAND = "cat /etc/shadow" + + class TestCommonBash(OdevTestCase): """Test the execution of system commands.""" @@ -26,37 +29,143 @@ def test_03_invalid_command_no_raise(self): exec_result = bash.execute("notacommand", raise_on_error=False) self.assertIsNone(exec_result) - def test_04_sudo_no_password(self): - """A command that fails should be re-executed with sudo and fail if the password is not set.""" - with self.assertRaises(CalledProcessError), self.patch(console, "secret") as mock_secret: - mock_secret.return_value = None - bash.execute("cat /etc/shadow", sudo=True) + def test_04_detached(self): + """A command that is run in detached mode should not block the program.""" + start = monotonic() + bash.detached("sleep 1") + self.assertLess(monotonic() - start, 1) + + +class TestCommonBashSudo(OdevTestCase): + """Elevation should be attempted once a command fails, and only with a password at hand. + + The subprocess and the effective user are both simulated: shelling out to `sudo` would make the result + depend on the local sudoers policy, and asking for a real elevation from a test suite is not something + a developer should have to accept. A machine granting passwordless sudo would run the privileged + command for real, and running the suite as root skips elevation entirely. + """ + + def setUp(self): + super().setUp() + self.addCleanup(self.restore_sudo_password) + self.commands: list[str] = [] + + def restore_sudo_password(self): + """Clear the session password cached in the module by the sudo tests.""" + bash.sudo_password = None - def test_05_sudo_no_password_no_raise(self): - """A command that fails should be re-executed with sudo and return None if the password is not set - and raise_on_error is False. + def failure(self, command: str) -> CalledProcessError: + """Build the error raised by a command the user is not allowed to run.""" + return CalledProcessError(1, command, output=b"", stderr=b"Permission denied") + + def patch_subprocess(self, sudo_succeeds: bool = False): + """Patch the subprocess call, recording commands and failing until sudo is used. + + :param sudo_succeeds: Whether the elevated command should succeed instead of failing again. """ - with self.patch(console, "secret", None): - exec_result = bash.execute("cat /etc/shadow", sudo=True, raise_on_error=False) + + def run(command: str, **kwargs): + self.commands.append(command) + + if sudo_succeeds and command.startswith("sudo "): + return CompletedProcess(command, 0, stdout=b"elevated", stderr=b"") + + raise self.failure(command) + + return self.patch(bash, "run_subprocess", side_effect=run) + + def patch_unprivileged_user(self): + """Pretend the suite runs as a regular user, so the elevation path is taken.""" + return self.patch(bash.os, "geteuid", return_value=1000) + + def test_01_password_is_asked_once_the_command_failed(self): + """The session password should only be requested after a first, unprivileged attempt.""" + with ( + self.patch_subprocess(), + self.patch_unprivileged_user(), + self.patch(console, "secret", return_value="secret") as mock_secret, + self.assertRaises(CalledProcessError), + ): + bash.execute(PRIVILEGED_COMMAND, sudo=True) + + mock_secret.assert_called_once() + self.assertEqual(self.commands, [PRIVILEGED_COMMAND, f"sudo -Sks {PRIVILEGED_COMMAND}"]) + + def test_02_no_password_does_not_elevate(self): + """Without a password there is nothing to elevate with, and the original error should surface.""" + with ( + self.patch_subprocess(), + self.patch_unprivileged_user(), + self.patch(console, "secret", return_value=None), + self.assertRaises(CalledProcessError), + ): + bash.execute(PRIVILEGED_COMMAND, sudo=True) + + self.assertEqual(self.commands, [PRIVILEGED_COMMAND]) + + def test_03_no_password_no_raise(self): + """A command failing without a password should return None when not raising.""" + with ( + self.patch_subprocess(), + self.patch_unprivileged_user(), + self.patch(console, "secret", return_value=None), + ): + exec_result = bash.execute(PRIVILEGED_COMMAND, sudo=True, raise_on_error=False) + self.assertIsNone(exec_result) - def test_06_sudo_wrong_password(self): - """A command that fails should be re-executed with sudo and fail again if the password is wrong.""" + def test_04_cached_password_is_reused(self): + """A password from an earlier command should be reused rather than asked again.""" + bash.sudo_password = "cached" # noqa: S105 + + with ( + self.patch_subprocess(sudo_succeeds=True), + self.patch_unprivileged_user(), + self.patch(console, "secret") as mock_secret, + ): + exec_result = bash.execute(PRIVILEGED_COMMAND, sudo=True) + + mock_secret.assert_not_called() + + if exec_result is None: + self.fail("the elevated command should have returned a result") + + self.assertEqual(exec_result.stdout, b"elevated") + + def test_05_wrong_password_raises(self): + """A password rejected by sudo should let the second failure surface.""" bash.sudo_password = "wrongpassword" # noqa: S105 - with self.assertRaises(CalledProcessError): - bash.execute("cat >> /etc/shadow", sudo=True) + with self.patch_subprocess(), self.patch_unprivileged_user(), self.assertRaises(CalledProcessError): + bash.execute(PRIVILEGED_COMMAND, sudo=True) - def test_07_sudo_wrong_password_no_raise(self): - """A command that fails should be re-executed with sudo and return None if the password is wrong - and raise_on_error is False. - """ + def test_06_wrong_password_no_raise(self): + """A password rejected by sudo should return None when not raising.""" bash.sudo_password = "wrongpassword" # noqa: S105 - exec_result = bash.execute("cat >> /etc/shadow", sudo=True, raise_on_error=False) + + with self.patch_subprocess(), self.patch_unprivileged_user(): + exec_result = bash.execute(PRIVILEGED_COMMAND, sudo=True, raise_on_error=False) + self.assertIsNone(exec_result) - def test_08_detached(self): - """A command that is run in detached mode should not block the program.""" - start = monotonic() - bash.detached("sleep 1") - self.assertLess(monotonic() - start, 1) + def test_07_wrong_password_is_forgotten(self): + """A rejected password should not be kept and asked again on the next command.""" + bash.sudo_password = "wrongpassword" # noqa: S105 + + with self.patch_subprocess(), self.patch_unprivileged_user(): + bash.execute(PRIVILEGED_COMMAND, sudo=True, raise_on_error=False) + + self.assertIsNone(bash.sudo_password) + + def test_08_root_does_not_elevate(self): + """Running as root already has the privileges, sudo would add nothing.""" + with ( + self.patch_subprocess(), + self.patch(bash.os, "geteuid", return_value=0), + self.patch(console, "secret") as mock_secret, + ): + exec_result = bash.execute(PRIVILEGED_COMMAND, sudo=True, raise_on_error=False) + + self.assertIsNone(exec_result) + mock_secret.assert_not_called() + self.assertEqual(self.commands, [PRIVILEGED_COMMAND]) diff --git a/tests/tests/common/test_git_worktree.py b/tests/tests/common/test_git_worktree.py new file mode 100644 index 000000000..e74fa76ce --- /dev/null +++ b/tests/tests/common/test_git_worktree.py @@ -0,0 +1,251 @@ +from pathlib import Path +from unittest.mock import MagicMock + +from git.exc import GitCommandError + +from odev.common.connectors.git import GitWorktree + +from tests.fixtures import OdevTestCase + + +WORKTREES_PATH = Path("/home/user/.local/share/odev/worktrees") +COMMIT = "0123456789abcdef0123456789abcdef01234567" + + +def porcelain(worktree: str, commit: str = COMMIT, *attributes: str) -> str: + """Build an entry as emitted by `git worktree list --porcelain`. + + :param worktree: The path of the worktree. + :param commit: The commit the worktree points to. + :param attributes: The trailing attribute lines, such as `branch refs/heads/17.0` or `detached`. + :return: The porcelain entry. + :rtype: str + """ + return "\n".join([f"worktree {worktree}", f"HEAD {commit}", *attributes]) + "\n" + + +class TestGitWorktreeParse(OdevTestCase): + """`git worktree list --porcelain` entries should be parsed into worktree objects.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.connector = MagicMock() + cls.connector.name = "odoo/odoo" + cls.connector.odev.worktrees_path = WORKTREES_PATH + + def parse(self, *arguments: str) -> GitWorktree: + """Parse a porcelain entry against the mocked connector.""" + return GitWorktree.parse(self.connector, porcelain(*arguments)) + + def test_01_branch(self): + """A worktree checked out on a branch should expose its path, commit and branch.""" + worktree = self.parse(f"{WORKTREES_PATH}/17.0/odoo", COMMIT, "branch refs/heads/17.0") + + self.assertEqual(worktree.path, WORKTREES_PATH / "17.0" / "odoo") + self.assertEqual(worktree.commit, COMMIT) + self.assertEqual(worktree.ref, "refs/heads/17.0") + self.assertEqual(worktree.branch, "17.0") + self.assertFalse(worktree.detached) + self.assertFalse(worktree.bare) + self.assertFalse(worktree.locked) + self.assertFalse(worktree.prunable) + + def test_02_detached(self): + """A detached worktree should be flagged as such and have no branch.""" + worktree = self.parse(f"{WORKTREES_PATH}/16.0/odoo", COMMIT, "detached") + + self.assertTrue(worktree.detached) + self.assertIsNone(worktree.branch) + self.assertIsNone(worktree.ref) + + def test_03_bare(self): + """A bare repository should be flagged as such.""" + worktree = self.parse("/home/user/repositories/odoo/odoo", COMMIT, "bare") + + self.assertTrue(worktree.bare) + self.assertFalse(worktree.detached) + + def test_04_locked_with_reason(self): + """A locked worktree should keep the reason given to `git worktree lock`.""" + worktree = self.parse(f"{WORKTREES_PATH}/15.0/odoo", COMMIT, "branch refs/heads/15.0", "locked on a usb drive") + + self.assertTrue(worktree.locked) + self.assertEqual(worktree.locked_reason, "on a usb drive") + + def test_05_locked_without_reason(self): + """A worktree may be locked without an explanation.""" + worktree = self.parse(f"{WORKTREES_PATH}/15.0/odoo", COMMIT, "branch refs/heads/15.0", "locked") + + self.assertTrue(worktree.locked) + self.assertIsNone(worktree.locked_reason) + + def test_06_prunable_with_reason(self): + """A prunable worktree should keep the reason reported by git.""" + worktree = self.parse( + f"{WORKTREES_PATH}/14.0/odoo", + COMMIT, + "branch refs/heads/14.0", + "prunable gitdir file points to non-existent location", + ) + + self.assertTrue(worktree.prunable) + self.assertEqual(worktree.prunable_reason, "gitdir file points to non-existent location") + + def test_07_flags_are_booleans(self): + """Flags should be coerced to booleans, not left as the matched text.""" + worktree = self.parse(f"{WORKTREES_PATH}/16.0/odoo", COMMIT, "detached") + + for flag in (worktree.bare, worktree.detached, worktree.locked, worktree.prunable): + self.assertIsInstance(flag, bool) + + +class TestGitWorktreeBranch(OdevTestCase): + """Worktrees created by odev carry a local branch suffixed with the worktree name.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.connector = MagicMock() + cls.connector.name = "odoo/odoo" + cls.connector.odev.worktrees_path = WORKTREES_PATH + + def test_01_odev_suffix_is_stripped(self): + """`GitConnector.create_worktree` names local branches `-odev-`. + + The upstream revision is what commands display and compare, the local branch is what git needs. + """ + worktree = GitWorktree.parse( + self.connector, + porcelain(f"{WORKTREES_PATH}/mydb/odoo", COMMIT, "branch refs/heads/17.0-odev-mydb"), + ) + + self.assertEqual(worktree.local_branch, "17.0-odev-mydb") + self.assertEqual(worktree.branch, "17.0") + + def test_02_branch_without_suffix_is_kept(self): + """A branch not created by odev should be reported unchanged.""" + worktree = GitWorktree.parse( + self.connector, + porcelain(f"{WORKTREES_PATH}/17.0/odoo", COMMIT, "branch refs/heads/17.0"), + ) + + self.assertEqual(worktree.local_branch, "17.0") + self.assertEqual(worktree.branch, "17.0") + + def test_03_only_the_first_suffix_is_split(self): + """A branch name containing the separator more than once should split on the first occurrence.""" + worktree = GitWorktree.parse( + self.connector, + porcelain(f"{WORKTREES_PATH}/mydb/odoo", COMMIT, "branch refs/heads/17.0-odev-my-odev-db"), + ) + + self.assertEqual(worktree.branch, "17.0") + + +class TestGitWorktreeIdentity(OdevTestCase): + """Worktrees are named after their directory and identified by their path.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.connector = MagicMock() + cls.connector.name = "odoo/odoo" + cls.connector.odev.worktrees_path = WORKTREES_PATH + + def worktree(self, path: str, branch: str = "17.0") -> GitWorktree: + """Build a worktree on a branch at the given path.""" + return GitWorktree.parse(self.connector, porcelain(path, COMMIT, f"branch refs/heads/{branch}")) + + def test_01_name_is_the_parent_directory(self): + """A worktree under the odev worktrees directory is named after the directory holding it.""" + self.assertEqual(self.worktree(f"{WORKTREES_PATH}/17.0/odoo").name, "17.0") + self.assertEqual(self.worktree(f"{WORKTREES_PATH}/mydb/enterprise").name, "mydb") + + def test_02_name_outside_worktrees_path(self): + """A checkout managed outside of odev is reported as the master worktree.""" + self.assertEqual(self.worktree("/home/user/repositories/odoo/odoo").name, "master") + + def test_03_equality_on_path(self): + """Two worktrees at the same path are the same worktree, whatever their revision.""" + self.assertEqual(self.worktree(f"{WORKTREES_PATH}/17.0/odoo"), self.worktree(f"{WORKTREES_PATH}/17.0/odoo")) + self.assertNotEqual(self.worktree(f"{WORKTREES_PATH}/17.0/odoo"), self.worktree(f"{WORKTREES_PATH}/16.0/odoo")) + self.assertNotEqual(self.worktree(f"{WORKTREES_PATH}/17.0/odoo"), f"{WORKTREES_PATH}/17.0/odoo") + + def test_04_hash_on_path(self): + """Worktrees should deduplicate on their path when collected in a set.""" + worktrees = { + self.worktree(f"{WORKTREES_PATH}/17.0/odoo", "17.0"), + self.worktree(f"{WORKTREES_PATH}/17.0/odoo", "saas-17.2"), + self.worktree(f"{WORKTREES_PATH}/16.0/odoo"), + } + + self.assertEqual(len(worktrees), 2) + + def test_05_repr(self): + """The representation should identify the worktree by name, repository and revision.""" + self.assertEqual( + repr(self.worktree(f"{WORKTREES_PATH}/17.0/odoo")), + "GitWorktree(name='17.0', repository='odoo/odoo', revision='17.0')", + ) + + +class TestGitWorktreePendingChanges(OdevTestCase): + """Pending changes are counted from the revision list against the tracked upstream branch.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.connector = MagicMock() + cls.connector.name = "odoo/odoo" + cls.connector.odev.worktrees_path = WORKTREES_PATH + + def worktree(self, *attributes: str) -> GitWorktree: + """Build a worktree with the given porcelain attributes.""" + return GitWorktree.parse(self.connector, porcelain(f"{WORKTREES_PATH}/17.0/odoo", COMMIT, *attributes)) + + def patch_rev_list(self, result: str | Exception): + """Patch `git.Repo` as used by `pending_changes` to return or raise on `rev_list`.""" + repository = MagicMock() + + if isinstance(result, Exception): + repository.git.rev_list.side_effect = result + else: + repository.git.rev_list.return_value = result + + return self.patch("odev.common.connectors.git", "Repo", return_value=repository) + + def test_01_detached_has_no_upstream(self): + """A detached worktree has nothing to compare against and reports no pending changes.""" + self.assertEqual(self.worktree("detached").pending_changes(), (0, 0)) + + def test_02_counts_behind_and_ahead(self): + """The counts should be read from `git rev-list --left-right --count`.""" + with self.patch_rev_list("12\t3"): + self.assertEqual(self.worktree("branch refs/heads/17.0").pending_changes(), (12, 3)) + + def test_03_up_to_date(self): + """A worktree in sync with its upstream should report no pending changes.""" + with self.patch_rev_list("0\t0"): + self.assertEqual(self.worktree("branch refs/heads/17.0").pending_changes(), (0, 0)) + + def test_04_no_upstream_configured(self): + """A branch without an upstream cannot be compared and reports no pending changes.""" + error = GitCommandError("rev-list", 128, b"fatal: no upstream configured for branch '17.0'") + + with self.patch_rev_list(error): + self.assertEqual(self.worktree("branch refs/heads/17.0").pending_changes(), (0, 0)) + + def test_05_head_does_not_point_to_a_branch(self): + """A HEAD not pointing to a branch cannot be compared and reports no pending changes.""" + error = GitCommandError("rev-list", 128, b"fatal: HEAD does not point to a branch") + + with self.patch_rev_list(error): + self.assertEqual(self.worktree("branch refs/heads/17.0").pending_changes(), (0, 0)) + + def test_06_unexpected_git_error_is_raised(self): + """Any other git failure should surface instead of being reported as no pending changes.""" + error = GitCommandError("rev-list", 128, b"fatal: bad revision") + + with self.patch_rev_list(error), self.assertRaises(GitCommandError): + self.worktree("branch refs/heads/17.0").pending_changes() diff --git a/tests/tests/common/test_odev.py b/tests/tests/common/test_odev.py index 146abed80..73d942b67 100644 --- a/tests/tests/common/test_odev.py +++ b/tests/tests/common/test_odev.py @@ -14,6 +14,13 @@ class TestCommonOdev(OdevTestCase): """Global sanity check of the odev framework.""" + def setUp(self): + super().setUp() + # Dispatching a command reads `sys.argv`; leaving a command line behind would feed it to whichever + # test runs next. + argv = sys.argv + self.addCleanup(setattr, sys, "argv", argv) + def test_01_config_file(self): """Config file should have been created in the correct directory.""" self.assertEqual(self.odev.config.name, "odev-test") diff --git a/tests/tests/common/test_postgres_table.py b/tests/tests/common/test_postgres_table.py new file mode 100644 index 000000000..1ccda851d --- /dev/null +++ b/tests/tests/common/test_postgres_table.py @@ -0,0 +1,240 @@ +from unittest.mock import MagicMock + +from psycopg2.errors import InvalidTableDefinition + +from odev.common.connectors import PostgresConnector +from odev.common.postgres import PostgresDatabase, PostgresTable + +from tests.fixtures import OdevTestCase + + +class TestPostgresColumnsExist(OdevTestCase): + """`columns_exist` reports the columns missing from a table, it does not need a live database.""" + + def columns_exist(self, existing: list[str], requested: list[str]) -> list[str]: + """Run `columns_exist` against a connector whose query returns the given existing columns.""" + connector = PostgresConnector.__new__(PostgresConnector) + connector.query = lambda _: [(column,) for column in existing] # type: ignore [method-assign] + return connector.columns_exist("table", requested) + + def test_01_some_columns_missing(self): + """Only the requested columns absent from the table should be returned.""" + self.assertEqual(self.columns_exist(["id"], ["id", "name", "date"]), ["name", "date"]) + + def test_02_no_column_exists(self): + """An empty result means none of the requested columns exist, so all of them are missing. + + A table created from an older definition holds none of the new columns; reporting nothing missing + would leave it unmigrated. + """ + self.assertEqual(self.columns_exist([], ["id", "name"]), ["id", "name"]) + + def test_03_all_columns_exist(self): + """A table holding every requested column should report nothing missing.""" + self.assertEqual(self.columns_exist(["id", "name"], ["id", "name"]), []) + + def test_04_order_is_preserved(self): + """Missing columns should be reported in the order they were requested.""" + self.assertEqual(self.columns_exist(["b"], ["a", "b", "c"]), ["a", "c"]) + + def test_05_no_column_requested(self): + """Asking for no column should report nothing missing without querying the database.""" + connector = PostgresConnector.__new__(PostgresConnector) + + def fail_on_query(_): + raise AssertionError("no query should be issued when no column is requested") + + connector.query = fail_on_query # type: ignore [method-assign] + self.assertEqual(connector.columns_exist("table", []), []) + + def test_06_columns_must_be_a_list(self): + """Passing a bare string would build a query over its characters and is rejected.""" + connector = PostgresConnector.__new__(PostgresConnector) + + with self.assertRaises(TypeError): + connector.columns_exist("table", "id") # type: ignore [arg-type] + + +class TestPostgresTable(OdevTestCase): + """Tables should be brought in line with the definition declared on their subclass.""" + + table_name = "odev_test_table" + + def setUp(self): + super().setUp() + self.database = MagicMock(spec=PostgresDatabase) + """The database the tables are built against, kept aside to assert the queries it received.""" + + self.database.name = "odev-test" + self.database.tables = {} + self.database.columns_exist.return_value = [] + + def build_table( + self, + columns: dict[str, str] | None, + constraints: dict[str, str] | None = None, + missing: list[str] | None = None, + ) -> PostgresTable: + """Build a table against the mocked database, reporting the given columns as missing. + + :param columns: The columns declared on the table subclass. + :param constraints: The constraints declared on the table subclass. + :param missing: The columns `columns_exist` should report as absent from the table. + """ + self.database.columns_exist.return_value = missing or [] + table_name = self.table_name + + class TestTable(PostgresTable): + name = table_name + _columns = columns + _constraints = constraints + + return TestTable(self.database) + + def test_01_registers_itself_on_the_database(self): + """A table should be reachable from the database it was built against.""" + table = self.build_table({"id": "SERIAL PRIMARY KEY"}) + + self.assertIs(self.database.tables[self.table_name], table) + + def test_02_prepare_creates_the_table(self): + """Preparing a table should create it from the columns of its definition.""" + columns = {"id": "SERIAL PRIMARY KEY", "name": "VARCHAR"} + self.build_table(columns).prepare_database_table() + + self.database.create_table.assert_called_once_with(self.table_name, columns) + + def test_03_prepare_without_columns_does_nothing(self): + """A table whose subclass declares no column has nothing to create.""" + self.build_table(None).prepare_database_table() + + self.database.create_table.assert_not_called() + self.database.columns_exist.assert_not_called() + + def test_04_prepare_adds_missing_columns(self): + """Columns absent from an existing table should be added to it. + + `CREATE TABLE IF NOT EXISTS` leaves an existing table alone, so a column added to the definition + can only reach the table through this pass. + """ + table = self.build_table({"id": "SERIAL PRIMARY KEY", "comment": "VARCHAR"}, missing=["comment"]) + table.prepare_database_table() + + self.database.create_column.assert_called_once_with(self.table_name, "comment", "VARCHAR") + + def test_05_prepare_without_missing_columns(self): + """A table already matching its definition should not be altered.""" + self.build_table({"id": "SERIAL PRIMARY KEY"}).prepare_database_table() + + self.database.create_column.assert_not_called() + + def test_06_prepare_applies_constraints(self): + """Constraints declared on the table should be applied when it is prepared.""" + table = self.build_table( + {"id": "SERIAL PRIMARY KEY", "name": "VARCHAR"}, + {"name_unique": "UNIQUE (name)"}, + ) + table.prepare_database_table() + + self.database.constraint.assert_called_once_with(self.table_name, "name_unique", "UNIQUE (name)") + + def test_07_constraints_need_columns(self): + """Constraints are applied from within the columns pass and are skipped without a definition.""" + self.build_table(None, {"name_unique": "UNIQUE (name)"}).prepare_database_table() + + self.database.constraint.assert_not_called() + + def test_08_clear_empties_the_table(self): + """Clearing a table should delete its rows, not the table itself.""" + self.build_table({"id": "SERIAL PRIMARY KEY"}).clear() + + self.database.query.assert_called_once_with(f"DELETE FROM {self.table_name}") + + +class TestPostgresTableMissingColumnErrors(OdevTestCase): + """Renaming a primary key column requires dropping the constraint left by the previous definition.""" + + table_name = "odev_test_table" + + def setUp(self): + super().setUp() + self.database = MagicMock(spec=PostgresDatabase) + """The database the table is built against, kept aside to assert the queries it received.""" + + self.database.name = "odev-test" + self.database.tables = {} + self.database.columns_exist.return_value = ["identifier"] + + def build_table(self, error: Exception) -> PostgresTable: + """Build a table whose first `create_column` call raises the given error.""" + self.database.create_column.side_effect = [error, None] + table_name = self.table_name + + class TestTable(PostgresTable): + name = table_name + _columns = {"identifier": "SERIAL PRIMARY KEY"} + + return TestTable(self.database) + + def runtime_error(self, cause: Exception) -> RuntimeError: + """Build the error raised by the connector thread when a query fails.""" + error = RuntimeError(f"Exception in thread: {cause}") + error.__cause__ = cause + + return error + + def test_01_drops_the_stale_primary_key(self): + """A leftover primary key should be dropped so the renamed column can be created.""" + cause = InvalidTableDefinition('multiple primary keys for table "odev_test_table" are not allowed') + + self.build_table(self.runtime_error(cause)).prepare_database_table() + + self.database.query.assert_called_once_with( + f"ALTER TABLE {self.table_name} DROP CONSTRAINT IF EXISTS {self.table_name}_pkey" + ) + self.assertEqual(self.database.create_column.call_count, 2) + + def test_02_other_table_definition_errors_are_swallowed(self): + """Another invalid definition should be logged without retrying nor dropping a constraint.""" + cause = InvalidTableDefinition("column is of type integer but default expression is of type text") + + self.build_table(self.runtime_error(cause)).prepare_database_table() + + self.database.query.assert_not_called() + self.assertEqual(self.database.create_column.call_count, 1) + + def test_03_unrelated_runtime_errors_are_raised(self): + """A failure unrelated to the table definition should surface to the caller.""" + table = self.build_table(self.runtime_error(ValueError("boom"))) + + with self.assertRaises(RuntimeError): + table.prepare_database_table() + + +class TestPostgresDatabaseTables(OdevTestCase): + """Each database should own the mapping of the tables registered against it.""" + + def test_01_tables_are_not_shared_between_databases(self): + """Registering a table on one database should not make it visible on another. + + The mapping used to be a class attribute, so every database instance shared a single registry and + tables collided across databases on their name alone. + """ + first = PostgresDatabase.__new__(PostgresDatabase) + first.tables = {} + second = PostgresDatabase.__new__(PostgresDatabase) + second.tables = {} + + first.tables["history"] = MagicMock(spec=PostgresTable) + + self.assertEqual(second.tables, {}) + self.assertIsNot(first.tables, second.tables) + + def test_02_tables_are_instance_attributes(self): + """The registry should not live on the class, where every database would share it.""" + self.assertNotIn("tables", PostgresDatabase.__dict__) + self.assertIn("tables", vars(self.odev.store)) + + def test_03_datastore_registers_its_tables(self): + """The odev datastore should hold the three tables it is made of.""" + self.assertEqual(set(self.odev.store.tables), {"databases", "history", "secrets"}) diff --git a/tests/tests/common/test_string.py b/tests/tests/common/test_string.py new file mode 100644 index 000000000..40d409b3b --- /dev/null +++ b/tests/tests/common/test_string.py @@ -0,0 +1,250 @@ +import datetime + +from odev.common import string + +from tests.fixtures import OdevTestCase + + +class TestCommonStringSizes(OdevTestCase): + """Byte sizes should be formatted and parsed back consistently.""" + + def test_01_bytes_size_units(self): + """Sizes should be scaled down to the largest unit under the 1024 factor.""" + self.assertEqual(string.bytes_size(0), "0.0 B") + self.assertEqual(string.bytes_size(512), "512.0 B") + self.assertEqual(string.bytes_size(1024), "1.0 KB") + self.assertEqual(string.bytes_size(1024**2), "1.0 MB") + self.assertEqual(string.bytes_size(1536**2), "2.2 MB") + self.assertEqual(string.bytes_size(1024**3), "1.0 GB") + + def test_02_bytes_size_largest_unit(self): + """Sizes above the largest known unit should fall back to yottabytes.""" + self.assertEqual(string.bytes_size(1024**8), "1.0 YB") + self.assertEqual(string.bytes_size(1024**9), "1024.0 YB") + + def test_03_bytes_size_negative(self): + """Negative sizes should be scaled on their absolute value and keep their sign.""" + self.assertEqual(string.bytes_size(-1024), "-1.0 KB") + + def test_04_bytes_from_string(self): + """Human readable sizes should be converted back to a number of bytes.""" + self.assertEqual(string.bytes_from_string("512"), 512) + self.assertEqual(string.bytes_from_string("512 B"), 512) + self.assertEqual(string.bytes_from_string("1 KB"), 1024) + self.assertEqual(string.bytes_from_string("1.5 MB"), 1572864) + self.assertEqual(string.bytes_from_string("2GB"), 2 * 1024**3) + + def test_05_bytes_from_string_invalid(self): + """A representation not starting with a number cannot be parsed.""" + with self.assertRaises(ValueError): + string.bytes_from_string("not a size") + + def test_06_bytes_size_roundtrip(self): + """Formatting a size and parsing it back should return the original value.""" + for size in (1024, 4 * 1024**2, 3 * 1024**3): + self.assertEqual(string.bytes_from_string(string.bytes_size(size)), size) + + +class TestCommonStringIndent(OdevTestCase): + """Indentation helpers back the layout of the `help` command output.""" + + text = " first line\n nested line\n last line" + + def test_01_min_indent(self): + """The smallest indentation of all non-blank lines should be returned.""" + self.assertEqual(string.min_indent(self.text), 4) + self.assertEqual(string.min_indent("no indent"), 0) + + def test_02_min_indent_without_content(self): + """A text without any non-blank line has no indentation to measure.""" + self.assertEqual(string.min_indent(""), 0) + self.assertEqual(string.min_indent("\n\n"), 0) + self.assertEqual(string.min_indent(" \n\t\n "), 0) + + def test_03_indent(self): + """Indenting should prefix every line with the requested number of spaces.""" + self.assertEqual(string.indent("one\ntwo", 2), " one\n two") + self.assertEqual(string.indent("one\ntwo"), "one\ntwo") + + def test_04_dedent(self): + """Dedenting by zero should keep the text as-is, relative indentation included.""" + self.assertEqual(string.dedent(self.text), self.text) + + def test_05_dedent_removes_indentation(self): + """Dedenting should remove the requested number of spaces from every line.""" + self.assertEqual(string.dedent(self.text, 4), "first line\n nested line\nlast line") + + def test_06_dedent_without_content(self): + """Dedenting a blank text should not fail on the absence of a minimum indentation.""" + self.assertEqual(string.dedent(""), "") + self.assertEqual(string.dedent("\n\n"), "\n\n") + + def test_07_normalize_indent(self): + """Normalizing should clean up a docstring-like text and strip its surrounding blanks.""" + self.assertEqual(string.normalize_indent("\n first line\n second line\n "), "first line\nsecond line") + self.assertEqual(string.normalize_indent(""), "") + + +class TestCommonStringJoin(OdevTestCase): + """Parts should be joined with the delimiters expected in user-facing messages.""" + + def test_01_join(self): + """Parts should be joined with commas when no last delimiter is given.""" + self.assertEqual(string.join([]), "") + self.assertEqual(string.join(["one"]), "one") + self.assertEqual(string.join(["one", "two", "three"]), "one, two, three") + + def test_02_join_and(self): + """The last two parts should be separated by "and".""" + self.assertEqual(string.join_and([]), "") + self.assertEqual(string.join_and(["one"]), "one") + self.assertEqual(string.join_and(["one", "two"]), "one and two") + self.assertEqual(string.join_and(["one", "two", "three"]), "one, two and three") + + def test_03_join_or(self): + """The last two parts should be separated by "or".""" + self.assertEqual(string.join_or(["one", "two", "three"]), "one, two or three") + + def test_04_join_bullet(self): + """Parts should be listed as bullets, without a leading blank line.""" + self.assertEqual(string.join_bullet([]), "") + self.assertEqual(string.join_bullet(["one"]), "• one") + self.assertEqual(string.join_bullet(["one", "two"]), "• one\n• two") + + +class TestCommonStringQuote(OdevTestCase): + """Quoting picks a delimiter absent from the string, it never escapes.""" + + def test_01_quote_default(self): + """A string without quotes should be wrapped in double quotes.""" + self.assertEqual(string.quote("plain"), '"plain"') + + def test_02_quote_containing_single(self): + """A string containing single quotes should be wrapped in double quotes.""" + self.assertEqual(string.quote("it's"), '"it\'s"') + + def test_03_quote_containing_double(self): + """A string containing double quotes should be wrapped in single quotes.""" + self.assertEqual(string.quote('say "hi"'), "'say \"hi\"'") + + def test_04_quote_force_single(self): + """Forcing single quotes should win over the automatic delimiter choice.""" + self.assertEqual(string.quote("plain", force_single=True), "'plain'") + self.assertEqual(string.quote('say "hi"', force_single=True), "'say \"hi\"'") + + def test_05_quote_dirty_only(self): + """Strings without any quote should be left untouched in `dirty_only` mode.""" + self.assertEqual(string.quote("plain", dirty_only=True), "plain") + self.assertEqual(string.quote("plain", dirty_only=True, force_single=True), "plain") + self.assertEqual(string.quote("it's", dirty_only=True), '"it\'s"') + + def test_06_quote_both_quote_characters(self): + """A string containing both delimiters cannot be represented and falls back to double quotes. + + Documented in `quote`: the helper selects a delimiter and never escapes, so callers must not + feed it untrusted input. + """ + self.assertEqual(string.quote("""a'b"c"""), '"a\'b"c"') + + +class TestCommonStringMarkup(OdevTestCase): + """Rich markup helpers should produce tags the console can render.""" + + def test_01_stylize_resolves_theme_styles(self): + """Aliased theme styles should be replaced by the value Rich understands.""" + self.assertEqual(string.stylize("text", "bold"), "[bold]text[/bold]") + self.assertNotIn("color.cyan", string.stylize("text", "color.cyan")) + + def test_02_list_styles(self): + """Opening tags should be listed in their order of appearance, closing ones ignored.""" + self.assertEqual(string.list_styles("[bold]one[/bold] [color.cyan]two[/color.cyan]"), ["bold", "color.cyan"]) + self.assertEqual(string.list_styles("[bold red]one[/bold red]"), ["bold red"]) + self.assertEqual(string.list_styles("no markup here"), []) + + def test_03_strip_styles(self): + """Markup tags should be removed, keeping the text they wrap.""" + self.assertEqual(string.strip_styles("[bold]text[/bold]"), "text") + self.assertEqual(string.strip_styles("plain text"), "plain text") + + def test_04_strip_styles_keeps_nested_tags(self): + """Only the outermost tag pair is removed, nested markup survives. + + `strip_styles` runs a single non-greedy substitution pass. Nothing in odev calls it today, so the + limitation is asserted rather than fixed. + """ + self.assertEqual( + string.strip_styles("[bold]one [color.cyan]two[/color.cyan] three[/bold]"), + "one [color.cyan]two[/color.cyan] three", + ) + + def test_05_resolve_styles(self): + """Aliased styles inside a text should be resolved to their theme value.""" + resolved = string.resolve_styles("[color.cyan]text[/color.cyan]") + self.assertNotIn("color.cyan", resolved) + self.assertIn("text", resolved) + + def test_06_strip_ansi_colors(self): + """ANSI color codes should be removed, leaving the text untouched.""" + self.assertEqual(string.strip_ansi_colors("\x1b[31mred\x1b[0m"), "red") + self.assertEqual(string.strip_ansi_colors("no colors"), "no colors") + + def test_07_link(self): + """Links should be rendered with the Rich link markup.""" + self.assertEqual( + string.link("odev", "https://github.com/odoo-odev"), "[link=https://github.com/odoo-odev]odev[/link]" + ) + + +class TestCommonStringHelpFormatting(OdevTestCase): + """Help formatting keeps the descriptions of the `help` command aligned in a column.""" + + def test_01_short_help(self): + """The name should be emphasized and the description aligned after the indentation.""" + self.assertEqual(string.short_help("run", "Run a database"), "[bold]run[/bold] Run a database") + self.assertEqual(string.short_help("run", "Run a database", 4), "[bold]run[/bold] Run a database") + + def test_02_format_options_list_aligns_descriptions(self): + """Descriptions should all start at the same column, driven by the longest name.""" + formatted = string.format_options_list([("run", "Run a database"), ("shell", "Open a shell")]) + descriptions = [line.index("Run a database") for line in formatted.splitlines() if "Run a database" in line] + descriptions += [line.index("Open a shell") for line in formatted.splitlines() if "Open a shell" in line] + + self.assertEqual(len(set(descriptions)), 1, "descriptions should be aligned on a single column") + + def test_03_format_options_list_blank_lines(self): + """Blank lines should be inserted between the elements of the list.""" + formatted = string.format_options_list([("run", "Run"), ("shell", "Shell")], blanks=1) + self.assertEqual(len(formatted.splitlines()), 3) + + +class TestCommonStringMisc(OdevTestCase): + """Remaining formatting helpers.""" + + def test_01_suid(self): + """Unique identifiers should be lowercase alphanumeric strings of a fixed length.""" + identifiers = {string.suid() for _ in range(100)} + + for identifier in identifiers: + self.assertRegex(identifier, r"^[a-z0-9]{8}$") + + self.assertGreater(len(identifiers), 1, "identifiers should not be constant") + + def test_02_seconds_to_time(self): + """Seconds should be rendered as a hours:minutes:seconds duration.""" + self.assertEqual(string.seconds_to_time(0), "0:00:00") + self.assertEqual(string.seconds_to_time(3661), "1:01:01") + + def test_03_ago(self): + """Past datetimes should be rendered relative to now.""" + self.assertEqual(string.ago(datetime.datetime.now() - datetime.timedelta(hours=2)), "2 hours ago") + + def test_04_float_to_hours_drops_minutes(self): + """Fractions of an hour are lost, minutes always come out as zero. + + `int(value - hours) * 60` truncates the fraction before scaling it, so it can only ever yield 0. + Nothing in odev nor in the plugins calls this helper, so the behaviour is asserted as-is rather + than fixed; correcting it would be `int((value - hours) * 60)`. + """ + self.assertEqual(string.float_to_hours(2.0), "2:00") + self.assertEqual(string.float_to_hours(1.5), "1:00") + self.assertEqual(string.float_to_hours(2.25), "2:00") diff --git a/tests/tests/common/test_version.py b/tests/tests/common/test_version.py index c4ee8bfab..18b7829a0 100644 --- a/tests/tests/common/test_version.py +++ b/tests/tests/common/test_version.py @@ -47,3 +47,92 @@ def test_04_master(self): def test_05_invalid(self): with self.assertRaises(InvalidVersion): OdooVersion("invalid") + + def test_06_major_only(self): + """A version without a minor number should default it to zero.""" + parsed = OdooVersion("17") + self.assertEqual(parsed.major, 17) + self.assertEqual(parsed.minor, 0) + self.assertEqual(str(parsed), "17.0") + + def test_07_enterprise(self): + """The enterprise marker should be parsed but left out of the string representation.""" + parsed = OdooVersion("17.0+e") + self.assertTrue(parsed.enterprise) + self.assertEqual(str(parsed), "17.0") + self.assertFalse(OdooVersion("17.0").enterprise) + + def test_08_repr(self): + """The representation should wrap the string version.""" + self.assertEqual(repr(OdooVersion("saas~16.4")), "OdooVersion(saas-16.4)") + + +class TestCommonVersionBool(OdevTestCase): + """A version should be falsy only when it carries no version information at all.""" + + def test_01_empty(self): + """An empty version has nothing set and should be falsy. + + `module` is padded to `MIN_VERSION_LENGTH`, so it is never an empty tuple and cannot be tested on + its own truthiness. + """ + self.assertFalse(OdooVersion("")) + self.assertFalse(OdooVersion("0.0")) + + def test_02_not_empty(self): + """Any version component being set should make the version truthy.""" + self.assertTrue(OdooVersion("17.0")) + self.assertTrue(OdooVersion("0.1")) + self.assertTrue(OdooVersion("master")) + self.assertTrue(OdooVersion("0.0.1.0.0")) + + +class TestCommonVersionOrdering(OdevTestCase): + """Versions should sort the way Odoo releases succeed each other. + + Ordering is what picks a revision when odev has several to choose from, so it matters as much as + parsing does. + """ + + def test_01_major_versions(self): + """Newer major versions should sort after older ones.""" + self.assertLess(OdooVersion("15.0"), OdooVersion("16.0")) + self.assertGreater(OdooVersion("17.0"), OdooVersion("16.0")) + + def test_02_saas_between_majors(self): + """A SaaS version should sort after the major it branches off, and before the next one.""" + self.assertGreater(OdooVersion("saas~16.4"), OdooVersion("16.0")) + self.assertLess(OdooVersion("saas~16.4"), OdooVersion("17.0")) + self.assertGreater(OdooVersion("saas~16.4"), OdooVersion("saas~16.2")) + + def test_03_saas_after_same_numbered_version(self): + """At equal numbers, a SaaS version should sort after the stable one.""" + self.assertGreater(OdooVersion("saas~16.0"), OdooVersion("16.0")) + + def test_04_master_is_the_newest(self): + """Master is the development version and should sort after every numbered version.""" + self.assertGreater(OdooVersion("master"), OdooVersion("17.0")) + self.assertGreater(OdooVersion("master"), OdooVersion("saas~17.4")) + + def test_05_enterprise_after_community(self): + """At equal versions, the enterprise edition should sort after the community one.""" + self.assertGreater(OdooVersion("17.0+e"), OdooVersion("17.0")) + + def test_06_module_versions(self): + """Module versions should be compared component by component, ignoring trailing zeros.""" + self.assertLess(OdooVersion("17.0.1.0.0"), OdooVersion("17.0.1.1.0")) + self.assertEqual(OdooVersion("17.0.1.0.0"), OdooVersion("17.0.1")) + + def test_07_equality_and_hash(self): + """Equal versions should compare equal and hash alike, whatever their notation.""" + self.assertEqual(OdooVersion("17.0"), OdooVersion("17.0")) + self.assertEqual(hash(OdooVersion("17.0")), hash(OdooVersion("17.0"))) + self.assertEqual(OdooVersion("saas~16.4"), OdooVersion("saas-16.4")) + self.assertNotEqual(OdooVersion("17.0"), OdooVersion("16.0")) + + def test_08_sorting(self): + """Sorting a set of versions should yield the chronological order of the releases.""" + versions = ["master", "16.0", "saas~16.4", "17.0", "15.0", "saas~17.2"] + expected = ["15.0", "16.0", "saas-16.4", "17.0", "saas-17.2", "master"] + + self.assertEqual([str(version) for version in sorted(map(OdooVersion, versions))], expected) From 87dc0e497c1085bd3ff759e920a5dcecf48e0550 Mon Sep 17 00:00:00 2001 From: brinkflew Date: Mon, 27 Jul 2026 00:59:43 +0200 Subject: [PATCH 3/7] [FIX] common: keep listing databases when one is dropped mid-inspection `is_odoo` checks that a database exists and then connects to it, and any process can drop it in between: `odev list` inspects every database in turn and would fail outright because one went away. A database that is gone is not an Odoo database, while anything else stays an error. The second check bypasses the query cache, since the cache is what claimed the database was still there. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- odev/common/databases/local.py | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/odev/common/databases/local.py b/odev/common/databases/local.py index d9a732646..84d9da044 100644 --- a/odev/common/databases/local.py +++ b/odev/common/databases/local.py @@ -23,6 +23,7 @@ from zipfile import ZipFile from packaging.version import Version +from psycopg2 import OperationalError from odev.common import bash, progress, string from odev.common.connectors import GitWorktree, PostgresConnector @@ -116,8 +117,20 @@ def is_odoo(self) -> bool: if not self.exists: return False - with self: - return self.table_exists("ir_module_module") + try: + with self: + return self.table_exists("ir_module_module") + + except OperationalError: + # Any process can drop a database at any time, including between the check above and this + # connection. Listing databases inspects each of them in turn and must not fail because one + # went away in the meantime; anything else is a genuine connection error. The check has to + # reach the server rather than the cache, which is what said the database was still there. + with self.psql() as psql, psql.nocache(): + if psql.database_exists(self.name): + raise + + return False @property def venv(self) -> PythonEnv: From 86630d345acaedd1bfaf8f5021526c7adc0ecca5 Mon Sep 17 00:00:00 2001 From: brinkflew Date: Mon, 27 Jul 2026 00:59:54 +0200 Subject: [PATCH 4/7] [IMP] common: let the framework namespace and the command link be chosen `Odev` derived its name from the test mode alone, and everything it owns is named after it: the configuration file and the datastore database were fixed paths that any two instances had to share. It can now be given a name of its own. The setup script computed the destination of the `odev` symlink halfway through creating it, leaving no way to point it elsewhere; the decision moves to `link_path`. Both make the test suite able to run against resources of its own rather than against those of the user. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- odev/common/odev.py | 18 +++++++++++++---- odev/setup/symlink.py | 47 ++++++++++++++++++++++++++++++------------- 2 files changed, 47 insertions(+), 18 deletions(-) diff --git a/odev/common/odev.py b/odev/common/odev.py index c84426e36..e57d2d7c3 100644 --- a/odev/common/odev.py +++ b/odev/common/odev.py @@ -23,7 +23,6 @@ Any, ClassVar, Generic, - Literal, NamedTuple, TypedDict, cast, @@ -131,10 +130,11 @@ class Odev(Generic[CommandType]): _command_stack: list[CommandType] = [] """Stack of current commands being executed. Last command in list is the one currently running.""" - def __init__(self, test: bool = False): + def __init__(self, test: bool = False, name: str | None = None): """Initialize the framework. :param test: Whether the framework is being initialized for testing purposes + :param name: Namespace of the framework, overriding the one inferred from the test mode """ self.start_time = monotonic() """Time when the framework was started.""" @@ -142,6 +142,9 @@ def __init__(self, test: bool = False): self.in_test_mode = test """Whether the framework is in testing mode.""" + self._name = name + """Namespace explicitly assigned to this instance, if any.""" + self._load_config() self.__class__.store = DataStore(self.name) self.telemetry = Telemetry(self) @@ -156,8 +159,15 @@ def git(self) -> GitConnector: return GitConnector(f"{self.path.parent.name}/{self.path.name}", self.path) @property - def name(self) -> Literal["odev", "odev-test"]: - """Name of the framework.""" + def name(self) -> str: + """Name of the framework, and the namespace of everything it owns. + + The configuration file and the datastore database are both named after it, so an instance given an + explicit name works on its own resources rather than on the ones of the user. + """ + if self._name is not None: + return self._name + return "odev" if not self.in_test_mode else "odev-test" @property diff --git a/odev/setup/symlink.py b/odev/setup/symlink.py index 7d2470e58..2ccdbf0ef 100644 --- a/odev/setup/symlink.py +++ b/odev/setup/symlink.py @@ -15,6 +15,27 @@ PRIORITY = 10 +LOCAL_BIN_PATH = Path("~/.local/bin").expanduser() +"""Directory holding the executables of the current user.""" + +SYSTEM_BIN_PATH = Path("/usr/local/bin") +"""Directory holding the executables of every user, used when the local one is not in `PATH`.""" + + +# --- Values ------------------------------------------------------------------- + + +def link_path() -> Path: + """Return the path the `odev` command should be linked from. + + :return: The link to create, in the local binaries directory of the user when it is in `PATH`, in the + system-wide one otherwise. + :rtype: Path + """ + known_env_paths = getenv("PATH", "").split(":") + + return (LOCAL_BIN_PATH if str(LOCAL_BIN_PATH) in known_env_paths else SYSTEM_BIN_PATH) / "odev" + # --- Setup -------------------------------------------------------------------- @@ -25,26 +46,24 @@ def setup(odev: Odev) -> None: :param config: Odev configuration """ main_path = Path(__file__).parents[2] / "odev.sh" - known_env_paths = getenv("PATH", "").split(":") - local_path = Path("~/.local/bin").expanduser() - link_path = (local_path if str(local_path) in known_env_paths else Path("/usr/local/bin")) / "odev" + command_path = link_path() - if not link_path.parent.exists(): - logger.debug(f"Directory {link_path.parent} does not exist, creating it") - link_path.parent.mkdir(parents=True) + if not command_path.parent.exists(): + logger.debug(f"Directory {command_path.parent} does not exist, creating it") + command_path.parent.mkdir(parents=True) - if link_path.exists() or link_path.is_symlink(): - logger.warning(f"Symlink path {link_path} already exists, this is used to run odev as a shell command") + if command_path.exists() or command_path.is_symlink(): + logger.warning(f"Symlink path {command_path} already exists, this is used to run odev as a shell command") if console.confirm("Would you like to overwrite it?"): - logger.debug(f"Removing symlink path {link_path}") - bash.execute(f"rm {link_path}", sudo=True) + logger.debug(f"Removing symlink path {command_path}") + bash.execute(f"rm {command_path}", sudo=True) - if not link_path.exists() and not link_path.is_symlink(): - logger.debug(f"Creating symlink from {link_path} to {main_path}") - bash.execute(f"ln -s {main_path} {link_path}", sudo=True) + if not command_path.exists() and not command_path.is_symlink(): + logger.debug(f"Creating symlink from {command_path} to {main_path}") + bash.execute(f"ln -s {main_path} {command_path}", sudo=True) logger.info("Symlink created") - for path in (main_path, link_path): + for path in (main_path, command_path): logger.debug(f"Checking execute permissions for {path}") path.chmod(path.stat().st_mode | stat.S_IXUSR) From 716c8e13d53b23559ab5e6fa13849d3ddcc4cf2f Mon Sep 17 00:00:00 2001 From: brinkflew Date: Mon, 27 Jul 2026 01:00:46 +0200 Subject: [PATCH 5/7] [IMP] tests: run each suite in a sandbox of its own and clean it up MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Everything the suite touched was named after a constant `odev-test`: the datastore database, the configuration file, the run directories and the databases created by the command tests. Two suites running at once shared all of it and destroyed each other's work — `odev delete --expression` removed the databases of the other run, dropping the datastore terminated its connections, and both wrote the same configuration file at once. A run now claims a sandbox named after itself and holds an exclusive lock on it for its whole life. The sandbox is removed when the session ends, including on `Ctrl+C` and on `SIGTERM`; the kernel drops the lock however the process dies, so a lock that can be taken marks leftovers the next run sweeps away. Nothing is written outside of it anymore: the configuration directory, the repositories and dumps directories, and the two symlinks the setup scripts create all point inside the sandbox, which also keeps the suite from repointing the `odev` command of the developer at whichever checkout it happens to run from. odev captures `SIGINT` around every query to cancel it rather than let it through, so an interrupt is recorded and acted upon at the next test boundary: pressing `Ctrl+C` stops the run without abandoning a connection mid-statement. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- tests/conftest.py | 95 ++++++++++ tests/fixtures/case.py | 49 ++++- tests/fixtures/sandbox.py | 251 +++++++++++++++++++++++++ tests/tests/commands/test_database.py | 4 +- tests/tests/commands/test_utilities.py | 2 +- tests/tests/common/test_interrupts.py | 56 ++++++ tests/tests/common/test_odev.py | 5 +- tests/tests/setup/test_setup.py | 30 ++- 8 files changed, 473 insertions(+), 19 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/fixtures/sandbox.py create mode 100644 tests/tests/common/test_interrupts.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 000000000..eb78b7c83 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,95 @@ +"""Session-wide setup for the test suite: give this run a private sandbox and make sure it is cleaned up. + +`pytest_sessionfinish` runs from a `finally` block in pytest's session wrapper, so the sandbox is removed +both when the suite completes and when it is interrupted. A `SIGTERM` and a `Ctrl+C` are both turned into +the same orderly exit, and anything that still escapes — a `SIGKILL`, a crashed interpreter — is picked up +by the sweep at the start of the next run. +""" + +import atexit +from collections.abc import Callable +from signal import SIGINT, SIGTERM, Signals, signal +from typing import Any +from unittest.mock import patch + +import pytest + +from tests.fixtures import sandbox + + +TERMINATED_EXIT_CODE = 2 +"""Exit code reported when the suite is stopped before it could complete.""" + +INTERRUPT_MESSAGE = "interrupted" +"""Reason reported when the suite is stopped before it could complete.""" + +SIGNAL_INSTALLER = "odev.common.signal_handling.signal" +"""Where odev installs the signal handlers it uses to cancel the operation it is running.""" + + +class InterruptRecorder: + """Signal handler noting an interrupt for the session, then handing it over to odev. + + odev captures `SIGINT` around every query and every subprocess, and its handlers cancel that single + operation instead of propagating. A suite spends much of its time inside one of those blocks, so a + `Ctrl+C` landing in one would be swallowed and the run would carry on. Letting the interrupt through + instead abandons the PostgreSQL connection mid-statement, and a suite interrupted that way exhausts + the connection slots of the server; so it is recorded here and acted upon at the next test boundary, + once odev has closed what it had open. + """ + + interrupted: bool = False + """Whether an interrupt was received while odev was holding the signal handlers.""" + + def __init__(self, handler: Callable[..., Any]): + self.handler: Callable[..., Any] = handler + """The handler odev installed, called once the interrupt has been recorded.""" + + def __call__(self, *args) -> Any: + InterruptRecorder.interrupted = True + + return self.handler(*args) + + +def pytest_configure(config: pytest.Config) -> None: + """Make the suite stoppable, whichever way it is asked to stop.""" + + def terminate(*args): + pytest.exit(INTERRUPT_MESSAGE, returncode=TERMINATED_EXIT_CODE) + + signal(SIGTERM, terminate) + + installer = patch(SIGNAL_INSTALLER, new=install_handler) + installer.start() + config.add_cleanup(installer.stop) + + +def pytest_sessionstart(session: pytest.Session) -> None: + """Claim a sandbox for this run, then clean up after the runs that no longer own theirs.""" + sandbox.acquire() + atexit.register(sandbox.release) + sandbox.sweep() + + +def pytest_runtest_setup(item: pytest.Item) -> None: + """Stop the session on an interrupt odev handled itself, now that it is between two tests.""" + if InterruptRecorder.interrupted: + pytest.exit(INTERRUPT_MESSAGE, returncode=TERMINATED_EXIT_CODE) + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + """Remove the sandbox of this run, whether the suite completed or was interrupted.""" + sandbox.release() + + +def install_handler(signal_number: Signals, handler: Any) -> Any: + """Install a signal handler for odev, recording the interrupts it would otherwise swallow. + + :param signal_number: The signal odev wants to handle. + :param handler: The handler it wants to install, or the one it restores once its block is over. + :return: The handler that was previously installed. + """ + if signal_number != SIGINT or not callable(handler) or isinstance(handler, InterruptRecorder): + return signal(signal_number, handler) + + return signal(signal_number, InterruptRecorder(handler)) diff --git a/tests/fixtures/case.py b/tests/fixtures/case.py index 99c301018..229e750f4 100644 --- a/tests/fixtures/case.py +++ b/tests/fixtures/case.py @@ -16,7 +16,7 @@ from odev.common.config import Config from odev.common.string import suid -from tests.fixtures import CaptureOutput +from tests.fixtures import CaptureOutput, sandbox class OdevTestCase(TestCase): @@ -32,7 +32,7 @@ class OdevTestCase(TestCase): """Name of the test case run, used for environment preparation.""" run_path: ClassVar[Path] - """Path to the test case run directory under `/tmp`.""" + """Path to the test case run directory, inside the sandbox of the current suite run.""" _patches: ClassVar[list[_patch]] """The patches applied to the test case. @@ -55,11 +55,16 @@ def tearDown(self): @classmethod def setUpClass(cls): cls._patches = [] - Config.parser = ConfigParser() - cls.odev = odev.Odev(test=True) cls.run_id = suid() - cls.run_name = f"{cls.odev.name}-{cls.run_id}" - cls.run_path = Path(f"/tmp/{cls.run_name}") # noqa: S108 + cls.run_path = sandbox.SESSION_PATH / cls.run_id + cls.run_name = f"{sandbox.SESSION_NAME}-{cls.run_id}" + + # The framework reads its name and its configuration directory while being constructed, so both + # have to point inside the sandbox before the instance exists. + cls.__patch_paths() + + Config.parser = ConfigParser() + cls.odev = odev.Odev(test=True, name=sandbox.SESSION_NAME) cls.res_path = cls.odev.tests_path / "resources" cls.replacer = Replacer() cls.__patch_cli() @@ -67,6 +72,7 @@ def setUpClass(cls): cls.__patch_framework() cls.addClassCleanup(cls.tearDownClass) cls.odev.start() + cls.__sandbox_config_paths() @classmethod def tearDownClass(cls): @@ -74,10 +80,11 @@ def tearDownClass(cls): cls.replacer.restore() cls.odev.commands.clear() cls.odev.store.drop() - cls.odev.config.path.unlink(missing_ok=True) - if cls.run_path.exists(): - shutil.rmtree(cls.run_path, ignore_errors=True) + # The configuration file lives in the run directory, and goes away with it. Whatever this misses, + # because the run was interrupted or because a test left a database behind, is picked up by the + # sandbox: either when the suite ends or at the start of the next one. + shutil.rmtree(cls.run_path, ignore_errors=True) odev.HOME_PATH = (Path.home() / ".local" / "share" / "odev").resolve() @@ -185,6 +192,29 @@ def _patch_object( cls._patches.append(patched) patched.start() + @classmethod + def __patch_paths(cls): + """Redirect the configuration directory into the run directory. + + The config file, and the plugin `config.py` modules `Config` discovers next to it, then come from + the sandbox instead of `~/.config/odev`: the suite writes nothing outside of it, and behaves the + same whether or not the developer running it has plugins installed. + """ + patched = patch("odev.common.config.CONFIG_DIR", cls.run_path) + cls._patches.append(patched) + patched.start() + + @classmethod + def __sandbox_config_paths(cls): + """Point the directories odev reads from its configuration at the sandbox. + + They default to `~/odoo`, where a test cloning a repository or downloading a dump would land in + the middle of the checkouts of the user — and in the way of a suite running alongside this one. + """ + cls.odev.config.paths.repositories = cls.run_path / "repositories" + cls.odev.config.paths.dumps = cls.run_path / "dumps" + cls.odev.config.paths.upgrade = cls.run_path / "repositories" / "odoo" / "upgrade" + @classmethod def __patch_cli(cls): """Patch interactions with the CLI to avoid waiting for user input or showing live status during tests.""" @@ -215,7 +245,6 @@ def __patch_odev(cls): ("_update", False), ], [ - ("name", "odev-test"), ("upgrades_path", cls.odev.tests_path / "resources" / "upgrades"), ("setup_path", cls.odev.tests_path / "resources" / "setup"), ("scripts_path", cls.odev.tests_path / "resources" / "scripts"), diff --git a/tests/fixtures/sandbox.py b/tests/fixtures/sandbox.py new file mode 100644 index 000000000..a6dc526fe --- /dev/null +++ b/tests/fixtures/sandbox.py @@ -0,0 +1,251 @@ +"""Private namespace owned by a single run of the test suite. + +Every resource the suite touches — the datastore database, the databases created by the command tests, +the configuration file, the temporary directories — is named after `SESSION_NAME` or nested under +`SESSION_PATH`. Two suites running at the same time therefore never share anything, and whatever a run +leaves behind can be identified and removed by the next one. + +A run holds an exclusive `flock` on its sandbox for its whole lifetime. The kernel releases that lock +when the process dies, whichever way it dies, so a lock that can be taken is proof that its owner is +gone and its leftovers are safe to remove. This is what makes cleanup survive `SIGKILL`, where no +handler of ours can run. +""" + +import shutil +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager +from fcntl import LOCK_EX, LOCK_NB, LOCK_UN, flock +from pathlib import Path +from typing import IO + +import psycopg2 +from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT, cursor as Cursor + +from odev.common.config import CONFIG_DIR +from odev.common.logging import logging +from odev.common.string import suid + + +logger = logging.getLogger(__name__) + + +SANDBOX_PREFIX = "odev-test" +"""Prefix shared by every sandbox, and the only namespace this module is ever allowed to delete.""" + +SANDBOX_ROOT = Path(tempfile.gettempdir()) +"""Directory holding the sandbox of each run.""" + +SESSION_NAME = f"{SANDBOX_PREFIX}-{suid()}" +"""Name of the sandbox owned by this run, used as the name of the odev framework it drives.""" + +SESSION_PATH = SANDBOX_ROOT / SESSION_NAME +"""Directory holding every file this run creates.""" + +REAL_CONFIG_DIR = CONFIG_DIR +"""The user's actual configuration directory, captured before the test case redirects `CONFIG_DIR`. + +Runs predating this module wrote their `odev-test*` files there; the sweep still cleans them up. +""" + +LOCK_NAME = ".lock" +"""Name of the lock file marking a sandbox as owned by a live process.""" + +MAINTENANCE_DATABASE = "postgres" +"""Database to connect to when listing and dropping the databases of a sandbox.""" + +SESSION_PARTS = 3 +"""Number of dash-separated components in a sandbox name, as in `odev-test-4kq2z81a`.""" + + +_lock: IO[str] | None = None +"""Open handle on this run's lock file, kept for as long as the run lives.""" + + +def acquire() -> None: + """Create the sandbox of this run and hold its lock until the process exits.""" + global _lock # noqa: PLW0603 - the lock lives as long as the module does + + if _lock is not None: + return + + # The sandbox is locked before it is published under its final name, and the rename is atomic: a + # sweep running in another process must never come across a sandbox that does not hold its lock yet, + # or it would take it for the leftovers of a dead run and delete it. + staging = SANDBOX_ROOT / f".{SESSION_NAME}" + staging.mkdir(parents=True, exist_ok=True) + + _lock = (staging / LOCK_NAME).open("w") + flock(_lock, LOCK_EX | LOCK_NB) + staging.rename(SESSION_PATH) + + +def release() -> None: + """Remove everything this run created, then release its lock. + + Safe to call more than once: pytest calls it at the end of the session and `atexit` calls it again if + the interpreter goes down another way. + """ + global _lock + + if _lock is None: + return + + handle, _lock = _lock, None + + try: + discard(SESSION_NAME) + finally: + flock(handle, LOCK_UN) + handle.close() + + +def sweep() -> None: + """Remove the sandboxes of runs that no longer hold their lock. + + Covers whatever escaped `release()`: a suite killed with `SIGKILL`, a crashed interpreter, a machine + that went down mid-run. + """ + # The sandboxes are listed before the live ones are: a run publishing its own between the two + # snapshots is then simply absent from the list, rather than present in it and seemingly unowned. + known = _known_sessions() + live = _live_sessions() + + for session in sorted(known - live): + try: + discard(session) + except Exception as error: # noqa: BLE001 - a sandbox we cannot clean must not fail the suite + logger.warning(f"Could not remove the leftovers of test session {session!r}: {error}") + + +def discard(session: str) -> None: + """Remove every trace of a sandbox: its databases, its configuration files and its directory. + + :param session: The name of the sandbox to remove. + :raises ValueError: If the name falls outside the sandbox namespace. + """ + if session != SANDBOX_PREFIX and not session.startswith(f"{SANDBOX_PREFIX}-"): + raise ValueError(f"Refusing to remove {session!r}, which is not a test sandbox") + + # `odev-test` on its own is the sandbox of runs predating this module. Everything below it belongs to + # other sandboxes, possibly live ones, so only its own name is removed. + owns_children = session != SANDBOX_PREFIX + + for database in _databases(session, children=owns_children): + _drop_database(database) + + for path in REAL_CONFIG_DIR.glob(f"{session}*" if owns_children else f"{session}.*"): + path.unlink(missing_ok=True) + + shutil.rmtree(SANDBOX_ROOT / session, ignore_errors=True) + + +def _live_sessions() -> set[str]: + """Return the sandboxes still owned by a running process.""" + return {path.name for path in _sandbox_directories() if _is_locked(path / LOCK_NAME)} + + +def _known_sessions() -> set[str]: + """Return every sandbox that left a trace on this machine, live or not.""" + sessions = {path.name for path in _sandbox_directories()} + sessions |= {_session_of(path.name) for path in REAL_CONFIG_DIR.glob(f"{SANDBOX_PREFIX}*")} + sessions |= {_session_of(database) for database in _databases(SANDBOX_PREFIX, children=True)} + + return sessions + + +def _sandbox_directories() -> Iterator[Path]: + """Yield the sandbox directories present on this machine.""" + return (path for path in SANDBOX_ROOT.glob(f"{SANDBOX_PREFIX}-*") if path.is_dir()) + + +def _is_locked(path: Path) -> bool: + """Check whether a lock file is held by a live process. + + `flock` conflicts between separate open file descriptions, including within a single process, so this + also reports the sandbox of the current run as live. A missing lock file means the sandbox predates + this module or its owner died before taking the lock: either way nobody owns it. + """ + if not path.exists(): + return False + + try: + with path.open("r") as handle: + flock(handle, LOCK_EX | LOCK_NB) + flock(handle, LOCK_UN) + + except OSError: + return True + + return False + + +def _session_of(artifact: str) -> str: + """Return the sandbox an artifact belongs to. + + Artifacts are named after their sandbox with a suffix of their own, so the sandbox is the first three + components of the name: the database `odev-test-4kq2z81a-9zf1z0aa` and the file + `odev-test-4kq2z81a.cfg` both belong to `odev-test-4kq2z81a`. `suid` only ever emits lowercase letters + and digits, so no component contains a dash of its own. + + :param artifact: The name of a database, or the name of a file including its extension. + :return: The name of the sandbox owning it. + :rtype: str + """ + return "-".join(artifact.split(".")[0].split("-")[:SESSION_PARTS]) + + +def _databases(session: str, children: bool) -> list[str]: + """List the existing databases belonging to a sandbox. + + :param session: The name of the sandbox. + :param children: Whether to also return the databases named after a sandbox nested below it. + :return: The names of the matching databases. + :rtype: list[str] + """ + with _maintenance() as cursor: + if cursor is None: + return [] + + cursor.execute( + "SELECT datname FROM pg_database WHERE datname = %s OR datname LIKE %s", + (session, f"{session}-%" if children else session), + ) + + return [name for (name,) in cursor.fetchall()] + + +def _drop_database(database: str) -> None: + """Drop a database left behind by a dead run, disconnecting whatever still holds it open.""" + with _maintenance() as cursor: + if cursor is None: + return + + cursor.execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = %s AND pid <> pg_backend_pid()", + (database,), + ) + cursor.execute(f'DROP DATABASE IF EXISTS "{database}"') + + +@contextmanager +def _maintenance() -> Iterator[Cursor | None]: + """Yield a cursor on the maintenance database, or `None` if PostgreSQL cannot be reached. + + Cleaning up is best-effort: a developer without a running PostgreSQL gets a warning rather than a + suite that refuses to start. + """ + try: + connection = psycopg2.connect(database=MAINTENANCE_DATABASE) + except psycopg2.Error as error: + logger.warning(f"Could not connect to PostgreSQL to clean up test sandboxes: {error}") + yield None + + return + + connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) + + try: + yield connection.cursor() + finally: + connection.close() diff --git a/tests/tests/commands/test_database.py b/tests/tests/commands/test_database.py index ade716e42..fe2ca3b57 100644 --- a/tests/tests/commands/test_database.py +++ b/tests/tests/commands/test_database.py @@ -568,11 +568,13 @@ def test_99_delete_expression(self): """Command `odev delete` should delete databases matching a regular expression.""" self.assertDatabaseExist(self.database_name) + # The expression runs against the real PostgreSQL instance, so it has to be scoped to the sandbox + # of this run: a broader one would delete the databases of a suite running alongside this one. with self.patch(self.odev.console, "confirm", return_value=True): stdout, _ = self.dispatch_command( "delete", "--expression", - "^odev-test-[a-z0-9]{8}", + f"^{self.odev.name}-[a-z0-9]{{8}}", "--include-whitelisted", ) diff --git a/tests/tests/commands/test_utilities.py b/tests/tests/commands/test_utilities.py index 7974c2a77..601fa1711 100644 --- a/tests/tests/commands/test_utilities.py +++ b/tests/tests/commands/test_utilities.py @@ -15,7 +15,7 @@ class TestCommandUtilities(OdevCommandTestCase): def test_version_01_no_argument(self): """Command `odev version` should print the version of the application.""" stdout, _ = self.dispatch_command("version") - self.assertIn(f"Odev-test version {__version__}", stdout) + self.assertIn(f"{self.odev.name.capitalize()} version {__version__}", stdout) def test_config_01_no_argument(self): """Run the command without arguments.""" diff --git a/tests/tests/common/test_interrupts.py b/tests/tests/common/test_interrupts.py new file mode 100644 index 000000000..6baf1cbbc --- /dev/null +++ b/tests/tests/common/test_interrupts.py @@ -0,0 +1,56 @@ +import os +from signal import SIGINT, getsignal + +from odev.common.signal_handling import capture_signals + +from tests.conftest import InterruptRecorder +from tests.fixtures import OdevTestCase + + +class TestSuiteInterrupts(OdevTestCase): + """The suite has to stay stoppable while odev is holding the signal handlers. + + odev captures `SIGINT` around every query and every subprocess to cancel that operation rather than + let the interrupt through. Without the recorder `conftest` installs, a `Ctrl+C` landing inside one of + those blocks would be swallowed and the run would carry on. + """ + + def setUp(self): + super().setUp() + self.addCleanup(self.clear_interrupt) + + def clear_interrupt(self): + """Forget the interrupt recorded by a test. + + Left set, the flag would stop the session before the next test rather than at the end of this one, + taking the rest of the suite with it. + """ + InterruptRecorder.interrupted = False + + def test_01_odev_handlers_are_wrapped(self): + """The handlers odev installs should be the ones recording interrupts.""" + with capture_signals(handler=lambda *args: None): + self.assertIsInstance(getsignal(SIGINT), InterruptRecorder) + + def test_02_interrupt_is_recorded_and_handled(self): + """An interrupt captured by odev should reach its handler and be noted for the session.""" + handled: list[int] = [] + + def handler(signal_number, *args): + handled.append(signal_number) + + with capture_signals(handler=handler): + os.kill(os.getpid(), SIGINT) + + self.assertEqual(handled, [SIGINT], "odev should still get a chance to cancel what it was doing") + self.assertTrue(InterruptRecorder.interrupted, "the session should be stopped at the next test boundary") + + def test_03_handlers_are_wrapped_once(self): + """Restoring a wrapped handler should not wrap it again, however many blocks are nested.""" + with capture_signals(handler=lambda *args: None): + outer = getsignal(SIGINT) + + with capture_signals(handler=lambda *args: None): + pass + + self.assertIs(getsignal(SIGINT), outer, "the outer handler should be restored as it was") diff --git a/tests/tests/common/test_odev.py b/tests/tests/common/test_odev.py index 73d942b67..040f9a945 100644 --- a/tests/tests/common/test_odev.py +++ b/tests/tests/common/test_odev.py @@ -23,8 +23,9 @@ def setUp(self): def test_01_config_file(self): """Config file should have been created in the correct directory.""" - self.assertEqual(self.odev.config.name, "odev-test") - self.assertEqual(self.odev.config.path, Path.home() / ".config/odev/odev-test.cfg") + self.assertEqual(self.odev.config.name, self.odev.name) + self.assertEqual(self.odev.config.path, self.run_path / f"{self.odev.name}.cfg") + self.assertTrue(self.odev.config.path.exists()) def test_02_config_get_set_reset_delete(self): """Config manager should be able to get, set and reset values, as well as delete a key or a section. diff --git a/tests/tests/setup/test_setup.py b/tests/tests/setup/test_setup.py index 148f7051e..0bf37ecc3 100644 --- a/tests/tests/setup/test_setup.py +++ b/tests/tests/setup/test_setup.py @@ -1,5 +1,5 @@ import shutil -from pathlib import Path +from unittest.mock import patch from odev.setup import completion, directories, symlink, update @@ -7,24 +7,44 @@ class TestSetup(OdevTestCase): + """Test the setup scripts run when installing odev. + + Both scripts link odev into the shell of the user, at `~/.local/bin/odev` and in the bash completion + directory. Left to their real destinations they would repoint the `odev` command of the developer at + whichever checkout the suite happens to run from, and two suites running at once would fight over the + same two links, so they are redirected into the sandbox of the run. + """ + def test_completion_01_completion(self): """Test the setup script responsible of creating a symlink to the bash completion script of odev. A symlink should be created on the file system. """ - with self.patch(completion.console, "confirm", return_value=True): + completion_path = self.run_path / "completions" / "complete_odev.sh" + + with ( + patch.object(completion, "comp_path", completion_path), + self.patch(completion.console, "confirm", return_value=True), + ): completion.setup(self.odev) - self.assertTrue(Path("~/.local/share/bash-completion/completions/complete_odev.sh").expanduser().is_symlink()) + self.assertTrue(completion_path.is_symlink()) + self.assertEqual(completion_path.resolve(), self.odev.path / "complete_odev.sh") def test_symlink_01_symlink(self): """Test the setup script responsible of creating a symlink to odev. A symlink should be created to map the "odev" command to the main file of this application. """ - with self.patch(symlink.console, "confirm", return_value=True): + command_path = self.run_path / "bin" / "odev" + + with ( + self.patch(symlink, "link_path", return_value=command_path), + self.patch(symlink.console, "confirm", return_value=True), + ): symlink.setup(self.odev) - self.assertTrue(Path("~/.local/bin/odev").expanduser().is_symlink()) + self.assertTrue(command_path.is_symlink()) + self.assertEqual(command_path.resolve(), self.odev.path / "odev.sh") def test_update_01_update(self): """Test the setup script responsible of setting the auto-update values for odev. From 47499cb79117525eb7ed61ec8a0f269f08b80f44 Mon Sep 17 00:00:00 2001 From: brinkflew Date: Mon, 27 Jul 2026 01:27:32 +0200 Subject: [PATCH 6/7] [FIX] common: close the PostgreSQL connection a database block opened MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both database context managers built a second, unconnected connector to close instead of the one they had connected, so `disconnect` did nothing and the connection stayed open until the garbage collector got to it. `ensure_connected` runs every database method inside its own block and those blocks nest — `is_odoo` opens one and then calls `table_exists`, which opens another — so this meant a fresh backend per call. Closing the right connector is not enough on its own: an inner block would close the connection the enclosing one is still using. The blocks are now reentrant and share a single connector, counted in the mixin so both classes get the same behaviour. The datastore keeps its connection instead of reopening it for each read: every command reads it and it lives as long as the process, which is not true of the databases odev walks through for `list` or `delete`. Over a full test suite run, the backends held at once drop from 42 to 3, and the suite goes from 55s to 34s. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- odev/common/databases/local.py | 11 ++++---- odev/common/mixins/connectors/postgres.py | 32 +++++++++++++++++++++++ odev/common/postgres.py | 12 +++++++-- odev/common/store/datastore.py | 6 +++++ 4 files changed, 53 insertions(+), 8 deletions(-) diff --git a/odev/common/databases/local.py b/odev/common/databases/local.py index 84d9da044..d0f209519 100644 --- a/odev/common/databases/local.py +++ b/odev/common/databases/local.py @@ -61,9 +61,6 @@ class LocalDatabase(PostgresConnectorMixin, Database): """Class for manipulating PostgreSQL (local) databases.""" - connector: PostgresConnector | None = None - """The PostgreSQL connector of the database.""" - _whitelisted: bool = False """Whether the database is whitelisted and should not be removed automatically.""" @@ -102,11 +99,11 @@ def __init__(self, name: str): self.whitelisted = info is not None and info.whitelisted def __enter__(self): - self.connector = self.psql(self.name).__enter__() # type: ignore [assignment] + self._enter_connector(self.name) return self def __exit__(self, *args): - self.psql(self.name).__exit__(*args) + self._exit_connector(*args) @property def rpc_port(self): @@ -609,7 +606,9 @@ def signal_handler_progress( tracker.stop() - if self.connector is not None: + # The attribute holds the connector class until a block connects for the first time, so the check + # is on the type rather than on `None`, as in `drop`. + if isinstance(self.connector, PostgresConnector): self.connector.invalidate_cache() def _replace_filestore(self, source_dir: Path) -> None: diff --git a/odev/common/mixins/connectors/postgres.py b/odev/common/mixins/connectors/postgres.py index 2cf96c3c4..a6718e38e 100644 --- a/odev/common/mixins/connectors/postgres.py +++ b/odev/common/mixins/connectors/postgres.py @@ -10,6 +10,38 @@ class PostgresConnectorMixin(ConnectorMixin): _connector_class: type[PostgresConnector] = PostgresConnector # type: ignore [assignment] connector: PostgresConnector # type: ignore [assignment] + _connection_depth: int = 0 + """Number of nested blocks currently sharing the connector.""" + def psql(self, name: str = "postgres") -> PostgresConnector: """Return a PostgreSQL connector to the selected database.""" return self._connector_class(name) + + def _enter_connector(self, name: str) -> PostgresConnector: + """Connect to a database, or join the connection an enclosing block already opened. + + `ensure_connected` runs every database method inside its own block, and those blocks nest, so + opening a connection per block would mean a new PostgreSQL backend for each call. + + :param name: The name of the database to connect to. + :return: The connector the caller should use. + :rtype: PostgresConnector + """ + # A held connection that was closed in the meantime is replaced rather than reused: `drop` + # disconnects explicitly, and the datastore holds its own connection across such a call. At depth + # zero the check is skipped, `connector` still being the connector class rather than an instance. + if not self._connection_depth or not self.connector.connected: + # Narrows what the base mixin types as a `Connector`, deliberately: everything reached through + # this mixin talks to PostgreSQL, and its callers query rather than use the base interface. + self.connector = self.psql(name).__enter__() # pyright: ignore[reportIncompatibleVariableOverride] + + self._connection_depth += 1 + + return self.connector + + def _exit_connector(self, *args) -> None: + """Close the connection once the outermost block sharing it is done with it.""" + self._connection_depth -= 1 + + if not self._connection_depth: + self.connector.__exit__(*args) diff --git a/odev/common/postgres.py b/odev/common/postgres.py index 3c34526db..6fe74ed27 100644 --- a/odev/common/postgres.py +++ b/odev/common/postgres.py @@ -35,11 +35,19 @@ def __init__(self, name: str): self.prepare_database() def __enter__(self): - self.connector = self._connector_class(self.name).__enter__() + self._enter_connector(self.name) return self def __exit__(self, *args): - self._connector_class(self.name).__exit__(*args) + self._exit_connector(*args) + + def hold_connection(self) -> None: + """Keep the connection open for the lifetime of this object rather than for that of a block. + + Meant for a database read often enough that reconnecting for each operation is wasteful, and + long-lived enough that holding a backend for it is not. + """ + self._enter_connector(self.name) def __repr__(self): """Return the representation of the database.""" diff --git a/odev/common/store/datastore.py b/odev/common/store/datastore.py index ab769ae8f..674b644c0 100644 --- a/odev/common/store/datastore.py +++ b/odev/common/store/datastore.py @@ -18,6 +18,12 @@ class DataStore(PostgresDatabase): def __init__(self, name: str = "odev"): super().__init__(name) + + # Every command reads the store, and it lives as long as the process does: its connection is + # opened once here and held, rather than reopened for each of the reads a single run makes. + # Holding it from this point also covers the tables prepared below. + self.hold_connection() + self.databases = DatabaseStore(self) self.history = HistoryStore(self) self.secrets = SecretStore(self) From cd070079a4fabd1a866a38ed48bb44fdbdac385c Mon Sep 17 00:00:00 2001 From: brinkflew Date: Mon, 27 Jul 2026 01:27:49 +0200 Subject: [PATCH 7/7] [IMP] tests: cover the lifetime of database connections Pin the behaviour a connection block is expected to have, since nothing failed loudly when it did not have it: leaving a block closes what it opened, a nested block joins the connection of the enclosing one rather than opening its own, repeated calls to decorated methods do not leave backends behind, and the datastore holds a single one throughout. The count is read from `pg_stat_activity`, which is what makes the third one a regression guard rather than a restatement of the code. Claude-Session: https://claude.ai/code/session_012tWyDuUYenE92nsBtG19Du --- tests/tests/common/test_connectors.py | 67 +++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/tests/tests/common/test_connectors.py b/tests/tests/common/test_connectors.py index c9a3d5d41..db5f9b8c8 100644 --- a/tests/tests/common/test_connectors.py +++ b/tests/tests/common/test_connectors.py @@ -5,6 +5,7 @@ from odev.common.connectors.postgres import Cursor, PostgresConnector from odev.common.connectors.rest import RestConnector +from odev.common.postgres import PostgresDatabase from tests.fixtures import OdevTestCase @@ -106,3 +107,69 @@ def execute(self, statement): raise RuntimeError("boom") self.assertEqual(cursor.calls, ["BEGIN", "ROLLBACK"]) + + +class TestPostgresConnectionLifecycle(OdevTestCase): + """A block has to close the connection it opened, and nested blocks have to share one. + + `ensure_connected` wraps every method of a database in `with self:`, so anything less means a fresh + PostgreSQL backend per call: enough of them at once and the server runs out of connection slots. + """ + + def setUp(self): + super().setUp() + self.database = PostgresDatabase(self.odev.name) + """A handle on the datastore database, connected and disconnected by the tests.""" + + def backends(self, name: str) -> int: + """Count the backends PostgreSQL currently holds for a database.""" + with PostgresConnector() as psql, psql.nocache(): + result = psql.query(f"SELECT count(*) FROM pg_stat_activity WHERE datname = '{name}'") + + return result[0][0] if isinstance(result, list) else 0 + + def test_01_block_closes_what_it_opened(self): + """Leaving a block should disconnect the connector the block connected.""" + with self.database: + self.assertTrue(self.database.connector.connected) + + self.assertFalse(self.database.connector.connected) + + def test_02_nested_blocks_share_one_connection(self): + """An inner block should join the connection of the outer one instead of opening its own.""" + with self.database: + connector = self.database.connector + + with self.database: + self.assertIs(self.database.connector, connector, "the inner block should reuse the connector") + + self.assertTrue(connector.connected, "the inner block should not close what the outer one uses") + + self.assertFalse(connector.connected, "the outermost block should close it") + + def test_03_repeated_calls_do_not_pile_up_backends(self): + """Decorated methods each open a block, and those should not accumulate connections.""" + baseline = self.backends(self.database.name) + + for _ in range(20): + self.database.table_exists("history") + + self.assertLessEqual( + self.backends(self.database.name), + baseline, + "connections opened by the calls should have been closed again", + ) + + def test_04_store_keeps_a_single_connection(self): + """The datastore is read by every command and holds its connection rather than reopening it.""" + connector = self.odev.store.connector + self.assertTrue(connector.connected, "the store should be connected as soon as it exists") + + backends = self.backends(self.odev.store.name) + + for _ in range(20): + self.odev.store.table_exists("history") + + self.assertIs(self.odev.store.connector, connector, "the store should keep the same connector") + self.assertTrue(connector.connected, "a block should not close the connection the store holds") + self.assertEqual(self.backends(self.odev.store.name), backends, "the store should hold a single backend")