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}" + )