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 15 16import pytest 17 18import isctest.mark 19 20pytest.register_assert_rewrite("generic") 21import generic 22 23requests = pytest.importorskip("requests") 24 25pytestmark = [ 26 isctest.mark.have_json_c, 27 pytest.mark.extra_artifacts( 28 [ 29 "ns2/*.jnl", 30 "ns2/*.signed", 31 "ns2/dsset-*", 32 "ns2/K*", 33 "ns2/dnssec.db.signed", 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# JSON helper functions 47def fetch_zones_json(statsip, statsport): 48 r = requests.get( 49 "http://{}:{}/json/v1/zones".format(statsip, statsport), timeout=600 50 ) 51 assert r.status_code == 200 52 53 data = r.json() 54 return data["views"]["_default"]["zones"] 55 56 57def fetch_traffic_json(statsip, statsport): 58 r = requests.get( 59 "http://{}:{}/json/v1/traffic".format(statsip, statsport), timeout=600 60 ) 61 assert r.status_code == 200 62 63 data = r.json() 64 65 return data["traffic"] 66 67 68def load_timers_json(zone, primary=True): 69 name = zone["name"] 70 71 # Check if the primary zone timer exists 72 assert "loaded" in zone 73 loaded = datetime.strptime(zone["loaded"], generic.fmt) 74 75 if primary: 76 # Check if the secondary zone timers does not exist 77 assert "expires" not in zone 78 assert "refresh" not in zone 79 expires = None 80 refresh = None 81 else: 82 assert "expires" in zone 83 assert "refresh" in zone 84 expires = datetime.strptime(zone["expires"], generic.fmt) 85 refresh = datetime.strptime(zone["refresh"], generic.fmt) 86 87 return (name, loaded, expires, refresh) 88 89 90def load_zone_json(zone): 91 name = zone["name"] 92 93 return name 94 95 96def test_zone_timers_primary_json(statsport): 97 generic.test_zone_timers_primary( 98 fetch_zones_json, 99 load_timers_json, 100 statsip="10.53.0.1", 101 statsport=statsport, 102 zonedir="ns1", 103 ) 104 105 106def test_zone_timers_secondary_json(statsport): 107 generic.test_zone_timers_secondary( 108 fetch_zones_json, 109 load_timers_json, 110 statsip="10.53.0.3", 111 statsport=statsport, 112 zonedir="ns3", 113 ) 114 115 116def test_zone_with_many_keys_json(statsport): 117 generic.test_zone_with_many_keys( 118 fetch_zones_json, load_zone_json, statsip="10.53.0.2", statsport=statsport 119 ) 120 121 122@isctest.mark.flaky(max_runs=2, rerun_filter=isctest.mark.with_tsan) 123def test_traffic_json(statsport): 124 generic.test_traffic(fetch_traffic_json, statsip="10.53.0.2", statsport=statsport) 125