Move get_soa_serial into isctest.query and make it retry

The pre-update SOA serial capture in the multisigner test was a
one-shot query outside any retry loop, so a single transient timeout
would fail the test that the timing fix is meant to deflake.  Make
get_soa_serial a shared isctest.query helper that retries internally
(for up to 10 seconds by default) and reuse it in wait_for_serial,
which duplicated the same SOA extraction logic.

Assisted-by: Claude:claude-fable-5

diff --git a/bin/tests/system/isctest/query.py b/bin/tests/system/isctest/query.py
index b73a407cdb59bf2c6d0f1163ff25da1132b9b808..74279fd5422695f2841c2c053a8c20e5a23552e7 100644 (file)
--- a/bin/tests/system/isctest/query.py
+++ b/bin/tests/system/isctest/query.py
@@ -169,28 +169,53 @@ def create(
     return msg
 
 
-def wait_for_serial(server_ip, zone, expected_serial, timeout=30):
-    """Wait until the server has the expected SOA serial for the zone.
-
-    Queries the server repeatedly until the SOA serial matches or the
-    timeout expires.
+def get_soa_serial(server_ip, zone, timeout=10):
+    """
+    Get the current SOA serial of a zone from a server.
 
-    'server_ip' is the IP address to query (string).
-    'zone' is the zone name (string, with or without trailing dot).
-    'expected_serial' is the expected SOA serial number (int).
-    'timeout' is the maximum time to wait in seconds (default 30).
+    Queries the server repeatedly until it responds with a well-formed
+    SOA answer or the timeout expires.
     """
     query = create(zone, "SOA", dnssec=False)
+    serial = None
 
     def check():
-        res = tcp(query, server_ip)
+        nonlocal serial
+        res = tcp(
+            query,
+            server_ip,
+            timeout=3,
+            attempts=1,
+            expected_rcode=dns.rcode.NOERROR,
+        )
         soa = res.get_rrset(
             res.answer,
             dns.name.from_text(zone),
             dns.rdataclass.IN,
             dns.rdatatype.SOA,
         )
-        return soa is not None and len(soa) == 1 and soa[0].serial == expected_serial
+        assert soa is not None and len(soa) == 1
+        serial = soa[0].serial
+        return True
+
+    isctest.run.retry_with_timeout(
+        check,
+        timeout=timeout,
+        msg=f"timed out getting SOA serial of {zone} from {server_ip}",
+    )
+    return serial
+
+
+def wait_for_serial(server_ip, zone, expected_serial, timeout=30):
+    """
+    Wait until the server has the expected SOA serial for the zone.
+
+    Queries the server repeatedly until the SOA serial matches or the
+    timeout expires.
+    """
+
+    def check():
+        return get_soa_serial(server_ip, zone) == expected_serial
 
     isctest.run.retry_with_timeout(
         check,

diff --git a/bin/tests/system/multisigner/tests_multisigner.py b/bin/tests/system/multisigner/tests_multisigner.py
index fa65449f28ca691c4e84eb98730334726601a737..82d80089b0b49239cf49b1831c35a12c46e1229a 100644 (file)
--- a/bin/tests/system/multisigner/tests_multisigner.py
+++ b/bin/tests/system/multisigner/tests_multisigner.py
@@ -14,10 +14,6 @@ from re import compile as Re
 
 import os
 
-import dns.name
-import dns.rcode
-import dns.rdataclass
-import dns.rdatatype
 import dns.update
 import pytest
 
@@ -112,36 +108,19 @@ def check_no_dnssec_in_journal(server, zone):
     ), "dnssec record found in journal"
 
 
-def get_soa_serial(server, zone):
-    fqdn = f"{zone}."
-    query = isctest.query.create(fqdn, dns.rdatatype.SOA)
-    response = isctest.query.tcp(
-        query, server.ip, server.ports.dns, timeout=3, attempts=1
-    )
-    assert response.rcode() == dns.rcode.NOERROR
-    soa = response.get_rrset(
-        response.answer,
-        dns.name.from_text(fqdn),
-        dns.rdataclass.IN,
-        dns.rdatatype.SOA,
-    )
-    assert soa is not None and len(soa) == 1
-    return soa[0].serial
-
-
 def wait_for_serial(primary, server, zone, previous_serial=None):
     if primary.identifier == server.identifier:
         assert previous_serial is not None
 
         def check_prev_serial():
-            return get_soa_serial(server, zone) != previous_serial
+            return isctest.query.get_soa_serial(server.ip, zone) != previous_serial
 
         isctest.run.retry_with_timeout(check_prev_serial, timeout=30)
         return
 
     def check_serial():
-        serial1 = get_soa_serial(primary, zone)
-        serial2 = get_soa_serial(server, zone)
+        serial1 = isctest.query.get_soa_serial(primary.ip, zone)
+        serial2 = isctest.query.get_soa_serial(server.ip, zone)
 
         return (
             f"zone {zone}/IN (signed): serial {serial2} (unsigned {serial1})"
@@ -154,7 +133,7 @@ def wait_for_serial(primary, server, zone, previous_serial=None):
 def nsupdate_and_wait(primary, server, zone, update_msg):
     previous_serial = None
     if primary.identifier == server.identifier:
-        previous_serial = get_soa_serial(server, zone)
+        previous_serial = isctest.query.get_soa_serial(server.ip, zone)
 
     primary.nsupdate(update_msg)
     wait_for_serial(primary, server, zone, previous_serial)