fix(timezone): add timezone-aware clock across agent, cron, and execute_code

tests/test_timezone.py

@@ -0,0 +1,269 @@
+"""
+Tests for timezone support (hermes_time module + integration points).
+
+Covers:
+  - Valid timezone applies correctly
+  - Invalid timezone falls back safely (no crash, warning logged)
+  - execute_code child env receives TZ
+  - Cron uses timezone-aware now()
+  - Backward compatibility with naive timestamps
+"""
+
+import os
+import logging
+import sys
+import pytest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch, MagicMock
+from zoneinfo import ZoneInfo
+
+import hermes_time
+
+
+# =========================================================================
+# hermes_time.now() — core helper
+# =========================================================================
+
+class TestHermesTimeNow:
+    """Test the timezone-aware now() helper."""
+
+    def setup_method(self):
+        hermes_time.reset_cache()
+
+    def teardown_method(self):
+        hermes_time.reset_cache()
+        os.environ.pop("HERMES_TIMEZONE", None)
+
+    def test_valid_timezone_applies(self):
+        """With a valid IANA timezone, now() returns time in that zone."""
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+        result = hermes_time.now()
+        assert result.tzinfo is not None
+        # IST is UTC+5:30
+        offset = result.utcoffset()
+        assert offset == timedelta(hours=5, minutes=30)
+
+    def test_utc_timezone(self):
+        """UTC timezone works."""
+        os.environ["HERMES_TIMEZONE"] = "UTC"
+        result = hermes_time.now()
+        assert result.utcoffset() == timedelta(0)
+
+    def test_us_eastern(self):
+        """US/Eastern timezone works (DST-aware zone)."""
+        os.environ["HERMES_TIMEZONE"] = "America/New_York"
+        result = hermes_time.now()
+        assert result.tzinfo is not None
+        # Offset is -5h or -4h depending on DST
+        offset_hours = result.utcoffset().total_seconds() / 3600
+        assert offset_hours in (-5, -4)
+
+    def test_invalid_timezone_falls_back(self, caplog):
+        """Invalid timezone logs warning and falls back to server-local."""
+        os.environ["HERMES_TIMEZONE"] = "Mars/Olympus_Mons"
+        with caplog.at_level(logging.WARNING, logger="hermes_time"):
+            result = hermes_time.now()
+        assert result.tzinfo is not None  # Still tz-aware (server-local)
+        assert "Invalid timezone" in caplog.text
+        assert "Mars/Olympus_Mons" in caplog.text
+
+    def test_empty_timezone_uses_local(self):
+        """No timezone configured → server-local time (still tz-aware)."""
+        os.environ.pop("HERMES_TIMEZONE", None)
+        result = hermes_time.now()
+        assert result.tzinfo is not None
+
+    def test_format_unchanged(self):
+        """Timestamp formatting matches original strftime pattern."""
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+        result = hermes_time.now()
+        formatted = result.strftime("%A, %B %d, %Y %I:%M %p")
+        # Should produce something like "Monday, March 03, 2026 05:30 PM"
+        assert len(formatted) > 10
+        # No timezone abbreviation in the format (matching original behavior)
+        assert "+" not in formatted
+
+    def test_cache_invalidation(self):
+        """Changing env var + reset_cache picks up new timezone."""
+        os.environ["HERMES_TIMEZONE"] = "UTC"
+        hermes_time.reset_cache()
+        r1 = hermes_time.now()
+        assert r1.utcoffset() == timedelta(0)
+
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+        hermes_time.reset_cache()
+        r2 = hermes_time.now()
+        assert r2.utcoffset() == timedelta(hours=5, minutes=30)
+
+
+class TestGetTimezone:
+    """Test get_timezone() and get_timezone_name()."""
+
+    def setup_method(self):
+        hermes_time.reset_cache()
+
+    def teardown_method(self):
+        hermes_time.reset_cache()
+        os.environ.pop("HERMES_TIMEZONE", None)
+
+    def test_returns_zoneinfo_for_valid(self):
+        os.environ["HERMES_TIMEZONE"] = "Europe/London"
+        tz = hermes_time.get_timezone()
+        assert isinstance(tz, ZoneInfo)
+        assert str(tz) == "Europe/London"
+
+    def test_returns_none_for_empty(self):
+        os.environ.pop("HERMES_TIMEZONE", None)
+        tz = hermes_time.get_timezone()
+        assert tz is None
+
+    def test_returns_none_for_invalid(self):
+        os.environ["HERMES_TIMEZONE"] = "Not/A/Timezone"
+        tz = hermes_time.get_timezone()
+        assert tz is None
+
+    def test_get_timezone_name(self):
+        os.environ["HERMES_TIMEZONE"] = "Asia/Tokyo"
+        assert hermes_time.get_timezone_name() == "Asia/Tokyo"
+
+
+# =========================================================================
+# execute_code child env — TZ injection
+# =========================================================================
+
+@pytest.mark.skipif(sys.platform == "win32", reason="UDS not available on Windows")
+class TestCodeExecutionTZ:
+    """Verify TZ env var is passed to sandboxed child process via real execute_code."""
+
+    @pytest.fixture(autouse=True)
+    def _import_execute_code(self):
+        """Lazy-import execute_code to avoid pulling in firecrawl at collection time."""
+        try:
+            from tools.code_execution_tool import execute_code
+            self._execute_code = execute_code
+        except ImportError:
+            pytest.skip("tools.code_execution_tool not importable (missing deps)")
+
+    def teardown_method(self):
+        os.environ.pop("HERMES_TIMEZONE", None)
+
+    def _mock_handle(self, function_name, function_args, task_id=None, user_task=None):
+        import json as _json
+        return _json.dumps({"error": f"unexpected tool call: {function_name}"})
+
+    def test_tz_injected_when_configured(self):
+        """When HERMES_TIMEZONE is set, child process sees TZ env var."""
+        import json as _json
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+
+        with patch("model_tools.handle_function_call", side_effect=self._mock_handle):
+            result = _json.loads(self._execute_code(
+                code='import os; print(os.environ.get("TZ", "NOT_SET"))',
+                task_id="tz-test",
+                enabled_tools=[],
+            ))
+        assert result["status"] == "success"
+        assert "Asia/Kolkata" in result["output"]
+
+    def test_tz_not_injected_when_empty(self):
+        """When HERMES_TIMEZONE is not set, child process has no TZ."""
+        import json as _json
+        os.environ.pop("HERMES_TIMEZONE", None)
+
+        with patch("model_tools.handle_function_call", side_effect=self._mock_handle):
+            result = _json.loads(self._execute_code(
+                code='import os; print(os.environ.get("TZ", "NOT_SET"))',
+                task_id="tz-test-empty",
+                enabled_tools=[],
+            ))
+        assert result["status"] == "success"
+        assert "NOT_SET" in result["output"]
+
+    def test_hermes_timezone_not_leaked_to_child(self):
+        """HERMES_TIMEZONE itself must NOT appear in child env (only TZ)."""
+        import json as _json
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+
+        with patch("model_tools.handle_function_call", side_effect=self._mock_handle):
+            result = _json.loads(self._execute_code(
+                code='import os; print(os.environ.get("HERMES_TIMEZONE", "NOT_SET"))',
+                task_id="tz-leak-test",
+                enabled_tools=[],
+            ))
+        assert result["status"] == "success"
+        assert "NOT_SET" in result["output"]
+
+
+# =========================================================================
+# Cron timezone-aware scheduling
+# =========================================================================
+
+class TestCronTimezone:
+    """Verify cron paths use timezone-aware now()."""
+
+    def setup_method(self):
+        hermes_time.reset_cache()
+
+    def teardown_method(self):
+        hermes_time.reset_cache()
+        os.environ.pop("HERMES_TIMEZONE", None)
+
+    def test_parse_schedule_duration_uses_tz_aware_now(self):
+        """parse_schedule('30m') should produce a tz-aware run_at."""
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+        from cron.jobs import parse_schedule
+        result = parse_schedule("30m")
+        run_at = datetime.fromisoformat(result["run_at"])
+        # The stored timestamp should be tz-aware
+        assert run_at.tzinfo is not None
+
+    def test_compute_next_run_tz_aware(self):
+        """compute_next_run returns tz-aware timestamps."""
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+        from cron.jobs import compute_next_run
+        schedule = {"kind": "interval", "minutes": 60}
+        result = compute_next_run(schedule)
+        next_dt = datetime.fromisoformat(result)
+        assert next_dt.tzinfo is not None
+
+    def test_get_due_jobs_handles_naive_timestamps(self, tmp_path, monkeypatch):
+        """Backward compat: naive timestamps from before tz support don't crash."""
+        import cron.jobs as jobs_module
+        monkeypatch.setattr(jobs_module, "CRON_DIR", tmp_path / "cron")
+        monkeypatch.setattr(jobs_module, "JOBS_FILE", tmp_path / "cron" / "jobs.json")
+        monkeypatch.setattr(jobs_module, "OUTPUT_DIR", tmp_path / "cron" / "output")
+
+        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
+        hermes_time.reset_cache()
+
+        # Create a job with a NAIVE past timestamp (simulating pre-tz data)
+        from cron.jobs import create_job, load_jobs, save_jobs, get_due_jobs
+        job = create_job(prompt="Test job", schedule="every 1h")
+        jobs = load_jobs()
+        # Force a naive (no timezone) past timestamp
+        naive_past = (datetime.now() - timedelta(minutes=5)).isoformat()
+        jobs[0]["next_run_at"] = naive_past
+        save_jobs(jobs)
+
+        # Should not crash — _ensure_aware handles the naive timestamp
+        due = get_due_jobs()
+        assert len(due) == 1
+
+    def test_create_job_stores_tz_aware_timestamps(self, tmp_path, monkeypatch):
+        """New jobs store timezone-aware created_at and next_run_at."""
+        import cron.jobs as jobs_module
+        monkeypatch.setattr(jobs_module, "CRON_DIR", tmp_path / "cron")
+        monkeypatch.setattr(jobs_module, "JOBS_FILE", tmp_path / "cron" / "jobs.json")
+        monkeypatch.setattr(jobs_module, "OUTPUT_DIR", tmp_path / "cron" / "output")
+
+        os.environ["HERMES_TIMEZONE"] = "US/Eastern"
+        hermes_time.reset_cache()
+
+        from cron.jobs import create_job
+        job = create_job(prompt="TZ test", schedule="every 2h")
+
+        created = datetime.fromisoformat(job["created_at"])
+        assert created.tzinfo is not None
+
+        next_run = datetime.fromisoformat(job["next_run_at"])
+        assert next_run.tzinfo is not None