diff --git a/src/praisonai-agents/praisonaiagents/session/store.py b/src/praisonai-agents/praisonaiagents/session/store.py index 61d46468c..20b1ee81a 100644 --- a/src/praisonai-agents/praisonaiagents/session/store.py +++ b/src/praisonai-agents/praisonaiagents/session/store.py @@ -15,10 +15,10 @@ import time # fcntl is Unix-only; on Windows, use msvcrt for file locking -if sys.platform != 'win32': +try: import fcntl _HAS_FCNTL = True -else: +except ImportError: _HAS_FCNTL = False from dataclasses import dataclass, field from datetime import datetime, timezone @@ -167,11 +167,11 @@ def acquire(self) -> bool: if _HAS_FCNTL: fcntl.flock(self._lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) else: - # Warn once about degraded locking (should never happen with current platform detection) + # Warn once about degraded locking on non-Windows platforms without fcntl global _WARNED_NO_FCNTL if not _WARNED_NO_FCNTL: logger.warning( - "File locking unavailable on this platform (no fcntl/msvcrt); " + "File locking unavailable on this platform (fcntl not available); " "concurrent writers may corrupt session files." ) _WARNED_NO_FCNTL = True diff --git a/src/praisonai-agents/tests/unit/session/test_session_store.py b/src/praisonai-agents/tests/unit/session/test_session_store.py index 368dc0c0e..0e3e0ad38 100644 --- a/src/praisonai-agents/tests/unit/session/test_session_store.py +++ b/src/praisonai-agents/tests/unit/session/test_session_store.py @@ -9,8 +9,11 @@ import tempfile import threading import time +import builtins +import importlib import pytest from concurrent.futures import ThreadPoolExecutor +from unittest.mock import patch from praisonaiagents.session.store import ( DefaultSessionStore, @@ -160,6 +163,26 @@ def worker(worker_id): if start_idx < other_start < end_idx: pytest.fail(f"Lock interleaving detected: {results}") + def test_import_without_fcntl(self): + """Test module import succeeds when fcntl is unavailable.""" + import praisonaiagents.session.store as store_module + import importlib.util + original_import = builtins.__import__ + + def import_without_fcntl(name, *args, **kwargs): + if name == "fcntl": + raise ImportError("fcntl unavailable") + return original_import(name, *args, **kwargs) + + try: + with patch("builtins.__import__", side_effect=import_without_fcntl): + reloaded_module = importlib.reload(store_module) + assert reloaded_module._HAS_FCNTL is False + finally: + restored_module = importlib.reload(store_module) + + assert restored_module._HAS_FCNTL is (importlib.util.find_spec("fcntl") is not None) + class TestDefaultSessionStore: """Tests for DefaultSessionStore."""