test_timezone.py
1 """ 2 Tests for timezone support (hermes_time module + integration points). 3 4 Covers: 5 - Valid timezone applies correctly 6 - Invalid timezone falls back safely (no crash, warning logged) 7 - execute_code child env receives TZ 8 - Cron uses timezone-aware now() 9 - Backward compatibility with naive timestamps 10 """ 11 12 import os 13 import logging 14 import sys 15 import pytest 16 from datetime import datetime, timedelta, timezone 17 from unittest.mock import patch, MagicMock 18 from zoneinfo import ZoneInfo 19 20 import hermes_time 21 22 23 def _reset_hermes_time_cache(): 24 """Reset the hermes_time module cache (replacement for removed reset_cache).""" 25 hermes_time._cached_tz = None 26 hermes_time._cached_tz_name = None 27 hermes_time._cache_resolved = False 28 29 30 # ========================================================================= 31 # hermes_time.now() — core helper 32 # ========================================================================= 33 34 class TestHermesTimeNow: 35 """Test the timezone-aware now() helper.""" 36 37 def setup_method(self): 38 _reset_hermes_time_cache() 39 40 def teardown_method(self): 41 _reset_hermes_time_cache() 42 os.environ.pop("HERMES_TIMEZONE", None) 43 44 def test_valid_timezone_applies(self): 45 """With a valid IANA timezone, now() returns time in that zone.""" 46 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 47 result = hermes_time.now() 48 assert result.tzinfo is not None 49 # IST is UTC+5:30 50 offset = result.utcoffset() 51 assert offset == timedelta(hours=5, minutes=30) 52 53 def test_utc_timezone(self): 54 """UTC timezone works.""" 55 os.environ["HERMES_TIMEZONE"] = "UTC" 56 result = hermes_time.now() 57 assert result.utcoffset() == timedelta(0) 58 59 def test_us_eastern(self): 60 """US/Eastern timezone works (DST-aware zone).""" 61 os.environ["HERMES_TIMEZONE"] = "America/New_York" 62 result = hermes_time.now() 63 assert result.tzinfo is not None 64 # Offset is -5h or -4h depending on DST 65 offset_hours = result.utcoffset().total_seconds() / 3600 66 assert offset_hours in (-5, -4) 67 68 def test_invalid_timezone_falls_back(self, caplog): 69 """Invalid timezone logs warning and falls back to server-local.""" 70 os.environ["HERMES_TIMEZONE"] = "Mars/Olympus_Mons" 71 with caplog.at_level(logging.WARNING, logger="hermes_time"): 72 result = hermes_time.now() 73 assert result.tzinfo is not None # Still tz-aware (server-local) 74 assert "Invalid timezone" in caplog.text 75 assert "Mars/Olympus_Mons" in caplog.text 76 77 def test_empty_timezone_uses_local(self): 78 """No timezone configured → server-local time (still tz-aware).""" 79 os.environ.pop("HERMES_TIMEZONE", None) 80 result = hermes_time.now() 81 assert result.tzinfo is not None 82 83 def test_format_unchanged(self): 84 """Timestamp formatting matches original strftime pattern.""" 85 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 86 result = hermes_time.now() 87 formatted = result.strftime("%A, %B %d, %Y %I:%M %p") 88 # Should produce something like "Monday, March 03, 2026 05:30 PM" 89 assert len(formatted) > 10 90 # No timezone abbreviation in the format (matching original behavior) 91 assert "+" not in formatted 92 93 def test_cache_invalidation(self): 94 """Changing env var + reset_cache picks up new timezone.""" 95 os.environ["HERMES_TIMEZONE"] = "UTC" 96 _reset_hermes_time_cache() 97 r1 = hermes_time.now() 98 assert r1.utcoffset() == timedelta(0) 99 100 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 101 _reset_hermes_time_cache() 102 r2 = hermes_time.now() 103 assert r2.utcoffset() == timedelta(hours=5, minutes=30) 104 105 106 class TestGetTimezone: 107 """Test get_timezone().""" 108 109 def setup_method(self): 110 _reset_hermes_time_cache() 111 112 def teardown_method(self): 113 _reset_hermes_time_cache() 114 os.environ.pop("HERMES_TIMEZONE", None) 115 116 def test_returns_zoneinfo_for_valid(self): 117 os.environ["HERMES_TIMEZONE"] = "Europe/London" 118 tz = hermes_time.get_timezone() 119 assert isinstance(tz, ZoneInfo) 120 assert str(tz) == "Europe/London" 121 122 def test_returns_none_for_empty(self): 123 os.environ.pop("HERMES_TIMEZONE", None) 124 tz = hermes_time.get_timezone() 125 assert tz is None 126 127 def test_returns_none_for_invalid(self): 128 os.environ["HERMES_TIMEZONE"] = "Not/A/Timezone" 129 tz = hermes_time.get_timezone() 130 assert tz is None 131 132 133 134 # ========================================================================= 135 # execute_code child env — TZ injection 136 # ========================================================================= 137 138 @pytest.mark.skipif(sys.platform == "win32", reason="UDS not available on Windows") 139 class TestCodeExecutionTZ: 140 """Verify TZ env var is passed to sandboxed child process via real execute_code.""" 141 142 @pytest.fixture(autouse=True) 143 def _import_execute_code(self, monkeypatch): 144 """Lazy-import execute_code to avoid pulling in firecrawl at collection time.""" 145 # Force local backend — other tests in the same xdist worker may leak 146 # TERMINAL_ENV=modal/docker which causes modal.exception.AuthError. 147 monkeypatch.setenv("TERMINAL_ENV", "local") 148 try: 149 from tools.code_execution_tool import execute_code 150 self._execute_code = execute_code 151 except ImportError: 152 pytest.skip("tools.code_execution_tool not importable (missing deps)") 153 154 def teardown_method(self): 155 os.environ.pop("HERMES_TIMEZONE", None) 156 157 def _mock_handle(self, function_name, function_args, task_id=None, user_task=None): 158 import json as _json 159 return _json.dumps({"error": f"unexpected tool call: {function_name}"}) 160 161 def test_tz_injected_when_configured(self): 162 """When HERMES_TIMEZONE is set, child process sees TZ env var. 163 164 Verified alongside leak-prevention + empty-TZ handling in one 165 subprocess call so we don't pay 3x the subprocess startup cost 166 (each execute_code spawns a real Python subprocess ~3s). 167 """ 168 import json as _json 169 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 170 171 # One subprocess, three things checked: 172 # 1) TZ is injected as "Asia/Kolkata" 173 # 2) HERMES_TIMEZONE itself does NOT leak into the child env 174 probe = ( 175 'import os; ' 176 'print("TZ=" + os.environ.get("TZ", "NOT_SET")); ' 177 'print("HERMES_TIMEZONE=" + os.environ.get("HERMES_TIMEZONE", "NOT_SET"))' 178 ) 179 with patch("model_tools.handle_function_call", side_effect=self._mock_handle): 180 result = _json.loads(self._execute_code( 181 code=probe, 182 task_id="tz-combined-test", 183 enabled_tools=[], 184 )) 185 assert result["status"] == "success" 186 assert "TZ=Asia/Kolkata" in result["output"] 187 assert "HERMES_TIMEZONE=NOT_SET" in result["output"], ( 188 "HERMES_TIMEZONE should not leak into child env (only TZ)" 189 ) 190 191 def test_tz_not_injected_when_empty(self): 192 """When HERMES_TIMEZONE is not set, child process has no TZ.""" 193 import json as _json 194 os.environ.pop("HERMES_TIMEZONE", None) 195 196 with patch("model_tools.handle_function_call", side_effect=self._mock_handle): 197 result = _json.loads(self._execute_code( 198 code='import os; print(os.environ.get("TZ", "NOT_SET"))', 199 task_id="tz-test-empty", 200 enabled_tools=[], 201 )) 202 assert result["status"] == "success" 203 assert "NOT_SET" in result["output"] 204 205 206 # ========================================================================= 207 # Cron timezone-aware scheduling 208 # ========================================================================= 209 210 class TestCronTimezone: 211 """Verify cron paths use timezone-aware now().""" 212 213 def setup_method(self): 214 _reset_hermes_time_cache() 215 216 def teardown_method(self): 217 _reset_hermes_time_cache() 218 os.environ.pop("HERMES_TIMEZONE", None) 219 220 def test_parse_schedule_duration_uses_tz_aware_now(self): 221 """parse_schedule('30m') should produce a tz-aware run_at.""" 222 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 223 from cron.jobs import parse_schedule 224 result = parse_schedule("30m") 225 run_at = datetime.fromisoformat(result["run_at"]) 226 # The stored timestamp should be tz-aware 227 assert run_at.tzinfo is not None 228 229 def test_compute_next_run_tz_aware(self): 230 """compute_next_run returns tz-aware timestamps.""" 231 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 232 from cron.jobs import compute_next_run 233 schedule = {"kind": "interval", "minutes": 60} 234 result = compute_next_run(schedule) 235 next_dt = datetime.fromisoformat(result) 236 assert next_dt.tzinfo is not None 237 238 def test_get_due_jobs_handles_naive_timestamps(self, tmp_path, monkeypatch): 239 """Backward compat: naive timestamps from before tz support don't crash.""" 240 import cron.jobs as jobs_module 241 monkeypatch.setattr(jobs_module, "CRON_DIR", tmp_path / "cron") 242 monkeypatch.setattr(jobs_module, "JOBS_FILE", tmp_path / "cron" / "jobs.json") 243 monkeypatch.setattr(jobs_module, "OUTPUT_DIR", tmp_path / "cron" / "output") 244 245 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 246 _reset_hermes_time_cache() 247 248 # Create a job with a NAIVE past timestamp (simulating pre-tz data) 249 from cron.jobs import create_job, load_jobs, save_jobs, get_due_jobs 250 job = create_job(prompt="Test job", schedule="every 1h") 251 jobs = load_jobs() 252 # Force a naive (no timezone) past timestamp 253 naive_past = (datetime.now() - timedelta(seconds=30)).isoformat() 254 jobs[0]["next_run_at"] = naive_past 255 save_jobs(jobs) 256 257 # Should not crash — _ensure_aware handles the naive timestamp 258 due = get_due_jobs() 259 assert len(due) == 1 260 261 def test_ensure_aware_naive_preserves_absolute_time(self): 262 """_ensure_aware must preserve the absolute instant for naive datetimes. 263 264 Regression: the old code used replace(tzinfo=hermes_tz) which shifted 265 absolute time when system-local tz != Hermes tz. The fix interprets 266 naive values as system-local wall time, then converts. 267 """ 268 from cron.jobs import _ensure_aware 269 270 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 271 _reset_hermes_time_cache() 272 273 # Create a naive datetime — will be interpreted as system-local time 274 naive_dt = datetime(2026, 3, 11, 12, 0, 0) 275 276 result = _ensure_aware(naive_dt) 277 278 # The result should be in Kolkata tz 279 assert result.tzinfo is not None 280 281 # The UTC equivalent must match what we'd get by correctly interpreting 282 # the naive dt as system-local time first, then converting 283 system_tz = datetime.now().astimezone().tzinfo 284 expected_utc = naive_dt.replace(tzinfo=system_tz).astimezone(timezone.utc) 285 actual_utc = result.astimezone(timezone.utc) 286 assert actual_utc == expected_utc, ( 287 f"Absolute time shifted: expected {expected_utc}, got {actual_utc}" 288 ) 289 290 def test_ensure_aware_normalizes_aware_to_hermes_tz(self): 291 """Already-aware datetimes should be normalized to Hermes tz.""" 292 from cron.jobs import _ensure_aware 293 294 os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata" 295 _reset_hermes_time_cache() 296 297 # Create an aware datetime in UTC 298 utc_dt = datetime(2026, 3, 11, 15, 0, 0, tzinfo=timezone.utc) 299 result = _ensure_aware(utc_dt) 300 301 # Must be in Hermes tz (Kolkata) but same absolute instant 302 kolkata = ZoneInfo("Asia/Kolkata") 303 assert result.utctimetuple()[:5] == (2026, 3, 11, 15, 0) 304 expected_local = utc_dt.astimezone(kolkata) 305 assert result == expected_local 306 307 def test_ensure_aware_due_job_not_skipped_when_system_ahead(self, tmp_path, monkeypatch): 308 """Reproduce the actual bug: system tz ahead of Hermes tz caused 309 overdue jobs to appear as not-yet-due. 310 311 Scenario: system is Asia/Kolkata (UTC+5:30), Hermes is UTC. 312 A naive timestamp from 5 minutes ago (local time) should still 313 be recognized as due after conversion. 314 """ 315 import cron.jobs as jobs_module 316 monkeypatch.setattr(jobs_module, "CRON_DIR", tmp_path / "cron") 317 monkeypatch.setattr(jobs_module, "JOBS_FILE", tmp_path / "cron" / "jobs.json") 318 monkeypatch.setattr(jobs_module, "OUTPUT_DIR", tmp_path / "cron" / "output") 319 320 os.environ["HERMES_TIMEZONE"] = "UTC" 321 _reset_hermes_time_cache() 322 323 from cron.jobs import create_job, load_jobs, save_jobs, get_due_jobs 324 325 job = create_job(prompt="Bug repro", schedule="every 1h") 326 jobs = load_jobs() 327 328 # Simulate a naive timestamp that was written by datetime.now() on a 329 # system running in UTC+5:30 — 5 minutes in the past (local time) 330 naive_past = (datetime.now() - timedelta(seconds=30)).isoformat() 331 jobs[0]["next_run_at"] = naive_past 332 save_jobs(jobs) 333 334 # Must be recognized as due regardless of tz mismatch 335 due = get_due_jobs() 336 assert len(due) == 1, ( 337 "Overdue job was skipped — _ensure_aware likely shifted absolute time" 338 ) 339 340 def test_get_due_jobs_naive_cross_timezone(self, tmp_path, monkeypatch): 341 """Naive past timestamps must be detected as due even when Hermes tz 342 is behind system local tz — the scenario that triggered #806.""" 343 import cron.jobs as jobs_module 344 monkeypatch.setattr(jobs_module, "CRON_DIR", tmp_path / "cron") 345 monkeypatch.setattr(jobs_module, "JOBS_FILE", tmp_path / "cron" / "jobs.json") 346 monkeypatch.setattr(jobs_module, "OUTPUT_DIR", tmp_path / "cron" / "output") 347 348 # Use a Hermes timezone far behind UTC so that the numeric wall time 349 # of the naive timestamp exceeds _hermes_now's wall time — this would 350 # have caused a false "not due" with the old replace(tzinfo=...) approach. 351 os.environ["HERMES_TIMEZONE"] = "Pacific/Midway" # UTC-11 352 _reset_hermes_time_cache() 353 354 from cron.jobs import create_job, load_jobs, save_jobs, get_due_jobs 355 create_job(prompt="Cross-tz job", schedule="every 1h") 356 jobs = load_jobs() 357 358 # Force a naive past timestamp (system-local wall time, 10 min ago) 359 naive_past = (datetime.now() - timedelta(seconds=30)).isoformat() 360 jobs[0]["next_run_at"] = naive_past 361 save_jobs(jobs) 362 363 due = get_due_jobs() 364 assert len(due) == 1, ( 365 "Naive past timestamp should be due regardless of Hermes timezone" 366 ) 367 368 def test_create_job_stores_tz_aware_timestamps(self, tmp_path, monkeypatch): 369 """New jobs store timezone-aware created_at and next_run_at.""" 370 import cron.jobs as jobs_module 371 monkeypatch.setattr(jobs_module, "CRON_DIR", tmp_path / "cron") 372 monkeypatch.setattr(jobs_module, "JOBS_FILE", tmp_path / "cron" / "jobs.json") 373 monkeypatch.setattr(jobs_module, "OUTPUT_DIR", tmp_path / "cron" / "output") 374 375 os.environ["HERMES_TIMEZONE"] = "US/Eastern" 376 _reset_hermes_time_cache() 377 378 from cron.jobs import create_job 379 job = create_job(prompt="TZ test", schedule="every 2h") 380 381 created = datetime.fromisoformat(job["created_at"]) 382 assert created.tzinfo is not None 383 384 next_run = datetime.fromisoformat(job["next_run_at"]) 385 assert next_run.tzinfo is not None