chg: test: Rewrite stub system test to pytest

Merge branch 'mnowak/pytest_rewrite_stub' into 'main'

See merge request isc-projects/bind9!9190

bin/tests/system/conftest.py

@@ -9,7 +9,6 @@
 # See the COPYRIGHT file distributed with this work for additional
 # information regarding copyright ownership.
 
-from functools import partial
 import filecmp
 import os
 from pathlib import Path
@@ -18,7 +17,7 @@
 import subprocess
 import tempfile
 import time
-from typing import Any, List, Optional
+from typing import Any
 
 import pytest
 
@@ -483,46 +482,6 @@ def templates(system_test_dir: Path):
     return isctest.template.TemplateEngine(system_test_dir)
 
 
-def _run_script(
-    system_test_dir: Path,
-    interpreter: str,
-    script: str,
-    args: Optional[List[str]] = None,
-):
-    """Helper function for the shell / perl script invocations (through fixtures below)."""
-    if args is None:
-        args = []
-    path = Path(script)
-    if not path.is_absolute():
-        # make sure relative paths are always relative to system_dir
-        path = system_test_dir.parent / path
-    script = str(path)
-    cwd = os.getcwd()
-    if not path.exists():
-        raise FileNotFoundError(f"script {script} not found in {cwd}")
-    isctest.log.debug("running script: %s %s %s", interpreter, script, " ".join(args))
-    isctest.log.debug("  workdir: %s", cwd)
-    returncode = 1
-
-    cmd = [interpreter, script] + args
-    with subprocess.Popen(
-        cmd,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        bufsize=1,
-        universal_newlines=True,
-        errors="backslashreplace",
-    ) as proc:
-        if proc.stdout:
-            for line in proc.stdout:
-                isctest.log.info("    %s", line.rstrip("\n"))
-        proc.communicate()
-        returncode = proc.returncode
-        if returncode:
-            raise subprocess.CalledProcessError(returncode, cmd)
-        isctest.log.debug("  exited with %d", returncode)
-
-
 def _get_node_path(node) -> Path:
     if isinstance(node.parent, pytest.Session):
         if _pytest_major_ver >= 8:
@@ -533,23 +492,11 @@ def _get_node_path(node) -> Path:
 
 
 @pytest.fixture(scope="module")
-def shell(system_test_dir):
-    """Function to call a shell script with arguments."""
-    return partial(_run_script, system_test_dir, os.environ["SHELL"])
-
-
-@pytest.fixture(scope="module")
-def perl(system_test_dir):
-    """Function to call a perl script with arguments."""
-    return partial(_run_script, system_test_dir, os.environ["PERL"])
-
-
-@pytest.fixture(scope="module")
-def run_tests_sh(system_test_dir, shell):
+def run_tests_sh(system_test_dir):
     """Utility function to execute tests.sh as a python test."""
 
     def run_tests():
-        shell(f"{system_test_dir}/tests.sh")
+        isctest.run.shell(f"{system_test_dir}/tests.sh")
 
     return run_tests
 
@@ -559,8 +506,6 @@ def system_test(
     request,
     system_test_dir,
     templates,
-    shell,
-    perl,
 ):
     """
     Driver of the test setup/teardown process. Used automatically for every test module.
@@ -586,14 +531,16 @@ def system_test(
 
     def check_net_interfaces():
         try:
-            perl("testsock.pl", ["-p", os.environ["PORT"]])
+            isctest.run.perl(
+                f"{os.environ['srcdir']}/testsock.pl", ["-p", os.environ["PORT"]]
+            )
         except subprocess.CalledProcessError as exc:
             isctest.log.error("testsock.pl: exited with code %d", exc.returncode)
             pytest.skip("Network interface aliases not set up.")
 
     def check_prerequisites():
         try:
-            shell(f"{system_test_dir}/prereq.sh")
+            isctest.run.shell(f"{system_test_dir}/prereq.sh")
         except FileNotFoundError:
             pass  # prereq.sh is optional
         except subprocess.CalledProcessError:
@@ -602,7 +549,7 @@ def check_prerequisites():
     def setup_test():
         templates.render_auto()
         try:
-            shell(f"{system_test_dir}/setup.sh")
+            isctest.run.shell(f"{system_test_dir}/setup.sh")
         except FileNotFoundError:
             pass  # setup.sh is optional
         except subprocess.CalledProcessError as exc:
@@ -611,22 +558,27 @@ def setup_test():
 
     def start_servers():
         try:
-            perl("start.pl", ["--port", os.environ["PORT"], system_test_dir.name])
+            isctest.run.perl(
+                f"{os.environ['srcdir']}/start.pl",
+                ["--port", os.environ["PORT"], system_test_dir.name],
+            )
         except subprocess.CalledProcessError as exc:
             isctest.log.error("Failed to start servers")
             pytest.fail(f"start.pl exited with {exc.returncode}")
 
     def stop_servers():
         try:
-            perl("stop.pl", [system_test_dir.name])
+            isctest.run.perl(f"{os.environ['srcdir']}/stop.pl", [system_test_dir.name])
         except subprocess.CalledProcessError as exc:
             isctest.log.error("Failed to stop servers")
             get_core_dumps()
             pytest.fail(f"stop.pl exited with {exc.returncode}")
 
     def get_core_dumps():
         try:
-            shell("get_core_dumps.sh", [system_test_dir.name])
+            isctest.run.shell(
+                f"{os.environ['srcdir']}/get_core_dumps.sh", [system_test_dir.name]
+            )
         except subprocess.CalledProcessError as exc:
             isctest.log.error("Found core dumps or sanitizer reports")
             pytest.fail(f"get_core_dumps.sh exited with {exc.returncode}")

bin/tests/system/isctest/check.py

@@ -102,6 +102,10 @@ def is_executable(cmd: str, errmsg: str) -> None:
     assert executable is not None, errmsg
 
 
+def notauth(message: dns.message.Message) -> None:
+    rcode(message, dns.rcode.NOTAUTH)
+
+
 def nxdomain(message: dns.message.Message) -> None:
     rcode(message, dns.rcode.NXDOMAIN)
 

bin/tests/system/isctest/instance.py

@@ -11,13 +11,15 @@
 # See the COPYRIGHT file distributed with this work for additional
 # information regarding copyright ownership.
 
-from typing import NamedTuple, Optional
+from typing import List, NamedTuple, Optional
 
 import logging
 import os
+from pathlib import Path
 import re
 
 from .rndc import RNDCBinaryExecutor, RNDCException, RNDCExecutor
+from .run import perl
 from .log import info, LogFile, WatchLogFromStart, WatchLogFromHere
 
 
@@ -48,13 +50,17 @@ def test_foo(servers):
     def __init__(
         self,
         identifier: str,
+        num: Optional[int] = None,
         ports: Optional[NamedPorts] = None,
         rndc_logger: Optional[logging.Logger] = None,
         rndc_executor: Optional[RNDCExecutor] = None,
     ) -> None:
         """
-        `identifier` must be an `ns<X>` string, where `<X>` is an integer
-        identifier of the `named` instance this object should represent.
+        `identifier` is the name of the instance's directory
+
+        `num` is optional if the identifier is in a form of `ns<X>`, in which
+        case `<X>` is assumed to be numeric identifier; otherwise it must be
+        provided to assign a numeric identification to the server
 
         `ports` is the `NamedPorts` instance listing the UDP/TCP ports on which
         this `named` instance is listening for various types of traffic (both
@@ -67,20 +73,35 @@ def __init__(
         `rndc_executor` is an object implementing the `RNDCExecutor` interface
         that is used for executing RNDC commands on this `named` instance.
         """
-        self.ip = self._identifier_to_ip(identifier)
+        self.directory = Path(identifier).absolute()
+        if not self.directory.is_dir():
+            raise ValueError(f"{self.directory} isn't a directory")
+        self.system_test_name = self.directory.parent.name
+
+        self.identifier = identifier
+        self.num = self._identifier_to_num(identifier, num)
         if ports is None:
             ports = NamedPorts.from_env()
         self.ports = ports
         self.log = LogFile(os.path.join(identifier, "named.run"))
         self._rndc_executor = rndc_executor or RNDCBinaryExecutor()
         self._rndc_logger = rndc_logger
 
+    @property
+    def ip(self) -> str:
+        """IPv4 address of the instance."""
+        return f"10.53.0.{self.num}"
+
     @staticmethod
-    def _identifier_to_ip(identifier: str) -> str:
+    def _identifier_to_num(identifier: str, num: Optional[int] = None) -> int:
         regex_match = re.match(r"^ns(?P<index>[0-9]{1,2})$", identifier)
         if not regex_match:
-            raise ValueError("Invalid named instance identifier" + identifier)
-        return "10.53.0." + regex_match.group("index")
+            if num is None:
+                raise ValueError(f'Can\'t parse numeric identifier from "{identifier}"')
+            return num
+        parsed_num = int(regex_match.group("index"))
+        assert num is None or num == parsed_num, "mismatched num and identifier"
+        return parsed_num
 
     def rndc(self, command: str, ignore_errors: bool = False, log: bool = True) -> str:
         """
@@ -175,3 +196,19 @@ def _rndc_log(self, command: str, response: str) -> None:
             info(fmt, args)
         else:
             self._rndc_logger.info(fmt, args)
+
+    def stop(self, args: Optional[List[str]] = None) -> None:
+        """Stop the instance."""
+        args = args or []
+        perl(
+            f"{os.environ['srcdir']}/stop.pl",
+            [self.system_test_name, self.identifier] + args,
+        )
+
+    def start(self, args: Optional[List[str]] = None) -> None:
+        """Start the instance."""
+        args = args or []
+        perl(
+            f"{os.environ['srcdir']}/start.pl",
+            [self.system_test_name, self.identifier] + args,
+        )

bin/tests/system/isctest/run.py

@@ -10,9 +10,10 @@
 # information regarding copyright ownership.
 
 import os
+from pathlib import Path
 import subprocess
 import time
-from typing import Optional
+from typing import List, Optional
 
 import isctest.log
 from isctest.compat import dns_rcode
@@ -65,6 +66,51 @@ def print_debug_logs(procdata):
         return exc
 
 
+def _run_script(
+    interpreter: str,
+    script: str,
+    args: Optional[List[str]] = None,
+):
+    if args is None:
+        args = []
+    path = Path(script)
+    script = str(path)
+    cwd = os.getcwd()
+    if not path.exists():
+        raise FileNotFoundError(f"script {script} not found in {cwd}")
+    isctest.log.debug("running script: %s %s %s", interpreter, script, " ".join(args))
+    isctest.log.debug("  workdir: %s", cwd)
+    returncode = 1
+
+    command = [interpreter, script] + args
+    with subprocess.Popen(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        bufsize=1,
+        universal_newlines=True,
+        errors="backslashreplace",
+    ) as proc:
+        if proc.stdout:
+            for line in proc.stdout:
+                isctest.log.info("    %s", line.rstrip("\n"))
+        proc.communicate()
+        returncode = proc.returncode
+        if returncode:
+            raise subprocess.CalledProcessError(returncode, command)
+        isctest.log.debug("  exited with %d", returncode)
+
+
+def shell(script: str, args: Optional[List[str]] = None) -> None:
+    """Run a given script with system's shell interpreter."""
+    _run_script(os.environ["SHELL"], script, args)
+
+
+def perl(script: str, args: Optional[List[str]] = None) -> None:
+    """Run a given script with system's perl interpreter."""
+    _run_script(os.environ["PERL"], script, args)
+
+
 def retry_with_timeout(func, timeout, delay=1, msg=None):
     start_time = time.time()
     while time.time() < start_time + timeout:
@@ -91,18 +137,6 @@ def get_named_cmdline(cfg_dir, cfg_file="named.conf"):
     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")

bin/tests/system/shutdown/tests_shutdown.py

@@ -170,7 +170,7 @@ def test_named_shutdown(kill_method):
     cfg_dir = "resolver"
 
     named_cmdline = isctest.run.get_named_cmdline(cfg_dir)
-    instance = isctest.run.get_custom_named_instance("ns3")
+    instance = isctest.instance.NamedInstance("resolver", num=3)
 
     with open(os.path.join(cfg_dir, "named.run"), "ab") as named_log:
         with subprocess.Popen(