Skip to content
Snippets Groups Projects
Select Git revision
  • develop default protected
  • master protected
  • feature/POL1-813-error-report-sensu-check
  • 0.21
  • 0.20
  • 0.19
  • 0.18
  • 0.17
  • 0.16
  • 0.15
  • 0.14
  • 0.13
  • 0.12
  • 0.11
  • 0.10
  • 0.9
  • 0.8
  • 0.7
  • 0.6
  • 0.5
  • 0.4
  • 0.3
  • 0.2
23 results
Blame Permalink
test_sensu_checks.py 4.29 KiB
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123 






import copy
import random

import jsonschema
import responses

from brian_polling_manager import sensu, inventory, interfaces


@responses.activate
def test_load_checks(config, mocked_sensu):
    checks = sensu.load_all_checks(config['sensu'])
    assert len(checks) > 0  # test data has checks in it


@responses.activate
def test_check_lifecycle(config, mocked_sensu, mocked_inventory):

    test_interface = random.choice(
        inventory.load_interfaces(config['inventory']))
    test_interface['name'] = 'xyz'

    new_check = interfaces.InterfaceCheck(
        config['sensu']['interface-check'],
        test_interface).to_dict()

    # create the new check
    check_name = new_check['metadata']['name']
    assert check_name not in mocked_sensu
    sensu.create_check(config['sensu'], new_check)
    assert check_name in mocked_sensu

    # modify interval & update, then verify the correct call was made
    updated_check = copy.copy(new_check)
    updated_check['interval'] += 1
    sensu.update_check(config['sensu'], updated_check)
    assert mocked_sensu[check_name]['interval'] == updated_check['interval']

    # delete the check and confirm the correct call was made
    sensu.delete_check(config['sensu'], check_name)
    assert check_name not in mocked_sensu


class DummyCheck(sensu.AbstractCheck):

    def __init__(self, name, command, proxy_entity_name):
        super().__init__()
        self._name = name
        self._command = command
        self._proxy_entity_name = proxy_entity_name

    @sensu.AbstractCheck.name.getter
    def name(self):
        return self._name

    @sensu.AbstractCheck.command.getter
    def command(self):
        return self._command

    @sensu.AbstractCheck.proxy_entity_name.getter
    def proxy_entity_name(self):
        return self._proxy_entity_name


def test_check_dict_schema():
    c = DummyCheck(name='x', command='y', proxy_entity_name='z')
    jsonschema.validate(c.to_dict(), sensu.AbstractCheck.CHECK_DICT_SCHEMA)


def test_check_compare():
    a = DummyCheck(name='x', command='y', proxy_entity_name='z')
    b = DummyCheck(name='x', command='y', proxy_entity_name='z')
    assert sensu.checks_match(a.to_dict(), b.to_dict())


def test_checks_differ():
    a = DummyCheck(name='x', command='x', proxy_entity_name='1')
    b = DummyCheck(name='x', command='x', proxy_entity_name='2')
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='1', proxy_entity_name='x')
    b = DummyCheck(name='x', command='2', proxy_entity_name='x')
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='1', command='x', proxy_entity_name='x')
    b = DummyCheck(name='2', command='x', proxy_entity_name='x')
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.publish = not a.publish
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.interval += 1
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.round_robin = not a.round_robin
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.output_metric_format = a.output_metric_format + 'x'
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.subscriptions.append('x')
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.output_metric_handlers.append('x')
    assert not sensu.checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name='x', command='x', proxy_entity_name='x')
    b = DummyCheck(name='x', command='x', proxy_entity_name='x')
    a.namespace = a.namespace + 'x'
    assert not sensu.checks_match(a.to_dict(), b.to_dict())