1#!/usr/bin/python3 2 3# Copyright (C) Internet Systems Consortium, Inc. ("ISC") 4# 5# SPDX-License-Identifier: MPL-2.0 6# 7# This Source Code Form is subject to the terms of the Mozilla Public 8# License, v. 2.0. If a copy of the MPL was not distributed with this 9# file, you can obtain one at https://mozilla.org/MPL/2.0/. 10# 11# See the COPYRIGHT file distributed with this work for additional 12# information regarding copyright ownership. 13 14from datetime import datetime 15import xml.etree.ElementTree as ET 16 17import pytest 18 19import isctest.mark 20 21pytest.register_assert_rewrite("generic") 22import generic 23 24requests = pytest.importorskip("requests") 25 26pytestmark = [ 27 isctest.mark.have_libxml2, 28 pytest.mark.extra_artifacts( 29 [ 30 "ns2/K*", 31 "ns2/*.jnl", 32 "ns2/*.signed", 33 "ns2/dsset-*", 34 "ns2/dnssec.*.id", 35 "ns2/manykeys.*.id", 36 "ns2/signzone.out.*", 37 "ns3/_default.nzd", 38 "ns3/example-tcp.db", 39 "ns3/example-tls.db", 40 "ns3/example.db", 41 ] 42 ), 43] 44 45 46# XML helper functions 47def fetch_zones_xml(statsip, statsport): 48 r = requests.get( 49 "http://{}:{}/xml/v3/zones".format(statsip, statsport), timeout=600 50 ) 51 assert r.status_code == 200 52 53 root = ET.fromstring(r.text) 54 55 default_view = None 56 for view in root.find("views").iter("view"): 57 if view.attrib["name"] == "_default": 58 default_view = view 59 break 60 assert default_view is not None 61 62 return default_view.find("zones").findall("zone") 63 64 65def fetch_traffic_xml(statsip, statsport): 66 def load_counters(data): 67 out = {} 68 for counter in data.findall("counter"): 69 out[counter.attrib["name"]] = int(counter.text) 70 71 return out 72 73 r = requests.get( 74 "http://{}:{}/xml/v3/traffic".format(statsip, statsport), timeout=600 75 ) 76 assert r.status_code == 200 77 78 root = ET.fromstring(r.text) 79 80 traffic = {} 81 for ip in ["ipv4", "ipv6"]: 82 for proto in ["udp", "tcp"]: 83 proto_root = root.find("traffic").find(ip).find(proto) 84 for counters in proto_root.findall("counters"): 85 if counters.attrib["type"] == "request-size": 86 key = "dns-{}-requests-sizes-received-{}".format(proto, ip) 87 else: 88 key = "dns-{}-responses-sizes-sent-{}".format(proto, ip) 89 90 values = load_counters(counters) 91 traffic[key] = values 92 93 return traffic 94 95 96def load_timers_xml(zone, primary=True): 97 name = zone.attrib["name"] 98 99 loaded_el = zone.find("loaded") 100 assert loaded_el is not None 101 loaded = datetime.strptime(loaded_el.text, generic.fmt) 102 103 expires_el = zone.find("expires") 104 refresh_el = zone.find("refresh") 105 if primary: 106 assert expires_el is None 107 assert refresh_el is None 108 expires = None 109 refresh = None 110 else: 111 assert expires_el is not None 112 assert refresh_el is not None 113 expires = datetime.strptime(expires_el.text, generic.fmt) 114 refresh = datetime.strptime(refresh_el.text, generic.fmt) 115 116 return (name, loaded, expires, refresh) 117 118 119def load_zone_xml(zone): 120 name = zone.attrib["name"] 121 122 return name 123 124 125def test_zone_timers_primary_xml(statsport): 126 generic.test_zone_timers_primary( 127 fetch_zones_xml, 128 load_timers_xml, 129 statsip="10.53.0.1", 130 statsport=statsport, 131 zonedir="ns1", 132 ) 133 134 135def test_zone_timers_secondary_xml(statsport): 136 generic.test_zone_timers_secondary( 137 fetch_zones_xml, 138 load_timers_xml, 139 statsip="10.53.0.3", 140 statsport=statsport, 141 zonedir="ns3", 142 ) 143 144 145def test_zone_with_many_keys_xml(statsport): 146 generic.test_zone_with_many_keys( 147 fetch_zones_xml, load_zone_xml, statsip="10.53.0.2", statsport=statsport 148 ) 149 150 151@isctest.mark.flaky(max_runs=2, rerun_filter=isctest.mark.with_tsan) 152def test_traffic_xml(statsport): 153 generic.test_traffic(fetch_traffic_xml, statsip="10.53.0.2", statsport=statsport) 154