diff --git a/odev/_version.py b/odev/_version.py index 930d25f6c..bef1c3a36 100644 --- a/odev/_version.py +++ b/odev/_version.py @@ -22,4 +22,4 @@ # or merged change. # ------------------------------------------------------------------------------ -__version__ = "4.31.1" +__version__ = "4.31.2" 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/databases/local.py b/odev/common/databases/local.py index c846e7bf1..56f05c7a8 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 GitConnector, GitWorktree, PostgresConnector @@ -60,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.""" @@ -101,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): @@ -116,8 +114,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: @@ -637,7 +647,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/odev.py b/odev/common/odev.py index 0f2477b73..636d5636c 100644 --- a/odev/common/odev.py +++ b/odev/common/odev.py @@ -25,7 +25,6 @@ Any, ClassVar, Generic, - Literal, NamedTuple, TypedDict, cast, @@ -197,10 +196,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.""" @@ -208,6 +208,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.commands = CommandRegistry(self) """Collection of existing commands, imported on demand.""" @@ -225,8 +228,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/common/postgres.py b/odev/common/postgres.py index b81ad4d23..34495314e 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,14 +29,25 @@ 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): - 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) 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 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) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 000000000..b877c8849 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,119 @@ +"""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 +import logging +from collections.abc import Callable +from signal import SIGINT, SIGTERM, Signals, signal +from typing import Any +from unittest.mock import patch + +import pytest + +from odev.common.logging import OdevRichHandler + +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, and leave the logs to pytest.""" + + 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) + + detach_odev_log_handler() + + +def detach_odev_log_handler() -> None: + """Take odev's own logging handler off the root logger for the duration of the suite. + + `odev.common.logging` configures logging when it is imported, but `logging.basicConfig` is a no-op + once the root logger has handlers: whether odev's handler ends up installed depends on whether that + import happens before or after pytest sets its own up. Importing anything from odev in this module, + as the sandbox does, is enough to tip it one way. + + The handler renders through the console, so leaving it on makes every record reach the output twice — + once rendered by odev and once through whatever the test is capturing — and turns a plain `logger.info` + into a `console.print` that tests patching the console then have to account for. Records are left to + the handlers pytest and the test cases install, which is what the suite asserts on. + """ + root = logging.getLogger() + + for handler in [handler for handler in root.handlers if isinstance(handler, OdevRichHandler)]: + root.removeHandler(handler) + + +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 81de06ac0..3765bf70a 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,10 +32,14 @@ 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.""" + _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,11 +54,17 @@ def tearDown(self): @classmethod def setUpClass(cls): - Config.parser = ConfigParser() - cls.odev = odev.Odev(test=True) + cls._patches = [] 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() @@ -62,6 +72,7 @@ def setUpClass(cls): cls.__patch_framework() cls.addClassCleanup(cls.tearDownClass) cls.odev.start() + cls.__sandbox_config_paths() @classmethod def tearDownClass(cls): @@ -69,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() @@ -152,8 +164,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( @@ -178,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.""" @@ -208,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 3ba11185f..bad4f871f 100644 --- a/tests/tests/commands/test_utilities.py +++ b/tests/tests/commands/test_utilities.py @@ -23,7 +23,7 @@ def test_version_01_no_argument(self): with self.patch(self.odev, "update_available", return_value=False): stdout, stderr = self.dispatch_command("version") - self.assertIn(f"Odev-test version {__version__}", stdout) + self.assertIn(f"{self.odev.name.capitalize()} version {__version__}", stdout) self.assertNotIn("A newer version is available", stderr) def test_version_02_update_available(self): 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_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") 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_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 8a6f02d5a..fb5a551bb 100644 --- a/tests/tests/common/test_odev.py +++ b/tests/tests/common/test_odev.py @@ -20,10 +20,18 @@ 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") - 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/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) 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.