| 1 | import os |
| 2 | import sys |
| 3 | import threading |
| 4 | from datetime import datetime |
| 5 | from pathlib import Path |
| 6 | |
| 7 | import pytest |
| 8 | import pytz |
| 9 | from langchain_core.documents import Document |
| 10 | |
| 11 | PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| 12 | if str(PROJECT_ROOT) not in sys.path: |
| 13 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 14 | |
| 15 | import helpers.localization as localization_module |
| 16 | import helpers.plugins as plugins_module |
| 17 | import helpers.settings as settings_module |
| 18 | from helpers.localization import Localization |
| 19 | from helpers.task_scheduler import ( |
| 20 | TaskPlan, |
| 21 | parse_task_plan, |
| 22 | serialize_task_plan, |
| 23 | ) |
| 24 | from plugins._memory.api.memory_dashboard import MemoryDashboard |
| 25 | from plugins._desktop.helpers import desktop_session |
| 26 | |
| 27 | |
| 28 | @pytest.fixture |
| 29 | def isolated_localization(monkeypatch): |
| 30 | saved: list[tuple[str, str]] = [] |
| 31 | |
| 32 | def fake_get_dotenv_value(key, default=None): |
| 33 | if key == "DEFAULT_USER_TIMEZONE": |
| 34 | return default or "UTC" |
| 35 | if key == "DEFAULT_USER_UTC_OFFSET_MINUTES": |
| 36 | return None |
| 37 | return default |
| 38 | |
| 39 | monkeypatch.setattr(localization_module, "get_dotenv_value", fake_get_dotenv_value) |
| 40 | monkeypatch.setattr( |
| 41 | localization_module, |
| 42 | "save_dotenv_value", |
| 43 | lambda key, value: saved.append((key, str(value))), |
| 44 | ) |
| 45 | monkeypatch.setattr( |
| 46 | localization_module.PrintStyle, |
| 47 | "error", |
| 48 | staticmethod(lambda *args, **kwargs: None), |
| 49 | ) |
| 50 | Localization._instance = None |
| 51 | original_tz = os.environ.get("TZ") |
| 52 | |
| 53 | yield saved |
| 54 | |
| 55 | Localization._instance = None |
| 56 | if original_tz is None: |
| 57 | monkeypatch.delenv("TZ", raising=False) |
| 58 | else: |
| 59 | monkeypatch.setenv("TZ", original_tz) |
| 60 | |
| 61 | |
| 62 | def set_test_timezone(timezone: str) -> Localization: |
| 63 | Localization._instance = Localization(timezone) |
| 64 | return Localization.get() |
| 65 | |
| 66 | |
| 67 | def test_invalid_timezone_preserves_current_user_timezone(isolated_localization): |
| 68 | saved = isolated_localization |
| 69 | localization = set_test_timezone("Europe/Rome") |
| 70 | saved.clear() |
| 71 | |
| 72 | localization.set_timezone("Mars/Olympus") |
| 73 | |
| 74 | assert localization.get_timezone() == "Europe/Rome" |
| 75 | assert os.environ["TZ"] == "Europe/Rome" |
| 76 | assert not any(item == ("DEFAULT_USER_TIMEZONE", "UTC") for item in saved) |
| 77 | assert not any(item == ("DEFAULT_USER_TIMEZONE", "Mars/Olympus") for item in saved) |
| 78 | |
| 79 | |
| 80 | def test_startup_refreshes_stale_persisted_offset(monkeypatch): |
| 81 | saved: list[tuple[str, str]] = [] |
| 82 | |
| 83 | def fake_get_dotenv_value(key, default=None): |
| 84 | if key == "DEFAULT_USER_TIMEZONE": |
| 85 | return "Europe/Rome" |
| 86 | if key == "DEFAULT_USER_UTC_OFFSET_MINUTES": |
| 87 | return "0" |
| 88 | return default |
| 89 | |
| 90 | monkeypatch.setattr(localization_module, "get_dotenv_value", fake_get_dotenv_value) |
| 91 | monkeypatch.setattr( |
| 92 | localization_module, |
| 93 | "save_dotenv_value", |
| 94 | lambda key, value: saved.append((key, str(value))), |
| 95 | ) |
| 96 | Localization._instance = None |
| 97 | |
| 98 | localization = Localization.get() |
| 99 | expected_offset = int( |
| 100 | datetime.now(pytz.timezone("Europe/Rome")).utcoffset().total_seconds() // 60 |
| 101 | ) |
| 102 | |
| 103 | assert localization.get_timezone() == "Europe/Rome" |
| 104 | assert localization.get_offset_minutes() == expected_offset |
| 105 | assert ("DEFAULT_USER_UTC_OFFSET_MINUTES", str(expected_offset)) in saved |
| 106 | |
| 107 | |
| 108 | def test_scheduler_naive_plan_times_round_trip_as_user_local(isolated_localization): |
| 109 | set_test_timezone("Europe/Rome") |
| 110 | |
| 111 | plan = parse_task_plan({"todo": ["2026-05-03T09:30:00"], "in_progress": None, "done": []}) |
| 112 | serialized = serialize_task_plan(plan) |
| 113 | |
| 114 | assert serialized["todo"] == ["2026-05-03T09:30:00+02:00"] |
| 115 | |
| 116 | |
| 117 | def test_settings_auto_timezone_resolves_to_browser_timezone(isolated_localization, monkeypatch): |
| 118 | set_test_timezone("UTC") |
| 119 | hooks: list[dict] = [] |
| 120 | base_settings = settings_module.get_default_settings() |
| 121 | monkeypatch.setattr( |
| 122 | settings_module, |
| 123 | "_settings", |
| 124 | {**base_settings, "timezone": settings_module.TIMEZONE_AUTO}, |
| 125 | ) |
| 126 | monkeypatch.setattr( |
| 127 | plugins_module, |
| 128 | "call_plugin_hook", |
| 129 | lambda plugin_name, hook_name, *args, **kwargs: hooks.append( |
| 130 | { |
| 131 | "plugin_name": plugin_name, |
| 132 | "hook_name": hook_name, |
| 133 | "kwargs": kwargs, |
| 134 | } |
| 135 | ), |
| 136 | ) |
| 137 | |
| 138 | settings_module._apply_timezone_setting( |
| 139 | {**base_settings, "timezone": settings_module.TIMEZONE_AUTO}, |
| 140 | browser_timezone="Europe/Rome", |
| 141 | ) |
| 142 | |
| 143 | assert Localization.get().get_timezone() == "Europe/Rome" |
| 144 | assert hooks[0]["plugin_name"] == "_office" |
| 145 | assert hooks[0]["hook_name"] == "timezone_changed" |
| 146 | assert hooks[0]["kwargs"]["previous_timezone"] == "UTC" |
| 147 | assert hooks[0]["kwargs"]["timezone"] == "Europe/Rome" |
| 148 | |
| 149 | |
| 150 | def test_settings_fixed_timezone_ignores_browser_timezone(isolated_localization, monkeypatch): |
| 151 | set_test_timezone("Europe/Rome") |
| 152 | base_settings = settings_module.get_default_settings() |
| 153 | monkeypatch.setattr( |
| 154 | settings_module, |
| 155 | "_settings", |
| 156 | {**base_settings, "timezone": "America/New_York"}, |
| 157 | ) |
| 158 | monkeypatch.setattr(plugins_module, "call_plugin_hook", lambda *args, **kwargs: None) |
| 159 | |
| 160 | settings_module._apply_timezone_setting( |
| 161 | {**base_settings, "timezone": settings_module.TIMEZONE_AUTO}, |
| 162 | browser_timezone="Europe/Rome", |
| 163 | ) |
| 164 | |
| 165 | assert Localization.get().get_timezone() == "America/New_York" |
| 166 | |
| 167 | |
| 168 | def test_settings_fixed_timezone_reapplies_when_runtime_drifted(isolated_localization, monkeypatch): |
| 169 | set_test_timezone("Europe/Rome") |
| 170 | base_settings = settings_module.get_default_settings() |
| 171 | fixed_settings = {**base_settings, "timezone": "America/New_York"} |
| 172 | hooks: list[dict] = [] |
| 173 | monkeypatch.setattr(settings_module, "_settings", fixed_settings) |
| 174 | monkeypatch.setattr( |
| 175 | plugins_module, |
| 176 | "call_plugin_hook", |
| 177 | lambda plugin_name, hook_name, *args, **kwargs: hooks.append( |
| 178 | { |
| 179 | "plugin_name": plugin_name, |
| 180 | "hook_name": hook_name, |
| 181 | "kwargs": kwargs, |
| 182 | } |
| 183 | ), |
| 184 | ) |
| 185 | |
| 186 | settings_module._apply_timezone_setting( |
| 187 | {**base_settings, "timezone": "America/New_York"}, |
| 188 | browser_timezone="Europe/Rome", |
| 189 | ) |
| 190 | |
| 191 | assert Localization.get().get_timezone() == "America/New_York" |
| 192 | assert hooks[0]["plugin_name"] == "_office" |
| 193 | assert hooks[0]["hook_name"] == "timezone_changed" |
| 194 | assert hooks[0]["kwargs"]["previous_timezone"] == "Europe/Rome" |
| 195 | assert hooks[0]["kwargs"]["timezone"] == "America/New_York" |
| 196 | |
| 197 | |
| 198 | def test_settings_rejects_invalid_timezone_value(): |
| 199 | settings_data = settings_module.get_default_settings() |
| 200 | normalized = settings_module.normalize_settings( |
| 201 | {**settings_data, "timezone": "Mars/Olympus"} |
| 202 | ) |
| 203 | |
| 204 | assert normalized["timezone"] == settings_module.TIMEZONE_AUTO |
| 205 | |
| 206 | |
| 207 | def test_settings_rejects_invalid_time_format_value(): |
| 208 | settings_data = settings_module.get_default_settings() |
| 209 | |
| 210 | invalid = settings_module.normalize_settings( |
| 211 | {**settings_data, "time_format": "bananas"} |
| 212 | ) |
| 213 | twenty_four = settings_module.normalize_settings( |
| 214 | {**settings_data, "time_format": settings_module.TIME_FORMAT_24H} |
| 215 | ) |
| 216 | |
| 217 | assert invalid["time_format"] == settings_module.TIME_FORMAT_12H |
| 218 | assert twenty_four["time_format"] == settings_module.TIME_FORMAT_24H |
| 219 | |
| 220 | |
| 221 | def test_scheduler_naive_plan_times_follow_changed_user_timezone(isolated_localization): |
| 222 | set_test_timezone("America/New_York") |
| 223 | |
| 224 | plan = parse_task_plan({"todo": ["2026-05-03T09:30:00"], "in_progress": None, "done": []}) |
| 225 | serialized = serialize_task_plan(plan) |
| 226 | |
| 227 | assert serialized["todo"] == ["2026-05-03T09:30:00-04:00"] |
| 228 | |
| 229 | |
| 230 | def test_task_plan_create_localizes_naive_datetimes(isolated_localization): |
| 231 | set_test_timezone("Europe/Rome") |
| 232 | |
| 233 | plan = TaskPlan.create(todo=[datetime(2026, 5, 3, 9, 30)]) |
| 234 | |
| 235 | assert plan.todo[0].isoformat() == "2026-05-03T09:30:00+02:00" |
| 236 | |
| 237 | |
| 238 | def test_memory_dashboard_normalizes_legacy_naive_timestamps(isolated_localization): |
| 239 | set_test_timezone("Europe/Rome") |
| 240 | dashboard = MemoryDashboard(app=None, thread_lock=threading.RLock()) |
| 241 | |
| 242 | formatted = dashboard._format_memory_for_dashboard( |
| 243 | Document( |
| 244 | page_content="legacy memory", |
| 245 | metadata={ |
| 246 | "id": "memory-1", |
| 247 | "area": "main", |
| 248 | "timestamp": "2026-05-02 18:27:51", |
| 249 | }, |
| 250 | ) |
| 251 | ) |
| 252 | |
| 253 | assert formatted["timestamp"] == "2026-05-02T18:27:51+02:00" |
| 254 | assert formatted["metadata"]["timestamp"] == "2026-05-02 18:27:51" |
| 255 | |
| 256 | |
| 257 | def test_memory_dashboard_converts_aware_timestamps_to_user_timezone(isolated_localization): |
| 258 | set_test_timezone("Europe/Rome") |
| 259 | dashboard = MemoryDashboard(app=None, thread_lock=threading.RLock()) |
| 260 | |
| 261 | assert ( |
| 262 | dashboard._serialize_memory_timestamp("2026-05-02T16:27:51+00:00") |
| 263 | == "2026-05-02T18:27:51+02:00" |
| 264 | ) |
| 265 | |
| 266 | |
| 267 | class FakeProcess: |
| 268 | pid = 4242 |
| 269 | |
| 270 | def poll(self): |
| 271 | return None |
| 272 | |
| 273 | |
| 274 | def test_desktop_session_env_uses_session_timezone(isolated_localization, tmp_path): |
| 275 | set_test_timezone("Europe/Rome") |
| 276 | session = desktop_session.DesktopSession( |
| 277 | session_id=desktop_session.SYSTEM_SESSION_ID, |
| 278 | file_id=desktop_session.SYSTEM_FILE_ID, |
| 279 | extension="desktop", |
| 280 | path=str(tmp_path), |
| 281 | title="Desktop", |
| 282 | display=120, |
| 283 | xpra_port=14500, |
| 284 | token=desktop_session.SYSTEM_SESSION_ID, |
| 285 | url="/desktop/session/agent-zero-desktop/index.html", |
| 286 | profile_dir=tmp_path / "profile", |
| 287 | timezone="America/New_York", |
| 288 | ) |
| 289 | |
| 290 | env = desktop_session.DesktopSessionManager()._session_env(session) |
| 291 | |
| 292 | assert env["TZ"] == "America/New_York" |
| 293 | |
| 294 | |
| 295 | def test_desktop_timezone_sync_restarts_active_system_desktop( |
| 296 | isolated_localization, |
| 297 | monkeypatch, |
| 298 | tmp_path, |
| 299 | ): |
| 300 | set_test_timezone("America/New_York") |
| 301 | manager = desktop_session.DesktopSessionManager() |
| 302 | old_session = desktop_session.DesktopSession( |
| 303 | session_id=desktop_session.SYSTEM_SESSION_ID, |
| 304 | file_id=desktop_session.SYSTEM_FILE_ID, |
| 305 | extension="desktop", |
| 306 | path=str(tmp_path), |
| 307 | title="Desktop", |
| 308 | display=120, |
| 309 | xpra_port=14500, |
| 310 | token=desktop_session.SYSTEM_SESSION_ID, |
| 311 | url="/desktop/session/agent-zero-desktop/index.html", |
| 312 | profile_dir=tmp_path / "profile-old", |
| 313 | timezone="Europe/Rome", |
| 314 | processes={"xpra": FakeProcess()}, |
| 315 | ) |
| 316 | replacement = desktop_session.DesktopSession( |
| 317 | session_id=desktop_session.SYSTEM_SESSION_ID, |
| 318 | file_id=desktop_session.SYSTEM_FILE_ID, |
| 319 | extension="desktop", |
| 320 | path=str(tmp_path), |
| 321 | title="Desktop", |
| 322 | display=120, |
| 323 | xpra_port=14500, |
| 324 | token=desktop_session.SYSTEM_SESSION_ID, |
| 325 | url="/desktop/session/agent-zero-desktop/index.html", |
| 326 | profile_dir=tmp_path / "profile-new", |
| 327 | timezone="America/New_York", |
| 328 | processes={"xpra": FakeProcess()}, |
| 329 | ) |
| 330 | manager._sessions[desktop_session.SYSTEM_SESSION_ID] = old_session |
| 331 | restarted: list[desktop_session.DesktopSession] = [] |
| 332 | |
| 333 | def fake_restart(session): |
| 334 | restarted.append(session) |
| 335 | manager._sessions[desktop_session.SYSTEM_SESSION_ID] = replacement |
| 336 | return replacement |
| 337 | |
| 338 | monkeypatch.setattr(manager, "_restart_system_desktop_for_timezone_locked", fake_restart) |
| 339 | |
| 340 | result = manager.sync_timezone("America/New_York") |
| 341 | |
| 342 | assert result == { |
| 343 | "ok": True, |
| 344 | "restarted": True, |
| 345 | "session_id": desktop_session.SYSTEM_SESSION_ID, |
| 346 | "timezone": "America/New_York", |
| 347 | } |
| 348 | assert restarted == [old_session] |