installers/charm/osm-mon/tests/integration/test_datasource_actions.py

   1 #!/usr/bin/env python3
   2 # Copyright 2022 Canonical Ltd.
   3 #
   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
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 #
  22 # Learn more about testing at: https://juju.is/docs/sdk/testing
  23
  24 import asyncio
  25 import json
  26 import logging
  27 from pathlib import Path
  28
  29 import pytest
  30 import yaml
  31 from pytest_operator.plugin import OpsTest
  32
  33 logger = logging.getLogger(__name__)
  34
  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"
  41
  42
  43 async def _run_action_create_datasource(ops_test: OpsTest, name: str, url: str):
  44     action = (
  45         await ops_test.model.applications[MON_APP]
  46         .units[0]
  47         .run_action(
  48             action_name="create-datasource",
  49             name=name,
  50             url=url,
  51         )
  52     )
  53     return await action.wait()
  54
  55
  56 async def _run_action_list_datasources(ops_test: OpsTest):
  57     action = (
  58         await ops_test.model.applications[MON_APP]
  59         .units[0]
  60         .run_action(action_name="list-datasources")
  61     )
  62     return await action.wait()
  63
  64
  65 async def _run_action_delete_datasource(ops_test: OpsTest, name: str):
  66     action = (
  67         await ops_test.model.applications[MON_APP]
  68         .units[0]
  69         .run_action(action_name="delete-datasource", name=name)
  70     )
  71     return await action.wait()
  72
  73
  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"]}
  78
  79     await asyncio.gather(
  80         ops_test.model.deploy(
  81             charm, resources=resources, application_name=MON_APP, series="jammy"
  82         ),
  83         ops_test.model.deploy(GRAFANA_CHARM, application_name=GRAFANA_APP, channel="stable"),
  84     )
  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")
  88
  89
  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
  95
  96
  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
 102
 103
 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
 109
 110
 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"}
 115     )
 116     async with ops_test.fast_forward():
 117         await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
 118
 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"
 122
 123
 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"}
 128     )
 129     async with ops_test.fast_forward():
 130         await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
 131
 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'"
 135
 136
 137 async def _get_grafana_admin_password(ops_test: OpsTest):
 138     action = (
 139         await ops_test.model.applications[GRAFANA_APP].units[0].run_action("get-admin-password")
 140     )
 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
 145
 146
 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)
 150
 151     default_password = None
 152     grafana_password_in_mon = (
 153         (await ops_test.model.applications[MON_APP].get_config())
 154         .get("grafana-password")
 155         .get("value")
 156     )
 157     assert grafana_password_in_mon == default_password
 158
 159     await ops_test.model.applications[MON_APP].set_config({"grafana-password": admin_password})
 160
 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")
 164
 165     grafana_password_in_mon = (
 166         (await ops_test.model.applications[MON_APP].get_config())
 167         .get("grafana-password")
 168         .get("value")
 169     )
 170     assert grafana_password_in_mon == admin_password
 171
 172
 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
 178
 179
 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"
 188
 189
 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"
 195
 196
 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"
 201
 202
 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
 209
 210
 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"
 216
 217
 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"}
 222     )
 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
 228
 229
 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
 235
 236
 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
 242
 243
 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"}
 248     )
 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
 254
 255
 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"}
 260     )
 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"
 266
 267
 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"