From: Daniele Varrazzo Date: Tue, 7 Jun 2022 20:10:36 +0000 (+0200) Subject: test: allow to mark tests to run only or skip on certain version ranges X-Git-Tag: 3.1~49^2~18 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=3e90e52752a3edbfe46040471dd0548222021d11;p=thirdparty%2Fpsycopg.git test: allow to mark tests to run only or skip on certain version ranges --- diff --git a/tests/fix_crdb.py b/tests/fix_crdb.py index 8a13c28c9..3209bd4e6 100644 --- a/tests/fix_crdb.py +++ b/tests/fix_crdb.py @@ -1,6 +1,8 @@ +from typing import Optional + import pytest -from .utils import check_version +from .utils import VersionCheck from psycopg.crdb import CrdbConnection @@ -8,8 +10,8 @@ def pytest_configure(config): # register libpq marker config.addinivalue_line( "markers", - "crdb(version_expr, reason=detail): run the test only with matching CockroachDB" - " (e.g. '>= 21.2.10', '< 22.1', 'skip')", + "crdb(version_expr, reason=detail): run/skip the test with matching CockroachDB" + " (e.g. '>= 21.2.10', '< 22.1', 'skip < 22')", ) @@ -19,27 +21,27 @@ def check_crdb_version(got, mark): This function is called on the tests marked with something like:: - @pytest.mark.crdb(">= 21.1") - @pytest.mark.crdb("only") - @pytest.mark.crdb("skip") + @pytest.mark.crdb("only") # run on CRDB only, any version + @pytest.mark.crdb # same as above + @pytest.mark.crdb("only >= 21.1") # run on CRDB only >= 21.1 (not on PG) + @pytest.mark.crdb(">= 21.1") # same as above + @pytest.mark.crdb("skip") # don't run on CRDB, any version + @pytest.mark.crdb("skip < 22") # don't run on CRDB < 22 (run on PG) and skips the test if the server version doesn't match what expected. """ assert len(mark.args) <= 1 assert not (set(mark.kwargs) - {"reason"}) - want = mark.args[0] if mark.args else "only" - msg = None - - if got is None: - if want == "only": - msg = "skipping test: CockroachDB only" - else: - if want == "only": - pass - elif want == "skip": - msg = crdb_skip_message(mark.kwargs.get("reason")) - else: - msg = check_version(got, want, "CockroachDB") + pred = VersionCheck.parse(mark.args[0] if mark.args else "only") + pred.whose = "CockroachDB" + + msg = pred.get_skip_message(got) + if not msg: + return None + + reason = crdb_skip_message(mark.kwargs.get("reason")) + if reason: + msg = f"{msg}: {reason}" return msg @@ -49,10 +51,10 @@ def check_crdb_version(got, mark): is_crdb = CrdbConnection.is_crdb -def crdb_skip_message(reason): - msg = "skipping test on CockroachDB" +def crdb_skip_message(reason: Optional[str]) -> str: + msg = "" if reason: - msg = f"{msg}: {reason}" + msg = reason if reason in _crdb_reasons: url = ( "https://github.com/cockroachdb/cockroach/" diff --git a/tests/fix_db.py b/tests/fix_db.py index c2c8527ee..49bd688d4 100644 --- a/tests/fix_db.py +++ b/tests/fix_db.py @@ -8,7 +8,7 @@ import psycopg from psycopg import pq from psycopg import sql -from .utils import check_libpq_version, check_server_version +from .utils import check_libpq_version, check_postgres_version # Set by warm_up_database() the first time the dsn fixture is used pg_version: int @@ -276,7 +276,7 @@ def check_connection_version(node): for mark in node.iter_markers(): if mark.name == "pg": assert len(mark.args) == 1 - msg = check_server_version(pg_version, mark.args[0]) + msg = check_postgres_version(pg_version, mark.args[0]) if msg: pytest.skip(msg) diff --git a/tests/utils.py b/tests/utils.py index f472d796c..496b241cc 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -1,6 +1,7 @@ import gc import re import operator +from typing import Callable, Optional, Tuple import pytest @@ -20,7 +21,7 @@ def check_libpq_version(got, want): return check_version(got, want, "libpq") -def check_server_version(got, want): +def check_postgres_version(got, want): """ Verify if the server version is a version accepted. @@ -30,51 +31,95 @@ def check_server_version(got, want): and skips the test if the server version doesn't match what expected. """ - return check_version(got, want, "server") + return check_version(got, want, "PostgreSQL") def check_version(got, want, whose_version): - """Check that a postgres-style version matches a desired spec. + pred = VersionCheck.parse(want) + pred.whose = whose_version + return pred.get_skip_message(got) - - The postgres-style version is a number such as 90603 for 9.6.3. - - The want version is a spec string such as "> 9.6" + +class VersionCheck: + """ + Helper to compare a version number with a test spec. """ - # convert 90603 to (9, 6, 3), 120003 to (12, 3) - got, got_fix = divmod(got, 100) - got_maj, got_min = divmod(got, 100) - if got_maj >= 10: - got = (got_maj, got_fix) - else: - got = (got_maj, got_min, got_fix) - - # Parse a spec like "> 9.6" - m = re.match( - r"^\s*(>=|<=|>|<|==)?\s*(?:(\d+)(?:\.(\d+)(?:\.(\d+))?)?)?\s*$", - want, - ) - if m is None: - pytest.fail(f"bad wanted version spec: {want}") - - # convert "9.6" into (9, 6, 0), "10.3" into (10, 3) - want_maj = int(m.group(2)) - want_min = int(m.group(3) or "0") - want_fix = int(m.group(4) or "0") - if want_maj >= 10: - if want_fix: - pytest.fail(f"bad version in {want}") - want = (want_maj, want_min) - else: - want = (want_maj, want_min, want_fix) - - opnames = {">=": "ge", "<=": "le", ">": "gt", "<": "lt", "==": "eq"} - op = getattr(operator, opnames[m.group(1) or "=="]) - - if not op(got, want): - revops = {">=": "<", "<=": ">", ">": "<=", "<": ">=", "==": "!="} - return ( - f"{whose_version} version is {'.'.join(map(str, got))}" - f" {revops[m.group(1)]} {'.'.join(map(str, want))}" + + def __init__( + self, + *, + skip: bool = False, + op: Optional[str] = None, + version_tuple: Tuple[int, ...] = (), + whose: str = "(wanted)", + ): + self.skip = skip + self.op = op or "==" + self.version_tuple = version_tuple + self.whose = whose + + @classmethod + def parse(cls, spec: str) -> "VersionCheck": + # Parse a spec like "> 9.6", "skip < 21.2.0" + m = re.match( + r"""(?ix) + ^\s* (skip|only)? + \s* (>=|<=|>|<)? + \s* (?:(\d+)(?:\.(\d+)(?:\.(\d+))?)?)? + \s* $ + """, + spec, ) + if m is None: + pytest.fail(f"bad wanted version spec: {spec}") + + skip = (m.group(1) or "only").lower() == "skip" + op = m.group(2) + version_tuple = tuple(int(n) for n in m.groups()[2:] if n) + return cls(skip=skip, op=op, version_tuple=version_tuple) + + def get_skip_message(self, version: Optional[int]) -> Optional[str]: + got_tuple = self._parse_int_version(version) + + msg: Optional[str] = None + if self.skip: + if got_tuple: + if not self.version_tuple: + msg = f"skip on {self.whose}" + elif self._match_version(got_tuple): + msg = ( + f"skip on {self.whose} {self.op}" + f" {'.'.join(map(str, self.version_tuple))}" + ) + else: + if not got_tuple: + msg = f"only for {self.whose}" + elif not self._match_version(got_tuple): + if self.version_tuple: + msg = ( + f"only for {self.whose} {self.op}" + f" {'.'.join(map(str, self.version_tuple))}" + ) + else: + msg = f"only for {self.whose}" + + return msg + + _OP_NAMES = {">=": "ge", "<=": "le", ">": "gt", "<": "lt", "==": "eq"} + + def _match_version(self, got_tuple: Tuple[int, ...]) -> bool: + if not self.version_tuple: + return True + op: Callable[[Tuple[int, ...], Tuple[int, ...]], bool] + op = getattr(operator, self._OP_NAMES[self.op]) + return op(got_tuple, self.version_tuple) + + def _parse_int_version(self, version: Optional[int]) -> Tuple[int, ...]: + if version is None: + return () + version, ver_fix = divmod(version, 100) + ver_maj, ver_min = divmod(version, 100) + return (ver_maj, ver_min, ver_fix) def gc_collect():