]> git.ipfire.org Git - thirdparty/bind9.git/commitdiff
Add options for query&response logging to pytest
authorNicki Křížek <nicki@isc.org>
Tue, 17 Jun 2025 15:40:07 +0000 (17:40 +0200)
committerNicki Křížek <nicki@isc.org>
Fri, 27 Jun 2025 14:31:49 +0000 (16:31 +0200)
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.

bin/tests/system/isctest/query.py

index 6e84c8188d209ae63727adcc58a057e1edb26280..4b8590fb2f65905a0d5f53ed50904da46b8ae401 100644 (file)
@@ -32,6 +32,8 @@ def generic_query(
     attempts: int = 10,
     expected_rcode: dns_rcode = None,
     verify: bool = False,
+    log_query: bool = False,
+    log_response: bool = False,
 ) -> Any:
     if port is None:
         if query_func.__name__ == "tls":
@@ -51,15 +53,25 @@ def generic_query(
 
     res = None
     for attempt in range(attempts):
-        isctest.log.debug(
-            f"{query_func.__name__}(): ip={ip}, port={port}, source={source}, "
+        log_msg = (
+            f"isc.query.{query_func.__name__}(): ip={ip}, port={port}, source={source}, "
             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:
             res = query_func(**query_args)
         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:
+            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:
                 return res
         time.sleep(1)
@@ -67,7 +79,7 @@ def generic_query(
     if expected_rcode is not None:
         last_rcode = dns_rcode.to_text(res.rcode()) if res else None
         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