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

Clean up the "checkds" system test

Browse Source 
The "checkds" system test contains a lot of duplicated code despite
carrying out the same set of actions for every tested scenario
(zone_check() → wait for logs to appear → keystate_check()).  Extract
the parts of the code shared between all tests into a new function,
test_checkds(), and use pytest's test parametrization capabilities to
pass distinct sets of test parameters to this new function, in an
attempt to cleanly separate the fixed parts of this system test from the
variable ones.  Replace format() calls with f-strings.
This commit is contained in:
Michał Kępień
2023-07-25 14:37:05 +02:00
committed by Štěpán Balážik
parent cf338a7ca3
commit aa31a872d0
1 changed files with 269 additions and 282 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -11,6 +11,8 @@
# See the COPYRIGHT file distributed with this work for additional # See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership. # information regarding copyright ownership.
from typing import NamedTuple, Tuple
import os import os
import subprocess import subprocess
import sys import sys
@@ -41,8 +43,8 @@ def has_signed_apex_nsec(zone, response):
nextname = "a." nextname = "a."
labelcount = zone.count(".") # zone is specified as FQDN labelcount = zone.count(".") # zone is specified as FQDN
types = "NS SOA RRSIG NSEC DNSKEY" types = "NS SOA RRSIG NSEC DNSKEY"
match = "{0} {1} IN NSEC {2}{0} {3}".format(zone, ttl, nextname, types) match = f"{zone} {ttl} IN NSEC {nextname}{zone} {types}"
sig = "{0} {1} IN RRSIG NSEC 13 {2} 300".format(zone, ttl, labelcount) sig = f"{zone} {ttl} IN RRSIG NSEC 13 {labelcount} 300"
for rr in response.answer: for rr in response.answer:
if match in rr.to_text(): if match in rr.to_text():
@@ -76,7 +78,7 @@ def verify_zone(zone, transfer):
verify = os.getenv("VERIFY") verify = os.getenv("VERIFY")
assert verify is not None assert verify is not None
filename = "{}out".format(zone) filename = f"{zone}out"
with open(filename, "w", encoding="utf-8") as file: with open(filename, "w", encoding="utf-8") as file:
for rr in transfer.answer: for rr in transfer.answer:
file.write(rr.to_text()) file.write(rr.to_text())
@@ -88,7 +90,7 @@ def verify_zone(zone, transfer):
verifier = subprocess.run(verify_cmd, capture_output=True, check=True) verifier = subprocess.run(verify_cmd, capture_output=True, check=True)
if verifier.returncode != 0: if verifier.returncode != 0:
print("error: dnssec-verify {} failed".format(zone)) print(f"error: dnssec-verify {zone} failed")
sys.stderr.buffer.write(verifier.stderr) sys.stderr.buffer.write(verifier.stderr)
return verifier.returncode == 0 return verifier.returncode == 0
@@ -102,7 +104,7 @@ def read_statefile(server, zone):
response = do_query(server, zone, "DS", tcp=True) response = do_query(server, zone, "DS", tcp=True)
if not isinstance(response, dns.message.Message): 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 {} return {}
if response.rcode() == dns.rcode.NOERROR: if response.rcode() == dns.rcode.NOERROR:
@@ -120,20 +122,16 @@ def read_statefile(server, zone):
if count != 1: if count != 1:
print( print(
"error: expected a single DS in response for {} from {}," f"error: expected a single DS in response for {zone} from {addr}, got {count}"
"got {}".format(zone, addr, count)
) )
return {} return {}
else: else:
print( rcode = dns.rcode.to_text(response.rcode())
"error: {} response for {} DNSKEY from {}".format( print(f"error: {rcode} response for {zone} DNSKEY from {addr}")
dns.rcode.to_text(response.rcode()), zone, addr
)
)
return {} return {}
filename = "ns9/K{}+013+{:05d}.state".format(zone, keyid) filename = f"ns9/K{zone}+013+{keyid:05d}.state"
print("read state file {}".format(filename)) print(f"read state file {filename}")
try: try:
with open(filename, "r", encoding="utf-8") as file: with open(filename, "r", encoding="utf-8") as file:
@@ -152,22 +150,19 @@ def read_statefile(server, zone):
def zone_check(server, zone): def zone_check(server, zone):
addr = server.ip addr = server.ip
fqdn = "{}.".format(zone) fqdn = f"{zone}."
# wait until zone is fully signed. # wait until zone is fully signed.
signed = False signed = False
for _ in range(10): for _ in range(10):
response = do_query(server, fqdn, "NSEC") response = do_query(server, fqdn, "NSEC")
if not isinstance(response, dns.message.Message): 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: elif response.rcode() == dns.rcode.NOERROR:
signed = has_signed_apex_nsec(fqdn, response) signed = has_signed_apex_nsec(fqdn, response)
else: else:
print( rcode = dns.rcode.to_text(response.rcode())
"error: {} response for {} NSEC from {}".format( print(f"error: {rcode} response for {fqdn} NSEC from {addr}")
dns.rcode.to_text(response.rcode()), fqdn, addr
)
)
if signed: if signed:
break break
@@ -180,21 +175,18 @@ def zone_check(server, zone):
verified = False verified = False
transfer = do_query(server, fqdn, "AXFR", tcp=True) transfer = do_query(server, fqdn, "AXFR", tcp=True)
if not isinstance(transfer, dns.message.Message): 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: elif transfer.rcode() == dns.rcode.NOERROR:
verified = verify_zone(fqdn, transfer) verified = verify_zone(fqdn, transfer)
else: else:
print( rcode = dns.rcode.to_text(transfer.rcode())
"error: {} response for {} AXFR from {}".format( print(f"error: {rcode} response for {fqdn} AXFR from {addr}")
dns.rcode.to_text(transfer.rcode()), fqdn, addr
)
)
assert verified assert verified
def keystate_check(server, zone, key): def keystate_check(server, zone, key):
fqdn = "{}.".format(zone) fqdn = f"{zone}."
val = 0 val = 0
deny = False deny = False
@@ -245,278 +237,273 @@ def rekey(zone):
controller = subprocess.run(rndc_cmd, capture_output=True, check=True) controller = subprocess.run(rndc_cmd, capture_output=True, check=True)
if controller.returncode != 0: if controller.returncode != 0:
print("error: rndc loadkeys {} failed".format(zone)) print(f"error: rndc loadkeys {zone} failed")
sys.stderr.buffer.write(controller.stderr) sys.stderr.buffer.write(controller.stderr)
assert controller.returncode == 0 assert controller.returncode == 0
def checkds_dspublished(named_port, servers, checkds, addr): class CheckDSTest(NamedTuple):
zone: str
logs_to_wait_for: Tuple[str]
expected_parent_state: str
parental_agents_tests = [
# Using a reference to parental-agents.
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).
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).
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 dspublished_tests(checkds, addr):
return [
# #
# 1.1.1: DS is correctly published in parent. # 1.1.1: DS is correctly published in parent.
# parental-agents: ns2 # parental-agents: ns2
# #
# The simple case. # The simple case.
zone = "good.{}.dspublish.ns2".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"good.{checkds}.dspublish.ns2",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(f"DS response from {addr}",),
line = f"zone {zone}/IN (signed): checkds: DS response from {addr}" expected_parent_state="DSPublish",
watcher.wait_for_line(line) ),
keystate_check(servers["ns2"], zone, "DSPublish")
# #
# 1.1.2: DS is not published in parent. # 1.1.2: DS is not published in parent.
# parental-agents: ns5 # parental-agents: ns5
# #
zone = "not-yet.{}.dspublish.ns5".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"not-yet.{checkds}.dspublish.ns5",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=("empty DS response from 10.53.0.5",),
line = f"zone {zone}/IN (signed): checkds: empty DS response from 10.53.0.5" expected_parent_state="!DSPublish",
watcher.wait_for_line(line) ),
keystate_check(servers["ns2"], zone, "!DSPublish")
# #
# 1.1.3: The parental agent is badly configured. # 1.1.3: The parental agent is badly configured.
# parental-agents: ns6 # parental-agents: ns6
# #
zone = "bad.{}.dspublish.ns6".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"bad.{checkds}.dspublish.ns6",
if checkds == "explicit": logs_to_wait_for=(
with servers["ns9"].watch_log_from_start() as watcher: "bad DS response from 10.53.0.6"
line = f"zone {zone}/IN (signed): checkds: bad DS response from 10.53.0.6" if checkds == "explicit"
watcher.wait_for_line(line) else "error during parental-agents processing",
elif checkds == "yes": ),
with servers["ns9"].watch_log_from_start() as watcher: expected_parent_state="!DSPublish",
line = f"zone {zone}/IN (signed): checkds: error during parental-agents processing" ),
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "!DSPublish")
# #
# 1.1.4: DS is published, but has bogus signature. # 1.1.4: DS is published, but has bogus signature.
# #
# TBD # TBD
# #
# 1.2.1: DS is correctly published in all parents. # 1.2.1: DS is correctly published in all parents.
# parental-agents: ns2, ns4 # parental-agents: ns2, ns4
# #
zone = "good.{}.dspublish.ns2-4".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"good.{checkds}.dspublish.ns2-4",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(f"DS response from {addr}", "DS response from 10.53.0.4"),
line = f"zone {zone}/IN (signed): checkds: DS response from {addr}" expected_parent_state="DSPublish",
watcher.wait_for_line(line) ),
with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.4"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "DSPublish")
# #
# 1.2.2: DS is not published in some parents. # 1.2.2: DS is not published in some parents.
# parental-agents: ns2, ns4, ns5 # parental-agents: ns2, ns4, ns5
# #
zone = "incomplete.{}.dspublish.ns2-4-5".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"incomplete.{checkds}.dspublish.ns2-4-5",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(
line = f"zone {zone}/IN (signed): checkds: DS response from {addr}" f"DS response from {addr}",
watcher.wait_for_line(line) "DS response from 10.53.0.4",
with servers["ns9"].watch_log_from_start() as watcher: "empty DS response from 10.53.0.5",
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.4" ),
watcher.wait_for_line(line) expected_parent_state="!DSPublish",
with servers["ns9"].watch_log_from_start() as watcher: ),
line = f"zone {zone}/IN (signed): checkds: empty DS response from 10.53.0.5"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "!DSPublish")
# #
# 1.2.3: One parental agent is badly configured. # 1.2.3: One parental agent is badly configured.
# parental-agents: ns2, ns4, ns6 # parental-agents: ns2, ns4, ns6
# #
zone = "bad.{}.dspublish.ns2-4-6".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"bad.{checkds}.dspublish.ns2-4-6",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(
line = f"zone {zone}/IN (signed): checkds: DS response from {addr}" f"DS response from {addr}",
watcher.wait_for_line(line) "DS response from 10.53.0.4",
with servers["ns9"].watch_log_from_start() as watcher: "bad DS response from 10.53.0.6",
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.4" ),
watcher.wait_for_line(line) expected_parent_state="!DSPublish",
with servers["ns9"].watch_log_from_start() as watcher: ),
line = f"zone {zone}/IN (signed): checkds: bad DS response from 10.53.0.6"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "!DSPublish")
# #
# 1.2.4: DS is completely published, bogus signature. # 1.2.4: DS is completely published, bogus signature.
# #
# TBD # TBD
# TBD: Check with TSIG # TBD: Check with TSIG
# TBD: Check with TLS # TBD: Check with TLS
]
def checkds_dswithdrawn(named_port, servers, checkds, addr): def dswithdrawn_tests(checkds, addr):
return [
# #
# 2.1.1: DS correctly withdrawn from the parent. # 2.1.1: DS correctly withdrawn from the parent.
# parental-agents: ns5 # parental-agents: ns5
# #
# The simple case. # The simple case.
zone = "good.{}.dsremoved.ns5".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"good.{checkds}.dsremoved.ns5",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(f"empty DS response from {addr}",),
line = f"zone {zone}/IN (signed): checkds: empty DS response from {addr}" expected_parent_state="DSRemoved",
watcher.wait_for_line(line) ),
keystate_check(servers["ns2"], zone, "DSRemoved")
# #
# 2.1.2: DS is published in the parent. # 2.1.2: DS is published in the parent.
# parental-agents: ns2 # parental-agents: ns2
# #
zone = "still-there.{}.dsremoved.ns2".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"still-there.{checkds}.dsremoved.ns2",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=("DS response from 10.53.0.2",),
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.2" expected_parent_state="!DSRemoved",
watcher.wait_for_line(line) ),
keystate_check(servers["ns2"], zone, "!DSRemoved")
# #
# 2.1.3: The parental agent is badly configured. # 2.1.3: The parental agent is badly configured.
# parental-agents: ns6 # parental-agents: ns6
# #
zone = "bad.{}.dsremoved.ns6".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"bad.{checkds}.dsremoved.ns6",
if checkds == "explicit": logs_to_wait_for=(
with servers["ns9"].watch_log_from_start() as watcher: "bad DS response from 10.53.0.6"
line = f"zone {zone}/IN (signed): checkds: bad DS response from 10.53.0.6" if checkds == "explicit"
watcher.wait_for_line(line) else "error during parental-agents processing",
elif checkds == "yes": ),
with servers["ns9"].watch_log_from_start() as watcher: expected_parent_state="!DSRemoved",
line = f"zone {zone}/IN (signed): checkds: error during parental-agents processing" ),
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "!DSRemoved")
# #
# 2.1.4: DS is withdrawn, but has bogus signature. # 2.1.4: DS is withdrawn, but has bogus signature.
# #
# TBD # TBD
# #
# 2.2.1: DS is correctly withdrawn from all parents. # 2.2.1: DS is correctly withdrawn from all parents.
# parental-agents: ns5, ns7 # parental-agents: ns5, ns7
# #
zone = "good.{}.dsremoved.ns5-7".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"good.{checkds}.dsremoved.ns5-7",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(
line = f"zone {zone}/IN (signed): checkds: empty DS response from {addr}" f"empty DS response from {addr}",
watcher.wait_for_line(line) "empty DS response from 10.53.0.7",
with servers["ns9"].watch_log_from_start() as watcher: ),
line = f"zone {zone}/IN (signed): checkds: empty DS response from 10.53.0.7" expected_parent_state="DSRemoved",
watcher.wait_for_line(line) ),
keystate_check(servers["ns2"], zone, "DSRemoved")
# #
# 2.2.2: DS is not withdrawn from some parents. # 2.2.2: DS is not withdrawn from some parents.
# parental-agents: ns2, ns5, ns7 # parental-agents: ns2, ns5, ns7
# #
zone = "incomplete.{}.dsremoved.ns2-5-7".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"incomplete.{checkds}.dsremoved.ns2-5-7",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.2" "DS response from 10.53.0.2",
watcher.wait_for_line(line) f"empty DS response from {addr}",
with servers["ns9"].watch_log_from_start() as watcher: "empty DS response from 10.53.0.7",
line = f"zone {zone}/IN (signed): checkds: empty DS response from {addr}" ),
watcher.wait_for_line(line) expected_parent_state="!DSRemoved",
with servers["ns9"].watch_log_from_start() as watcher: ),
line = f"zone {zone}/IN (signed): checkds: empty DS response from 10.53.0.7"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "!DSRemoved")
# #
# 2.2.3: One parental agent is badly configured. # 2.2.3: One parental agent is badly configured.
# parental-agents: ns5, ns6, ns7 # parental-agents: ns5, ns6, ns7
# #
zone = "bad.{}.dsremoved.ns5-6-7".format(checkds) CheckDSTest(
zone_check(servers["ns9"], zone) zone=f"bad.{checkds}.dsremoved.ns5-6-7",
with servers["ns9"].watch_log_from_start() as watcher: logs_to_wait_for=(
line = f"zone {zone}/IN (signed): checkds: empty DS response from {addr}" f"empty DS response from {addr}",
watcher.wait_for_line(line) "empty DS response from 10.53.0.7",
with servers["ns9"].watch_log_from_start() as watcher: "bad DS response from 10.53.0.6",
line = f"zone {zone}/IN (signed): checkds: empty DS response from 10.53.0.7" ),
watcher.wait_for_line(line) expected_parent_state="!DSRemoved",
with servers["ns9"].watch_log_from_start() as watcher: ),
line = f"zone {zone}/IN (signed): checkds: bad DS response from 10.53.0.6"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "!DSRemoved")
# #
# 2.2.4:: DS is removed completely, bogus signature. # 2.2.4:: DS is removed completely, bogus signature.
# #
# TBD # TBD
]
def test_checkds_reference(named_port, servers): checkds_no_tests = [
# Using a reference to parental-agents. CheckDSTest(
zone = "reference.explicit.dspublish.ns2" zone="good.no.dspublish.ns2",
zone_check(servers["ns9"], zone) 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",
),
]
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
)
@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)
# 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: with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.8" line = f"zone {params.zone}/IN (signed): checkds: {log_string}"
watcher.wait_for_line(line) try:
keystate_check(servers["ns2"], zone, "DSPublish") watcher.wait_for_line(line, timeout=1)
except TimeoutError:
rekey(params.zone)
else:
break
else:
raise TimeoutError
# Check whether key states on the parent server provided match
def test_checkds_resolver(named_port, servers): # expectations.
# Using a resolver as parental-agent (ns3). keystate_check(servers["ns2"], params.zone, params.expected_parent_state)
zone = "resolver.explicit.dspublish.ns2"
zone_check(servers["ns9"], zone)
with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.3"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "DSPublish")
# Using a resolver as parental-agent (ns3).
zone = "resolver.explicit.dsremoved.ns5"
zone_check(servers["ns9"], zone)
with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {zone}/IN (signed): checkds: empty DS response from 10.53.0.3"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "DSRemoved")
def test_checkds_no_ent(named_port, servers):
zone = "no-ent.ns2"
zone_check(servers["ns9"], zone)
with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.2"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "DSPublish")
zone = "no-ent.ns5"
zone_check(servers["ns9"], zone)
with servers["ns9"].watch_log_from_start() as watcher:
line = f"zone {zone}/IN (signed): checkds: DS response from 10.53.0.5"
watcher.wait_for_line(line)
keystate_check(servers["ns2"], zone, "DSRemoved")
def test_checkds_dspublished(named_port, servers):
checkds_dspublished(named_port, servers, "explicit", "10.53.0.8")
checkds_dspublished(named_port, servers, "yes", "10.53.0.2")
def test_checkds_dswithdrawn(named_port, servers):
checkds_dswithdrawn(named_port, servers, "explicit", "10.53.0.10")
checkds_dswithdrawn(named_port, servers, "yes", "10.53.0.5")
def test_checkds_no(named_port, servers):
zone_check(servers["ns9"], "good.no.dspublish.ns2")
keystate_check(servers["ns2"], "good.no.dspublish.ns2", "!DSPublish")
zone_check(servers["ns9"], "good.no.dspublish.ns2-4")
keystate_check(servers["ns2"], "good.no.dspublish.ns2-4", "!DSPublish")
zone_check(servers["ns9"], "good.no.dsremoved.ns5")
keystate_check(servers["ns2"], "good.no.dsremoved.ns5", "!DSRemoved")
zone_check(servers["ns9"], "good.no.dsremoved.ns5-7")
keystate_check(servers["ns2"], "good.no.dsremoved.ns5-7", "!DSRemoved")