integrations-extras/ping/tests/test_ping.py at 67c63df1727877f1cf6592cb9d422c1cb53315d3 · DataDog/integrations-extras · GitHub

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
105
106
107
108
import mock
import pytest

from datadog_checks.base import AgentCheck
from datadog_checks.base.errors import CheckException
from datadog_checks.ping import PingCheck


def mock_exec_ping():
    return """FAKEPING 127.0.0.1 (127.0.0.1): 56 data bytes
64 bytes from 127.0.0.1: icmp_seq=0 ttl=64 time=0.093 ms

--- 127.0.0.1 ping statistics ---
1 packets transmitted, 1 packets received, 0.0% packet loss
round-trip min/avg/max/stddev = 0.093/0.093/0.093/0.000 ms"""


def mock_exec_ping_german():
    return (
        "Antwort von 127.0.0.1: Bytes=32 Zeit=3ms TTL=117\n"
        "Ping-Statistik für 127.0.0.1:\n"
        "    Minimum = 3ms, Maximum = 3ms, Mittelwert = 3ms"
    )


def test_empty_check(empty_instance):
    check = PingCheck("ping", {}, {})

    with pytest.raises(CheckException):
        check.check(empty_instance)


def test_incorrect_ip_check(incorrect_ip_instance):
    check = PingCheck("ping", {}, {})

    with pytest.raises(CheckException):
        check.check(incorrect_ip_instance)


def test_valid_check(aggregator, instance):
    check = PingCheck("ping", {}, {})

    with mock.patch.object(check, "_exec_ping", return_value=mock_exec_ping()):
        check.check(instance)
    aggregator.assert_service_check("network.ping.can_connect", AgentCheck.OK)
    aggregator.assert_metric("network.ping.can_connect", value=1)
    aggregator.assert_all_metrics_covered()


def test_valid_check_ipv6(aggregator, instance_ipv6):
    check = PingCheck("ping", {}, {})

    with mock.patch.object(check, "_exec_ping", return_value=mock_exec_ping()):
        check.check(instance_ipv6)
    aggregator.assert_service_check("network.ping.can_connect", AgentCheck.OK)
    aggregator.assert_metric("network.ping.can_connect", value=1)
    aggregator.assert_all_metrics_covered()


def test_localized_output(aggregator, instance_response_time):
    check = PingCheck("ping", {}, {})

    with mock.patch.object(check, "_exec_ping", return_value=mock_exec_ping_german()):
        check.check(instance_response_time)
    aggregator.assert_service_check("network.ping.can_connect", AgentCheck.OK)
    aggregator.assert_metric("network.ping.can_connect", value=1)
    aggregator.assert_metric("network.ping.response_time", value=3)
    aggregator.assert_all_metrics_covered()


@pytest.mark.usefixtures("dd_environment")
def test_integration(aggregator, instance):
    check = PingCheck("ping", {}, {})
    check.check(instance)

    tags = ["ping1", "ping2"]
    all_tags = tags.append("target_host:127.0.0.1")

    aggregator.assert_service_check("network.ping.can_connect", AgentCheck.OK)
    aggregator.assert_metric("network.ping.can_connect", value=1, tags=all_tags)
    aggregator.assert_all_metrics_covered()


@pytest.mark.usefixtures("dd_environment")
def test_integration_ipv6(aggregator, instance_ipv6):
    check = PingCheck("ping", {}, {})
    check.check(instance_ipv6)

    tags = ["ping1", "ping2"]
    all_tags = tags.append("target_host:0000:0000:0000:0000:0000:0000:0000:0001")

    aggregator.assert_service_check("network.ping.can_connect", AgentCheck.OK)
    aggregator.assert_metric("network.ping.can_connect", value=1, tags=all_tags)
    aggregator.assert_all_metrics_covered()


@pytest.mark.usefixtures("dd_environment")
def test_integration_response_time(aggregator, instance_response_time):
    check = PingCheck("ping", {}, {})
    check.check(instance_response_time)

    tags = ["response_time:yes"]
    all_tags = tags.append("target_host:127.0.0.1")

    aggregator.assert_service_check("network.ping.can_connect", AgentCheck.OK)
    aggregator.assert_metric("network.ping.can_connect", value=1, tags=all_tags)
    aggregator.assert_metric("network.ping.response_time", tags=all_tags)
    aggregator.assert_all_metrics_covered()