"""How far a host's clock is from Steward's (milestone 490): the agent stamps each post with sent_at, Steward compares it with the moment it arrived, and the host page says it in words, graded by Settings → Thresholds' clock drift (30 s warn, 5 min critical by default).""" from datetime import datetime, timezone import pytest from plugins.host_agent import agent as a from plugins.host_agent.routes import clock_offset, clock_words from steward.core.settings import to_thresholds_cfg RECEIVED = datetime(2026, 10, 8, 10, 0, 0, tzinfo=timezone.utc) TH = to_thresholds_cfg({}) def test_the_agent_stamps_each_post_with_its_own_clock(): payload = a.build_payload(samples=[{"ts": "2026-10-08T10:00:00+00:00"}], hostname="h", metadata={}) sent = datetime.fromisoformat(payload["sent_at"]) assert sent.tzinfo is not None assert abs((sent - datetime.now(timezone.utc)).total_seconds()) < 5 assert a.AGENT_VERSION >= "1.12.0" @pytest.mark.parametrize("sent_at, offset", [ ("2026-10-08T10:00:05+00:00", 5.0), # the host is ahead ("2026-10-08T09:57:50Z", -130.0), # behind, Z suffix ("2026-10-08T06:00:00.250-04:00", 0.25), # another zone, the same moment ]) def test_the_offset_is_the_hosts_clock_against_steward_on_arrival(sent_at, offset): assert clock_offset(sent_at, RECEIVED) == offset @pytest.mark.parametrize("sent_at", [None, "", "yesterday", 1728381600, "2026-10-08T10:00:05"]) def test_no_usable_stamp_is_no_offset(sent_at): # An agent before 1.12.0 sends none; a naive time says nothing about a zone. assert clock_offset(sent_at, RECEIVED) is None @pytest.mark.parametrize("offset, words, tone", [ (None, "", "unknown"), (0.3, "clock in step", "ok"), (-0.9, "clock in step", "ok"), (12.4, "clock 12s ahead", "ok"), (29.9, "clock 29s ahead", "ok"), (-130, "clock 2m 10s behind", "warn"), (31, "clock 31s ahead", "warn"), (-3600, "clock 1h behind", "crit"), ]) def test_the_host_page_says_it_in_words(offset, words, tone): assert clock_words(offset, TH) == (words, tone) def test_the_cutoff_is_the_operators(): th = to_thresholds_cfg({"thresholds.clock_warn_s": 5, "thresholds.clock_crit_s": 60}) assert clock_words(12.4, th) == ("clock 12s ahead", "warn") assert clock_words(-90, th)[1] == "crit"