rolling_update = os.environ.get("ROLLING_UPDATE", "False")
group_names = ansible_vars["group_names"]
docker = ansible_vars.get("docker")
+ dashboard = ansible_vars.get("dashboard_enabled", True)
radosgw_num_instances = ansible_vars.get("radosgw_num_instances", 1)
ceph_release_num = {
'jewel': 10,
pytest.skip(
"Not a valid test for non-containerized deployments or atomic hosts") # noqa E501
+ if request.node.get_closest_marker("dashboard") and not dashboard:
+ pytest.skip(
+ "Not a valid test with dashboard disabled")
data = dict(
vars=ansible_vars,
item.add_marker(pytest.mark.nfss)
elif "iscsi" in test_path:
item.add_marker(pytest.mark.iscsigws)
+ elif "grafana" in test_path:
+ item.add_marker(pytest.mark.grafanas)
else:
item.add_marker(pytest.mark.all)
--- /dev/null
+import pytest
+
+
+class TestGrafanas(object):
+
+ @pytest.mark.dashboard
+ @pytest.mark.no_docker
+ def test_grafana_dashboard_is_installed(self, node, host):
+ assert host.package("ceph-grafana-dashboards").is_installed
+
+ @pytest.mark.dashboard
+ @pytest.mark.parametrize('svc', [
+ 'alertmanager', 'grafana-server', 'prometheus'
+ ])
+ def test_grafana_service_enabled_and_running(self, node, host, svc):
+ s = host.service(svc)
+ assert s.is_enabled
+ assert s.is_running
+
+ @pytest.mark.dashboard
+ @pytest.mark.parametrize('port', [
+ '3000', '9092', '9093'
+ ])
+ def test_grafana_socket(self, node, host, setup, port):
+ s = host.socket('tcp://%s:%s' % (setup["address"], port))
+ assert s.is_listening
def test_mgr_is_installed(self, node, host):
assert host.package("ceph-mgr").is_installed
- def test_mgr_service_is_running(self, node, host):
- service_name = "ceph-mgr@{hostname}".format(
- hostname=node["vars"]["inventory_hostname"]
- )
- assert host.service(service_name).is_running
+ @pytest.mark.dashboard
+ @pytest.mark.no_docker
+ def test_mgr_dashboard_is_installed(self, node, host):
+ assert host.package("ceph-mgr-dashboard").is_installed
- def test_mgr_service_is_enabled(self, node, host):
+ def test_mgr_service_is_enabled_and_running(self, node, host):
service_name = "ceph-mgr@{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
- assert host.service(service_name).is_enabled
+ s = host.service(service_name)
+ assert s.is_enabled
+ assert s.is_running
+
+ @pytest.mark.dashboard
+ @pytest.mark.parametrize('port', [
+ '8443', '9283'
+ ])
+ def test_mgr_dashboard_is_listening(self, node, host, port):
+ assert host.socket('tcp://%s' % port).is_listening
def test_mgr_is_up(self, node, host, setup):
hostname = node["vars"]["inventory_hostname"]
--- /dev/null
+import pytest
+
+
+class TestNodeExporter(object):
+
+ @pytest.mark.dashboard
+ def test_node_exporter_service_enabled_and_running(self, node, host):
+ s = host.service("node_exporter")
+ assert s.is_enabled
+ assert s.is_running
+
+ @pytest.mark.dashboard
+ def test_node_exporter_socket(self, node, host):
+ assert host.socket('tcp://9100').is_listening