get-debug-mode-information:
description: Get information to debug the container
+
+create-datasource:
+ description: Create a new prometheus datasource.
+ params:
+ name:
+ type: string
+ description: The name of the datasource.
+ url:
+ type: string
+ description: URL to prometheus.
+ required: [name, url]
+ additionalProperties: false
+
+list-datasources:
+ description: List datasources.
+ additionalProperties: false
+
+delete-datasource:
+ description: Delete a given datasource.
+ params:
+ name:
+ type: string
+ description: The name of the datasource.
+ required: [name]
+ additionalProperties: false
grafana-url:
description: Grafana URL
type: string
- default: http://grafana:3000
grafana-user:
description: Grafana user
type: string
- default: admin
grafana-password:
description: Grafana password
type: string
- default: admin
keystone-enabled:
description: MON will use Keystone backend
type: boolean
ops < 2.2
lightkube
lightkube-models
+requests
# git+https://github.com/charmed-osm/config-validator/
import logging
from typing import Any, Dict
+from urllib.parse import urlparse
from charms.data_platform_libs.v0.data_interfaces import DatabaseRequires
from charms.kafka_k8s.v0.kafka import KafkaRequires, _KafkaAvailableEvent
from ops.main import main
from ops.model import ActiveStatus, Container
+from grafana_datasource_handler import (
+ DatasourceConfig,
+ GrafanaConfig,
+ GrafanaDataSourceHandler,
+)
from legacy_interfaces import KeystoneClient, PrometheusClient
HOSTPATHS = [
}
event.set_results(debug_info)
+ def _on_create_datasource_action(self, event: ActionEvent) -> None:
+ """Handler for the create-datasource action event."""
+ url = event.params["url"]
+ if not self._is_valid_url(url):
+ event.fail(f"Invalid datasource url '{url}'")
+ return
+ grafana_config = self._get_grafana_config()
+ datasource_config = DatasourceConfig(event.params["name"], url)
+ response = GrafanaDataSourceHandler.create_datasource(grafana_config, datasource_config)
+ logger.debug(response)
+ if response.is_success:
+ event.set_results(response.results)
+ return
+ event.fail(response.message)
+
+ def _on_list_datasources_action(self, event: ActionEvent) -> None:
+ """Handler for the list-datasource action event."""
+ grafana_config = self._get_grafana_config()
+ response = GrafanaDataSourceHandler.list_datasources(grafana_config)
+ logger.debug(response)
+ if response.is_success:
+ event.set_results(response.results)
+ return
+ event.fail(response.message)
+
+ def _on_delete_datasource_action(self, event: ActionEvent) -> None:
+ """Handler for the delete-datasource action event."""
+ datasource_name = event.params["name"]
+ grafana_config = self._get_grafana_config()
+ response = GrafanaDataSourceHandler.delete_datasource(grafana_config, datasource_name)
+ logger.debug(response)
+ if not response.is_success:
+ event.fail(response.message)
+
+ def _get_grafana_config(self) -> GrafanaConfig:
+ return GrafanaConfig(
+ self.config.get("grafana-user", ""),
+ self.config.get("grafana-password", ""),
+ self.config.get("grafana-url", ""),
+ )
+
# ---------------------------------------------------------------------------
# Validation and configuration and more
# ---------------------------------------------------------------------------
self.on["mongodb"].relation_broken: self._on_required_relation_broken,
# Action events
self.on.get_debug_mode_information_action: self._on_get_debug_mode_information_action,
+ self.on.create_datasource_action: self._on_create_datasource_action,
+ self.on.list_datasources_action: self._on_list_datasources_action,
+ self.on.delete_datasource_action: self._on_delete_datasource_action,
}
for relation in [self.on[rel_name] for rel_name in ["prometheus", "keystone"]]:
event_handler_mapping[relation.relation_changed] = self._on_config_changed
CharmError: if charm configuration is invalid.
"""
logger.debug("validating charm config")
+ missing_configs = []
+ grafana_configs = ["grafana-url", "grafana-user", "grafana-password"]
+ for config in grafana_configs:
+ if not self.config.get(config):
+ missing_configs.append(config)
+
+ if missing_configs:
+ config_str = ", ".join(missing_configs)
+ error_msg = f"need {config_str} config"
+ logger.warning(error_msg)
+ raise CharmError(error_msg)
+
+ grafana_url = self.config["grafana-url"]
+ if not self._is_valid_url(grafana_url):
+ raise CharmError(f"Invalid value for grafana-url config: '{grafana_url}'")
+
+ def _is_valid_url(self, url) -> bool:
+ return urlparse(url).hostname is not None
def _check_relations(self) -> None:
"""Validate charm relations.
"OSMMON_PROMETHEUS_URL": f"http://{self.prometheus_client.hostname}:{self.prometheus_client.port}",
"OSMMON_PROMETHEUS_USER": self.prometheus_client.user,
"OSMMON_PROMETHEUS_PASSWORD": self.prometheus_client.password,
- "OSMMON_GRAFANA_URL": self.config["grafana-url"],
- "OSMMON_GRAFANA_USER": self.config["grafana-user"],
- "OSMMON_GRAFANA_PASSWORD": self.config["grafana-password"],
+ "OSMMON_GRAFANA_URL": self.config.get("grafana-url", ""),
+ "OSMMON_GRAFANA_USER": self.config.get("grafana-user", ""),
+ "OSMMON_GRAFANA_PASSWORD": self.config.get("grafana-password", ""),
"OSMMON_KEYSTONE_ENABLED": self.config["keystone-enabled"],
"OSMMON_KEYSTONE_URL": self.keystone_client.host,
"OSMMON_KEYSTONE_DOMAIN_NAME": self.keystone_client.user_domain_name,
--- /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 at: https://juju.is/docs/sdk
+
+"""Module to handle Grafana datasource operations."""
+
+import json
+import logging
+from dataclasses import dataclass
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class GrafanaConfig:
+ """Values needed to make grafana API calls."""
+
+ user: str
+ password: str
+ url: str
+
+
+@dataclass
+class DatasourceConfig:
+ """Information about the datasource to create."""
+
+ name: str
+ url: str
+
+
+@dataclass
+class DatasourceResponse:
+ """Return value used by GrafanaDataSourceHandler operations."""
+
+ is_success: bool
+ message: str
+ results: dict
+
+
+class GrafanaDataSourceHandler:
+ """Handle grafana datasource pperations."""
+
+ @staticmethod
+ def create_datasource(
+ grafana_config: GrafanaConfig, datasource_config: DatasourceConfig
+ ) -> DatasourceResponse:
+ """Calls the Grafana API to create a new prometheus datasource."""
+ try:
+ auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
+ url = grafana_config.url + "/api/datasources"
+ datasource_data = {
+ "name": datasource_config.name,
+ "type": "prometheus",
+ "url": datasource_config.url,
+ "access": "proxy",
+ "readOnly": False,
+ "basicAuth": False,
+ }
+ response = requests.post(url, json=datasource_data, auth=auth)
+ response_content = response.json()
+ results = {"datasource-name": response_content.get("name")}
+ return DatasourceResponse(
+ response.ok, response_content.get("message"), results=results
+ )
+ except Exception as e:
+ logger.debug(f"Exception processing request for creating datasource: {e}")
+ return DatasourceResponse(False, str(e), results={})
+
+ @staticmethod
+ def list_datasources(grafana_config: GrafanaConfig) -> DatasourceResponse:
+ """Calls the Grafana API to get a list of datasources."""
+ try:
+ auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
+ url = grafana_config.url + "/api/datasources"
+ response = requests.get(url, auth=auth)
+ response_content = response.json()
+ results = {"datasources": json.dumps(response_content)}
+ message = response_content.get("message") if not response.ok else ""
+ return DatasourceResponse(response.ok, message=message, results=results)
+ except Exception as e:
+ logger.debug(f"Exception processing request to list datasources: {e}")
+ return DatasourceResponse(False, str(e), results={})
+
+ @staticmethod
+ def delete_datasource(
+ grafana_config: GrafanaConfig, datasource_name: str
+ ) -> DatasourceResponse:
+ """Calls the Grafana API to delete a given datasource by name."""
+ try:
+ auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
+ url = grafana_config.url + f"/api/datasources/name/{datasource_name}"
+ response = requests.delete(url, auth=auth)
+ response_content = response.json()
+ return DatasourceResponse(response.ok, response_content.get("message"), results={})
+ except Exception as e:
+ logger.debug(f"Exception processing request for deleting datasource: {e}")
+ return DatasourceResponse(False, str(e), results={})
@pytest.mark.abort_on_fail
-async def test_mon_is_deployed(ops_test: OpsTest):
+async def test_mon_and_other_charms_are_idle(ops_test: OpsTest):
charm = await ops_test.build_charm(".")
resources = {"mon-image": METADATA["resources"]["mon-image"]["upstream-source"]}
await ops_test.run(*shlex.split(cmd), check=True)
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=APPS,
- )
+ await ops_test.model.wait_for_idle(apps=APPS)
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_is_blocked_due_to_missing_grafana_config(ops_test: OpsTest):
assert ops_test.model.applications[MON_APP].status == "blocked"
unit = ops_test.model.applications[MON_APP].units[0]
+ assert (
+ unit.workload_status_message == "need grafana-url, grafana-user, grafana-password config"
+ )
+
+ await ops_test.model.applications[MON_APP].set_config({"grafana-url": "new_value"})
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+ assert unit.workload_status_message == "need grafana-user, grafana-password config"
+
+ await ops_test.model.applications[MON_APP].set_config({"grafana-password": "new_value"})
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+ assert unit.workload_status_message == "need grafana-user config"
+
+ await ops_test.model.applications[MON_APP].set_config({"grafana-user": "new_value"})
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+
+ assert unit.workload_status_message == "Invalid value for grafana-url config: 'new_value'"
+ await ops_test.model.applications[MON_APP].set_config({"grafana-url": "http://valid:92"})
+
+
+@pytest.mark.abort_on_fail
+async def test_mon_is_blocked_due_to_missing_relations(ops_test: OpsTest):
+ async with ops_test.fast_forward():
+ await ops_test.model.wait_for_idle(apps=[MON_APP], status="blocked")
+ unit = ops_test.model.applications[MON_APP].units[0]
assert unit.workload_status_message == "need kafka, mongodb, prometheus, keystone relations"
logger.info("Adding relations for other components")
await ops_test.model.add_relation(MON_APP, PROMETHEUS_APP)
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=APPS,
- status="active",
- )
+ await ops_test.model.wait_for_idle(apps=APPS, status="active")
@pytest.mark.abort_on_fail
assert unit.workload_status_message == f"need {relation_to_remove} relation"
await ops_test.model.add_relation(MON_APP, relation_to_remove)
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=APPS,
- status="active",
- )
+ await ops_test.model.wait_for_idle(apps=APPS, status="active")
@pytest.mark.abort_on_fail
async def test_mon_action_debug_mode_disabled(ops_test: OpsTest):
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=APPS,
- status="active",
- )
+ await ops_test.model.wait_for_idle(apps=APPS, status="active")
logger.info("Running action 'get-debug-mode-information'")
action = (
await ops_test.model.applications[MON_APP]
),
)
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=[VCA_APP],
- )
+ await ops_test.model.wait_for_idle(apps=[VCA_APP])
controllers = (Path.home() / ".local/share/juju/controllers.yaml").read_text()
accounts = (Path.home() / ".local/share/juju/accounts.yaml").read_text()
public_key = (Path.home() / ".local/share/juju/ssh/juju_id_rsa.pub").read_text()
}
)
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=APPS + [VCA_APP],
- status="active",
- )
+ await ops_test.model.wait_for_idle(apps=APPS + [VCA_APP], status="active")
await ops_test.model.add_relation(MON_APP, VCA_APP)
async with ops_test.fast_forward():
- await ops_test.model.wait_for_idle(
- apps=APPS + [VCA_APP],
- status="active",
- )
+ await ops_test.model.wait_for_idle(apps=APPS + [VCA_APP], status="active")
--- /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"
harness.cleanup()
+def _set_grafana_config(harness: Harness):
+ harness.update_config(
+ {
+ "grafana-url": "http://prometheus:1234",
+ "grafana-user": "user",
+ "grafana-password": "password",
+ }
+ )
+
+
+def test_default_grafana_config_is_invalid_charm_is_blocked(harness: Harness):
+ harness.charm.on.config_changed.emit()
+ assert type(harness.charm.unit.status) == BlockedStatus
+ assert (
+ "need grafana-url, grafana-user, grafana-password config"
+ == harness.charm.unit.status.message
+ )
+
+
def test_missing_relations(harness: Harness):
+ _set_grafana_config(harness)
harness.charm.on.config_changed.emit()
assert type(harness.charm.unit.status) == BlockedStatus
assert all(
)
+def test_grafana_url_without_schema_block_status(harness: Harness):
+ harness.update_config(
+ {"grafana-url": "foo.com", "grafana-user": "user", "grafana-password": "password"}
+ )
+ assert type(harness.charm.unit.status) == BlockedStatus
+ assert "Invalid value for grafana-url config: 'foo.com'" == harness.charm.unit.status.message
+
+
+def test_grafana_url_with_port_without_schema_block_status(harness: Harness):
+ harness.update_config(
+ {"grafana-url": "foo.com:9090", "grafana-user": "user", "grafana-password": "password"}
+ )
+ assert type(harness.charm.unit.status) == BlockedStatus
+ assert (
+ "Invalid value for grafana-url config: 'foo.com:9090'" == harness.charm.unit.status.message
+ )
+
+
+def test_grafana_url_without_port_is_valid(harness: Harness):
+ _add_relations(harness)
+ harness.update_config(
+ {"grafana-url": "http://foo", "grafana-user": "user", "grafana-password": "password"}
+ )
+ assert harness.charm.unit.status == ActiveStatus()
+
+
+def test_grafana_url_with_port_is_valid(harness: Harness):
+ _add_relations(harness)
+ harness.update_config(
+ {"grafana-url": "http://foo:90", "grafana-user": "user", "grafana-password": "password"}
+ )
+ assert harness.charm.unit.status == ActiveStatus()
+
+
def test_ready(harness: Harness):
+ _set_grafana_config(harness)
_add_relations(harness)
assert harness.charm.unit.status == ActiveStatus()
def test_container_stops_after_relation_broken(harness: Harness):
+ _set_grafana_config(harness)
harness.charm.on[container_name].pebble_ready.emit(container_name)
container = harness.charm.unit.get_container(container_name)
relation_ids = _add_relations(harness)
projects / osm / devops.git / commitdiff