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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
from flexmock import flexmock
from borgmatic.hooks.monitoring import cronhub as module
def test_ping_monitor_rewrites_ping_url_for_start_state():
hook_config = {'ping_url': 'https://example.com/start/abcdef'}
flexmock(module.requests).should_receive('get').with_args(
'https://example.com/start/abcdef',
timeout=int,
headers={'User-Agent': 'borgmatic'},
).and_return(flexmock(ok=True))
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.START,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_rewrites_ping_url_and_state_for_start_state():
hook_config = {'ping_url': 'https://example.com/ping/abcdef'}
flexmock(module.requests).should_receive('get').with_args(
'https://example.com/start/abcdef',
timeout=int,
headers={'User-Agent': 'borgmatic'},
).and_return(flexmock(ok=True))
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.START,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_rewrites_ping_url_for_finish_state():
hook_config = {'ping_url': 'https://example.com/start/abcdef'}
flexmock(module.requests).should_receive('get').with_args(
'https://example.com/finish/abcdef',
timeout=int,
headers={'User-Agent': 'borgmatic'},
).and_return(flexmock(ok=True))
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.FINISH,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_rewrites_ping_url_for_fail_state():
hook_config = {'ping_url': 'https://example.com/start/abcdef'}
flexmock(module.requests).should_receive('get').with_args(
'https://example.com/fail/abcdef',
timeout=int,
headers={'User-Agent': 'borgmatic'},
).and_return(flexmock(ok=True))
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_dry_run_does_not_hit_ping_url():
hook_config = {'ping_url': 'https://example.com'}
flexmock(module.requests).should_receive('get').never()
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.START,
monitoring_log_level=1,
dry_run=True,
)
def test_ping_monitor_with_connection_error_logs_warning():
hook_config = {'ping_url': 'https://example.com/start/abcdef'}
flexmock(module.requests).should_receive('get').and_raise(
module.requests.exceptions.ConnectionError,
)
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
(),
'config.yaml',
module.monitor.State.START,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_with_other_error_logs_warning():
hook_config = {'ping_url': 'https://example.com/start/abcdef'}
response = flexmock(ok=False)
response.should_receive('raise_for_status').and_raise(
module.requests.exceptions.RequestException,
)
flexmock(module.requests).should_receive('get').with_args(
'https://example.com/start/abcdef',
timeout=int,
headers={'User-Agent': 'borgmatic'},
).and_return(response)
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.START,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_with_unsupported_monitoring_state_bails():
hook_config = {'ping_url': 'https://example.com'}
flexmock(module.requests).should_receive('get').never()
module.ping_monitor(
hook_config,
{},
'config.yaml',
module.monitor.State.LOG,
monitoring_log_level=1,
dry_run=False,
)
|
