Merge branch '4452-pytest-improve-assert-messages' into 'main'

Improve assert messages in pytests

Closes #4452

See merge request isc-projects/bind9!8518

bin/tests/system/conftest.py

@@ -23,6 +23,9 @@ from typing import Any, Dict, List, Optional
 import pytest
 
 
+pytest.register_assert_rewrite("isctest")
+
+
 # Silence warnings caused by passing a pytest fixture to another fixture.
 # pylint: disable=redefined-outer-name
 

bin/tests/system/doth/stress_http_quota.py

@@ -161,12 +161,12 @@ class SubDIG:
 # and examining their statuses in one logical operation.
 class MultiDIG:
     def __init__(self, numdigs, http_secure=None, extra_args=None):
-        assert int(numdigs) > 0
+        assert int(numdigs) > 0, f"numdigs={numdigs}"
         digs = []
         for _ in range(1, int(numdigs) + 1):
             digs.append(SubDIG(http_secure=http_secure, extra_args=extra_args))
         self.digs = digs
-        assert len(self.digs) == int(numdigs)
+        assert len(self.digs) == int(numdigs), f"len={len(self.digs)} numdigs={numdigs}"
 
     def run(self):
         for p in self.digs:

bin/tests/system/isctest/__init__.py

@@ -0,0 +1,13 @@
+# 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 . import check
+from . import query

bin/tests/system/isctest/check.py

@@ -0,0 +1,34 @@
+# 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 Any
+
+import dns.rcode
+import dns.message
+
+
+# compatiblity with dnspython<2.0.0
+try:
+    # In dnspython>=2.0.0, dns.rcode.Rcode class is available
+    # pylint: disable=invalid-name
+    dns_rcode = dns.rcode.Rcode  # type: Any
+except AttributeError:
+    # In dnspython<2.0.0, selected rcodes are available as integers directly
+    # from dns.rcode
+    dns_rcode = dns.rcode
+
+
+def rcode(message: dns.message.Message, expected_rcode) -> None:
+    assert message.rcode() == expected_rcode, str(message)
+
+
+def noerror(message: dns.message.Message) -> None:
+    rcode(message, dns_rcode.NOERROR)

bin/tests/system/isctest/query.py

@@ -0,0 +1,35 @@
+# 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 os
+from typing import Optional
+
+import dns.query
+import dns.message
+
+
+QUERY_TIMEOUT = 10
+
+
+def udp(
+    message: dns.message.Message, ip: str, port: Optional[int] = None
+) -> dns.message.Message:
+    if port is None:
+        port = int(os.environ["PORT"])
+    return dns.query.udp(message, ip, QUERY_TIMEOUT, port=port)
+
+
+def tcp(
+    message: dns.message.Message, ip: str, port: Optional[int] = None
+) -> dns.message.Message:
+    if port is None:
+        port = int(os.environ["PORT"])
+    return dns.query.tcp(message, ip, QUERY_TIMEOUT, port=port)

bin/tests/system/statschannel/conftest.py

@@ -9,17 +9,9 @@
 # See the COPYRIGHT file distributed with this work for additional
 # information regarding copyright ownership.
 
-import os
 import pytest
 
 
-@pytest.fixture
-def statsport(request):
-    # pylint: disable=unused-argument
-    env_port = os.getenv("EXTRAPORT1")
-    if env_port is None:
-        env_port = 5301
-    else:
-        env_port = int(env_port)
-
-    return env_port
+@pytest.fixture(scope="module")
+def statsport(ports):
+    return ports["EXTRAPORT1"]

bin/tests/system/statschannel/generic.py

@@ -10,8 +10,15 @@
 # information regarding copyright ownership.
 
 from datetime import datetime, timedelta
+from collections import defaultdict
 import os
 
+import dns.message
+import dns.query
+import dns.rcode
+
+import isctest
+
 
 # ISO datetime format without msec
 fmt = "%Y-%m-%dT%H:%M:%SZ"
@@ -104,3 +111,101 @@ def test_zone_with_many_keys(fetch_zones, load_zone, **kwargs):
         name = load_zone(zone)
         if name == "manykeys":
             check_manykeys(name)
+
+
+def create_msg(qname, qtype):
+    msg = dns.message.make_query(
+        qname, qtype, want_dnssec=True, use_edns=0, payload=4096
+    )
+
+    return msg
+
+
+def create_expected(data):
+    expected = {
+        "dns-tcp-requests-sizes-received-ipv4": defaultdict(int),
+        "dns-tcp-responses-sizes-sent-ipv4": defaultdict(int),
+        "dns-tcp-requests-sizes-received-ipv6": defaultdict(int),
+        "dns-tcp-responses-sizes-sent-ipv6": defaultdict(int),
+        "dns-udp-requests-sizes-received-ipv4": defaultdict(int),
+        "dns-udp-requests-sizes-received-ipv6": defaultdict(int),
+        "dns-udp-responses-sizes-sent-ipv4": defaultdict(int),
+        "dns-udp-responses-sizes-sent-ipv6": defaultdict(int),
+    }
+
+    for k, v in data.items():
+        for kk, vv in v.items():
+            expected[k][kk] += vv
+
+    return expected
+
+
+def update_expected(expected, key, msg):
+    msg_len = len(msg.to_wire())
+    bucket_num = (msg_len // 16) * 16
+    bucket = "{}-{}".format(bucket_num, bucket_num + 15)
+
+    expected[key][bucket] += 1
+
+
+def check_traffic(data, expected):
+    def ordered(obj):
+        if isinstance(obj, dict):
+            return sorted((k, ordered(v)) for k, v in obj.items())
+        if isinstance(obj, list):
+            return sorted(ordered(x) for x in obj)
+        return obj
+
+    ordered_data = ordered(data)
+    ordered_expected = ordered(expected)
+
+    assert len(ordered_data) == 8
+    assert len(ordered_expected) == 8
+    assert len(data) == len(ordered_data)
+    assert len(expected) == len(ordered_expected)
+
+    assert ordered_data == ordered_expected
+
+
+def test_traffic(fetch_traffic, **kwargs):
+    statsip = kwargs["statsip"]
+    statsport = kwargs["statsport"]
+
+    data = fetch_traffic(statsip, statsport)
+    exp = create_expected(data)
+
+    msg = create_msg("short.example.", "TXT")
+    update_expected(exp, "dns-udp-requests-sizes-received-ipv4", msg)
+    ans = isctest.query.udp(msg, statsip)
+    isctest.check.noerror(ans)
+    update_expected(exp, "dns-udp-responses-sizes-sent-ipv4", ans)
+    data = fetch_traffic(statsip, statsport)
+
+    check_traffic(data, exp)
+
+    msg = create_msg("long.example.", "TXT")
+    update_expected(exp, "dns-udp-requests-sizes-received-ipv4", msg)
+    ans = isctest.query.udp(msg, statsip)
+    isctest.check.noerror(ans)
+    update_expected(exp, "dns-udp-responses-sizes-sent-ipv4", ans)
+    data = fetch_traffic(statsip, statsport)
+
+    check_traffic(data, exp)
+
+    msg = create_msg("short.example.", "TXT")
+    update_expected(exp, "dns-tcp-requests-sizes-received-ipv4", msg)
+    ans = isctest.query.tcp(msg, statsip)
+    isctest.check.noerror(ans)
+    update_expected(exp, "dns-tcp-responses-sizes-sent-ipv4", ans)
+    data = fetch_traffic(statsip, statsport)
+
+    check_traffic(data, exp)
+
+    msg = create_msg("long.example.", "TXT")
+    update_expected(exp, "dns-tcp-requests-sizes-received-ipv4", msg)
+    ans = isctest.query.tcp(msg, statsip)
+    isctest.check.noerror(ans)
+    update_expected(exp, "dns-tcp-responses-sizes-sent-ipv4", ans)
+    data = fetch_traffic(statsip, statsport)
+
+    check_traffic(data, exp)

bin/tests/system/statschannel/generic_dnspython.py

@@ -1,128 +0,0 @@
-# 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 collections import defaultdict
-
-import dns.message
-import dns.query
-import dns.rcode
-
-
-TIMEOUT = 10
-
-
-def create_msg(qname, qtype):
-    msg = dns.message.make_query(
-        qname, qtype, want_dnssec=True, use_edns=0, payload=4096
-    )
-
-    return msg
-
-
-def udp_query(ip, port, msg):
-    ans = dns.query.udp(msg, ip, TIMEOUT, port=port)
-    assert ans.rcode() == dns.rcode.NOERROR
-
-    return ans
-
-
-def tcp_query(ip, port, msg):
-    ans = dns.query.tcp(msg, ip, TIMEOUT, port=port)
-    assert ans.rcode() == dns.rcode.NOERROR
-
-    return ans
-
-
-def create_expected(data):
-    expected = {
-        "dns-tcp-requests-sizes-received-ipv4": defaultdict(int),
-        "dns-tcp-responses-sizes-sent-ipv4": defaultdict(int),
-        "dns-tcp-requests-sizes-received-ipv6": defaultdict(int),
-        "dns-tcp-responses-sizes-sent-ipv6": defaultdict(int),
-        "dns-udp-requests-sizes-received-ipv4": defaultdict(int),
-        "dns-udp-requests-sizes-received-ipv6": defaultdict(int),
-        "dns-udp-responses-sizes-sent-ipv4": defaultdict(int),
-        "dns-udp-responses-sizes-sent-ipv6": defaultdict(int),
-    }
-
-    for k, v in data.items():
-        for kk, vv in v.items():
-            expected[k][kk] += vv
-
-    return expected
-
-
-def update_expected(expected, key, msg):
-    msg_len = len(msg.to_wire())
-    bucket_num = (msg_len // 16) * 16
-    bucket = "{}-{}".format(bucket_num, bucket_num + 15)
-
-    expected[key][bucket] += 1
-
-
-def check_traffic(data, expected):
-    def ordered(obj):
-        if isinstance(obj, dict):
-            return sorted((k, ordered(v)) for k, v in obj.items())
-        if isinstance(obj, list):
-            return sorted(ordered(x) for x in obj)
-        return obj
-
-    ordered_data = ordered(data)
-    ordered_expected = ordered(expected)
-
-    assert len(ordered_data) == 8
-    assert len(ordered_expected) == 8
-    assert len(data) == len(ordered_data)
-    assert len(expected) == len(ordered_expected)
-
-    assert ordered_data == ordered_expected
-
-
-def test_traffic(fetch_traffic, **kwargs):
-    statsip = kwargs["statsip"]
-    statsport = kwargs["statsport"]
-    port = kwargs["port"]
-
-    data = fetch_traffic(statsip, statsport)
-    exp = create_expected(data)
-
-    msg = create_msg("short.example.", "TXT")
-    update_expected(exp, "dns-udp-requests-sizes-received-ipv4", msg)
-    ans = udp_query(statsip, port, msg)
-    update_expected(exp, "dns-udp-responses-sizes-sent-ipv4", ans)
-    data = fetch_traffic(statsip, statsport)
-
-    check_traffic(data, exp)
-
-    msg = create_msg("long.example.", "TXT")
-    update_expected(exp, "dns-udp-requests-sizes-received-ipv4", msg)
-    ans = udp_query(statsip, port, msg)
-    update_expected(exp, "dns-udp-responses-sizes-sent-ipv4", ans)
-    data = fetch_traffic(statsip, statsport)
-
-    check_traffic(data, exp)
-
-    msg = create_msg("short.example.", "TXT")
-    update_expected(exp, "dns-tcp-requests-sizes-received-ipv4", msg)
-    ans = tcp_query(statsip, port, msg)
-    update_expected(exp, "dns-tcp-responses-sizes-sent-ipv4", ans)
-    data = fetch_traffic(statsip, statsport)
-
-    check_traffic(data, exp)
-
-    msg = create_msg("long.example.", "TXT")
-    update_expected(exp, "dns-tcp-requests-sizes-received-ipv4", msg)
-    ans = tcp_query(statsip, port, msg)
-    update_expected(exp, "dns-tcp-responses-sizes-sent-ipv4", ans)
-    data = fetch_traffic(statsip, statsport)
-
-    check_traffic(data, exp)

bin/tests/system/statschannel/tests_json.py

@@ -15,9 +15,11 @@ from datetime import datetime
 
 import pytest
 
-import generic
 import pytest_custom_markers
 
+pytest.register_assert_rewrite("generic")
+import generic
+
 pytestmark = pytest_custom_markers.have_json_c
 requests = pytest.importorskip("requests")
 
@@ -98,8 +100,5 @@ def test_zone_with_many_keys_json(statsport):
     )
 
 
-def test_traffic_json(named_port, statsport):
-    generic_dnspython = pytest.importorskip("generic_dnspython")
-    generic_dnspython.test_traffic(
-        fetch_traffic_json, statsip="10.53.0.2", statsport=statsport, port=named_port
-    )
+def test_traffic_json(statsport):
+    generic.test_traffic(fetch_traffic_json, statsip="10.53.0.2", statsport=statsport)

bin/tests/system/statschannel/tests_xml.py

@@ -16,9 +16,11 @@ import xml.etree.ElementTree as ET
 
 import pytest
 
-import generic
 import pytest_custom_markers
 
+pytest.register_assert_rewrite("generic")
+import generic
+
 pytestmark = pytest_custom_markers.have_libxml2
 requests = pytest.importorskip("requests")
 
@@ -128,8 +130,5 @@ def test_zone_with_many_keys_xml(statsport):
     )
 
 
-def test_traffic_xml(named_port, statsport):
-    generic_dnspython = pytest.importorskip("generic_dnspython")
-    generic_dnspython.test_traffic(
-        fetch_traffic_xml, statsip="10.53.0.2", statsport=statsport, port=named_port
-    )
+def test_traffic_xml(statsport):
+    generic.test_traffic(fetch_traffic_xml, statsip="10.53.0.2", statsport=statsport)

dangerfile.py

@@ -462,6 +462,7 @@ for testname in testnames:
         not os.path.isdir(dirpath)
         or testname.startswith(".")
         or testname.startswith("_")
+        or testname == "isctest"
     ):
         continue
     if "_" in testname: