Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,3 +38,19 @@ Or run tests from a specific submodule.
```sh
py.test --pyargs stackhpc_cloud_tests.test_prometheus
```

To skip specific tests, set the SKIP_SCT variable:
| Test to skip | SKIP_SCT value |
|---------------|----------------|
|test_docker.py | docker |
|test_ntp.py | ntp |
|test_selinux.py| selinux |
|test_grafana.py| grafana |
|test_opensearch.py| opensearch |
|test_prometheus.py| prometheus |

To skip more than one test, set SKIP_SCT to a comma separated list:
```sh
# Skip both test_docker.py and test_ntp.py
export SKIP_SCT=docker,ntp
```
18 changes: 12 additions & 6 deletions stackhpc_cloud_tests/host/test_docker.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@
from packaging.version import parse
import pytest

skip = False
if os.environ.get("SKIP_SCT") is not None:
skip_sct = os.environ.get("SKIP_SCT")
if "docker" in skip_sct:
skip = True

@pytest.fixture
def docker_info(host, scope="session"):
Expand All @@ -25,7 +30,7 @@ def docker_info(host, scope="session"):
docker_info = host.check_output("docker info --format json")
return json.loads(docker_info)


@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_docker_version(host):
"""Check that Docker is accessible and optionally check version."""
# An optional inclusive minimum version.
Expand All @@ -44,7 +49,7 @@ def test_docker_version(host):
assert client_version < max_version
assert server_version < max_version


@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_docker_containers(subtests, host):
"""Check that Docker containers are healthy."""
with host.sudo("stack"):
Expand All @@ -65,12 +70,12 @@ def test_docker_containers(subtests, host):
assert "max-file" in state["HostConfig"]["LogConfig"]["Config"]
assert "max-size" in state["HostConfig"]["LogConfig"]["Config"]


@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_docker_driver(docker_info):
"""Check that Docker is using the overlay2 storage driver."""
assert docker_info["Driver"] == "overlay2"


@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_no_bridge_network_exists(host):
"""Check that no bridge network exists."""
with host.sudo("stack"):
Expand All @@ -80,12 +85,12 @@ def test_no_bridge_network_exists(host):
assert network["Name"] != "bridge"
assert network["Driver"] != "bridge"


@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_ip_forwarding_disabled(docker_info):
"""Check that IP forwarding is disabled."""
assert not docker_info["IPv4Forwarding"]


@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_iptables_disabled(host):
"""Check that IPTables manipulation is disabled."""
# (MaxN) "docker system info" for version 27.4.1 will report "true" for "BridgeNfIptables" and "BridgeNfIp6tables"
Expand All @@ -95,6 +100,7 @@ def test_iptables_disabled(host):
assert "FORWARD" in iptables_chains
assert "DOCKER" not in iptables_chains

@pytest.mark.skipif(skip, reason="Docker tests skipped.")
def test_live_restore_enabled(docker_info):
"""Check that live restore is enabled."""
assert docker_info["LiveRestoreEnabled"]
14 changes: 10 additions & 4 deletions stackhpc_cloud_tests/host/test_ntp.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,21 @@

MAX_OFFSET_SECONDS = 0.5 # 500 ms limit

skip = False
if os.environ.get("SKIP_SCT") is not None:
skip_sct = os.environ.get("SKIP_SCT")
if "ntp" in skip_sct:
skip = True

@pytest.mark.skipif(skip, reason="NTP tests skipped.")
def _get_parsed_ntp_sources():
# fetch and parse NTP_SOURCES
raw_ntp_sources = os.environ.get("NTP_SOURCES")
if not raw_ntp_sources or not raw_ntp_sources.strip():
return []
return [s.strip() for s in raw_ntp_sources.split(",") if s.strip()]


@pytest.mark.skipif(skip, reason="NTP tests skipped.")
def test_ntp_sources_env_var():
"""Verify NTP_SOURCES is set."""
raw_ntp_sources = os.environ.get("NTP_SOURCES")
Expand All @@ -38,7 +44,7 @@ def test_ntp_sources_env_var():
ntp_sources = _get_parsed_ntp_sources()
assert ntp_sources, "NTP_SOURCES contains no valid source entries."


@pytest.mark.skipif(skip, reason="NTP tests skipped.")
def test_chrony_service_and_sync_status(host):
"""Check that chrony is running and syncing properly"""
# Check timedatectl
Expand All @@ -56,7 +62,7 @@ def test_chrony_service_and_sync_status(host):
f"Output:\n{sources_stdout}"
)


@pytest.mark.skipif(skip, reason="NTP tests skipped.")
def test_active_source_matches_expected(host):
"""Check that one of the NTP_SOURCES values is actively used by chrony (*=current source)"""
ntp_sources = _get_parsed_ntp_sources()
Expand All @@ -77,7 +83,7 @@ def test_active_source_matches_expected(host):
f"Chronyc Sources Output:\n{sources_stdout}"
)


@pytest.mark.skipif(skip, reason="NTP tests skipped.")
def test_ntp_time_offset(host):
"""Check time offset"""
chrony_stdout = host.check_output("chronyc tracking")
Expand Down
6 changes: 6 additions & 0 deletions stackhpc_cloud_tests/host/test_selinux.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,13 @@
import os
import pytest

skip = False
if os.environ.get("SKIP_SCT") is not None:
skip_sct = os.environ.get("SKIP_SCT")
if "selinux" in skip_sct:
skip = True

@pytest.mark.skipif(skip, reason="SELinux tests skipped.")
def test_selinux(host):
"""Check that SELinux is enabled and permissive on supported systems."""
# Adapted from Kayobe host configure tests:
Expand Down
7 changes: 6 additions & 1 deletion stackhpc_cloud_tests/monitoring/test_grafana.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@
import os
import pytest

skip = False
if os.environ.get("SKIP_SCT") is not None:
skip_sct = os.environ.get("SKIP_SCT")
if "grafana" in skip_sct:
skip = True

@pytest.fixture
def grafana() -> GrafanaApi:
Expand All @@ -31,7 +36,7 @@ def grafana() -> GrafanaApi:
credential=(grafana_username, grafana_password),
)


@pytest.mark.skipif(skip, reason="Grafana tests skipped.")
def test_grafana_api_stats(grafana):
"""Test that Grafana API stats are accessible."""
# https://grafana.com/docs/grafana/latest/developers/http_api/admin/#grafana-stats
Expand Down
9 changes: 7 additions & 2 deletions stackhpc_cloud_tests/monitoring/test_opensearch.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,11 @@

from stackhpc_cloud_tests import utils

skip = False
if os.environ.get("SKIP_SCT") is not None:
skip_sct = os.environ.get("SKIP_SCT")
if "grafana" in skip_sct:
skip = True

@pytest.fixture
def opensearch() -> OpenSearch:
Expand All @@ -47,7 +52,7 @@ def opensearch() -> OpenSearch:
ssl_show_warn=False,
)


@pytest.mark.skipif(skip, reason="OpenSearch tests skipped.")
def test_opensearch_has_info_logs(opensearch):
"""Check that OpenSearch has some INFO level logs."""
query = {
Expand All @@ -61,7 +66,7 @@ def test_opensearch_has_info_logs(opensearch):
result = opensearch.search(body=query, index="flog-*", size=1)
assert len(result["hits"]["hits"]) == 1


@pytest.mark.skipif(skip, reason="OpenSearch tests skipped.")
def test_opensearch_dashboards_status():
"""Check that OpenSearch Dashboards is accessible and is in a green state."""
dashboard_url = os.environ["OPENSEARCH_DASHBOARDS_URL"]
Expand Down
11 changes: 8 additions & 3 deletions stackhpc_cloud_tests/monitoring/test_prometheus.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,11 @@
from prometheus_api_client import PrometheusConnect
import pytest

skip = False
if os.environ.get("SKIP_SCT") is not None:
skip_sct = os.environ.get("SKIP_SCT")
if "prometheus" in skip_sct:
skip = True

@pytest.fixture
def prom() -> PrometheusConnect:
Expand All @@ -29,18 +34,18 @@ def prom() -> PrometheusConnect:
kwargs["auth"] = (prometheus_username, prometheus_password)
return PrometheusConnect(url=prometheus_url, disable_ssl=True, **kwargs)


@pytest.mark.skipif(skip, reason="Prometheus tests skipped.")
def test_prometheus_connection(prom):
"""Check that Prometheus is accessible."""
assert prom.check_prometheus_connection()


@pytest.mark.skipif(skip, reason="Prometheus tests skipped.")
def test_prometheus_node_exporter_metrics(prom):
"""Check that expected node exporter metrics exist."""
metrics = prom.all_metrics()
assert "node_cpu_seconds_total" in metrics


@pytest.mark.skipif(skip, reason="Prometheus tests skipped.")
def test_prometheus_alerts_inactive(prom):
"""Check that no Prometheus alerts are active."""
# https://prometheus.io/docs/prometheus/latest/querying/api/#alerts
Expand Down