Commit 0c9bd01729 for bind
commit 0c9bd017297966845b1671d35d929344de02c97e
Author: Štěpán Balážik <stepan@isc.org>
Date: Tue Sep 8 16:46:32 2026 +0200
Match on a QNAME label with a matcher
LabelPredicate declares a test of one label of the QNAME a handler
answers, LeftmostLabel and LeftmostLabelPrefix the common forms of it,
which was previously done with a match() method.
Assisted-by: Claude:claude-fable-5
diff --git a/bin/tests/system/cookie/cookie_ans.py b/bin/tests/system/cookie/cookie_ans.py
index 063ee175f7..1e186d73b3 100644
--- a/bin/tests/system/cookie/cookie_ans.py
+++ b/bin/tests/system/cookie/cookie_ans.py
@@ -25,7 +25,7 @@ from isctest.asyncserver import (
ResponseHandler,
)
from isctest.asyncserver.actions import DnsResponseSend
-from isctest.asyncserver.matchers import Protocol, Qtype
+from isctest.asyncserver.matchers import LeftmostLabel, Protocol, Qtype
from isctest.name import prepend_label
from isctest.vars.algorithms import ALG_VARS
@@ -135,12 +135,9 @@ class TcpAHandler(ResponseHandler):
class WithtsigUdpAHandler(ResponseHandler):
- def match(self, qctx: QueryContext) -> bool:
- return (
- qctx.qtype == dns.rdatatype.A
- and qctx.protocol == DnsProtocol.UDP
- and _first_label(qctx) == "withtsig"
- )
+ matcher = (
+ Qtype(dns.rdatatype.A) & Protocol(DnsProtocol.UDP) & LeftmostLabel(b"withtsig")
+ )
async def get_responses(
self, qctx: QueryContext
diff --git a/bin/tests/system/isctest/asyncserver/matchers.py b/bin/tests/system/isctest/asyncserver/matchers.py
index 8c52fb0c51..97c630bc64 100644
--- a/bin/tests/system/isctest/asyncserver/matchers.py
+++ b/bin/tests/system/isctest/asyncserver/matchers.py
@@ -9,7 +9,7 @@
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
from typing import TypeVar
import abc
@@ -243,3 +243,53 @@ class EdnsOptions(Matcher):
def __str__(self) -> str:
return "with EDNS options"
+
+
+class LabelPredicate(Matcher):
+ """
+ Match queries whose QNAME label at the given index passes the given test,
+ described for the log; a QNAME without such a label does not match.
+ """
+
+ def __init__(
+ self, index: int, test: Callable[[bytes], bool], description: str
+ ) -> None:
+ self._index = index
+ self._test = test
+ self._description = description
+
+ def match(self, qctx: QueryContext) -> bool:
+ try:
+ label = qctx.qname.labels[self._index]
+ except IndexError:
+ return False
+ return self._test(label)
+
+ def __str__(self) -> str:
+ return f"label {self._index} {self._description}"
+
+
+class LeftmostLabel(LabelPredicate):
+ """
+ Match queries whose leftmost label is one of the given labels.
+ """
+
+ def __init__(self, *labels: bytes) -> None:
+ super().__init__(
+ 0,
+ lambda label: label in labels,
+ f"in [{', '.join(label.decode() for label in labels)}]",
+ )
+
+
+class LeftmostLabelPrefix(LabelPredicate):
+ """
+ Match queries whose leftmost label starts with the given prefix.
+ """
+
+ def __init__(self, prefix: bytes) -> None:
+ super().__init__(
+ 0,
+ lambda label: label.startswith(prefix),
+ f"starting with {prefix.decode()}",
+ )
diff --git a/bin/tests/system/resolver/ans3/ans.py b/bin/tests/system/resolver/ans3/ans.py
index 38eb5ff14f..a15400dbfa 100644
--- a/bin/tests/system/resolver/ans3/ans.py
+++ b/bin/tests/system/resolver/ans3/ans.py
@@ -18,7 +18,7 @@ import dns.rdatatype
from isctest.asyncserver import AsyncDnsServer, QueryContext, ResponseHandler
from isctest.asyncserver.actions import DnsResponseSend
from isctest.asyncserver.handlers import IgnoreAllQueries, StaticResponseHandler
-from isctest.asyncserver.matchers import Domain, Qname, Qtype
+from isctest.asyncserver.matchers import Domain, LeftmostLabelPrefix, Qname, Qtype
from ..resolver_ans import (
DelegationHandler,
@@ -128,8 +128,7 @@ class LargeReferralHandler(StaticResponseHandler):
class LongCnameHandler(ResponseHandler):
- def match(self, qctx: QueryContext) -> bool:
- return qctx.qname.labels[0].startswith(b"longcname")
+ matcher = LeftmostLabelPrefix(b"longcname")
async def get_responses(
self, qctx: QueryContext