/
test_charm.py
108 lines (89 loc) · 3.71 KB
/
test_charm.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
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
#!/usr/bin/env python3
# Copyright 2021 Canonical Ltd.
# See LICENSE file for licensing details.
import asyncio
import logging
import pytest
from helpers import (
check_prometheus_is_ready,
get_job_config_for,
get_prometheus_config,
get_prometheus_rules,
get_rules_for,
initial_workload_is_ready,
oci_image,
)
from pytest_operator.plugin import OpsTest
logger = logging.getLogger(__name__)
prometheus_app_name = "prometheus"
tester_app_name = "prometheus-tester"
app_names = [prometheus_app_name, tester_app_name]
initial_config = None
initial_rules = None
@pytest.mark.abort_on_fail
async def test_prometheus_scrape_relation_with_prometheus_tester(
ops_test: OpsTest, prometheus_charm, prometheus_tester_charm
):
"""Test basic functionality of prometheus_scrape relation interface."""
await asyncio.gather(
ops_test.model.deploy(
prometheus_charm,
resources={"prometheus-image": oci_image("./metadata.yaml", "prometheus-image")},
application_name=prometheus_app_name,
trust=True, # otherwise errors on ghwf (persistentvolumeclaims ... is forbidden)
),
ops_test.model.deploy(
prometheus_tester_charm,
resources={
"prometheus-tester-image": oci_image(
"./tests/integration/prometheus-tester/metadata.yaml",
"prometheus-tester-image",
)
},
application_name=tester_app_name,
),
)
await ops_test.model.wait_for_idle(apps=app_names, status="active", wait_for_units=1)
assert initial_workload_is_ready(ops_test, app_names)
assert await check_prometheus_is_ready(ops_test, prometheus_app_name, 0)
global initial_config, initial_rules
initial_config, initial_rules = await asyncio.gather(
get_prometheus_config(ops_test, prometheus_app_name, 0),
get_prometheus_rules(ops_test, prometheus_app_name, 0),
)
await ops_test.model.add_relation(
f"{prometheus_app_name}:metrics-endpoint", f"{tester_app_name}:metrics-endpoint"
)
await ops_test.model.wait_for_idle(apps=app_names, status="active")
config_with_relation = await get_prometheus_config(ops_test, prometheus_app_name, 0)
tester_job = get_job_config_for(tester_app_name, config_with_relation)
assert tester_job != {}
rules_with_relation = await get_prometheus_rules(ops_test, prometheus_app_name, 0)
tester_rules = get_rules_for(tester_app_name, rules_with_relation)
assert len(tester_rules) == 1
async def test_alert_rule_path_can_be_changed(ops_test, prometheus_tester_charm):
"""Ensure scrape alert rules can be updated.
This test upgrades the metrics provider charm and checks that
updates to alert rules are propagated correctly.
"""
# Change the alert rule path and ensure that there are 2
# after refreshing so they are reloaded
await ops_test.model.applications[tester_app_name].set_config(
{"alert-rules-path": "src/with_extra_alert_rule"}
)
await ops_test.model.wait_for_idle(apps=app_names, status="active")
await ops_test.model.applications[tester_app_name].refresh(
path=prometheus_tester_charm,
resources={
"prometheus-tester-image": oci_image(
"./tests/integration/prometheus-tester/metadata.yaml",
"prometheus-tester-image",
)
},
)
await ops_test.model.wait_for_idle(
apps=[prometheus_app_name, tester_app_name], status="active", idle_period=60
)
rules_with_relation = await get_prometheus_rules(ops_test, prometheus_app_name, 0)
tester_rules = get_rules_for(tester_app_name, rules_with_relation)
assert len(tester_rules) == 2