Move check_nsec3_case to common code

With the nsec and nsec3 test cases being nearly identical for all
modules, these can be be unified and moved to common.py.

diff --git a/bin/tests/system/nsec3/common.py b/bin/tests/system/nsec3/common.py
index 12a176fea5157b3d9538937b5a719e1d47adddac..678cc4cbed53dc2dbf5d086289c809c6116d0fd4 100644 (file)
--- a/bin/tests/system/nsec3/common.py
+++ b/bin/tests/system/nsec3/common.py
@@ -16,6 +16,8 @@ from datetime import timedelta
 import dns
 import pytest
 
+import isctest
+
 pytestmark = pytest.mark.extra_artifacts(
     [
         "*.axfr",
@@ -103,3 +105,63 @@ def check_nsec3param(response, match, saltlen):
     assert len(rrs) != 0
 
     return salt
+
+
+def check_nsec3_case(server, params, nsec3=True):
+    # Get test parameters.
+    zone = params["zone"]
+    fqdn = f"{zone}."
+    policy = params["policy"]
+    keydir = server.identifier
+    config = default_config
+    ttl = int(config["dnskey-ttl"].total_seconds())
+    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=params["key-properties"])
+
+    # Test case.
+    isctest.log.info(f"check nsec case zone {zone} policy {policy}")
+
+    # Key files.
+    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
+    if "external-keys" in params:
+        expected2 = isctest.kasp.policy_to_properties(ttl, keys=params["external-keys"])
+        for ek in expected2:
+            ek.private = False  # noqa
+            ek.legacy = True  # noqa
+        expected = expected + expected2
+        assert "external-keydir" in params
+        extkeys = isctest.kasp.keydir_to_keylist(zone, params["external-keydir"])
+        keys = keys + extkeys
+
+    isctest.kasp.check_keys(zone, keys, expected)
+    isctest.kasp.check_dnssec_verify(server, zone)
+    isctest.kasp.check_apex(server, zone, keys, [])
+
+    query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
+    nsec3param_response = isctest.query.tcp(query, server.ip)
+    assert nsec3param_response.rcode() == dns.rcode.NOERROR
+
+    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
+    response = isctest.query.tcp(query, server.ip)
+    assert response.rcode() == dns.rcode.NXDOMAIN
+
+    if nsec3:
+        # NSEC3
+        minimum = params.get("soa-minimum", 3600)
+        iterations = 0
+        optout = 0
+        saltlen = 0
+        if "nsec3param" in params:
+            optout = params["nsec3param"].get("optout", 0)
+            saltlen = params["nsec3param"].get("salt-length", 0)
+
+        match = f"{fqdn} {minimum} IN NSEC3PARAM 1 0 {iterations}"
+
+        salt = check_nsec3param(nsec3param_response, match, saltlen)
+
+        check_auth_nsec3(response, iterations, optout, salt)
+    else:
+        # NSEC
+        assert len(nsec3param_response.answer) == 0
+        check_auth_nsec(nsec3param_response)
+
+        check_auth_nsec(response)

diff --git a/bin/tests/system/nsec3/tests_nsec3_initial.py b/bin/tests/system/nsec3/tests_nsec3_initial.py
index 99247f8973512027d9bc1affc70cc56fa772ec9e..a8f52350fe19c1ef305c5f0abd07ba723f83fbe5 100644 (file)
--- a/bin/tests/system/nsec3/tests_nsec3_initial.py
+++ b/bin/tests/system/nsec3/tests_nsec3_initial.py
@@ -23,9 +23,7 @@ from nsec3.common import (
     SIZE,
     default_config,
     pytestmark,
-    check_auth_nsec,
-    check_auth_nsec3,
-    check_nsec3param,
+    check_nsec3_case,
 )
 
 
@@ -95,45 +93,12 @@ from nsec3.common import (
 def test_nsec_case(ns3, params):
     # Get test parameters.
     zone = params["zone"]
-    fqdn = f"{zone}."
-    policy = params["policy"]
-    keydir = ns3.identifier
-    config = default_config
-    ttl = int(config["dnskey-ttl"].total_seconds())
-    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=params["key-properties"])
-
-    # Test case.
-    isctest.log.info(f"check nsec case zone {zone} policy {policy}")
 
     # First make sure the zone is properly signed.
     isctest.kasp.wait_keymgr_done(ns3, zone)
 
-    # Key files.
-    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
-    if "external-keys" in params:
-        expected2 = isctest.kasp.policy_to_properties(ttl, keys=params["external-keys"])
-        for ek in expected2:
-            ek.private = False  # noqa
-            ek.legacy = True  # noqa
-        expected = expected + expected2
-        assert "external-keydir" in params
-        extkeys = isctest.kasp.keydir_to_keylist(zone, params["external-keydir"])
-        keys = keys + extkeys
-
-    isctest.kasp.check_keys(zone, keys, expected)
-    isctest.kasp.check_dnssec_verify(ns3, zone)
-    isctest.kasp.check_apex(ns3, zone, keys, [])
-
-    query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NOERROR
-    assert len(response.answer) == 0
-    check_auth_nsec(response)
-
-    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NXDOMAIN
-    check_auth_nsec(response)
+    # Test case.
+    check_nsec3_case(ns3, params, nsec3=False)
 
     # Extra test for nsec3-dynamic-update-inline.kasp.
     if zone == "nsec3-dynamic-update-inline.kasp":
@@ -291,40 +256,9 @@ def test_nsec_case(ns3, params):
 def test_nsec3_case(ns3, params):
     # Get test parameters.
     zone = params["zone"]
-    fqdn = f"{zone}."
-    policy = params["policy"]
-    keydir = ns3.identifier
-    config = default_config
-    ttl = int(config["dnskey-ttl"].total_seconds())
-    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=params["key-properties"])
-
-    iterations = 0
-    optout = 0
-    saltlen = 0
-    if "nsec3param" in params:
-        optout = params["nsec3param"].get("optout", 0)
-        saltlen = params["nsec3param"].get("salt-length", 0)
-
-    match = f"{fqdn} 3600 IN NSEC3PARAM 1 0 {iterations}"
-
-    # Test case.
-    isctest.log.info(f"check nsec3 case zone {zone} policy {policy}")
 
     # First make sure the zone is properly signed.
     isctest.kasp.wait_keymgr_done(ns3, zone)
 
-    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
-    isctest.kasp.check_keys(zone, keys, expected)
-    isctest.kasp.check_dnssec_verify(ns3, zone)
-    isctest.kasp.check_apex(ns3, zone, keys, [])
-
-    query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NOERROR
-
-    salt = check_nsec3param(response, match, saltlen)
-
-    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NXDOMAIN
-    check_auth_nsec3(response, iterations, optout, salt)
+    # Test case.
+    check_nsec3_case(ns3, params)

diff --git a/bin/tests/system/nsec3/tests_nsec3_reconfig.py b/bin/tests/system/nsec3/tests_nsec3_reconfig.py
index 5dd27e281a7febc72f2525fce04a4b2301288b9b..3525a264449c2fb5d86408b53e82546b07637d90 100644 (file)
--- a/bin/tests/system/nsec3/tests_nsec3_reconfig.py
+++ b/bin/tests/system/nsec3/tests_nsec3_reconfig.py
@@ -27,9 +27,7 @@ from nsec3.common import (
     SIZE,
     default_config,
     pytestmark,
-    check_auth_nsec,
-    check_auth_nsec3,
-    check_nsec3param,
+    check_nsec3_case,
 )
 
 
@@ -130,38 +128,13 @@ def after_servers_start(ns3, templates):
     ],
 )
 def test_nsec_case(ns3, params):
-    # Get test parameters.
     zone = params["zone"]
-    fqdn = f"{zone}."
-    policy = params["policy"]
-    keydir = ns3.identifier
-    config = default_config
-    ttl = int(config["dnskey-ttl"].total_seconds())
-    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=params["key-properties"])
-
-    # Test case.
-    isctest.log.info(f"check nsec case zone {zone} policy {policy}")
 
     # First make sure the zone is properly signed.
     isctest.kasp.wait_keymgr_done(ns3, zone, reconfig=True)
 
-    # Key files.
-    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
-
-    isctest.kasp.check_keys(zone, keys, expected)
-    isctest.kasp.check_dnssec_verify(ns3, zone)
-    isctest.kasp.check_apex(ns3, zone, keys, [])
-
-    query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NOERROR
-    assert len(response.answer) == 0
-    check_auth_nsec(response)
-
-    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NXDOMAIN
-    check_auth_nsec(response)
+    # Test case.
+    check_nsec3_case(ns3, params, nsec3=False)
 
 
 @pytest.mark.parametrize(
@@ -307,44 +280,12 @@ def test_nsec_case(ns3, params):
 def test_nsec3_case(ns3, params):
     # Get test parameters.
     zone = params["zone"]
-    fqdn = f"{zone}."
-    policy = params["policy"]
-    keydir = ns3.identifier
-    config = default_config
-    ttl = int(config.get("dnskey-ttl", 3600).total_seconds())
-    minimum = params.get("soa-minimum", 3600)
-    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=params["key-properties"])
-
-    iterations = 0
-    optout = 0
-    saltlen = 0
-    if "nsec3param" in params:
-        optout = params["nsec3param"].get("optout", 0)
-        saltlen = params["nsec3param"].get("salt-length", 0)
-
-    match = f"{fqdn} {minimum} IN NSEC3PARAM 1 0 {iterations}"
-
-    # Test case.
-    isctest.log.info(f"check nsec3 case zone {zone} policy {policy}")
 
     # First make sure the zone is properly signed.
     isctest.kasp.wait_keymgr_done(ns3, zone, reconfig=True)
 
-    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
-    isctest.kasp.check_keys(zone, keys, expected)
-    isctest.kasp.check_dnssec_verify(ns3, zone)
-    isctest.kasp.check_apex(ns3, zone, keys, [])
-
-    query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NOERROR
-
-    salt = check_nsec3param(response, match, saltlen)
-
-    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
-    response = isctest.query.tcp(query, ns3.ip)
-    assert response.rcode() == dns.rcode.NXDOMAIN
-    check_auth_nsec3(response, iterations, optout, salt)
+    # Test case.
+    check_nsec3_case(ns3, params)
 
     # Extra test for nsec3-change.kasp.
     if zone == "nsec3-change.kasp":
@@ -358,41 +299,24 @@ def test_nsec3_case(ns3, params):
 
 def test_nsec3_ent(ns3, templates):
     # Zone: nsec3-ent.kasp (regression test for #5108)
-    zone = "nsec3-ent.kasp"
-    fqdn = f"{zone}."
-    policy = "nsec3"
-    keydir = ns3.identifier
-    config = default_config
-    ttl = int(config.get("dnskey-ttl", 3600).total_seconds())
-    minimum = 3600
-    keyprops = [
-        f"csk 0 {ALGORITHM} {SIZE} goal:omnipresent dnskey:rumoured krrsig:rumoured zrrsig:rumoured ds:hidden",
-    ]
-    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=keyprops)
+    params = {
+        "zone": "nsec3-ent.kasp",
+        "policy": "nsec3",
+        "key-properties": [
+            f"csk 0 {ALGORITHM} {SIZE} goal:omnipresent dnskey:rumoured krrsig:rumoured zrrsig:rumoured ds:hidden",
+        ],
+    }
 
-    # Test case.
-    isctest.log.info(f"check nsec3 case zone {zone} policy {policy}")
+    zone = params["zone"]
+    fqdn = f"{zone}."
 
     # First make sure the zone is properly signed.
     isctest.kasp.wait_keymgr_done(ns3, zone, reconfig=True)
 
-    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
-    isctest.kasp.check_keys(zone, keys, expected)
-    isctest.kasp.check_dnssec_verify(ns3, zone)
-    isctest.kasp.check_apex(ns3, zone, keys, [])
-
-    query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
-    response = isctest.query.tcp(query, ns3.ip, ns3.ports.dns, timeout=3)
-    assert response.rcode() == dns.rcode.NOERROR
-
-    match = f"{fqdn} {minimum} IN NSEC3PARAM 1 0 0"
-    salt = check_nsec3param(response, match, 0)
-
-    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
-    response = isctest.query.tcp(query, ns3.ip, ns3.ports.dns, timeout=3)
-    assert response.rcode() == dns.rcode.NXDOMAIN
-    check_auth_nsec3(response, 0, 0, salt)
+    # Test case.
+    check_nsec3_case(ns3, params)
 
+    # Test empty non-terminals do not trigger a crash.
     isctest.log.info("check query for newly empty name does not crash")
 
     # confirm the pre-existing name still exists

diff --git a/bin/tests/system/nsec3/tests_nsec3_restart.py b/bin/tests/system/nsec3/tests_nsec3_restart.py
index 9415040eea9715103ca561adb057e9b44b3f846f..cae21f4413e5896a42bc77f06583fad6b392bc2d 100644 (file)
--- a/bin/tests/system/nsec3/tests_nsec3_restart.py
+++ b/bin/tests/system/nsec3/tests_nsec3_restart.py
@@ -24,7 +24,7 @@ from nsec3.common import (
     SIZE,
     default_config,
     pytestmark,
-    check_auth_nsec3,
+    check_nsec3_case,
     check_nsec3param,
 )
 
@@ -33,45 +33,25 @@ def perform_nsec3_tests(server, params):
     # Get test parameters.
     zone = params["zone"]
     fqdn = f"{zone}."
-    policy = params["policy"]
-    keydir = server.identifier
-    config = default_config
-    ttl = int(config.get("dnskey-ttl", 3600).total_seconds())
-    minimum = params.get("soa-minimum", 3600)
-    expected = isctest.kasp.policy_to_properties(ttl=ttl, keys=params["key-properties"])
 
+    # First make sure the zone is properly signed.
+    isctest.kasp.wait_keymgr_done(server, zone)
+
+    # Test case.
+    check_nsec3_case(server, params)
+
+    # Return salt.
+    minimum = params.get("soa-minimum", 3600)
     iterations = 0
-    optout = 0
     saltlen = 0
     if "nsec3param" in params:
-        optout = params["nsec3param"].get("optout", 0)
         saltlen = params["nsec3param"].get("salt-length", 0)
 
     match = f"{fqdn} {minimum} IN NSEC3PARAM 1 0 {iterations}"
 
-    # Test case.
-    isctest.log.info(f"check nsec3 case zone {zone} policy {policy}")
-
-    # First make sure the zone is properly signed.
-    isctest.kasp.wait_keymgr_done(server, zone)
-
-    keys = isctest.kasp.keydir_to_keylist(zone, keydir)
-    isctest.kasp.check_keys(zone, keys, expected)
-    isctest.kasp.check_dnssec_verify(server, zone)
-    isctest.kasp.check_apex(server, zone, keys, [])
-
     query = isctest.query.create(fqdn, dns.rdatatype.NSEC3PARAM)
     response = isctest.query.tcp(query, server.ip)
-    assert response.rcode() == dns.rcode.NOERROR
-
-    salt = check_nsec3param(response, match, saltlen)
-
-    query = isctest.query.create(f"nosuchname.{fqdn}", dns.rdatatype.A)
-    response = isctest.query.tcp(query, server.ip)
-    assert response.rcode() == dns.rcode.NXDOMAIN
-    check_auth_nsec3(response, iterations, optout, salt)
-
-    return salt
+    return check_nsec3param(response, match, saltlen)
 
 
 @pytest.mark.parametrize(