hermes-agent/tests/cron/test_timezone.py

"""
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 sys
import pytest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

import hermes_time


def _reset_hermes_time_cache():
    """Reset the hermes_time module cache."""
    hermes_time.reset_cache()


# =========================================================================
# hermes_time.now() — core helper
# =========================================================================

class TestHermesTimeNow:
    """Test the timezone-aware now() helper."""

    def setup_method(self):
        _reset_hermes_time_cache()

    def teardown_method(self):
        _reset_hermes_time_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)


# Windows zone name in cp1252 bytes, decoded under a UTF-8 LC_CTYPE with surrogateescape (#102910).
_ESCAPED_ZONE = "Paris, Madrid (heure d'\udce9t\udce9)"


class TestSafeStrftime:
    def test_valid_locale_text_is_byte_identical_to_strftime(self):
        """The system prompt embeds this output, so healthy locales must not change a byte."""
        from zoneinfo import ZoneInfo
        fmt = "%a %A %b %B %d %Y %H:%M:%S %Z %z %%Z"
        for value in (
            datetime(2026, 7, 14, 13, 5),
            datetime(2026, 7, 14, 13, 5, tzinfo=ZoneInfo("Europe/Paris")),
            datetime(2026, 7, 14, 13, 5, tzinfo=timezone(timedelta(hours=-3), "Hora estándar de Argentina")),
        ):
            assert hermes_time.safe_strftime(value, fmt) == value.strftime(fmt)

    def test_surrogate_zone_name_renders_json_safe(self):
        import json
        value = datetime(2026, 7, 14, 13, 5, tzinfo=timezone(timedelta(hours=2), _ESCAPED_ZONE))
        rendered = hermes_time.safe_strftime(value, "%a %Y-%m-%d %H:%M %Z %z %%Z")
        json.dumps(rendered, ensure_ascii=False).encode("utf-8")
        assert rendered.startswith("Tue 2026-07-14 13:05 Paris, Madrid (heure d'")
        assert rendered.endswith(") +0200 %Z")
        # The escaped bytes decode back through the Windows ANSI code page.
        assert hermes_time._repair_surrogates(_ESCAPED_ZONE, "cp1252") == "Paris, Madrid (heure d'été)"


class TestGetTimezone:
    """Test get_timezone()."""

    def setup_method(self):
        _reset_hermes_time_cache()

    def teardown_method(self):
        _reset_hermes_time_cache()
        os.environ.pop("HERMES_TIMEZONE", None)


    def test_cache_isolated_by_active_profile_config(self, tmp_path, monkeypatch):
        """Switching HERMES_HOME must not reuse another profile's timezone."""
        first_home = tmp_path / "first"
        second_home = tmp_path / "second"
        first_home.mkdir()
        second_home.mkdir()
        (first_home / "config.yaml").write_text("timezone: Asia/Tokyo\n", encoding="utf-8")
        (second_home / "config.yaml").write_text(
            "timezone: America/New_York\n", encoding="utf-8"
        )
        monkeypatch.delenv("HERMES_TIMEZONE", raising=False)

        monkeypatch.setenv("HERMES_HOME", str(first_home))
        assert str(hermes_time.get_timezone()) == "Asia/Tokyo"

        # Multiplexed profile runtime scopes switch HERMES_HOME in one process.
        monkeypatch.setenv("HERMES_HOME", str(second_home))
        assert str(hermes_time.get_timezone()) == "America/New_York"

        # Switching BACK must return the first profile's zone (per-identity
        # entries stay hot; no single-slot ping-pong).
        monkeypatch.setenv("HERMES_HOME", str(first_home))
        assert str(hermes_time.get_timezone()) == "Asia/Tokyo"

    def test_multiplex_prefers_routed_profile_config_over_env(self, tmp_path, monkeypatch):
        """Under the multiplexed gateway HERMES_TIMEZONE holds only the DEFAULT profile's value
        (bridged at startup), so a routed profile must resolve from its own config.yaml."""
        from agent.secret_scope import set_multiplex_active
        from hermes_constants import reset_hermes_home_override, set_hermes_home_override

        default_home, routed_home = tmp_path / "default", tmp_path / "routed"
        default_home.mkdir()
        routed_home.mkdir()
        (default_home / "config.yaml").write_text("timezone: America/New_York\n", encoding="utf-8")
        (routed_home / "config.yaml").write_text("timezone: Asia/Tokyo\n", encoding="utf-8")
        monkeypatch.setenv("HERMES_HOME", str(default_home))
        monkeypatch.setenv("HERMES_TIMEZONE", "America/New_York")

        # Single-profile process: env stays authoritative.
        assert hermes_time.get_timezone_name() == "America/New_York"

        set_multiplex_active(True)
        try:
            assert hermes_time.get_timezone_name() == "America/New_York"  # default profile turn
            token = set_hermes_home_override(str(routed_home))
            try:
                assert hermes_time.get_timezone_name() == "Asia/Tokyo"
                assert str(hermes_time.get_timezone()) == "Asia/Tokyo"
            finally:
                reset_hermes_home_override(token)
        finally:
            set_multiplex_active(False)

    def test_concurrent_profile_resolution_never_mixes_zones(
        self, tmp_path, monkeypatch
    ):
        """Racing profile-scoped threads must never observe a foreign zone.

        The multiplex cron ticker lets profile-A work (mark_job_run /
        compute_next_run) overlap the ticker advancing to profile B. The
        cache publication must be atomic per identity: identity A can never
        be paired with profile B's ZoneInfo (#97905 review finding on
        PR #92489).
        """
        import threading

        from hermes_constants import (
            reset_hermes_home_override,
            set_hermes_home_override,
        )

        zones = {"a": "Asia/Tokyo", "b": "America/New_York"}
        homes = {}
        for key, zone in zones.items():
            home = tmp_path / key
            home.mkdir()
            (home / "config.yaml").write_text(
                f"timezone: {zone}\n", encoding="utf-8"
            )
            homes[key] = home
        monkeypatch.delenv("HERMES_TIMEZONE", raising=False)

        errors = []
        barrier = threading.Barrier(2)

        def worker(key: str) -> None:
            barrier.wait()
            for _ in range(200):
                token = set_hermes_home_override(str(homes[key]))
                try:
                    tz = hermes_time.get_timezone()
                    if str(tz) != zones[key]:
                        errors.append((key, str(tz)))
                        return
                finally:
                    reset_hermes_home_override(token)

        threads = [
            threading.Thread(target=worker, args=(key,)) for key in zones
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        assert not errors, f"foreign timezone observed: {errors}"


# =========================================================================

# execute_code child env — TZ injection
# =========================================================================

@pytest.mark.platforms("posix")  # 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, monkeypatch):
        """Lazy-import execute_code to avoid pulling in firecrawl at collection time."""
        # Force local backend — other tests in the same process may leak
        # TERMINAL_ENV=modal/docker which causes modal.exception.AuthError.
        monkeypatch.setenv("TERMINAL_ENV", "local")
        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.

        Verified alongside leak-prevention + empty-TZ handling in one
        subprocess call so we don't pay 3x the subprocess startup cost
        (each execute_code spawns a real Python subprocess ~3s).
        """
        import json as _json
        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"

        # One subprocess, three things checked:
        #   1) TZ is injected as "Asia/Kolkata"
        #   2) HERMES_TIMEZONE itself does NOT leak into the child env
        probe = (
            'import os; '
            'print("TZ=" + os.environ.get("TZ", "NOT_SET")); '
            'print("HERMES_TIMEZONE=" + os.environ.get("HERMES_TIMEZONE", "NOT_SET"))'
        )
        with patch("model_tools.handle_function_call", side_effect=self._mock_handle):
            result = _json.loads(self._execute_code(
                code=probe,
                task_id="tz-combined-test",
                enabled_tools=[],
            ))
        assert result["status"] == "success"
        assert "TZ=Asia/Kolkata" in result["output"]
        assert "HERMES_TIMEZONE=NOT_SET" in result["output"], (
            "HERMES_TIMEZONE should not leak into child env (only TZ)"
        )



# =========================================================================
# Cron timezone-aware scheduling
# =========================================================================

class TestCronTimezone:
    """Verify cron paths use timezone-aware now()."""

    def setup_method(self):
        _reset_hermes_time_cache()

    def teardown_method(self):
        _reset_hermes_time_cache()
        os.environ.pop("HERMES_TIMEZONE", None)

    def test_parse_schedule_one_shot_duration_uses_tz_aware_now(self):
        """parse_schedule('in 30m') should produce a tz-aware run_at."""
        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
        from cron.jobs import parse_schedule
        result = parse_schedule("in 30m")
        run_at = datetime.fromisoformat(result["run_at"])
        # The stored timestamp should be tz-aware
        assert run_at.tzinfo is not None



    def test_ensure_aware_naive_preserves_absolute_time(self):
        """_ensure_aware must preserve the absolute instant for naive datetimes.

        Regression: the old code used replace(tzinfo=hermes_tz) which shifted
        absolute time when system-local tz != Hermes tz.  The fix interprets
        naive values as system-local wall time, then converts.
        """
        from cron.jobs import _ensure_aware

        os.environ["HERMES_TIMEZONE"] = "Asia/Kolkata"
        _reset_hermes_time_cache()

        # Create a naive datetime — will be interpreted as system-local time
        naive_dt = datetime(2026, 3, 11, 12, 0, 0)

        result = _ensure_aware(naive_dt)

        # The result should be in Kolkata tz
        assert result.tzinfo is not None

        # The UTC equivalent must match what we'd get by correctly interpreting
        # the naive dt as system-local time first, then converting
        system_tz = datetime.now().astimezone().tzinfo
        expected_utc = naive_dt.replace(tzinfo=system_tz).astimezone(timezone.utc)
        actual_utc = result.astimezone(timezone.utc)
        assert actual_utc == expected_utc, (
            f"Absolute time shifted: expected {expected_utc}, got {actual_utc}"
        )



    def test_get_due_jobs_naive_cross_timezone(self, tmp_path, monkeypatch):
        """Naive past timestamps must be detected as due even when Hermes tz
        is behind system local tz — the scenario that triggered #806."""
        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")

        # Use a Hermes timezone far behind UTC so that the numeric wall time
        # of the naive timestamp exceeds _hermes_now's wall time — this would
        # have caused a false "not due" with the old replace(tzinfo=...) approach.
        os.environ["HERMES_TIMEZONE"] = "Pacific/Midway"  # UTC-11
        _reset_hermes_time_cache()

        from cron.jobs import create_job, load_jobs, save_jobs, get_due_jobs
        create_job(prompt="Cross-tz job", schedule="every 1h")
        jobs = load_jobs()

        # Force a naive past timestamp (system-local wall time, 10 min ago)
        naive_past = (datetime.now() - timedelta(seconds=30)).isoformat()
        jobs[0]["next_run_at"] = naive_past
        save_jobs(jobs)

        due = get_due_jobs()
        assert len(due) == 1, (
            "Naive past timestamp should be due regardless of Hermes timezone"
        )

    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"
        _reset_hermes_time_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