Use Re() for creating regular expressions

It's a fairly common pattern to use regular expression in our tests.
Instead of using the fairly verbose re.compile(), import that function
as Re() instead to allow for more brevity in the test syntax.

(cherry picked from commit ac7127d620)
This commit is contained in:
Nicki Křížek 2025-10-02 11:47:56 +02:00
parent deb8b39dc6
commit fe53a4eb62
3 changed files with 15 additions and 10 deletions

View file

@ -13,7 +13,7 @@ from functools import partial
import filecmp
import os
from pathlib import Path
import re
from re import compile as Re
import shutil
import subprocess
import tempfile
@ -49,7 +49,7 @@ else:
XDIST_WORKER = os.environ.get("PYTEST_XDIST_WORKER", "")
FILE_DIR = os.path.abspath(Path(__file__).parent)
ENV_RE = re.compile(b"([^=]+)=(.*)")
ENV_RE = Re(b"([^=]+)=(.*)")
PORT_MIN = 5001
PORT_MAX = 32767
PORTS_PER_TEST = 20
@ -62,10 +62,10 @@ PRIORITY_TESTS = [
"timeouts/",
"upforwd/",
]
PRIORITY_TESTS_RE = re.compile("|".join(PRIORITY_TESTS))
PRIORITY_TESTS_RE = Re("|".join(PRIORITY_TESTS))
SYSTEM_TEST_DIR_GIT_PATH = "bin/tests/system"
SYSTEM_TEST_NAME_RE = re.compile(f"{SYSTEM_TEST_DIR_GIT_PATH}" + r"/([^/]+)")
SYMLINK_REPLACEMENT_RE = re.compile(r"/tests(_.*)\.py")
SYSTEM_TEST_NAME_RE = Re(f"{SYSTEM_TEST_DIR_GIT_PATH}" + r"/([^/]+)")
SYMLINK_REPLACEMENT_RE = Re(r"/tests(_.*)\.py")
# ---------------------- Module initialization ---------------------------

View file

@ -14,6 +14,7 @@ from typing import Any, Iterator, List, Match, Optional, Pattern, TextIO, TypeVa
import abc
import os
import re
from re import compile as Re
import time
@ -213,7 +214,7 @@ class WatchLog(abc.ABC):
if isinstance(string, Pattern):
patterns.append(string)
elif isinstance(string, str):
pattern = re.compile(re.escape(string))
pattern = Re(re.escape(string))
patterns.append(pattern)
else:
raise WatchLogException(
@ -256,13 +257,14 @@ class WatchLog(abc.ABC):
Recommended use:
```python
from re import compile as Re
import isctest
def test_foo(servers):
with servers["ns1"].watch_log_from_start() as watcher:
watcher.wait_for_line("all zones loaded")
pattern = re.compile(r"next key event in ([0-9]+) seconds")
pattern = Re(r"next key event in ([0-9]+) seconds")
with servers["ns1"].watch_log_from_here() as watcher:
# ... do stuff here ...
match = watcher.wait_for_line(pattern)
@ -321,7 +323,8 @@ class WatchLog(abc.ABC):
>>> # Different values must be returned depending on which line is
>>> # found in the log file.
>>> import tempfile
>>> patterns = [re.compile(r"bar ([0-9])"), "qux"]
>>> from re import compile as Re
>>> patterns = [Re(r"bar ([0-9])"), "qux"]
>>> with tempfile.NamedTemporaryFile("w") as file:
... print("foo bar 3", file=file, flush=True)
... with WatchLogFromStart(file.name) as watcher:
@ -443,7 +446,8 @@ class WatchLog(abc.ABC):
>>> assert ret[1].group(0) == "foo"
>>> import tempfile
>>> bar_pattern = re.compile('bar')
>>> from re import compile as Re
>>> bar_pattern = Re('bar')
>>> patterns = ['foo', bar_pattern]
>>> with tempfile.NamedTemporaryFile("w") as file:
... print("bar", file=file, flush=True)

View file

@ -12,6 +12,7 @@
import glob
import os
import re
from re import compile as Re
import shutil
import signal
import time
@ -71,7 +72,7 @@ def test_xferquota(named_port, servers):
isctest.check.rrsets_equal(ns1response.answer, ns2response.answer)
query_and_compare(axfr_msg)
pattern = re.compile(
pattern = Re(
f"transfer of 'changing/IN' from 10.53.0.1#{named_port}: "
f"Transfer completed: .*\\(serial 2\\)"
)