mir/bind
2
0
Fork 0
mirror of https://gitlab.isc.org/isc-projects/bind9 synced 2025-09-01 06:55:30 +00:00
Code Issues Releases Wiki Activity

Add options for query&response logging to pytest

Browse Source 
In some cases, it's useful to log the sent and received DNS messages.
Add options to enable this on demand. Query is only logged the first
time it's sent, since it doesn't change. If response logging is turned
on, then each response is logged, since it might be different every
time.

(cherry picked from commit 1e87b5ffc6)
This commit is contained in:
Nicki Křížek
2025-06-17 17:40:07 +02:00
parent a6599a0822
commit 03af371948
1 changed files with 16 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
bin/tests/system/isctest/query.py
View File

@@ -32,6 +32,8 @@ def generic_query(
attempts: int = 10, attempts: int = 10,
expected_rcode: dns_rcode = None, expected_rcode: dns_rcode = None,
verify: bool = False, verify: bool = False,
log_query: bool = False,
log_response: bool = False,
) -> Any: ) -> Any:
if port is None: if port is None:
if query_func.__name__ == "tls": if query_func.__name__ == "tls":
@@ -51,15 +53,25 @@ def generic_query(
res = None res = None
for attempt in range(attempts): for attempt in range(attempts):
isctest.log.debug( log_msg = (
f"{query_func.__name__}(): ip={ip}, port={port}, source={source}, " f"isc.query.{query_func.__name__}(): ip={ip}, port={port}, source={source}, "
f"timeout={timeout}, attempts left={attempts-attempt}" f"timeout={timeout}, attempts left={attempts-attempt}"
) )
if log_query:
log_msg += f"\n{message.to_text()}"
log_query = False # only log query on first attempt
isctest.log.debug(log_msg)
try: try:
res = query_func(**query_args) res = query_func(**query_args)
except (dns.exception.Timeout, ConnectionRefusedError) as e: except (dns.exception.Timeout, ConnectionRefusedError) as e:
isctest.log.debug(f"{query_func.__name__}(): the '{e}' exception raised") isctest.log.debug(
f"isc.query.{query_func.__name__}(): the '{e}' exception raised"
)
else: else:
if log_response:
isctest.log.debug(
f"isc.query.{query_func.__name__}(): response\n{res.to_text()}"
)
if res.rcode() == expected_rcode or expected_rcode is None: if res.rcode() == expected_rcode or expected_rcode is None:
return res return res
time.sleep(1) time.sleep(1)
@@ -67,7 +79,7 @@ def generic_query(
if expected_rcode is not None: if expected_rcode is not None:
last_rcode = dns_rcode.to_text(res.rcode()) if res else None last_rcode = dns_rcode.to_text(res.rcode()) if res else None
isctest.log.debug( isctest.log.debug(
f"{query_func.__name__}(): expected rcode={dns_rcode.to_text(expected_rcode)}, last rcode={last_rcode}" f"isc.query.{query_func.__name__}(): expected rcode={dns_rcode.to_text(expected_rcode)}, last rcode={last_rcode}"
) )
raise dns.exception.Timeout raise dns.exception.Timeout