Merge pull request #172 from PR0M3TH3AN/codex/improve-inactivity-timer-test

Fix inactivity timer and add tests

src/main.py

@@ -18,6 +18,7 @@ from password_manager.manager import PasswordManager
 from nostr.client import NostrClient
 from constants import INACTIVITY_TIMEOUT, initialize_app
 from utils.password_prompt import PasswordPromptError
+from utils import timed_input
 from local_bip85.bip85 import Bip85Error
 
 
@@ -568,7 +569,15 @@ def display_menu(
         for handler in logging.getLogger().handlers:
             handler.flush()
         print(colored(menu, "cyan"))
-        choice = input("Enter your choice (1-5): ").strip()
+        try:
+            choice = timed_input(
+                "Enter your choice (1-5): ", inactivity_timeout
+            ).strip()
+        except TimeoutError:
+            print(colored("Session timed out. Vault locked.", "yellow"))
+            password_manager.lock_vault()
+            password_manager.unlock_vault()
+            continue
         password_manager.update_activity()
         if not choice:
             print(

src/tests/test_auto_sync.py

@@ -31,7 +31,7 @@ def test_auto_sync_triggers_post(monkeypatch):
         called = True
 
     monkeypatch.setattr(main, "handle_post_to_nostr", fake_post)
-    monkeypatch.setattr("builtins.input", lambda _: "5")
+    monkeypatch.setattr(main, "timed_input", lambda *_: "5")
 
     with pytest.raises(SystemExit):
         main.display_menu(pm, sync_interval=0.1)

src/tests/test_cli_invalid_input.py

@@ -52,7 +52,7 @@ def test_empty_and_non_numeric_choice(monkeypatch, capsys):
     called = {"add": False, "retrieve": False, "modify": False}
     pm, _ = _make_pm(called)
     inputs = iter(["", "abc", "5"])
-    monkeypatch.setattr("builtins.input", lambda *_: next(inputs))
+    monkeypatch.setattr(main, "timed_input", lambda *_: next(inputs))
     with pytest.raises(SystemExit):
         main.display_menu(pm, sync_interval=1000, inactivity_timeout=1000)
     out = capsys.readouterr().out
@@ -65,7 +65,7 @@ def test_out_of_range_menu(monkeypatch, capsys):
     called = {"add": False, "retrieve": False, "modify": False}
     pm, _ = _make_pm(called)
     inputs = iter(["9", "5"])
-    monkeypatch.setattr("builtins.input", lambda *_: next(inputs))
+    monkeypatch.setattr(main, "timed_input", lambda *_: next(inputs))
     with pytest.raises(SystemExit):
         main.display_menu(pm, sync_interval=1000, inactivity_timeout=1000)
     out = capsys.readouterr().out
@@ -77,6 +77,7 @@ def test_invalid_add_entry_submenu(monkeypatch, capsys):
     called = {"add": False, "retrieve": False, "modify": False}
     pm, _ = _make_pm(called)
     inputs = iter(["1", "3", "2", "5"])
+    monkeypatch.setattr(main, "timed_input", lambda *_: next(inputs))
     monkeypatch.setattr("builtins.input", lambda *_: next(inputs))
     with pytest.raises(SystemExit):
         main.display_menu(pm, sync_interval=1000, inactivity_timeout=1000)
@@ -90,7 +91,7 @@ def test_inactivity_timeout_loop(monkeypatch, capsys):
     pm, locked = _make_pm(called)
     pm.last_activity = 0
     monkeypatch.setattr(time, "time", lambda: 100.0)
-    monkeypatch.setattr("builtins.input", lambda *_: "5")
+    monkeypatch.setattr(main, "timed_input", lambda *_: "5")
     with pytest.raises(SystemExit):
         main.display_menu(pm, sync_interval=1000, inactivity_timeout=0.1)
     out = capsys.readouterr().out

src/tests/test_inactivity_lock.py

@@ -36,10 +36,54 @@ def test_inactivity_triggers_lock(monkeypatch):
         unlock_vault=unlock_vault,
     )
 
-    monkeypatch.setattr("builtins.input", lambda _: "5")
+    monkeypatch.setattr(main, "timed_input", lambda *_: "5")
 
     with pytest.raises(SystemExit):
         main.display_menu(pm, sync_interval=1000, inactivity_timeout=0.1)
 
     assert locked["locked"]
     assert locked["unlocked"]
+
+
+def test_input_timeout_triggers_lock(monkeypatch):
+    """Ensure locking occurs if no input is provided before timeout."""
+    locked = {"locked": 0, "unlocked": 0}
+
+    def update_activity():
+        pm.last_activity = time.time()
+
+    def lock_vault():
+        locked["locked"] += 1
+
+    def unlock_vault():
+        locked["unlocked"] += 1
+        update_activity()
+
+    pm = SimpleNamespace(
+        is_dirty=False,
+        last_update=time.time(),
+        last_activity=time.time(),
+        nostr_client=SimpleNamespace(close_client_pool=lambda: None),
+        handle_add_password=lambda: None,
+        handle_retrieve_entry=lambda: None,
+        handle_modify_entry=lambda: None,
+        update_activity=update_activity,
+        lock_vault=lock_vault,
+        unlock_vault=unlock_vault,
+    )
+
+    responses = iter([TimeoutError(), "5"])
+
+    def fake_input(*_args, **_kwargs):
+        val = next(responses)
+        if isinstance(val, Exception):
+            raise val
+        return val
+
+    monkeypatch.setattr(main, "timed_input", fake_input)
+
+    with pytest.raises(SystemExit):
+        main.display_menu(pm, sync_interval=1000, inactivity_timeout=0.1)
+
+    assert locked["locked"] == 1
+    assert locked["unlocked"] == 1

src/utils/__init__.py

@@ -21,6 +21,7 @@ try:
         canonical_json_dumps,
     )
     from .password_prompt import prompt_for_password
+    from .input_utils import timed_input
 
     if logger.isEnabledFor(logging.DEBUG):
         logger.info("Modules imported successfully.")
@@ -41,4 +42,5 @@ __all__ = [
     "exclusive_lock",
     "shared_lock",
     "prompt_for_password",
+    "timed_input",
 ]

src/utils/input_utils.py

@@ -0,0 +1,19 @@
+import sys
+import select
+import io
+from typing import Optional
+
+
+def timed_input(prompt: str, timeout: Optional[float]) -> str:
+    """Read input from the user with a timeout."""
+    print(prompt, end="", flush=True)
+    if timeout is None or timeout <= 0:
+        return sys.stdin.readline().strip()
+    try:
+        sys.stdin.fileno()
+    except (AttributeError, io.UnsupportedOperation):
+        return input().strip()
+    ready, _, _ = select.select([sys.stdin], [], [], timeout)
+    if ready:
+        return sys.stdin.readline().strip()
+    raise TimeoutError("input timed out")