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 24pytestmark = isctest.mark.have_libxml2 25requests = pytest.importorskip("requests") 26 27 28# XML helper functions 29def fetch_zones_xml(statsip, statsport): 30 r = requests.get( 31 "http://{}:{}/xml/v3/zones".format(statsip, statsport), timeout=600 32 ) 33 assert r.status_code == 200 34 35 root = ET.fromstring(r.text) 36 37 default_view = None 38 for view in root.find("views").iter("view"): 39 if view.attrib["name"] == "_default": 40 default_view = view 41 break 42 assert default_view is not None 43 44 return default_view.find("zones").findall("zone") 45 46 47def fetch_traffic_xml(statsip, statsport): 48 def load_counters(data): 49 out = {} 50 for counter in data.findall("counter"): 51 out[counter.attrib["name"]] = int(counter.text) 52 53 return out 54 55 r = requests.get( 56 "http://{}:{}/xml/v3/traffic".format(statsip, statsport), timeout=600 57 ) 58 assert r.status_code == 200 59 60 root = ET.fromstring(r.text) 61 62 traffic = {} 63 for ip in ["ipv4", "ipv6"]: 64 for proto in ["udp", "tcp"]: 65 proto_root = root.find("traffic").find(ip).find(proto) 66 for counters in proto_root.findall("counters"): 67 if counters.attrib["type"] == "request-size": 68 key = "dns-{}-requests-sizes-received-{}".format(proto, ip) 69 else: 70 key = "dns-{}-responses-sizes-sent-{}".format(proto, ip) 71 72 values = load_counters(counters) 73 traffic[key] = values 74 75 return traffic 76 77 78def load_timers_xml(zone, primary=True): 79 name = zone.attrib["name"] 80 81 loaded_el = zone.find("loaded") 82 assert loaded_el is not None 83 loaded = datetime.strptime(loaded_el.text, generic.fmt) 84 85 expires_el = zone.find("expires") 86 refresh_el = zone.find("refresh") 87 if primary: 88 assert expires_el is None 89 assert refresh_el is None 90 expires = None 91 refresh = None 92 else: 93 assert expires_el is not None 94 assert refresh_el is not None 95 expires = datetime.strptime(expires_el.text, generic.fmt) 96 refresh = datetime.strptime(refresh_el.text, generic.fmt) 97 98 return (name, loaded, expires, refresh) 99 100 101def load_zone_xml(zone): 102 name = zone.attrib["name"] 103 104 return name 105 106 107def test_zone_timers_primary_xml(statsport): 108 generic.test_zone_timers_primary( 109 fetch_zones_xml, 110 load_timers_xml, 111 statsip="10.53.0.1", 112 statsport=statsport, 113 zonedir="ns1", 114 ) 115 116 117def test_zone_timers_secondary_xml(statsport): 118 generic.test_zone_timers_secondary( 119 fetch_zones_xml, 120 load_timers_xml, 121 statsip="10.53.0.3", 122 statsport=statsport, 123 zonedir="ns3", 124 ) 125 126 127def test_zone_with_many_keys_xml(statsport): 128 generic.test_zone_with_many_keys( 129 fetch_zones_xml, load_zone_xml, statsip="10.53.0.2", statsport=statsport 130 ) 131 132 133@isctest.mark.flaky(max_runs=2, rerun_filter=isctest.mark.with_tsan) 134def test_traffic_xml(statsport): 135 generic.test_traffic(fetch_traffic_xml, statsip="10.53.0.2", statsport=statsport) 136