From f503bfe555bbcf2325425b942ba598f6493dfbae Mon Sep 17 00:00:00 2001 From: Yoda Monplub Date: Tue, 18 Aug 2026 16:42:07 +0100 Subject: [PATCH 1/2] add ntp test --- stackhpc_cloud_tests/host/test_ntp.py | 78 +++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 stackhpc_cloud_tests/host/test_ntp.py diff --git a/stackhpc_cloud_tests/host/test_ntp.py b/stackhpc_cloud_tests/host/test_ntp.py new file mode 100644 index 0000000..3a4cbe8 --- /dev/null +++ b/stackhpc_cloud_tests/host/test_ntp.py @@ -0,0 +1,78 @@ +import os +import re +import subprocess + +MAX_OFFSET_SECONDS = 0.5 # 500 ms limit + + +def test_ntp_configuration_and_sync(): + raw_ntp_sources = os.environ.get("NTP_SOURCES") + + # 1. Verify NTP_SOURCES environment variable and parse into a list + assert ( + raw_ntp_sources is not None and raw_ntp_sources.strip() != "" + ), "NTP_SOURCES environment variable is not set or empty." + + # Split comma-separated string into a clean list of individual source hosts/IPs + ntp_sources = [s.strip() for s in raw_ntp_sources.split(",") if s.strip()] + assert ntp_sources, "NTP_SOURCES contains no valid source entries." + + # 2. Basic service check via timedatectl + timedate_res = subprocess.run( + ["timedatectl", "status"], + capture_output=True, + text=True, + check=True, + ) + assert ( + "NTP service: active" in timedate_res.stdout + or "System clock synchronized: yes" in timedate_res.stdout + ), f"System clock is not synchronized according to timedatectl:\n{timedate_res.stdout}" + + # 3. Check that one of the NTP_SOURCES values is actively used by chrony (* = current synchronized source) + sources_res = subprocess.run( + ["chronyc", "-n", "sources"], + capture_output=True, + text=True, + check=True, + ) + + # In 'chronyc sources', the line starting with '*' or '^*' indicates the active reference source. + active_source_match = re.search( + r"^\^?\*\s+([^\s]+)", sources_res.stdout, re.MULTILINE + ) + assert active_source_match is not None, ( + f"Chrony has no active reference source (no source marked with '*').\n" + f"Output:\n{sources_res.stdout}" + ) + + active_source = active_source_match.group(1) + + # Validate that active_source matches ANY server in your ntp_sources list + is_valid_source = any( + src in active_source or active_source in src for src in ntp_sources + ) + assert is_valid_source, ( + f"Active chrony source '{active_source}' does not match any expected NTP_SOURCES in {ntp_sources}.\n" + f"Chronyc Sources Output:\n{sources_res.stdout}" + ) + + # 4. Check exact time offset using chronyc tracking + chrony_res = subprocess.run( + ["chronyc", "tracking"], + capture_output=True, + text=True, + check=True, + ) + + # Output line example: "System time : 0.000012345 seconds slow of NTP time" + match = re.search(r"System time\s+:\s+([0-9.]+)\s+seconds", chrony_res.stdout) + assert ( + match is not None + ), f"Could not parse system time offset from chronyc output:\n{chrony_res.stdout}" + + offset = float(match.group(1)) + assert offset <= MAX_OFFSET_SECONDS, ( + f"NTP time offset {offset:.4f}s exceeds maximum threshold of {MAX_OFFSET_SECONDS}s (500ms).\n" + f"Chronyc Tracking:\n{chrony_res.stdout}" + ) From 7a0a11bfa0df586caa5008c6f42de123e8185f7b Mon Sep 17 00:00:00 2001 From: Yoda Monplub Date: Wed, 26 Aug 2026 12:02:40 +0100 Subject: [PATCH 2/2] split into 4 functions and change to use host.check_output(...) --- stackhpc_cloud_tests/host/test_ntp.py | 85 ++++++++++++++------------- 1 file changed, 44 insertions(+), 41 deletions(-) diff --git a/stackhpc_cloud_tests/host/test_ntp.py b/stackhpc_cloud_tests/host/test_ntp.py index 3a4cbe8..65c26cf 100644 --- a/stackhpc_cloud_tests/host/test_ntp.py +++ b/stackhpc_cloud_tests/host/test_ntp.py @@ -1,78 +1,81 @@ import os import re -import subprocess +import pytest MAX_OFFSET_SECONDS = 0.5 # 500 ms limit -def test_ntp_configuration_and_sync(): +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()] - # 1. Verify NTP_SOURCES environment variable and parse into a list + +# 1. Verify NTP_SOURCES is set +def test_ntp_sources_env_var(): + raw_ntp_sources = os.environ.get("NTP_SOURCES") assert ( raw_ntp_sources is not None and raw_ntp_sources.strip() != "" ), "NTP_SOURCES environment variable is not set or empty." - # Split comma-separated string into a clean list of individual source hosts/IPs - ntp_sources = [s.strip() for s in raw_ntp_sources.split(",") if s.strip()] + ntp_sources = _get_parsed_ntp_sources() assert ntp_sources, "NTP_SOURCES contains no valid source entries." - # 2. Basic service check via timedatectl - timedate_res = subprocess.run( - ["timedatectl", "status"], - capture_output=True, - text=True, - check=True, - ) + +# 2. Check that chrony is running and syncing properly +def test_chrony_service_and_sync_status(host): + # Check timedatectl + timedate_stdout = host.check_output("timedatectl status") assert ( - "NTP service: active" in timedate_res.stdout - or "System clock synchronized: yes" in timedate_res.stdout - ), f"System clock is not synchronized according to timedatectl:\n{timedate_res.stdout}" - - # 3. Check that one of the NTP_SOURCES values is actively used by chrony (* = current synchronized source) - sources_res = subprocess.run( - ["chronyc", "-n", "sources"], - capture_output=True, - text=True, - check=True, - ) + "NTP service: active" in timedate_stdout + or "System clock synchronized: yes" in timedate_stdout + ), f"System clock is not synchronized according to timedatectl:\n{timedate_stdout}" - # In 'chronyc sources', the line starting with '*' or '^*' indicates the active reference source. + # Check chrony has an active reference source marked with '*' + sources_stdout = host.check_output("chronyc -n sources") active_source_match = re.search( - r"^\^?\*\s+([^\s]+)", sources_res.stdout, re.MULTILINE + r"^\^?\*\s+([^\s]+)", sources_stdout, re.MULTILINE ) assert active_source_match is not None, ( f"Chrony has no active reference source (no source marked with '*').\n" - f"Output:\n{sources_res.stdout}" + f"Output:\n{sources_stdout}" ) - active_source = active_source_match.group(1) - # Validate that active_source matches ANY server in your ntp_sources list +# 3. Check that one of the NTP_SOURCES values is actively used by chrony (*=current source) +def test_active_source_matches_expected(host): + ntp_sources = _get_parsed_ntp_sources() + assert ntp_sources, "Cannot verify active source because NTP_SOURCES is missing or invalid." + + sources_stdout = host.check_output("chronyc -n sources") + active_source_match = re.search( + r"^\^?\*\s+([^\s]+)", sources_stdout, re.MULTILINE + ) + assert active_source_match is not None, "No active chrony source found to validate." + + active_source = active_source_match.group(1) is_valid_source = any( src in active_source or active_source in src for src in ntp_sources ) assert is_valid_source, ( f"Active chrony source '{active_source}' does not match any expected NTP_SOURCES in {ntp_sources}.\n" - f"Chronyc Sources Output:\n{sources_res.stdout}" + f"Chronyc Sources Output:\n{sources_stdout}" ) - # 4. Check exact time offset using chronyc tracking - chrony_res = subprocess.run( - ["chronyc", "tracking"], - capture_output=True, - text=True, - check=True, - ) - # Output line example: "System time : 0.000012345 seconds slow of NTP time" - match = re.search(r"System time\s+:\s+([0-9.]+)\s+seconds", chrony_res.stdout) +# 4. Check time offset +def test_ntp_time_offset(host): + chrony_stdout = host.check_output("chronyc tracking") + + match = re.search(r"System time\s+:\s+([0-9.]+)\s+seconds", chrony_stdout) assert ( match is not None - ), f"Could not parse system time offset from chronyc output:\n{chrony_res.stdout}" + ), f"Could not parse system time offset from chronyc output:\n{chrony_stdout}" offset = float(match.group(1)) assert offset <= MAX_OFFSET_SECONDS, ( f"NTP time offset {offset:.4f}s exceeds maximum threshold of {MAX_OFFSET_SECONDS}s (500ms).\n" - f"Chronyc Tracking:\n{chrony_res.stdout}" - ) + f"Chronyc Tracking:\n{chrony_stdout}" + ) \ No newline at end of file