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.get("grafana-url", ""),
- "OSMMON_GRAFANA_USER": self.config.get("grafana-user", ""),
- "OSMMON_GRAFANA_PASSWORD": self.config.get("grafana-password", ""),
+ "OSMMON_GRAFANA_URL": self.config["grafana-url"],
+ "OSMMON_GRAFANA_USER": self.config["grafana-user"],
+ "OSMMON_GRAFANA_PASSWORD": self.config["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_and_other_charms_are_idle(ops_test: OpsTest):
+async def test_mon_is_deployed(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)
-
-
-@pytest.mark.abort_on_fail
-async def test_mon_is_blocked_due_to_missing_grafana_config(ops_test: OpsTest):
+ await ops_test.model.wait_for_idle(
+ apps=APPS,
+ )
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