upstream/bind9
1
0
Fork 0
mirror of https://github.com/isc-projects/bind9.git synced 2026-07-15 00:40:49 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

new: test: Add AnsInstance as the ans counterpart of NamedInstance

Browse source 
Tests interacting with mock ans servers had to hardcode their IP
addresses and open ans.run directly, while named instances already
had the NamedInstance abstraction with `.ip`, `.log` and the
watch_log_*() helpers.  Factor the parts of NamedInstance that are
not named-specific into a ServerInstance base class and add an
AnsInstance subclass for ans servers, exposed through the `servers`
fixture and new ans1-ans11 convenience fixtures.

Assisted-by: Claude:claude-fable-5

Merge branch 'nicki/pytest-ans-instance' into 'main'

See merge request isc-projects/bind9!12241
This commit is contained in:
Nicki Křížek 2026-06-16 10:28:36 +02:00
commit 139b0fc48e
3 changed files with 199 additions and 80 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
bin/tests/system/conftest.py
View file

@ -637,12 +637,16 @@ def servers(system_test_dir):
instances = {}
for entry in system_test_dir.rglob("*"):
if entry.is_dir():
try:
dir_name = entry.name
instance = isctest.instance.NamedInstance(dir_name)
instances[dir_name] = instance
except ValueError:
continue
dir_name = entry.name
for instance_class in (
isctest.instance.NamedInstance,
isctest.instance.AnsInstance,
):
try:
instances[dir_name] = instance_class(dir_name)
break
except ValueError:
continue
return instances
@ -699,3 +703,58 @@ def ns10(servers):
@pytest.fixture(scope="module")
def ns11(servers):
return servers["ns11"]
@pytest.fixture(scope="module")
def ans1(servers):
return servers["ans1"]
@pytest.fixture(scope="module")
def ans2(servers):
return servers["ans2"]
@pytest.fixture(scope="module")
def ans3(servers):
return servers["ans3"]
@pytest.fixture(scope="module")
def ans4(servers):
return servers["ans4"]
@pytest.fixture(scope="module")
def ans5(servers):
return servers["ans5"]
@pytest.fixture(scope="module")
def ans6(servers):
return servers["ans6"]
@pytest.fixture(scope="module")
def ans7(servers):
return servers["ans7"]
@pytest.fixture(scope="module")
def ans8(servers):
return servers["ans8"]
@pytest.fixture(scope="module")
def ans9(servers):
return servers["ans9"]
@pytest.fixture(scope="module")
def ans10(servers):
return servers["ans10"]
@pytest.fixture(scope="module")
def ans11(servers):
return servers["ans11"]

11
bin/tests/system/dispatch/tests_tcponly.py
View file

@ -27,13 +27,12 @@ pytestmark = pytest.mark.extra_artifacts(
)
def _count_received(path, qname, protocol):
def _count_received(ans, qname, protocol):
pattern = Re(rf"Received {escape(qname)}/IN/A .* \({protocol}\)$")
with open(path, encoding="utf-8") as fh:
return sum(1 for line in fh if pattern.search(line.rstrip()))
return len(ans.log.grep(pattern))
def test_tcponly_fallback():
def test_tcponly_fallback(ans4):
"""
A resolver must fall back to TCP after repeated UDP timeouts to the
same authoritative server. ans4 drops every UDP query and answers
@ -51,7 +50,7 @@ def test_tcponly_fallback():
)
assert str(rdataset[0]) == "127.0.0.1"
udp = _count_received("ans4/ans.run", "foo.tcp-only", "UDP")
tcp = _count_received("ans4/ans.run", "foo.tcp-only", "TCP")
udp = _count_received(ans4, "foo.tcp-only", "UDP")
tcp = _count_received(ans4, "foo.tcp-only", "TCP")
assert udp == 2, f"expected exactly 2 UDP queries, got {udp}"
assert tcp == 1, f"expected exactly 1 TCP query, got {tcp}"

197
bin/tests/system/isctest/instance.py
View file

@ -14,6 +14,7 @@
from pathlib import Path
from typing import NamedTuple
import abc
import os
import re
@ -39,7 +40,109 @@ class NamedPorts(NamedTuple):
)
class NamedInstance:
class AnsPorts(NamedTuple):
dns: int = 53
@staticmethod
def from_env():
return AnsPorts(
dns=int(os.environ["PORT"]),
)
class ServerInstance(abc.ABC):
"""
Common base class for the server instances used in a system test.
This class should not be used directly; instead, its subclasses,
`NamedInstance` and `AnsInstance`, should be used.
"""
@property
@abc.abstractmethod
def log_filename(self) -> str:
"""Name of the log file in the instance's directory."""
@property
@abc.abstractmethod
def identifier_prefix(self) -> str:
"""Directory name prefix used to derive the numeric identifier."""
def __init__(self, identifier: str, num: int | None = None) -> None:
"""
`identifier` is the name of the instance's directory
`num` is optional if the identifier starts with `identifier_prefix`
followed by a number, in which case the number is assumed to be the
numeric identifier; otherwise it must be provided to assign a numeric
identification to the server
"""
self.directory = Path(identifier).absolute()
if not self.directory.is_dir():
raise ValueError(f"{self.directory} isn't a directory")
self.system_test_name = self.directory.parent.name
self.identifier = identifier
self.num = self._identifier_to_num(identifier, num)
self.log = TextFile(os.path.join(identifier, self.log_filename))
@property
def ip(self) -> str:
"""IPv4 address of the instance."""
return f"10.53.0.{self.num}"
@classmethod
def _identifier_to_num(cls, identifier: str, num: int | None = None) -> int:
regex_match = re.match(
rf"^{cls.identifier_prefix}(?P<index>[0-9]{{1,2}})$", identifier
)
if not regex_match:
if num is None:
raise ValueError(f'Can\'t parse numeric identifier from "{identifier}"')
return num
parsed_num = int(regex_match.group("index"))
assert num is None or num == parsed_num, "mismatched num and identifier"
return parsed_num
def watch_log_from_start(
self, timeout: float = WatchLogFromStart.DEFAULT_TIMEOUT
) -> WatchLogFromStart:
"""
Return an instance of the `WatchLogFromStart` context manager for this
instance's log file.
"""
return WatchLogFromStart(self.log.path, timeout)
def watch_log_from_here(
self, timeout: float = WatchLogFromHere.DEFAULT_TIMEOUT
) -> WatchLogFromHere:
"""
Return an instance of the `WatchLogFromHere` context manager for this
instance's log file.
"""
return WatchLogFromHere(self.log.path, timeout)
def stop(self, args: list[str] | None = None) -> None:
"""Stop the instance."""
args = args or []
perl(
f"{os.environ['srcdir']}/stop.pl",
[self.system_test_name, self.identifier] + args,
)
def start(self, args: list[str] | None = None) -> None:
"""Start the instance."""
args = args or []
perl(
f"{os.environ['srcdir']}/start.pl",
[self.system_test_name, self.identifier] + args,
)
def __repr__(self):
return self.identifier
class NamedInstance(ServerInstance):
"""
A class representing a `named` instance used in a system test.
@ -51,11 +154,13 @@ class NamedInstance:
```
"""
log_filename = "named.run"
identifier_prefix = "ns"
def __init__(
self,
identifier: str,
num: int | None = None,
ports: NamedPorts | None = None,
) -> None:
"""
`identifier` is the name of the instance's directory
@ -63,23 +168,9 @@ class NamedInstance:
`num` is optional if the identifier is in a form of `ns<X>`, in which
case `<X>` is assumed to be numeric identifier; otherwise it must be
provided to assign a numeric identification to the server
`ports` is the `NamedPorts` instance listing the UDP/TCP ports on which
this `named` instance is listening for various types of traffic (both
DNS traffic and RNDC commands). Defaults to ports set by the test
framework.
"""
self.directory = Path(identifier).absolute()
if not self.directory.is_dir():
raise ValueError(f"{self.directory} isn't a directory")
self.system_test_name = self.directory.parent.name
self.identifier = identifier
self.num = self._identifier_to_num(identifier, num)
if ports is None:
ports = NamedPorts.from_env()
self.ports = ports
self.log = TextFile(os.path.join(identifier, "named.run"))
super().__init__(identifier, num)
self.ports = NamedPorts.from_env()
self._rndc_conf = Path("../_common/rndc.conf").absolute()
self._rndc = EnvCmd("RNDC", self.rndc_args)
@ -89,22 +180,6 @@ class NamedInstance:
"""Base arguments for calling RNDC to control the instance."""
return f"-c {self._rndc_conf} -s {self.ip} -p {self.ports.rndc}"
@property
def ip(self) -> str:
"""IPv4 address of the instance."""
return f"10.53.0.{self.num}"
@staticmethod
def _identifier_to_num(identifier: str, num: int | None = None) -> int:
regex_match = re.match(r"^ns(?P<index>[0-9]{1,2})$", identifier)
if not regex_match:
if num is None:
raise ValueError(f'Can\'t parse numeric identifier from "{identifier}"')
return num
parsed_num = int(regex_match.group("index"))
assert num is None or num == parsed_num, "mismatched num and identifier"
return parsed_num
def rndc(self, command: str, timeout=10, **kwargs) -> CmdResult:
"""
Send `command` to this named instance using RNDC. Return the server's
@ -139,24 +214,6 @@ class NamedInstance:
)
return response
def watch_log_from_start(
self, timeout: float = WatchLogFromStart.DEFAULT_TIMEOUT
) -> WatchLogFromStart:
"""
Return an instance of the `WatchLogFromStart` context manager for this
`named` instance's log file.
"""
return WatchLogFromStart(self.log.path, timeout)
def watch_log_from_here(
self, timeout: float = WatchLogFromHere.DEFAULT_TIMEOUT
) -> WatchLogFromHere:
"""
Return an instance of the `WatchLogFromHere` context manager for this
`named` instance's log file.
"""
return WatchLogFromHere(self.log.path, timeout)
def reconfigure(self, **kwargs) -> CmdResult:
"""
Reconfigure this named `instance` and wait until reconfiguration is
@ -176,21 +233,25 @@ class NamedInstance:
watcher.wait_for_line("all zones loaded")
return cmd
def stop(self, args: list[str] | None = None) -> None:
"""Stop the instance."""
args = args or []
perl(
f"{os.environ['srcdir']}/stop.pl",
[self.system_test_name, self.identifier] + args,
)
def start(self, args: list[str] | None = None) -> None:
"""Start the instance."""
args = args or []
perl(
f"{os.environ['srcdir']}/start.pl",
[self.system_test_name, self.identifier] + args,
)
class AnsInstance(ServerInstance):
"""
A class representing a mock `ans` server instance used in a system test.
def __repr__(self):
return self.identifier
This class is expected to be instantiated as part of the `servers` fixture:
```python
def test_foo(servers):
assert "query received" in servers["ans4"].log
```
"""
log_filename = "ans.run"
identifier_prefix = "ans"
def __init__(self, identifier: str) -> None:
"""
`identifier` is the name of the instance's directory
"""
super().__init__(identifier)
self.ports = AnsPorts.from_env()