summaryrefslogtreecommitdiffstats
path: root/bin/tests/system/statschannel/tests_json.py
blob: 4f7c4a7d7cd96228b9e935bf9b2f471750eaf0a9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
#!/usr/bin/python3

# 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 datetime import datetime

import pytest

import pytest_custom_markers

pytest.register_assert_rewrite("generic")
import generic

pytestmark = pytest_custom_markers.have_json_c
requests = pytest.importorskip("requests")


# JSON helper functions
def fetch_zones_json(statsip, statsport):
    r = requests.get(
        "http://{}:{}/json/v1/zones".format(statsip, statsport), timeout=600
    )
    assert r.status_code == 200

    data = r.json()
    return data["views"]["_default"]["zones"]


def fetch_traffic_json(statsip, statsport):
    r = requests.get(
        "http://{}:{}/json/v1/traffic".format(statsip, statsport), timeout=600
    )
    assert r.status_code == 200

    data = r.json()

    return data["traffic"]


def load_timers_json(zone, primary=True):
    name = zone["name"]

    # Check if the primary zone timer exists
    assert "loaded" in zone
    loaded = datetime.strptime(zone["loaded"], generic.fmt)

    if primary:
        # Check if the secondary zone timers does not exist
        assert "expires" not in zone
        assert "refresh" not in zone
        expires = None
        refresh = None
    else:
        assert "expires" in zone
        assert "refresh" in zone
        expires = datetime.strptime(zone["expires"], generic.fmt)
        refresh = datetime.strptime(zone["refresh"], generic.fmt)

    return (name, loaded, expires, refresh)


def load_zone_json(zone):
    name = zone["name"]

    return name


def test_zone_timers_primary_json(statsport):
    generic.test_zone_timers_primary(
        fetch_zones_json,
        load_timers_json,
        statsip="10.53.0.1",
        statsport=statsport,
        zonedir="ns1",
    )


def test_zone_timers_secondary_json(statsport):
    generic.test_zone_timers_secondary(
        fetch_zones_json,
        load_timers_json,
        statsip="10.53.0.3",
        statsport=statsport,
        zonedir="ns3",
    )


def test_zone_with_many_keys_json(statsport):
    generic.test_zone_with_many_keys(
        fetch_zones_json, load_zone_json, statsip="10.53.0.2", statsport=statsport
    )


def test_traffic_json(statsport):
    generic.test_traffic(fetch_traffic_json, statsip="10.53.0.2", statsport=statsport)