--- /dev/null
+#!/usr/bin/env python3
+# Copyright 2022 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: legal@canonical.com
+#
+# To get in touch with the maintainers, please contact:
+# osm-charmers@lists.launchpad.net
+#
+# Learn more about testing at: https://juju.is/docs/sdk/testing
+
+import asyncio
+import json
+import logging
+from pathlib import Path
+
+import pytest
+import yaml
+from pytest_operator.plugin import OpsTest
+
+logger = logging.getLogger(__name__)
+
+METADATA = yaml.safe_load(Path("./metadata.yaml").read_text())
+MON_APP = METADATA["name"]
+GRAFANA_CHARM = "grafana-k8s"
+GRAFANA_APP = "grafana"
+DS_NAME = "osm_prometheus"
+DS_URL = "http://prometheus:9090"
+
+
+async def _run_action_create_datasource(ops_test: OpsTest, name: str, url: str):
+ action = (
+ await ops_test.model.applications[MON_APP]
+ .units[0]
+ .run_action(
+ action_name="create-datasource",
+ name=name,
+ url=url,
+ )
+ )
+ return await action.wait()
+
+
+async def _run_action_list_datasources(ops_test: OpsTest):
+ action = (
+ await ops_test.model.applications[MON_APP]
+ .units[0]
+ .run_action(action_name="list-datasources")
+ )
+ return await action.wait()
+
+
+async def _run_action_delete_datasource(ops_test: OpsTest, name: str):
+ action = (
+ await ops_test.model.applications[MON_APP]
+ .units[0]
+ .run_action(action_name="delete-datasource", name=name)
+ )
+ return await action.wait()
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_and_grafana_are_idle(ops_test: OpsTest):
+ charm = await ops_test.build_charm(".")
+ resources = {"mon-image": METADATA["resources"]["mon-image"]["upstream-source"]}
+
+ await asyncio.gather(
+ ops_test.model.deploy(
+ charm, resources=resources, application_name=MON_APP, series="jammy"
+ ),
+ ops_test.model.deploy(GRAFANA_CHARM, application_name=GRAFANA_APP, channel="stable"),
+ )
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[GRAFANA_APP], status="active")
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_cannot_create_datasource_grafana_url_is_not_set(ops_test: OpsTest):
+ action = await _run_action_create_datasource(ops_test, DS_NAME, DS_URL)
+ assert action.status == "failed"
+ assert "Invalid URL" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_cannot_list_datasources_grafana_url_is_not_set(ops_test: OpsTest):
+ action = await _run_action_list_datasources(ops_test)
+ assert action.status == "failed"
+ assert "Invalid URL" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_cannot_delete_datasource_grafana_url_is_not_set(ops_test: OpsTest):
+ action = await _run_action_delete_datasource(ops_test, "prometheus")
+ assert action.status == "failed"
+ assert "Invalid URL" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_cannot_create_datasource_due_to_invalid_grafana_password(ops_test: OpsTest):
+ await ops_test.model.applications[MON_APP].set_config(
+ {"grafana-url": f"http://{GRAFANA_APP}:3000", "grafana-user": "admin"}
+ )
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+
+ action = await _run_action_create_datasource(ops_test, DS_NAME, DS_URL)
+ assert action.status == "failed"
+ assert action.message == "invalid username or password"
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_cannot_create_datasource_invalid_url_fails(ops_test: OpsTest):
+ await ops_test.model.applications[MON_APP].set_config(
+ {"grafana-url": f"http://{GRAFANA_APP}:3000"}
+ )
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+
+ action = await _run_action_create_datasource(ops_test, DS_NAME, "prometheus:9090")
+ assert action.status == "failed"
+ assert action.message == "Invalid datasource url 'prometheus:9090'"
+
+
+async def _get_grafana_admin_password(ops_test: OpsTest):
+ action = (
+ await ops_test.model.applications[GRAFANA_APP].units[0].run_action("get-admin-password")
+ )
+ admin_password = (await action.wait()).results["admin-password"]
+ logger.info(f"Password obtained from {GRAFANA_APP} : {admin_password}")
+ assert admin_password != "Admin password has been changed by an administrator"
+ return admin_password
+
+
+@pytest.mark.abort_on_fail
+async def test_grafana_password_is_set_in_mon_config(ops_test: OpsTest):
+ admin_password = await _get_grafana_admin_password(ops_test)
+
+ default_password = None
+ grafana_password_in_mon = (
+ (await ops_test.model.applications[MON_APP].get_config())
+ .get("grafana-password")
+ .get("value")
+ )
+ assert grafana_password_in_mon == default_password
+
+ await ops_test.model.applications[MON_APP].set_config({"grafana-password": admin_password})
+
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[GRAFANA_APP], status="active")
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+
+ grafana_password_in_mon = (
+ (await ops_test.model.applications[MON_APP].get_config())
+ .get("grafana-password")
+ .get("value")
+ )
+ assert grafana_password_in_mon == admin_password
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_create_datasource_successfully(ops_test: OpsTest):
+ action = await _run_action_create_datasource(ops_test, DS_NAME, DS_URL)
+ assert action.status == "completed"
+ assert action.results["datasource-name"] == DS_NAME
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_list_datasources_returns_osm_prometheus(ops_test: OpsTest):
+ action = await _run_action_list_datasources(ops_test)
+ assert action.status == "completed"
+ datasources = json.loads(action.results.get("datasources"))
+ assert len(datasources) == 1
+ assert datasources[0].get("name") == DS_NAME
+ assert datasources[0].get("type") == "prometheus"
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_create_datasource_that_already_exists_returns_error_message(ops_test: OpsTest):
+ action = await _run_action_create_datasource(ops_test, DS_NAME, DS_URL)
+ assert action.status == "failed"
+ assert action.message == "data source with the same name already exists"
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_delete_existing_datasource_returns_success_message(ops_test: OpsTest):
+ action = await _run_action_delete_datasource(ops_test, DS_NAME)
+ assert action.status == "completed"
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_list_datasources_is_empty(ops_test: OpsTest):
+ action = await _run_action_list_datasources(ops_test)
+ assert action.status == "completed"
+ datasources = json.loads(action.results.get("datasources"))
+ assert len(datasources) == 0
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_delete_non_existing_datasource_returns_error_message(ops_test: OpsTest):
+ action = await _run_action_delete_datasource(ops_test, DS_NAME)
+ assert action.status == "failed"
+ assert action.message == "Data source not found"
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_create_datasource_incorrect_grafana_host_fails(ops_test: OpsTest):
+ await ops_test.model.applications[MON_APP].set_config(
+ {"grafana-url": f"http://{GRAFANA_APP}-k8s:3000"}
+ )
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+ action = await _run_action_create_datasource(ops_test, DS_NAME, DS_URL)
+ assert action.status == "failed"
+ assert f"Failed to resolve '{GRAFANA_APP}-k8s'" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_list_datasources_incorrect_grafana_host_fails(ops_test: OpsTest):
+ action = await _run_action_list_datasources(ops_test)
+ assert action.status == "failed"
+ assert f"Failed to resolve '{GRAFANA_APP}-k8s'" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_delete_datasource_incorrect_grafana_host_fails(ops_test: OpsTest):
+ action = await _run_action_delete_datasource(ops_test, DS_NAME)
+ assert action.status == "failed"
+ assert f"Failed to resolve '{GRAFANA_APP}-k8s'" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_incorrect_grafana_port_returns_timeout_message(ops_test: OpsTest):
+ await ops_test.model.applications[MON_APP].set_config(
+ {"grafana-url": f"http://{GRAFANA_APP}:3001"}
+ )
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+ action = await _run_action_create_datasource(ops_test, DS_NAME, DS_URL)
+ assert action.status == "failed"
+ assert f"Connection to {GRAFANA_APP} timed out" in action.message
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_list_datasources_incorrect_user_fails(ops_test: OpsTest):
+ await ops_test.model.applications[MON_APP].set_config(
+ {"grafana-url": f"http://{GRAFANA_APP}:3000", "grafana-user": "some_user"}
+ )
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+ action = await _run_action_list_datasources(ops_test)
+ assert action.status == "failed"
+ assert action.message == "invalid username or password"
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_delete_datasource_incorrect_user_fails(ops_test: OpsTest):
+ action = await _run_action_delete_datasource(ops_test, DS_NAME)
+ assert action.status == "failed"
+ assert action.message == "invalid username or password"
projects / osm / devops.git / blobdiff