+++ /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