From f503bfe555bbcf2325425b942ba598f6493dfbae Mon Sep 17 00:00:00 2001 From: Yoda Monplub Date: Tue, 18 Aug 2026 16:42:07 +0100 Subject: [PATCH 1/3] 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/3] 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 From 6ad19973574c60fbb351d170ff403f04230f0183 Mon Sep 17 00:00:00 2001 From: Yoda Monplub Date: Thu, 27 Aug 2026 10:57:51 +0100 Subject: [PATCH 3/3] add license, add docstrings, add newline at the end, reformat using black --- stackhpc_cloud_tests/host/test_ntp.py | 37 ++++++++++++++++++--------- 1 file changed, 25 insertions(+), 12 deletions(-) diff --git a/stackhpc_cloud_tests/host/test_ntp.py b/stackhpc_cloud_tests/host/test_ntp.py index 65c26cf..227c21f 100644 --- a/stackhpc_cloud_tests/host/test_ntp.py +++ b/stackhpc_cloud_tests/host/test_ntp.py @@ -1,3 +1,18 @@ +# Copyright (c) 2026 StackHPC 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. + + import os import re import pytest @@ -13,8 +28,8 @@ def _get_parsed_ntp_sources(): return [s.strip() for s in raw_ntp_sources.split(",") if s.strip()] -# 1. Verify NTP_SOURCES is set def test_ntp_sources_env_var(): + """Verify NTP_SOURCES is set.""" raw_ntp_sources = os.environ.get("NTP_SOURCES") assert ( raw_ntp_sources is not None and raw_ntp_sources.strip() != "" @@ -24,8 +39,8 @@ def test_ntp_sources_env_var(): assert ntp_sources, "NTP_SOURCES contains no valid source entries." -# 2. Check that chrony is running and syncing properly def test_chrony_service_and_sync_status(host): + """Check that chrony is running and syncing properly""" # Check timedatectl timedate_stdout = host.check_output("timedatectl status") assert ( @@ -35,24 +50,22 @@ def test_chrony_service_and_sync_status(host): # 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_stdout, re.MULTILINE - ) + active_source_match = re.search(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_stdout}" ) -# 3. Check that one of the NTP_SOURCES values is actively used by chrony (*=current source) 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() - assert ntp_sources, "Cannot verify active source because NTP_SOURCES is missing or invalid." + 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 - ) + 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) @@ -65,8 +78,8 @@ def test_active_source_matches_expected(host): ) -# 4. Check time offset def test_ntp_time_offset(host): + """Check time offset""" chrony_stdout = host.check_output("chronyc tracking") match = re.search(r"System time\s+:\s+([0-9.]+)\s+seconds", chrony_stdout) @@ -78,4 +91,4 @@ def test_ntp_time_offset(host): 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_stdout}" - ) \ No newline at end of file + )