2 # Copyright 2022 Canonical Ltd.
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
16 # For those usages not covered by the Apache License, Version 2.0 please
17 # contact: legal@canonical.com
19 # To get in touch with the maintainers, please contact:
20 # osm-charmers@lists.launchpad.net
22 # Learn more about testing at: https://juju.is/docs/sdk/testing
27 from pathlib
import Path
31 from pytest_operator
.plugin
import OpsTest
33 logger
= logging
.getLogger(__name__
)
35 METADATA
= yaml
.safe_load(Path("./metadata.yaml").read_text())
36 MON_APP
= METADATA
["name"]
37 GRAFANA_CHARM
= "grafana-k8s"
38 GRAFANA_APP
= "grafana"
39 DS_NAME
= "osm_prometheus"
40 DS_URL
= "http://prometheus:9090"
43 async def _run_action_create_datasource(ops_test
: OpsTest
, name
: str, url
: str):
45 await ops_test
.model
.applications
[MON_APP
]
48 action_name
="create-datasource",
53 return await action
.wait()
56 async def _run_action_list_datasources(ops_test
: OpsTest
):
58 await ops_test
.model
.applications
[MON_APP
]
60 .run_action(action_name
="list-datasources")
62 return await action
.wait()
65 async def _run_action_delete_datasource(ops_test
: OpsTest
, name
: str):
67 await ops_test
.model
.applications
[MON_APP
]
69 .run_action(action_name
="delete-datasource", name
=name
)
71 return await action
.wait()
74 @pytest.mark
.abort_on_fail
75 async def test_mon_and_grafana_are_idle(ops_test
: OpsTest
):
76 charm
= await ops_test
.build_charm(".")
77 resources
= {"mon-image": METADATA
["resources"]["mon-image"]["upstream-source"]}
80 ops_test
.model
.deploy(
81 charm
, resources
=resources
, application_name
=MON_APP
, series
="jammy"
83 ops_test
.model
.deploy(GRAFANA_CHARM
, application_name
=GRAFANA_APP
, channel
="stable"),
85 async with ops_test
.fast_forward():
86 await ops_test
.model
.wait_for_idle(apps
=[GRAFANA_APP
], status
="active")
87 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
90 @pytest.mark
.abort_on_fail
91 async def test_mon_cannot_create_datasource_grafana_url_is_not_set(ops_test
: OpsTest
):
92 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, DS_URL
)
93 assert action
.status
== "failed"
94 assert "Invalid URL" in action
.message
97 @pytest.mark
.abort_on_fail
98 async def test_mon_cannot_list_datasources_grafana_url_is_not_set(ops_test
: OpsTest
):
99 action
= await _run_action_list_datasources(ops_test
)
100 assert action
.status
== "failed"
101 assert "Invalid URL" in action
.message
104 @pytest.mark
.abort_on_fail
105 async def test_mon_cannot_delete_datasource_grafana_url_is_not_set(ops_test
: OpsTest
):
106 action
= await _run_action_delete_datasource(ops_test
, "prometheus")
107 assert action
.status
== "failed"
108 assert "Invalid URL" in action
.message
111 @pytest.mark
.abort_on_fail
112 async def test_mon_cannot_create_datasource_due_to_invalid_grafana_password(ops_test
: OpsTest
):
113 await ops_test
.model
.applications
[MON_APP
].set_config(
114 {"grafana-url": f
"http://{GRAFANA_APP}:3000", "grafana-user": "admin"}
116 async with ops_test
.fast_forward():
117 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
119 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, DS_URL
)
120 assert action
.status
== "failed"
121 assert action
.message
== "invalid username or password"
124 @pytest.mark
.abort_on_fail
125 async def test_mon_cannot_create_datasource_invalid_url_fails(ops_test
: OpsTest
):
126 await ops_test
.model
.applications
[MON_APP
].set_config(
127 {"grafana-url": f
"http://{GRAFANA_APP}:3000"}
129 async with ops_test
.fast_forward():
130 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
132 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, "prometheus:9090")
133 assert action
.status
== "failed"
134 assert action
.message
== "Invalid datasource url 'prometheus:9090'"
137 async def _get_grafana_admin_password(ops_test
: OpsTest
):
139 await ops_test
.model
.applications
[GRAFANA_APP
].units
[0].run_action("get-admin-password")
141 admin_password
= (await action
.wait()).results
["admin-password"]
142 logger
.info(f
"Password obtained from {GRAFANA_APP} : {admin_password}")
143 assert admin_password
!= "Admin password has been changed by an administrator"
144 return admin_password
147 @pytest.mark
.abort_on_fail
148 async def test_grafana_password_is_set_in_mon_config(ops_test
: OpsTest
):
149 admin_password
= await _get_grafana_admin_password(ops_test
)
151 default_password
= None
152 grafana_password_in_mon
= (
153 (await ops_test
.model
.applications
[MON_APP
].get_config())
154 .get("grafana-password")
157 assert grafana_password_in_mon
== default_password
159 await ops_test
.model
.applications
[MON_APP
].set_config({"grafana-password": admin_password
})
161 async with ops_test
.fast_forward():
162 await ops_test
.model
.wait_for_idle(apps
=[GRAFANA_APP
], status
="active")
163 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
165 grafana_password_in_mon
= (
166 (await ops_test
.model
.applications
[MON_APP
].get_config())
167 .get("grafana-password")
170 assert grafana_password_in_mon
== admin_password
173 @pytest.mark
.abort_on_fail
174 async def test_mon_create_datasource_successfully(ops_test
: OpsTest
):
175 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, DS_URL
)
176 assert action
.status
== "completed"
177 assert action
.results
["datasource-name"] == DS_NAME
180 @pytest.mark
.abort_on_fail
181 async def test_mon_list_datasources_returns_osm_prometheus(ops_test
: OpsTest
):
182 action
= await _run_action_list_datasources(ops_test
)
183 assert action
.status
== "completed"
184 datasources
= json
.loads(action
.results
.get("datasources"))
185 assert len(datasources
) == 1
186 assert datasources
[0].get("name") == DS_NAME
187 assert datasources
[0].get("type") == "prometheus"
190 @pytest.mark
.abort_on_fail
191 async def test_mon_create_datasource_that_already_exists_returns_error_message(ops_test
: OpsTest
):
192 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, DS_URL
)
193 assert action
.status
== "failed"
194 assert action
.message
== "data source with the same name already exists"
197 @pytest.mark
.abort_on_fail
198 async def test_mon_delete_existing_datasource_returns_success_message(ops_test
: OpsTest
):
199 action
= await _run_action_delete_datasource(ops_test
, DS_NAME
)
200 assert action
.status
== "completed"
203 @pytest.mark
.abort_on_fail
204 async def test_mon_list_datasources_is_empty(ops_test
: OpsTest
):
205 action
= await _run_action_list_datasources(ops_test
)
206 assert action
.status
== "completed"
207 datasources
= json
.loads(action
.results
.get("datasources"))
208 assert len(datasources
) == 0
211 @pytest.mark
.abort_on_fail
212 async def test_mon_delete_non_existing_datasource_returns_error_message(ops_test
: OpsTest
):
213 action
= await _run_action_delete_datasource(ops_test
, DS_NAME
)
214 assert action
.status
== "failed"
215 assert action
.message
== "Data source not found"
218 @pytest.mark
.abort_on_fail
219 async def test_mon_create_datasource_incorrect_grafana_host_fails(ops_test
: OpsTest
):
220 await ops_test
.model
.applications
[MON_APP
].set_config(
221 {"grafana-url": f
"http://{GRAFANA_APP}-k8s:3000"}
223 async with ops_test
.fast_forward():
224 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
225 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, DS_URL
)
226 assert action
.status
== "failed"
227 assert f
"Failed to resolve '{GRAFANA_APP}-k8s'" in action
.message
230 @pytest.mark
.abort_on_fail
231 async def test_mon_list_datasources_incorrect_grafana_host_fails(ops_test
: OpsTest
):
232 action
= await _run_action_list_datasources(ops_test
)
233 assert action
.status
== "failed"
234 assert f
"Failed to resolve '{GRAFANA_APP}-k8s'" in action
.message
237 @pytest.mark
.abort_on_fail
238 async def test_mon_delete_datasource_incorrect_grafana_host_fails(ops_test
: OpsTest
):
239 action
= await _run_action_delete_datasource(ops_test
, DS_NAME
)
240 assert action
.status
== "failed"
241 assert f
"Failed to resolve '{GRAFANA_APP}-k8s'" in action
.message
244 @pytest.mark
.abort_on_fail
245 async def test_mon_incorrect_grafana_port_returns_timeout_message(ops_test
: OpsTest
):
246 await ops_test
.model
.applications
[MON_APP
].set_config(
247 {"grafana-url": f
"http://{GRAFANA_APP}:3001"}
249 async with ops_test
.fast_forward():
250 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
251 action
= await _run_action_create_datasource(ops_test
, DS_NAME
, DS_URL
)
252 assert action
.status
== "failed"
253 assert f
"Connection to {GRAFANA_APP} timed out" in action
.message
256 @pytest.mark
.abort_on_fail
257 async def test_mon_list_datasources_incorrect_user_fails(ops_test
: OpsTest
):
258 await ops_test
.model
.applications
[MON_APP
].set_config(
259 {"grafana-url": f
"http://{GRAFANA_APP}:3000", "grafana-user": "some_user"}
261 async with ops_test
.fast_forward():
262 await ops_test
.model
.wait_for_idle(apps
=[MON_APP
], status
="blocked")
263 action
= await _run_action_list_datasources(ops_test
)
264 assert action
.status
== "failed"
265 assert action
.message
== "invalid username or password"
268 @pytest.mark
.abort_on_fail
269 async def test_mon_delete_datasource_incorrect_user_fails(ops_test
: OpsTest
):
270 action
= await _run_action_delete_datasource(ops_test
, DS_NAME
)
271 assert action
.status
== "failed"
272 assert action
.message
== "invalid username or password"