Extract "custom" named instances support to isctest.run module

diff --git a/bin/tests/system/isctest/run.py b/bin/tests/system/isctest/run.py
index e48e1c1b57c225e45cea64596e333287f64b033f..60703749ce0ff9637493f6dda1bb8202249efc7f 100644 (file)
--- a/bin/tests/system/isctest/run.py
+++ b/bin/tests/system/isctest/run.py
@@ -9,12 +9,26 @@
 # See the COPYRIGHT file distributed with this work for additional
 # information regarding copyright ownership.
 
+import os
 import subprocess
 import time
-from typing import Optional
+from typing import Any, Optional
 
 import isctest.log
 
+import dns.message
+
+
+# compatiblity with dnspython<2.0.0
+try:
+    # In dnspython>=2.0.0, dns.rcode.Rcode class is available
+    # pylint: disable=invalid-name
+    dns_rcode = dns.rcode.Rcode  # type: Any
+except AttributeError:
+    # In dnspython<2.0.0, selected rcodes are available as integers directly
+    # from dns.rcode
+    dns_rcode = dns.rcode
+
 
 def cmd(  # pylint: disable=too-many-arguments
     args,
@@ -70,3 +84,36 @@ def retry_with_timeout(func, timeout, delay=1, msg=None):
     if msg is None:
         msg = f"{func.__module__}.{func.__qualname__} timed out after {timeout} s"
     assert False, msg
+
+
+def get_named_cmdline(cfg_dir, cfg_file="named.conf"):
+    cfg_dir = os.path.join(os.getcwd(), cfg_dir)
+    assert os.path.isdir(cfg_dir)
+
+    cfg_file = os.path.join(cfg_dir, cfg_file)
+    assert os.path.isfile(cfg_file)
+
+    named = os.getenv("NAMED")
+    assert named is not None
+
+    named_cmdline = [named, "-c", cfg_file, "-d", "99", "-g"]
+
+    return named_cmdline
+
+
+def get_custom_named_instance(assumed_ns):
+    # This test launches and monitors a named instance itself rather than using
+    # bin/tests/system/start.pl, so manually defining a NamedInstance here is
+    # necessary for sending RNDC commands to that instance. If this "custom"
+    # instance listens on 10.53.0.3, use "ns3" as the identifier passed to
+    # the NamedInstance constructor.
+    named_ports = isctest.instance.NamedPorts.from_env()
+    instance = isctest.instance.NamedInstance(assumed_ns, named_ports)
+
+    return instance
+
+
+def assert_custom_named_is_alive(named_proc, resolver_ip):
+    assert named_proc.poll() is None, "named isn't running"
+    msg = dns.message.make_query("version.bind", "TXT", "CH")
+    isctest.query.tcp(msg, resolver_ip, expected_rcode=dns_rcode.NOERROR)

diff --git a/bin/tests/system/shutdown/tests_shutdown.py b/bin/tests/system/shutdown/tests_shutdown.py
index 0fc2b1c5e9189e9f09333a7a1646f90d93f41418..121feed7b7aaf56033a7337969c99c854123e875 100755 (executable)
--- a/bin/tests/system/shutdown/tests_shutdown.py
+++ b/bin/tests/system/shutdown/tests_shutdown.py
@@ -156,35 +156,19 @@ def wait_for_proc_termination(proc, max_timeout=10):
     ["rndc", "sigterm"],
 )
 def test_named_shutdown(kill_method):
-    # pylint: disable-msg=too-many-locals
-    cfg_dir = os.path.join(os.getcwd(), "resolver")
-    assert os.path.isdir(cfg_dir)
-
-    cfg_file = os.path.join(cfg_dir, "named.conf")
-    assert os.path.isfile(cfg_file)
+    resolver_ip = "10.53.0.3"
 
-    named = os.getenv("NAMED")
-    assert named is not None
+    cfg_dir = "resolver"
 
-    # This test launches and monitors a named instance itself rather than using
-    # bin/tests/system/start.pl, so manually defining a NamedInstance here is
-    # necessary for sending RNDC commands to that instance.  This "custom"
-    # instance listens on 10.53.0.3, so use "ns3" as the identifier passed to
-    # the NamedInstance constructor.
-    named_ports = isctest.instance.NamedPorts.from_env()
-    instance = isctest.instance.NamedInstance("ns3", named_ports)
+    named_cmdline = isctest.run.get_named_cmdline(cfg_dir)
+    instance = isctest.run.get_custom_named_instance("ns3")
 
-    resolver_ip = "10.53.0.3"
-    named_cmdline = [named, "-c", cfg_file, "-d", "99", "-g"]
     with open(os.path.join(cfg_dir, "named.run"), "ab") as named_log:
         with subprocess.Popen(
             named_cmdline, cwd=cfg_dir, stderr=named_log
         ) as named_proc:
             try:
-                assert named_proc.poll() is None, "named isn't running"
-                msg = dns.message.make_query("version.bind", "TXT", "CH")
-                res = isctest.query.tcp(msg, resolver_ip)
-                isctest.check.noerror(res)
+                isctest.run.assert_custom_named_is_alive(named_proc, resolver_ip)
                 do_work(
                     named_proc,
                     resolver_ip,