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

Merge branch 'stepan/rndc-fixtures-for-pytest' into 'main'

Browse Source 
Implement Python helpers for using RNDC in tests

See merge request isc-projects/bind9!8357
This commit is contained in:
Štěpán Balážik
2023-12-21 19:33:43 +00:00
parent e997a738d6 fdc45c70e8
commit 46b7cc2ed2
9 changed files with 914 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.gitlab-ci.yml
View File

@@ -579,6 +579,8 @@ coccinelle:
pylint:
<<: *precheck_job
needs: []
variables:
PYTHONPATH: "${CI_PROJECT_DIR}/bin/tests/system"
script:
- pylint --rcfile $CI_PROJECT_DIR/.pylintrc $(git ls-files '*.py' | grep -vE '(ans\.py|dangerfile\.py|^bin/tests/system/)')
# Ignore Pylint wrong-import-position error in system test to enable use of pytest.importorskip
@@ -629,6 +631,11 @@ checkbashisms:
script:
- checkbashisms $(find . -path './.git' -prune -o -type f -exec sh -c 'head -n 1 "{}" | grep -qsF "#!/bin/sh"' \; -print)
mypy:
<<: *precheck_job
script:
- mypy "bin/tests/system/isctest/"
tarball-create:
stage: precheck
<<: *base_image

35
bin/tests/system/addzone/tests_rndc_deadlock.py
View File

@@ -10,25 +10,12 @@
# information regarding copyright ownership.
import concurrent.futures
import os
import subprocess
import time
def run_rndc(server, rndc_command):
"""
Send the specified 'rndc_command' to 'server' with a timeout of 10 seconds
"""
rndc = os.getenv("RNDC")
port = os.getenv("CONTROLPORT")
cmdline = [rndc, "-c", "../_common/rndc.conf", "-p", port, "-s", server]
cmdline.extend(rndc_command)
subprocess.check_output(cmdline, stderr=subprocess.STDOUT, timeout=10)
import isctest
def rndc_loop(test_state, domain):
def rndc_loop(test_state, domain, ns3):
"""
Run "rndc addzone", "rndc modzone", and "rndc delzone" in a tight loop
until the test is considered finished, ignoring errors
@@ -45,35 +32,33 @@ def rndc_loop(test_state, domain):
while not test_state["finished"]:
for command in rndc_commands:
try:
run_rndc("10.53.0.3", command)
except subprocess.SubprocessError:
pass
ns3.rndc(" ".join(command), ignore_errors=True, log=False)
def check_if_server_is_responsive():
def check_if_server_is_responsive(ns3):
"""
Check if server status can be successfully retrieved using "rndc status"
"""
try:
run_rndc("10.53.0.3", ["status"])
ns3.rndc("status", log=False)
return True
except subprocess.SubprocessError:
except isctest.rndc.RNDCException:
return False
def test_rndc_deadlock():
def test_rndc_deadlock(servers):
"""
Test whether running "rndc addzone", "rndc modzone", and "rndc delzone"
commands concurrently does not trigger a deadlock
"""
test_state = {"finished": False}
ns3 = servers["ns3"]
# Create 4 worker threads running "rndc" commands in a loop.
with concurrent.futures.ThreadPoolExecutor() as executor:
for i in range(1, 5):
domain = "example%d" % i
executor.submit(rndc_loop, test_state, domain)
executor.submit(rndc_loop, test_state, domain, ns3)
# Run "rndc status" 10 times, with 1-second pauses between attempts.
# Each "rndc status" invocation has a timeout of 10 seconds. If any of
@@ -81,7 +66,7 @@ def test_rndc_deadlock():
server_is_responsive = True
attempts = 10
while server_is_responsive and attempts > 0:
server_is_responsive = check_if_server_is_responsive()
server_is_responsive = check_if_server_is_responsive(ns3)
attempts -= 1
time.sleep(1)

711
bin/tests/system/checkds/tests_checkds.py
View File

@@ -11,7 +11,8 @@
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
import mmap
from typing import NamedTuple, Tuple
import os
import subprocess
import sys
@@ -27,7 +28,6 @@ import dns.query
import dns.rcode
import dns.rdataclass
import dns.rdatatype
import dns.resolver
pytestmark = pytest.mark.skipif(
@@ -43,8 +43,8 @@ def has_signed_apex_nsec(zone, response):
nextname = "a."
labelcount = zone.count(".") # zone is specified as FQDN
types = "NS SOA RRSIG NSEC DNSKEY"
match = "{0} {1} IN NSEC {2}{0} {3}".format(zone, ttl, nextname, types)
sig = "{0} {1} IN RRSIG NSEC 13 {2} 300".format(zone, ttl, labelcount)
match = f"{zone} {ttl} IN NSEC {nextname}{zone} {types}"
sig = f"{zone} {ttl} IN RRSIG NSEC 13 {labelcount} 300"
for rr in response.answer:
if match in rr.to_text():
@@ -64,19 +64,11 @@ def do_query(server, qname, qtype, tcp=False):
query = dns.message.make_query(qname, qtype, use_edns=True, want_dnssec=True)
try:
if tcp:
response = dns.query.tcp(
query, server.nameservers[0], timeout=3, port=server.port
)
response = dns.query.tcp(query, server.ip, timeout=3, port=server.ports.dns)
else:
response = dns.query.udp(
query, server.nameservers[0], timeout=3, port=server.port
)
response = dns.query.udp(query, server.ip, timeout=3, port=server.ports.dns)
except dns.exception.Timeout:
print(
"error: query timeout for query {} {} to {}".format(
qname, qtype, server.nameservers[0]
)
)
print(f"error: query timeout for query {qname} {qtype} to {server.ip}")
return None
return response
@@ -86,7 +78,7 @@ def verify_zone(zone, transfer):
verify = os.getenv("VERIFY")
assert verify is not None
filename = "{}out".format(zone)
filename = f"{zone}out"
with open(filename, "w", encoding="utf-8") as file:
for rr in transfer.answer:
file.write(rr.to_text())
@@ -98,21 +90,21 @@ def verify_zone(zone, transfer):
verifier = subprocess.run(verify_cmd, capture_output=True, check=True)
if verifier.returncode != 0:
print("error: dnssec-verify {} failed".format(zone))
print(f"error: dnssec-verify {zone} failed")
sys.stderr.buffer.write(verifier.stderr)
return verifier.returncode == 0
def read_statefile(server, zone):
addr = server.nameservers[0]
addr = server.ip
count = 0
keyid = 0
state = {}
response = do_query(server, zone, "DS", tcp=True)
if not isinstance(response, dns.message.Message):
print("error: no response for {} DS from {}".format(zone, addr))
print(f"error: no response for {zone} DS from {addr}")
return {}
if response.rcode() == dns.rcode.NOERROR:
@@ -130,20 +122,16 @@ def read_statefile(server, zone):
if count != 1:
print(
"error: expected a single DS in response for {} from {},"
"got {}".format(zone, addr, count)
f"error: expected a single DS in response for {zone} from {addr}, got {count}"
)
return {}
else:
print(
"error: {} response for {} DNSKEY from {}".format(
dns.rcode.to_text(response.rcode()), zone, addr
)
)
rcode = dns.rcode.to_text(response.rcode())
print(f"error: {rcode} response for {zone} DNSKEY from {addr}")
return {}
filename = "ns9/K{}+013+{:05d}.state".format(zone, keyid)
print("read state file {}".format(filename))
filename = f"ns9/K{zone}+013+{keyid:05d}.state"
print(f"read state file {filename}")
try:
with open(filename, "r", encoding="utf-8") as file:
@@ -161,23 +149,20 @@ def read_statefile(server, zone):
def zone_check(server, zone):
addr = server.nameservers[0]
fqdn = "{}.".format(zone)
addr = server.ip
fqdn = f"{zone}."
# wait until zone is fully signed.
signed = False
for _ in range(10):
response = do_query(server, fqdn, "NSEC")
if not isinstance(response, dns.message.Message):
print("error: no response for {} NSEC from {}".format(fqdn, addr))
print(f"error: no response for {fqdn} NSEC from {addr}")
elif response.rcode() == dns.rcode.NOERROR:
signed = has_signed_apex_nsec(fqdn, response)
else:
print(
"error: {} response for {} NSEC from {}".format(
dns.rcode.to_text(response.rcode()), fqdn, addr
)
)
rcode = dns.rcode.to_text(response.rcode())
print(f"error: {rcode} response for {fqdn} NSEC from {addr}")
if signed:
break
@@ -190,21 +175,18 @@ def zone_check(server, zone):
verified = False
transfer = do_query(server, fqdn, "AXFR", tcp=True)
if not isinstance(transfer, dns.message.Message):
print("error: no response for {} AXFR from {}".format(fqdn, addr))
print(f"error: no response for {fqdn} AXFR from {addr}")
elif transfer.rcode() == dns.rcode.NOERROR:
verified = verify_zone(fqdn, transfer)
else:
print(
"error: {} response for {} AXFR from {}".format(
dns.rcode.to_text(transfer.rcode()), fqdn, addr
)
)
rcode = dns.rcode.to_text(transfer.rcode())
print(f"error: {rcode} response for {fqdn} AXFR from {addr}")
assert verified
def keystate_check(server, zone, key):
fqdn = "{}.".format(zone)
fqdn = f"{zone}."
val = 0
deny = False
@@ -255,420 +237,273 @@ def rekey(zone):
controller = subprocess.run(rndc_cmd, capture_output=True, check=True)
if controller.returncode != 0:
print("error: rndc loadkeys {} failed".format(zone))
print(f"error: rndc loadkeys {zone} failed")
sys.stderr.buffer.write(controller.stderr)
assert controller.returncode == 0
def wait_for_log(filename, zone, log):
found = False
for _ in range(10):
print("read log file {}".format(filename))
try:
with open(filename, "r", encoding="utf-8") as file:
s = mmap.mmap(file.fileno(), 0, access=mmap.ACCESS_READ)
if s.find(bytes(log, "ascii")) != -1:
found = True
except FileNotFoundError:
print("file not found {}".format(filename))
if found:
break
print("rekey")
rekey(zone)
print("sleep")
time.sleep(1)
assert found
class CheckDSTest(NamedTuple):
zone: str
logs_to_wait_for: Tuple[str]
expected_parent_state: str
def checkds_dspublished(named_port, checkds, addr):
# We create resolver instances that will be used to send queries.
server = dns.resolver.Resolver()
server.nameservers = ["10.53.0.9"]
server.port = named_port
parent = dns.resolver.Resolver()
parent.nameservers = ["10.53.0.2"]
parent.port = named_port
#
# 1.1.1: DS is correctly published in parent.
# parental-agents: ns2
#
# The simple case.
zone = "good.{}.dspublish.ns2".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from {}".format(zone, addr),
)
keystate_check(parent, zone, "DSPublish")
#
# 1.1.2: DS is not published in parent.
# parental-agents: ns5
#
zone = "not-yet.{}.dspublish.ns5".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from 10.53.0.5".format(zone),
)
keystate_check(parent, zone, "!DSPublish")
#
# 1.1.3: The parental agent is badly configured.
# parental-agents: ns6
#
zone = "bad.{}.dspublish.ns6".format(checkds)
zone_check(server, zone)
if checkds == "explicit":
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: bad DS response from 10.53.0.6".format(zone),
)
elif checkds == "yes":
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: error during parental-agents processing".format(
zone
),
)
keystate_check(parent, zone, "!DSPublish")
#
# 1.1.4: DS is published, but has bogus signature.
#
# TBD
#
# 1.2.1: DS is correctly published in all parents.
# parental-agents: ns2, ns4
#
zone = "good.{}.dspublish.ns2-4".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from {}".format(zone, addr),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.4".format(zone),
)
keystate_check(parent, zone, "DSPublish")
#
# 1.2.2: DS is not published in some parents.
# parental-agents: ns2, ns4, ns5
#
zone = "incomplete.{}.dspublish.ns2-4-5".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from {}".format(zone, addr),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.4".format(zone),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from 10.53.0.5".format(zone),
)
keystate_check(parent, zone, "!DSPublish")
#
# 1.2.3: One parental agent is badly configured.
# parental-agents: ns2, ns4, ns6
#
zone = "bad.{}.dspublish.ns2-4-6".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from {}".format(zone, addr),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.4".format(zone),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: bad DS response from 10.53.0.6".format(zone),
)
keystate_check(parent, zone, "!DSPublish")
#
# 1.2.4: DS is completely published, bogus signature.
#
# TBD
# TBD: Check with TSIG
# TBD: Check with TLS
def checkds_dswithdrawn(named_port, checkds, addr):
# We create resolver instances that will be used to send queries.
server = dns.resolver.Resolver()
server.nameservers = ["10.53.0.9"]
server.port = named_port
parent = dns.resolver.Resolver()
parent.nameservers = ["10.53.0.2"]
parent.port = named_port
#
# 2.1.1: DS correctly withdrawn from the parent.
# parental-agents: ns5
#
# The simple case.
zone = "good.{}.dsremoved.ns5".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from {}".format(zone, addr),
)
keystate_check(parent, zone, "DSRemoved")
#
# 2.1.2: DS is published in the parent.
# parental-agents: ns2
#
zone = "still-there.{}.dsremoved.ns2".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.2".format(zone),
)
keystate_check(parent, zone, "!DSRemoved")
#
# 2.1.3: The parental agent is badly configured.
# parental-agents: ns6
#
zone = "bad.{}.dsremoved.ns6".format(checkds)
zone_check(server, zone)
if checkds == "explicit":
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: bad DS response from 10.53.0.6".format(zone),
)
elif checkds == "yes":
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: error during parental-agents processing".format(
zone
),
)
keystate_check(parent, zone, "!DSRemoved")
#
# 2.1.4: DS is withdrawn, but has bogus signature.
#
# TBD
#
# 2.2.1: DS is correctly withdrawn from all parents.
# parental-agents: ns5, ns7
#
zone = "good.{}.dsremoved.ns5-7".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from {}".format(zone, addr),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from 10.53.0.7".format(zone),
)
keystate_check(parent, zone, "DSRemoved")
#
# 2.2.2: DS is not withdrawn from some parents.
# parental-agents: ns2, ns5, ns7
#
zone = "incomplete.{}.dsremoved.ns2-5-7".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.2".format(zone),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from {}".format(zone, addr),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from 10.53.0.7".format(zone),
)
keystate_check(parent, zone, "!DSRemoved")
#
# 2.2.3: One parental agent is badly configured.
# parental-agents: ns5, ns6, ns7
#
zone = "bad.{}.dsremoved.ns5-6-7".format(checkds)
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from {}".format(zone, addr),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from 10.53.0.7".format(zone),
)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: bad DS response from 10.53.0.6".format(zone),
)
keystate_check(parent, zone, "!DSRemoved")
#
# 2.2.4:: DS is removed completely, bogus signature.
#
# TBD
def test_checkds_reference(named_port):
# We create resolver instances that will be used to send queries.
server = dns.resolver.Resolver()
server.nameservers = ["10.53.0.9"]
server.port = named_port
parent = dns.resolver.Resolver()
parent.nameservers = ["10.53.0.2"]
parent.port = named_port
parental_agents_tests = [
# Using a reference to parental-agents.
zone = "reference.explicit.dspublish.ns2"
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.8".format(zone),
)
keystate_check(parent, zone, "DSPublish")
def test_checkds_resolver(named_port):
# We create resolver instances that will be used to send queries.
server = dns.resolver.Resolver()
server.nameservers = ["10.53.0.9"]
server.port = named_port
parent = dns.resolver.Resolver()
parent.nameservers = ["10.53.0.2"]
parent.port = named_port
CheckDSTest(
zone="reference.explicit.dspublish.ns2",
logs_to_wait_for=("DS response from 10.53.0.8",),
expected_parent_state="DSPublish",
),
# Using a resolver as parental-agent (ns3).
zone = "resolver.explicit.dspublish.ns2"
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.3".format(zone),
)
keystate_check(parent, zone, "DSPublish")
CheckDSTest(
zone="resolver.explicit.dspublish.ns2",
logs_to_wait_for=("DS response from 10.53.0.3",),
expected_parent_state="DSPublish",
),
# Using a resolver as parental-agent (ns3).
zone = "resolver.explicit.dsremoved.ns5"
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: empty DS response from 10.53.0.3".format(zone),
)
keystate_check(parent, zone, "DSRemoved")
CheckDSTest(
zone="resolver.explicit.dsremoved.ns5",
logs_to_wait_for=("empty DS response from 10.53.0.3",),
expected_parent_state="DSRemoved",
),
]
no_ent_tests = [
CheckDSTest(
zone="no-ent.ns2",
logs_to_wait_for=("DS response from 10.53.0.2",),
expected_parent_state="DSPublish",
),
CheckDSTest(
zone="no-ent.ns5",
logs_to_wait_for=("DS response from 10.53.0.5",),
expected_parent_state="DSRemoved",
),
]
def test_checkds_no_ent(named_port):
# We create resolver instances that will be used to send queries.
server = dns.resolver.Resolver()
server.nameservers = ["10.53.0.9"]
server.port = named_port
parent = dns.resolver.Resolver()
parent.nameservers = ["10.53.0.2"]
parent.port = named_port
zone = "no-ent.ns2"
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.2".format(zone),
)
keystate_check(parent, zone, "DSPublish")
zone = "no-ent.ns5"
zone_check(server, zone)
wait_for_log(
"ns9/named.run",
zone,
"zone {}/IN (signed): checkds: DS response from 10.53.0.5".format(zone),
)
keystate_check(parent, zone, "DSRemoved")
def dspublished_tests(checkds, addr):
return [
#
# 1.1.1: DS is correctly published in parent.
# parental-agents: ns2
#
# The simple case.
CheckDSTest(
zone=f"good.{checkds}.dspublish.ns2",
logs_to_wait_for=(f"DS response from {addr}",),
expected_parent_state="DSPublish",
),
#
# 1.1.2: DS is not published in parent.
# parental-agents: ns5
#
CheckDSTest(
zone=f"not-yet.{checkds}.dspublish.ns5",
logs_to_wait_for=("empty DS response from 10.53.0.5",),
expected_parent_state="!DSPublish",
),
#
# 1.1.3: The parental agent is badly configured.
# parental-agents: ns6
#
CheckDSTest(
zone=f"bad.{checkds}.dspublish.ns6",
logs_to_wait_for=(
"bad DS response from 10.53.0.6"
if checkds == "explicit"
else "error during parental-agents processing",
),
expected_parent_state="!DSPublish",
),
#
# 1.1.4: DS is published, but has bogus signature.
#
# TBD
#
# 1.2.1: DS is correctly published in all parents.
# parental-agents: ns2, ns4
#
CheckDSTest(
zone=f"good.{checkds}.dspublish.ns2-4",
logs_to_wait_for=(f"DS response from {addr}", "DS response from 10.53.0.4"),
expected_parent_state="DSPublish",
),
#
# 1.2.2: DS is not published in some parents.
# parental-agents: ns2, ns4, ns5
#
CheckDSTest(
zone=f"incomplete.{checkds}.dspublish.ns2-4-5",
logs_to_wait_for=(
f"DS response from {addr}",
"DS response from 10.53.0.4",
"empty DS response from 10.53.0.5",
),
expected_parent_state="!DSPublish",
),
#
# 1.2.3: One parental agent is badly configured.
# parental-agents: ns2, ns4, ns6
#
CheckDSTest(
zone=f"bad.{checkds}.dspublish.ns2-4-6",
logs_to_wait_for=(
f"DS response from {addr}",
"DS response from 10.53.0.4",
"bad DS response from 10.53.0.6",
),
expected_parent_state="!DSPublish",
),
#
# 1.2.4: DS is completely published, bogus signature.
#
# TBD
# TBD: Check with TSIG
# TBD: Check with TLS
]
def test_checkds_dspublished(named_port):
checkds_dspublished(named_port, "explicit", "10.53.0.8")
checkds_dspublished(named_port, "yes", "10.53.0.2")
def dswithdrawn_tests(checkds, addr):
return [
#
# 2.1.1: DS correctly withdrawn from the parent.
# parental-agents: ns5
#
# The simple case.
CheckDSTest(
zone=f"good.{checkds}.dsremoved.ns5",
logs_to_wait_for=(f"empty DS response from {addr}",),
expected_parent_state="DSRemoved",
),
#
# 2.1.2: DS is published in the parent.
# parental-agents: ns2
#
CheckDSTest(
zone=f"still-there.{checkds}.dsremoved.ns2",
logs_to_wait_for=("DS response from 10.53.0.2",),
expected_parent_state="!DSRemoved",
),
#
# 2.1.3: The parental agent is badly configured.
# parental-agents: ns6
#
CheckDSTest(
zone=f"bad.{checkds}.dsremoved.ns6",
logs_to_wait_for=(
"bad DS response from 10.53.0.6"
if checkds == "explicit"
else "error during parental-agents processing",
),
expected_parent_state="!DSRemoved",
),
#
# 2.1.4: DS is withdrawn, but has bogus signature.
#
# TBD
#
# 2.2.1: DS is correctly withdrawn from all parents.
# parental-agents: ns5, ns7
#
CheckDSTest(
zone=f"good.{checkds}.dsremoved.ns5-7",
logs_to_wait_for=(
f"empty DS response from {addr}",
"empty DS response from 10.53.0.7",
),
expected_parent_state="DSRemoved",
),
#
# 2.2.2: DS is not withdrawn from some parents.
# parental-agents: ns2, ns5, ns7
#
CheckDSTest(
zone=f"incomplete.{checkds}.dsremoved.ns2-5-7",
logs_to_wait_for=(
"DS response from 10.53.0.2",
f"empty DS response from {addr}",
"empty DS response from 10.53.0.7",
),
expected_parent_state="!DSRemoved",
),
#
# 2.2.3: One parental agent is badly configured.
# parental-agents: ns5, ns6, ns7
#
CheckDSTest(
zone=f"bad.{checkds}.dsremoved.ns5-6-7",
logs_to_wait_for=(
f"empty DS response from {addr}",
"empty DS response from 10.53.0.7",
"bad DS response from 10.53.0.6",
),
expected_parent_state="!DSRemoved",
),
#
# 2.2.4:: DS is removed completely, bogus signature.
#
# TBD
]
def test_checkds_dswithdrawn(named_port):
checkds_dswithdrawn(named_port, "explicit", "10.53.0.10")
checkds_dswithdrawn(named_port, "yes", "10.53.0.5")
checkds_no_tests = [
CheckDSTest(
zone="good.no.dspublish.ns2",
logs_to_wait_for=(),
expected_parent_state="!DSPublish",
),
CheckDSTest(
zone="good.no.dspublish.ns2-4",
logs_to_wait_for=(),
expected_parent_state="!DSPublish",
),
CheckDSTest(
zone="good.no.dsremoved.ns5",
logs_to_wait_for=(),
expected_parent_state="!DSRemoved",
),
CheckDSTest(
zone="good.no.dsremoved.ns5-7",
logs_to_wait_for=(),
expected_parent_state="!DSRemoved",
),
]
def test_checkds_no(named_port):
# We create resolver instances that will be used to send queries.
server = dns.resolver.Resolver()
server.nameservers = ["10.53.0.9"]
server.port = named_port
checkds_tests = (
parental_agents_tests
+ no_ent_tests
+ dspublished_tests("explicit", "10.53.0.8")
+ dspublished_tests("yes", "10.53.0.2")
+ dswithdrawn_tests("explicit", "10.53.0.10")
+ dswithdrawn_tests("yes", "10.53.0.5")
+ checkds_no_tests
)
parent = dns.resolver.Resolver()
parent.nameservers = ["10.53.0.2"]
parent.port = named_port
zone_check(server, "good.no.dspublish.ns2")
keystate_check(parent, "good.no.dspublish.ns2", "!DSPublish")
@pytest.mark.parametrize("params", checkds_tests, ids=lambda t: t.zone)
def test_checkds(servers, params):
# Wait until the provided zone is signed and then verify its DNSSEC data.
zone_check(servers["ns9"], params.zone)
zone_check(server, "good.no.dspublish.ns2-4")
keystate_check(parent, "good.no.dspublish.ns2-4", "!DSPublish")
# Wait until all the expected log lines are found in the log file for the
# provided server.
for log_string in params.logs_to_wait_for:
for _ in range(10):
with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {params.zone}/IN (signed): checkds: {log_string}"
try:
watcher.wait_for_line(line, timeout=1)
except TimeoutError:
rekey(params.zone)
else:
break
else:
raise TimeoutError
zone_check(server, "good.no.dsremoved.ns5")
keystate_check(parent, "good.no.dsremoved.ns5", "!DSRemoved")
zone_check(server, "good.no.dsremoved.ns5-7")
keystate_check(parent, "good.no.dsremoved.ns5-7", "!DSRemoved")
# Check whether key states on the parent server provided match
# expectations.
keystate_check(servers["ns2"], params.zone, params.expected_parent_state)

21
bin/tests/system/conftest.py
View File

@@ -22,9 +22,10 @@ from typing import Any, Dict, List, Optional
import pytest
pytest.register_assert_rewrite("isctest")
import isctest
# Silence warnings caused by passing a pytest fixture to another fixture.
# pylint: disable=redefined-outer-name
@@ -647,3 +648,21 @@ def system_test( # pylint: disable=too-many-arguments,too-many-statements
stop_servers()
get_core_dumps()
request.node.stash[FIXTURE_OK] = True
@pytest.fixture
def servers(ports, logger, system_test_dir):
instances = {}
for entry in system_test_dir.rglob("*"):
if entry.is_dir():
try:
dir_name = entry.name
# LATER: Make ports fixture return NamedPorts directly
named_ports = isctest.instance.NamedPorts(
dns=int(ports["PORT"]), rndc=int(ports["CONTROLPORT"])
)
instance = isctest.instance.NamedInstance(dir_name, named_ports, logger)
instances[dir_name] = instance
except ValueError:
continue
return instances

3
bin/tests/system/isctest/__init__.py
View File

@@ -10,4 +10,7 @@
# information regarding copyright ownership.
from . import check
from . import instance
from . import query
from . import rndc
from . import watchlog

168
bin/tests/system/isctest/instance.py Normal file
View File

@@ -0,0 +1,168 @@
#!/usr/bin/python3
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
#
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
from typing import NamedTuple, Optional
import logging
import os
import re
from .rndc import RNDCBinaryExecutor, RNDCException, RNDCExecutor
from .watchlog import WatchLogFromStart, WatchLogFromHere
class NamedPorts(NamedTuple):
dns: int = 53
rndc: int = 953
class NamedInstance:
"""
A class representing a `named` instance used in a system test.
This class is expected to be instantiated as part of the `servers` fixture:
```python
def test_foo(servers):
servers["ns1"].rndc("status")
```
"""
# pylint: disable=too-many-arguments
def __init__(
self,
identifier: str,
ports: NamedPorts = NamedPorts(),
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.
`ports` is the `NamedPorts` instance listing the UDP/TCP ports on which
this `named` instance is listening for various types of traffic (both
DNS traffic and RNDC commands).
`rndc_logger` is the `logging.Logger` to use for logging RNDC
commands sent to this `named` instance.
`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.ports = ports
self._log_file = os.path.join(identifier, "named.run")
self._rndc_executor = rndc_executor or RNDCBinaryExecutor()
self._rndc_logger = rndc_logger or logging.getLogger()
@staticmethod
def _identifier_to_ip(identifier: str) -> str:
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")
def rndc(self, command: str, ignore_errors: bool = False, log: bool = True) -> str:
"""
Send `command` to this named instance using RNDC. Return the server's
response.
If the RNDC command fails, an `RNDCException` is raised unless
`ignore_errors` is set to `True`.
The RNDC command will be logged to `rndc.log` (along with the server's
response) unless `log` is set to `False`.
>>> # Instances of the `NamedInstance` class are expected to be passed
>>> # to pytest tests as fixtures; here, some instances are created
>>> # directly (with a fake RNDC executor) so that doctest can work.
>>> import unittest.mock
>>> mock_rndc_executor = unittest.mock.Mock()
>>> ns1 = NamedInstance("ns1", rndc_executor=mock_rndc_executor)
>>> ns2 = NamedInstance("ns2", rndc_executor=mock_rndc_executor)
>>> ns3 = NamedInstance("ns3", rndc_executor=mock_rndc_executor)
>>> ns4 = NamedInstance("ns4", rndc_executor=mock_rndc_executor)
>>> # Send the "status" command to ns1. An `RNDCException` will be
>>> # raised if the RNDC command fails. This command will be logged.
>>> response = ns1.rndc("status")
>>> # Send the "thaw foo" command to ns2. No exception will be raised
>>> # in case the RNDC command fails. This command will be logged
>>> # (even if it fails).
>>> response = ns2.rndc("thaw foo", ignore_errors=True)
>>> # Send the "stop" command to ns3. An `RNDCException` will be
>>> # raised if the RNDC command fails, but this command will not be
>>> # logged (the server's response will still be returned to the
>>> # caller, though).
>>> response = ns3.rndc("stop", log=False)
>>> # Send the "halt" command to ns4 in "fire & forget mode": no
>>> # exceptions will be raised and no logging will take place (the
>>> # server's response will still be returned to the caller, though).
>>> response = ns4.rndc("stop", ignore_errors=True, log=False)
"""
try:
response = self._rndc_executor.call(self.ip, self.ports.rndc, command)
if log:
self._rndc_log(command, response)
except RNDCException as exc:
response = str(exc)
if log:
self._rndc_log(command, response)
if not ignore_errors:
raise
return response
def watch_log_from_start(self) -> WatchLogFromStart:
"""
Return an instance of the `WatchLogFromStart` context manager for this
`named` instance's log file.
"""
return WatchLogFromStart(self._log_file)
def watch_log_from_here(self) -> WatchLogFromHere:
"""
Return an instance of the `WatchLogFromHere` context manager for this
`named` instance's log file.
"""
return WatchLogFromHere(self._log_file)
def reconfigure(self) -> None:
"""
Reconfigure this named `instance` and wait until reconfiguration is
finished. Raise an `RNDCException` if reconfiguration fails.
"""
with self.watch_log_from_here() as watcher:
self.rndc("reconfig")
watcher.wait_for_line("any newly configured zones are now loaded")
def _rndc_log(self, command: str, response: str) -> None:
"""
Log an `rndc` invocation (and its output) to the `rndc.log` file in the
current working directory.
"""
fmt = '%(ip)s: "%(command)s"\n%(separator)s\n%(response)s%(separator)s'
self._rndc_logger.info(
fmt,
{
"ip": self.ip,
"command": command,
"separator": "-" * 80,
"response": response,
},
)

72
bin/tests/system/isctest/rndc.py Normal file
View File

@@ -0,0 +1,72 @@
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
#
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
import abc
import os
import subprocess
# pylint: disable=too-few-public-methods
class RNDCExecutor(abc.ABC):
"""
An interface which RNDC executors have to implement in order for the
`NamedInstance` class to be able to use them.
"""
@abc.abstractmethod
def call(self, ip: str, port: int, command: str) -> str:
"""
Send RNDC `command` to the `named` instance at `ip:port` and return the
server's response.
"""
class RNDCException(Exception):
"""
Raised by classes implementing the `RNDCExecutor` interface when sending an
RNDC command fails for any reason.
"""
class RNDCBinaryExecutor(RNDCExecutor):
"""
An `RNDCExecutor` which sends RNDC commands to servers using the `rndc`
binary.
"""
def __init__(self) -> None:
"""
This class needs the `RNDC` environment variable to be set to the path
to the `rndc` binary to use.
"""
rndc_path = os.environ.get("RNDC", "/bin/false")
rndc_conf = os.path.join("..", "_common", "rndc.conf")
self._base_cmdline = [rndc_path, "-c", rndc_conf]
def call(self, ip: str, port: int, command: str) -> str:
"""
Send RNDC `command` to the `named` instance at `ip:port` and return the
server's response.
"""
cmdline = self._base_cmdline[:]
cmdline.extend(["-s", ip])
cmdline.extend(["-p", str(port)])
cmdline.extend(command.split())
try:
return subprocess.check_output(
cmdline, stderr=subprocess.STDOUT, timeout=10, encoding="utf-8"
)
except subprocess.SubprocessError as exc:
msg = getattr(exc, "output", "RNDC exception occurred")
raise RNDCException(msg) from exc

335
bin/tests/system/isctest/watchlog.py Normal file
View File

@@ -0,0 +1,335 @@
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
#
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
from typing import Optional, TextIO, Dict, Any, overload, List, Union
import abc
import os
import shlex
import subprocess
import time
class WatchLogException(Exception):
pass
class WatchLog(abc.ABC):
"""
Wait for a log message to appear in a text file.
This class should not be used directly; instead, its subclasses,
`WatchLogFromStart` and `WatchLogFromHere`, should be used. For `named`
instances used in system tests, it is recommended to use the
`watch_log_from_start()` and `watch_log_from_here()` helper methods exposed
by the `NamedInstance` class (see below for recommended usage patterns).
"""
def __init__(self, path: str) -> None:
"""
`path` is the path to the log file to watch.
Every instance of this class must call one of the `wait_for_*()`
methods exactly once or else an `Exception` is thrown.
>>> with WatchLogFromStart("/dev/null") as watcher:
... print("Just print something without waiting for a log line")
Traceback (most recent call last):
...
Exception: wait_for_*() was not called
>>> with WatchLogFromHere("/dev/null") as watcher:
... try:
... watcher.wait_for_line("foo", timeout=0)
... except TimeoutError:
... pass
... try:
... watcher.wait_for_lines({"bar": 42}, timeout=0)
... except TimeoutError:
... pass
Traceback (most recent call last):
...
Exception: wait_for_*() was already called
"""
self._fd = None # type: Optional[TextIO]
self._path = path
self._wait_function_called = False
def wait_for_line(self, string: str, timeout: int = 10) -> None:
"""
Block execution until a line containing the provided `string` appears
in the log file. Return `None` once the line is found or raise a
`TimeoutError` after `timeout` seconds (default: 10) if `string` does
not appear in the log file. (Catching this exception is discouraged as
it indicates that the test code did not behave as expected.)
Recommended use:
```python
import isctest
def test_foo(servers):
with servers["ns1"].watch_log_from_here() as watcher:
# ... do stuff here ...
watcher.wait_for_line("foo bar")
```
One of `wait_for_line()` or `wait_for_lines()` must be called exactly
once for every `WatchLogFrom*` instance.
>>> # For `WatchLogFromStart`, `wait_for_line()` returns without
>>> # raising an exception as soon as the line being looked for appears
>>> # anywhere in the file, no matter whether that happens before of
>>> # after the `with` statement is reached.
>>> import tempfile
>>> with tempfile.NamedTemporaryFile("w") as file:
... print("foo", file=file, flush=True)
... with WatchLogFromStart(file.name) as watcher:
... retval = watcher.wait_for_line("foo", timeout=1)
>>> print(retval)
None
>>> with tempfile.NamedTemporaryFile("w") as file:
... with WatchLogFromStart(file.name) as watcher:
... print("foo", file=file, flush=True)
... retval = watcher.wait_for_line("foo", timeout=1)
>>> print(retval)
None
>>> # For `WatchLogFromHere`, `wait_for_line()` only returns without
>>> # raising an exception if the string being looked for appears in
>>> # the log file after the `with` statement is reached.
>>> import tempfile
>>> with tempfile.NamedTemporaryFile("w") as file:
... print("foo", file=file, flush=True)
... with WatchLogFromHere(file.name) as watcher:
... watcher.wait_for_line("foo", timeout=1) #doctest: +ELLIPSIS
Traceback (most recent call last):
...
TimeoutError: Timeout reached watching ...
>>> with tempfile.NamedTemporaryFile("w") as file:
... print("foo", file=file, flush=True)
... with WatchLogFromHere(file.name) as watcher:
... print("foo", file=file, flush=True)
... retval = watcher.wait_for_line("foo", timeout=1)
>>> print(retval)
None
"""
return self._wait_for({string: None}, timeout)
def wait_for_lines(self, strings: Dict[str, Any], timeout: int = 10) -> None:
"""
Block execution until a line of interest appears in the log file. This
function is a "multi-match" variant of `wait_for_line()` which is
useful when some action may cause several different (mutually
exclusive) messages to appear in the log file.
`strings` is a `dict` associating each string to look for with the
value this function should return when that string is found in the log
file. If none of the `strings` being looked for appear in the log file
after `timeout` seconds, a `TimeoutError` is raised.
(Catching this exception is discouraged as it indicates that the test
code did not behave as expected.)
Since `strings` is a `dict` and preserves key order (in CPython 3.6 as
implementation detail, since 3.7 by language design), each line is
checked against each key in order until the first match. Values provided
in the `strings` dictionary (i.e. values which this function is expected
to return upon a successful match) can be of any type.
Recommended use:
```python
import isctest
def test_foo(servers):
triggers = {
"message A": "value returned when message A is found",
"message B": "value returned when message B is found",
}
with servers["ns1"].watch_log_from_here() as watcher:
# ... do stuff here ...
retval = watcher.wait_for_lines(triggers)
```
One of `wait_for_line()` or `wait_for_lines()` must be called exactly
once for every `WatchLogFromHere` instance.
>>> # Different values must be returned depending on which line is
>>> # found in the log file.
>>> import tempfile
>>> triggers = {"foo": 42, "bar": 1337}
>>> with tempfile.NamedTemporaryFile("w") as file:
... print("foo", file=file, flush=True)
... with WatchLogFromStart(file.name) as watcher:
... retval1 = watcher.wait_for_lines(triggers, timeout=1)
... with WatchLogFromHere(file.name) as watcher:
... print("bar", file=file, flush=True)
... retval2 = watcher.wait_for_lines(triggers, timeout=1)
>>> print(retval1)
42
>>> print(retval2)
1337
"""
return self._wait_for(strings, timeout)
def _wait_for(self, strings: Dict[str, Any], timeout: int) -> Any:
"""
Block execution until one of the `strings` being looked for appears in
the log file. Raise a `TimeoutError` if none of the `strings` being
looked for are found in the log file for `timeout` seconds.
"""
if self._wait_function_called:
raise WatchLogException("wait_for_*() was already called")
self._wait_function_called = True
if not self._fd:
raise WatchLogException("No file to watch")
leftover = ""
deadline = time.time() + timeout
while time.time() < deadline:
for line in self._fd.readlines():
if line[-1] != "\n":
# Line is not completely written yet, buffer and keep on waiting
leftover += line
else:
line = leftover + line
leftover = ""
for string, retval in strings.items():
if string in line:
return retval
time.sleep(0.1)
raise TimeoutError(
"Timeout reached watching {} for {}".format(
self._path, list(strings.keys())
)
)
def __enter__(self) -> Any:
self._fd = open(self._path, encoding="utf-8")
self._seek_on_enter()
return self
@abc.abstractmethod
def _seek_on_enter(self) -> None:
"""
This method is responsible for setting the file position indicator for
the file being watched when execution reaches the __enter__() method.
It is expected to be set differently depending on which `WatchLog`
subclass is used. Since the base `WatchLog` class should not be used
directly, raise an exception upon any attempt of such use.
"""
raise NotImplementedError
def __exit__(self, *_: Any) -> None:
if not self._wait_function_called:
raise WatchLogException("wait_for_*() was not called")
if self._fd:
self._fd.close()
class WatchLogFromStart(WatchLog):
"""
A `WatchLog` subclass which looks for the provided string(s) in the entire
log file.
"""
def _seek_on_enter(self) -> None:
pass
class WatchLogFromHere(WatchLog):
"""
A `WatchLog` subclass which only looks for the provided string(s) in the
portion of the log file which is appended to it after the `with` statement
is reached.
"""
def _seek_on_enter(self) -> None:
if self._fd:
self._fd.seek(0, os.SEEK_END)
# pylint: disable=too-few-public-methods
class RNDCExecutor(abc.ABC):
"""
An interface which RNDC executors have to implement in order for the
`NamedInstance` class to be able to use them.
"""
@overload
def call(self, ip: str, port: int, command: str, timeout: int = 10) -> str:
...
@overload
def call(self, ip: str, port: int, command: List[str], timeout: int = 10) -> str:
...
@abc.abstractmethod
def call(
self, ip: str, port: int, command: Union[str, List[str]], timeout: int = 10
) -> str:
"""
Send RNDC `command` to the `named` instance at `ip:port` and return the
server's response.
"""
class RNDCException(Exception):
"""
Raised by classes implementing the `RNDCExecutor` interface when sending an
RNDC command fails for any reason.
"""
class RNDCBinaryExecutor(RNDCExecutor):
"""
An `RNDCExecutor` which sends RNDC commands to servers using the `rndc`
binary.
"""
def __init__(self) -> None:
"""
This class needs the `RNDC` environment variable to be set to the path
to the `rndc` binary to use.
"""
rndc_path = os.environ.get("RNDC", "/bin/false")
rndc_conf = os.path.join("..", "_common", "rndc.conf")
self._base_cmdline = [rndc_path, "-c", rndc_conf]
@overload
def call(self, ip: str, port: int, command: str, timeout: int = 10) -> str:
...
@overload
def call(self, ip: str, port: int, command: List[str], timeout: int = 10) -> str:
...
def call(
self, ip: str, port: int, command: Union[str, List[str]], timeout: int = 10
) -> str:
"""
Send RNDC `command` to the `named` instance at `ip:port` and return the
server's response.
"""
cmdline = self._base_cmdline[:]
cmdline.extend(["-s", ip])
cmdline.extend(["-p", str(port)])
cmdline.extend(shlex.split(command) if isinstance(command, str) else command)
try:
return subprocess.check_output(
cmdline, stderr=subprocess.STDOUT, timeout=timeout, encoding="utf-8"
)
except subprocess.SubprocessError as exc:
msg = getattr(exc, "output", "RNDC exception occurred")
raise RNDCException(msg) from exc

46
bin/tests/system/shutdown/tests_shutdown.py
View File

@@ -25,8 +25,10 @@ pytest.importorskip("dns", minversion="2.0.0")
import dns.exception
import dns.resolver
import isctest
def do_work(named_proc, resolver, rndc_cmd, kill_method, n_workers, n_queries):
def do_work(named_proc, resolver, instance, kill_method, n_workers, n_queries):
"""Creates a number of A queries to run in parallel
in order simulate a slightly more realistic test scenario.
@@ -48,8 +50,8 @@ def do_work(named_proc, resolver, rndc_cmd, kill_method, n_workers, n_queries):
:param resolver: target resolver
:type resolver: dns.resolver.Resolver
:param rndc_cmd: rndc command with default arguments
:type rndc_cmd: list of strings, e.g. ["rndc", "-p", "23750"]
:param instance: the named instance to send RNDC commands to
:type instance: isctest.instance.NamedInstance
:kill_method: "rndc" or "sigterm"
:type kill_method: str
@@ -63,9 +65,13 @@ def do_work(named_proc, resolver, rndc_cmd, kill_method, n_workers, n_queries):
# pylint: disable-msg=too-many-arguments
# pylint: disable-msg=too-many-locals
# helper function, args must be a list or tuple with arguments to rndc.
def launch_rndc(args):
return subprocess.call(rndc_cmd + args, timeout=10)
# helper function, 'command' is the rndc command to run
def launch_rndc(command):
try:
instance.rndc(command, log=False)
return 0
except isctest.rndc.RNDCException:
return -1
# We're going to execute queries in parallel by means of a thread pool.
# dnspython functions block, so we need to circunvent that.
@@ -99,13 +105,13 @@ def do_work(named_proc, resolver, rndc_cmd, kill_method, n_workers, n_queries):
elif shutdown: # We attempt to stop named in the middle
shutdown = False
if kill_method == "rndc":
futures[executor.submit(launch_rndc, ["stop"])] = "stop"
futures[executor.submit(launch_rndc, "stop")] = "stop"
else:
futures[executor.submit(named_proc.terminate)] = "kill"
else:
# We attempt to send couple rndc commands while named is
# being shutdown
futures[executor.submit(launch_rndc, ["-t", "5", "status"])] = "status"
futures[executor.submit(launch_rndc, "-t 5 status")] = "status"
ret_code = -1
for future in as_completed(futures):
@@ -164,7 +170,7 @@ def wait_for_proc_termination(proc, max_timeout=10):
"kill_method",
[pytest.param("rndc", marks=pytest.mark.xfail(reason="GL#4060")), "sigterm"],
)
def test_named_shutdown(named_port, control_port, kill_method):
def test_named_shutdown(ports, kill_method):
# pylint: disable-msg=too-many-locals
cfg_dir = os.path.join(os.getcwd(), "resolver")
assert os.path.isdir(cfg_dir)
@@ -175,20 +181,20 @@ def test_named_shutdown(named_port, control_port, kill_method):
named = os.getenv("NAMED")
assert named is not None
rndc = os.getenv("RNDC")
assert rndc is not None
# rndc configuration resides in ../_common/rndc.conf
rndc_cfg = os.path.join("..", "_common", "rndc.conf")
assert os.path.isfile(rndc_cfg)
# rndc command with default arguments.
rndc_cmd = [rndc, "-c", rndc_cfg, "-p", str(control_port), "-s", "10.53.0.3"]
# 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(
dns=ports["PORT"], rndc=ports["CONTROLPORT"]
)
instance = isctest.instance.NamedInstance("ns3", named_ports)
# We create a resolver instance that will be used to send queries.
resolver = dns.resolver.Resolver()
resolver.nameservers = ["10.53.0.3"]
resolver.port = named_port
resolver.port = named_ports.dns
named_cmdline = [named, "-c", cfg_file, "-f"]
with subprocess.Popen(named_cmdline, cwd=cfg_dir) as named_proc:
@@ -198,7 +204,7 @@ def test_named_shutdown(named_port, control_port, kill_method):
do_work(
named_proc,
resolver,
rndc_cmd,
instance,
kill_method,
n_workers=12,
n_queries=16,