-
Notifications
You must be signed in to change notification settings - Fork 3
/
test_log_aggregation.py
executable file
·59 lines (49 loc) · 1.66 KB
/
test_log_aggregation.py
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
#!/usr/bin/env python3
import requests
import time
def send_opa_decision_request():
response = requests.post(
'http://test-opa:8081/v1/data/test/world'
)
assert response.status_code == 200, \
'Cannot access the API of the opa cluster.'
def check_sent_events():
response = requests.post(
'http://opa-vector-aggregator:8686/graphql',
json={
'query': """
{
transforms(first:100) {
nodes {
componentId
metrics {
sentEventsTotal {
sentEventsTotal
}
}
}
}
}
"""
}
)
assert response.status_code == 200, \
'Cannot access the API of the vector aggregator.'
result = response.json()
transforms = result['data']['transforms']['nodes']
for transform in transforms:
sentEvents = transform['metrics']['sentEventsTotal']
componentId = transform['componentId']
if componentId == 'filteredInvalidEvents':
assert sentEvents is None or \
sentEvents['sentEventsTotal'] == 0, \
'Invalid log events were sent.'
else:
assert sentEvents is not None and \
sentEvents['sentEventsTotal'] > 0, \
f'No events were sent in "{componentId}".'
if __name__ == '__main__':
send_opa_decision_request()
time.sleep(10)
check_sent_events()
print('Test successful!')