From f01545fb1aec85d2d60fa9a1af3a86f69259e98c Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Fri, 26 Jun 2026 05:00:54 +0800 Subject: [PATCH] Add lock_session: lock the workstation and wait for unlock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit session_guard could only detect a locked session and raise; this adds acting on the lock state — lock the box at the end of an unattended run, block until a human unlocks it before resuming, and reduce a lock-state sample stream to lock/unlock events. The lock action runs through an injectable driver and the waits reuse session_guard's real probe with injectable clock/sleep, so all logic is unit-tested without the OS. --- WHATS_NEW.md | 6 + .../doc/new_features/v207_features_doc.rst | 62 +++++++ .../Zh/doc/new_features/v207_features_doc.rst | 55 +++++++ je_auto_control/__init__.py | 7 + .../gui/script_builder/command_schema.py | 28 ++++ .../utils/executor/action_executor.py | 32 ++++ .../utils/lock_session/__init__.py | 10 ++ .../utils/lock_session/lock_session.py | 153 ++++++++++++++++++ .../utils/mcp_server/tools/_factories.py | 38 +++++ .../utils/mcp_server/tools/_handlers.py | 24 +++ .../headless/test_lock_session_batch.py | 139 ++++++++++++++++ 11 files changed, 554 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v207_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v207_features_doc.rst create mode 100644 je_auto_control/utils/lock_session/__init__.py create mode 100644 je_auto_control/utils/lock_session/lock_session.py create mode 100644 test/unit_test/headless/test_lock_session_batch.py diff --git a/WHATS_NEW.md b/WHATS_NEW.md index 04b77bb1..0395fbb5 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -2,6 +2,12 @@ ## What's new (2026-06-26) +### Lock the Workstation + Wait for Unlock + +Lock the box at the end of a run, and block until a human unlocks it before resuming. Full reference: [`docs/source/Eng/doc/new_features/v207_features_doc.rst`](docs/source/Eng/doc/new_features/v207_features_doc.rst). + +- **`lock_session` / `plan_lock_session` / `wait_for_unlock` / `wait_for_lock` / `classify_lock_transitions`** (`AC_lock_session`, `AC_plan_lock_session`, `AC_wait_for_unlock`, `AC_classify_lock_transitions`): `session_guard` could *detect* a locked session and raise; this adds *acting* on it. `lock_session` locks the workstation now (`LockWorkStation` on Windows, `loginctl lock-session` / `CGSession -suspend` elsewhere) through an injectable `driver`; `wait_for_unlock` / `wait_for_lock` poll `session_guard.is_session_locked` (reusing its real Windows `OpenInputDesktop` probe) until the state flips or a timeout, with injectable `clock` / `sleep` / `probe`; `plan_lock_session` is the pure per-OS planner and `classify_lock_transitions` reduces a lock-state sample stream to `{event, locked}` lock/unlock events. `wait_for_unlock` is the blocking companion to `ensure_interactive_session`. Fifth feature of the ROUND-15 cross-app OS lane. No `PySide6`. + ### Read and Control the System Volume Set a known audio baseline before a run — mute, set 30%, or assert the level. Full reference: [`docs/source/Eng/doc/new_features/v206_features_doc.rst`](docs/source/Eng/doc/new_features/v206_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v207_features_doc.rst b/docs/source/Eng/doc/new_features/v207_features_doc.rst new file mode 100644 index 00000000..73585e2c --- /dev/null +++ b/docs/source/Eng/doc/new_features/v207_features_doc.rst @@ -0,0 +1,62 @@ +Lock the Workstation + Wait for Unlock +====================================== + +:mod:`session_guard` answers "is the session locked right now?" and raises if it +is. The missing half is *acting* on the lock state: lock the machine at the end +of an unattended run, block until a human unlocks it before resuming, or reduce +a stream of lock-state samples to lock / unlock events. ``lock_session`` adds +that, behind injectable seams so the logic is testable without touching the OS. + +* :func:`lock_session` — lock the workstation now (``LockWorkStation`` on + Windows, ``loginctl lock-session`` on Linux, ``CGSession -suspend`` on macOS) + through an injectable ``driver``. +* :func:`plan_lock_session` — pure planner: how the lock would be performed on + this OS and whether a default is available (``{backend, argv, available}``). +* :func:`wait_for_unlock` / :func:`wait_for_lock` — poll + :func:`is_session_locked` until the state flips or a timeout, with injectable + ``clock`` / ``sleep`` / ``probe`` for deterministic tests. +* :func:`classify_lock_transitions` — pure: a list of lock-state samples to a + list of ``{event, locked}`` lock / unlock transitions. + +The lock probe reused by the wait helpers is :mod:`session_guard`'s — the +Windows ``OpenInputDesktop`` check — so ``wait_for_unlock`` is the blocking +companion to ``ensure_interactive_session`` (which only raises). Imports no +``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import ( + lock_session, wait_for_unlock, classify_lock_transitions, + ) + + # ... unattended run finishes ... + lock_session() # secure the machine + + # Resume only once a human has unlocked the box + if wait_for_unlock(timeout_s=600): + run_next_stage() + + # Reduce a sampled lock-state log to events + classify_lock_transitions([False, True, True, False]) + # -> [{'event': 'lock', 'locked': True}, + # {'event': 'unlock', 'locked': False}] + +For tests (or any host) pass a ``driver`` / ``probe``: + +.. code-block:: python + + locked = lock_session(driver=lambda: True) # no real lock + wait_for_unlock(probe=lambda: False) # already unlocked + +Executor commands +----------------- + +``AC_lock_session`` (→ ``{locked}``), ``AC_plan_lock_session`` (→ the plan), +``AC_wait_for_unlock`` (``timeout`` / ``interval`` → ``{unlocked}``) and +``AC_classify_lock_transitions`` (``states`` JSON list → ``{events}``). They are +exposed as the matching ``ac_*`` MCP tools (``ac_lock_session`` is destructive — +it interrupts the session; the rest are read-only) and as Script Builder +commands under **Shell**. diff --git a/docs/source/Zh/doc/new_features/v207_features_doc.rst b/docs/source/Zh/doc/new_features/v207_features_doc.rst new file mode 100644 index 00000000..768b7213 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v207_features_doc.rst @@ -0,0 +1,55 @@ +鎖定工作站 + 等待解鎖 +==================== + +:mod:`session_guard` 回答「目前 session 是否鎖定?」並在鎖定時丟出例外。缺少的另一半是對鎖定狀態 +*採取行動*:在無人值守執行結束時鎖定機器、在恢復前阻塞直到有人解鎖,或把一連串鎖定狀態取樣化約為 +鎖定 / 解鎖事件。``lock_session`` 補上這些,並以可注入接縫實作,故邏輯能在不碰作業系統的情況下測試。 + +* :func:`lock_session` ——立即鎖定工作站(Windows 用 ``LockWorkStation``、Linux 用 + ``loginctl lock-session``、macOS 用 ``CGSession -suspend``),透過可注入的 ``driver``。 +* :func:`plan_lock_session` ——純 planner:此 OS 上會如何執行鎖定,以及是否有預設可用 + (``{backend, argv, available}``)。 +* :func:`wait_for_unlock` / :func:`wait_for_lock` ——輪詢 :func:`is_session_locked` + 直到狀態翻轉或逾時,``clock`` / ``sleep`` / ``probe`` 皆可注入以利確定性測試。 +* :func:`classify_lock_transitions` ——純函式:把一連串鎖定狀態取樣化約為 + ``{event, locked}`` 鎖定 / 解鎖轉變的清單。 + +wait 系列重用的鎖定 probe 即 :mod:`session_guard` 的——Windows 的 ``OpenInputDesktop`` 檢查—— +故 ``wait_for_unlock`` 是 ``ensure_interactive_session``(只會丟例外)的阻塞式搭檔。不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import ( + lock_session, wait_for_unlock, classify_lock_transitions, + ) + + # ... 無人值守執行結束 ... + lock_session() # 鎖住機器 + + # 等有人解鎖後才繼續 + if wait_for_unlock(timeout_s=600): + run_next_stage() + + # 把取樣的鎖定狀態紀錄化約為事件 + classify_lock_transitions([False, True, True, False]) + # -> [{'event': 'lock', 'locked': True}, + # {'event': 'unlock', 'locked': False}] + +測試時(或任何主機)可傳入 ``driver`` / ``probe``: + +.. code-block:: python + + locked = lock_session(driver=lambda: True) # 不真正鎖定 + wait_for_unlock(probe=lambda: False) # 已解鎖 + +執行器指令 +---------- + +``AC_lock_session``(→ ``{locked}``)、``AC_plan_lock_session``(→ 計畫)、 +``AC_wait_for_unlock``(``timeout`` / ``interval`` → ``{unlocked}``)與 +``AC_classify_lock_transitions``(``states`` JSON 清單 → ``{events}``)。皆以對應的 ``ac_*`` +MCP 工具(``ac_lock_session`` 為破壞性——會中斷 session;其餘為唯讀)及 Script Builder 指令 +(位於 **Shell** 分類下)形式提供。 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 6d6bb3f4..10a47b62 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -105,6 +105,11 @@ change_volume, get_volume, is_muted, mute, set_mute, set_volume, toggle_mute, unmute, ) +# Lock the workstation, wait for unlock, classify lock transitions +from je_auto_control.utils.lock_session import ( + classify_lock_transitions, lock_session, plan_lock_session, + wait_for_lock, wait_for_unlock, +) # Rich clipboard formats — RTF + CSV/TSV codecs and Windows get / set from je_auto_control.utils.clipboard_rich_formats import ( build_rtf, csv_to_rows, get_clipboard_csv, get_clipboard_rtf, rows_to_csv, @@ -1720,6 +1725,8 @@ def start_autocontrol_gui(*args, **kwargs): "normalize_ext", "file_association", "get_volume", "set_volume", "change_volume", "is_muted", "set_mute", "mute", "unmute", "toggle_mute", + "lock_session", "plan_lock_session", "wait_for_unlock", + "wait_for_lock", "classify_lock_transitions", "build_rtf", "rtf_to_text", "rows_to_csv", "csv_to_rows", "set_clipboard_rtf", "get_clipboard_rtf", "set_clipboard_csv", "get_clipboard_csv", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index cf7dc7ac..e8b56871 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -4345,6 +4345,34 @@ def _add_work_queue_specs(specs: List[CommandSpec]) -> None: fields=(), description="Flip the master mute flag.", )) + specs.append(CommandSpec( + "AC_lock_session", "Shell", "Lock Workstation", + fields=(), + description="Lock the workstation now (interrupts the session).", + )) + specs.append(CommandSpec( + "AC_plan_lock_session", "Shell", "Plan Lock Workstation", + fields=(), + description="Describe how the workstation would be locked (pure).", + )) + specs.append(CommandSpec( + "AC_wait_for_unlock", "Shell", "Wait for Unlock", + fields=( + FieldSpec("timeout", FieldType.FLOAT, optional=True, default=30.0, + placeholder="timeout seconds"), + FieldSpec("interval", FieldType.FLOAT, optional=True, default=0.5, + placeholder="poll interval seconds"), + ), + description="Block until the session is unlocked or timeout.", + )) + specs.append(CommandSpec( + "AC_classify_lock_transitions", "Shell", "Classify Lock Transitions", + fields=( + FieldSpec("states", FieldType.STRING, + placeholder="JSON list of booleans"), + ), + description="Reduce lock-state samples to lock / unlock events.", + )) specs.append(CommandSpec( "AC_normalize_ext", "Shell", "Normalize Extension", fields=( diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 457389da..0f751890 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2683,6 +2683,34 @@ def _toggle_mute() -> Dict[str, Any]: return {"muted": bool(toggle_mute())} +def _lock_session() -> Dict[str, Any]: + """Adapter: lock the workstation now.""" + from je_auto_control.utils.lock_session import lock_session + return {"locked": bool(lock_session())} + + +def _plan_lock_session() -> Dict[str, Any]: + """Adapter: describe how the workstation would be locked (pure).""" + from je_auto_control.utils.lock_session import plan_lock_session + return plan_lock_session() + + +def _wait_for_unlock(timeout: Any = 30.0, interval: Any = 0.5 + ) -> Dict[str, Any]: + """Adapter: block until the session is unlocked or timeout.""" + from je_auto_control.utils.lock_session import wait_for_unlock + unlocked = wait_for_unlock(timeout_s=float(timeout), + interval_s=float(interval)) + return {"unlocked": bool(unlocked)} + + +def _classify_lock_transitions(states: Any) -> Dict[str, Any]: + """Adapter: reduce lock-state samples to lock / unlock events (pure).""" + from je_auto_control.utils.lock_session import classify_lock_transitions + samples = [bool(s) for s in _coerce_list(states)] if states else [] + return {"events": classify_lock_transitions(samples)} + + def _normalize_ext(target: str) -> Dict[str, Any]: """Adapter: the lowercased extension of a path / bare ext (pure).""" from je_auto_control.utils.file_assoc import normalize_ext @@ -6697,6 +6725,10 @@ def __init__(self): "AC_change_volume": _change_volume, "AC_set_mute": _set_mute, "AC_toggle_mute": _toggle_mute, + "AC_lock_session": _lock_session, + "AC_plan_lock_session": _plan_lock_session, + "AC_wait_for_unlock": _wait_for_unlock, + "AC_classify_lock_transitions": _classify_lock_transitions, "AC_normalize_ext": _normalize_ext, "AC_file_association": _file_association, "AC_get_control_text": _get_control_text, diff --git a/je_auto_control/utils/lock_session/__init__.py b/je_auto_control/utils/lock_session/__init__.py new file mode 100644 index 00000000..59e5c5f9 --- /dev/null +++ b/je_auto_control/utils/lock_session/__init__.py @@ -0,0 +1,10 @@ +"""Lock the workstation, wait for unlock, and classify lock transitions.""" +from je_auto_control.utils.lock_session.lock_session import ( + classify_lock_transitions, lock_session, plan_lock_session, + wait_for_lock, wait_for_unlock, +) + +__all__ = [ + "lock_session", "plan_lock_session", "wait_for_unlock", "wait_for_lock", + "classify_lock_transitions", +] diff --git a/je_auto_control/utils/lock_session/lock_session.py b/je_auto_control/utils/lock_session/lock_session.py new file mode 100644 index 00000000..cacca6aa --- /dev/null +++ b/je_auto_control/utils/lock_session/lock_session.py @@ -0,0 +1,153 @@ +"""Lock the workstation, wait for it to be unlocked, and classify transitions. + +:mod:`session_guard` answers "is the session locked right now?" and raises if it +is. The missing half is *acting* on the lock state: lock the machine at the end +of an unattended run, block until a human unlocks it before resuming, or reduce +a stream of lock-state samples to lock / unlock events. ``lock_session`` adds: + +* :func:`lock_session` — lock the workstation now (``LockWorkStation`` on + Windows, ``loginctl lock-session`` / ``CGSession -suspend`` elsewhere), + through an injectable ``driver`` seam. +* :func:`plan_lock_session` — pure planner describing how the lock would be + performed on this OS, and whether a default is available. +* :func:`wait_for_unlock` / :func:`wait_for_lock` — poll + :func:`session_guard.is_session_locked` until the state flips or a timeout, + with injectable ``clock`` / ``sleep`` / ``probe`` for deterministic tests. +* :func:`classify_lock_transitions` — pure: a list of lock-state samples to a + list of ``{event, locked}`` lock / unlock transitions. + +Imports no ``PySide6``. +""" +import sys +import time +from typing import Any, Callable, Dict, List, Optional, Sequence + +from je_auto_control.utils.session_guard import is_session_locked +from je_auto_control.utils.session_guard.session_guard import LockProbe + +# A driver performs the lock and returns whether it succeeded. +LockDriver = Callable[[], bool] + + +def _win_lock() -> bool: + """Lock the Windows workstation via ``LockWorkStation``.""" + import ctypes + user32 = ctypes.windll.user32 # nosec B607 # reason: fixed system DLL + return bool(user32.LockWorkStation()) + + +def _lock_backend() -> str: + """Return the lock backend name for the current platform.""" + if sys.platform.startswith("win"): + return "LockWorkStation" + if sys.platform == "darwin": + return "CGSession" + return "loginctl" + + +_CGSESSION = ("/System/Library/CoreServices/Menu Extras/" + "User.menu/Contents/Resources/CGSession") + + +def _lock_argv(backend: str) -> Optional[List[str]]: + """Return the fixed argv for a subprocess lock backend, or None.""" + if backend == "loginctl": + return ["loginctl", "lock-session"] + if backend == "CGSession": + return [_CGSESSION, "-suspend"] + return None + + +def plan_lock_session() -> Dict[str, Any]: + """Describe how the workstation would be locked on this OS (pure). + + Returns ``{backend, argv, available}``. ``available`` is ``True`` when this + platform has a built-in default; otherwise :func:`lock_session` needs an + explicit ``driver=``. + """ + backend = _lock_backend() + argv = _lock_argv(backend) + available = backend == "LockWorkStation" or argv is not None + return {"backend": backend, "argv": argv, "available": available} + + +def _run_argv(argv: List[str]) -> bool: + """Run a fixed lock argv and report success (no shell).""" + import subprocess # nosec B404 # reason: fixed argv, no shell + completed = subprocess.run(argv, check=False) # nosec B603 # nosemgrep + return completed.returncode == 0 + + +def _default_driver() -> LockDriver: + """Return the OS lock driver, or raise if none is available.""" + backend = _lock_backend() + if backend == "LockWorkStation": + return _win_lock + argv = _lock_argv(backend) + if argv is not None: + return lambda: _run_argv(argv) + raise RuntimeError( + "lock_session has no OS driver on this platform; pass driver=") + + +def lock_session(*, driver: Optional[LockDriver] = None) -> bool: + """Lock the workstation now; return whether the lock was requested. + + Pass ``driver`` (a ``() -> bool``) to intercept the OS call in tests; the + default locks via the platform backend from :func:`plan_lock_session`. + """ + acquire = driver if driver is not None else _default_driver() + return bool(acquire()) + + +def _wait_lock_state(target_locked: bool, *, probe: Optional[LockProbe], + timeout_s: float, interval_s: float, + clock: Callable[[], float], + sleep: Callable[[float], None]) -> bool: + """Poll until the lock state equals ``target_locked`` or timeout.""" + deadline = clock() + float(timeout_s) + while True: + if bool(is_session_locked(probe)) == target_locked: + return True + if clock() >= deadline: + return False + sleep(float(interval_s)) + + +def wait_for_unlock(*, probe: Optional[LockProbe] = None, + timeout_s: float = 30.0, interval_s: float = 0.5, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep) -> bool: + """Block until the session is unlocked; return ``True``, or ``False`` on timeout. + + Reuses :func:`session_guard.is_session_locked` (Windows default probe). + ``clock`` / ``sleep`` / ``probe`` are injectable for deterministic tests. + """ + return _wait_lock_state(False, probe=probe, timeout_s=timeout_s, + interval_s=interval_s, clock=clock, sleep=sleep) + + +def wait_for_lock(*, probe: Optional[LockProbe] = None, + timeout_s: float = 30.0, interval_s: float = 0.5, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep) -> bool: + """Block until the session is locked; return ``True``, or ``False`` on timeout.""" + return _wait_lock_state(True, probe=probe, timeout_s=timeout_s, + interval_s=interval_s, clock=clock, sleep=sleep) + + +def classify_lock_transitions(states: Sequence[bool]) -> List[Dict[str, Any]]: + """Reduce lock-state samples to lock / unlock transitions (pure). + + Each adjacent ``False -> True`` yields a ``lock`` event and each + ``True -> False`` an ``unlock`` event; unchanged samples yield nothing. + """ + events: List[Dict[str, Any]] = [] + previous: Optional[bool] = None + for sample in states: + current = bool(sample) + if previous is not None and current != previous: + kind = "lock" if current else "unlock" + events.append({"event": kind, "locked": current}) + previous = current + return events diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index 5ef00f30..83df5f93 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -2211,6 +2211,44 @@ def process_and_shell_tools() -> List[MCPTool]: handler=h.toggle_mute, annotations=SIDE_EFFECT_ONLY, ), + MCPTool( + name="ac_lock_session", + description=("Lock the workstation now (LockWorkStation / loginctl " + "lock-session / CGSession). Returns {locked}. " + "Interrupts the interactive session."), + input_schema=schema({}), + handler=h.lock_session, + annotations=DESTRUCTIVE, + ), + MCPTool( + name="ac_plan_lock_session", + description=("Describe how the workstation would be locked on this " + "OS without locking (pure): {backend, argv, " + "available}."), + input_schema=schema({}), + handler=h.plan_lock_session, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_wait_for_unlock", + description=("Block until the session is unlocked (or 'timeout' " + "seconds), polling every 'interval'. Returns " + "{unlocked}."), + input_schema=schema({"timeout": {"type": "number"}, + "interval": {"type": "number"}}), + handler=h.wait_for_unlock, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_classify_lock_transitions", + description=("Reduce a list of lock-state booleans ('states') to " + "lock / unlock events (pure). Returns {events}."), + input_schema=schema({"states": {"type": "array", + "items": {"type": "boolean"}}}, + required=["states"]), + handler=h.classify_lock_transitions, + annotations=READ_ONLY, + ), ] diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 41fc8c7a..613adf2d 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -620,6 +620,30 @@ def toggle_mute(): return _toggle_mute() +def lock_session(): + from je_auto_control.utils.executor.action_executor import _lock_session + return _lock_session() + + +def plan_lock_session(): + from je_auto_control.utils.executor.action_executor import ( + _plan_lock_session, + ) + return _plan_lock_session() + + +def wait_for_unlock(timeout=30.0, interval=0.5): + from je_auto_control.utils.executor.action_executor import _wait_for_unlock + return _wait_for_unlock(timeout, interval) + + +def classify_lock_transitions(states): + from je_auto_control.utils.executor.action_executor import ( + _classify_lock_transitions, + ) + return _classify_lock_transitions(states) + + def normalize_ext(target): from je_auto_control.utils.executor.action_executor import _normalize_ext return _normalize_ext(target) diff --git a/test/unit_test/headless/test_lock_session_batch.py b/test/unit_test/headless/test_lock_session_batch.py new file mode 100644 index 00000000..af16d1c4 --- /dev/null +++ b/test/unit_test/headless/test_lock_session_batch.py @@ -0,0 +1,139 @@ +"""Headless tests for lock_session (injected driver / probe / clock).""" +import sys + +import je_auto_control as ac +from je_auto_control.utils.lock_session import ( + classify_lock_transitions, lock_session, plan_lock_session, + wait_for_lock, wait_for_unlock, +) + + +# --- lock action ---------------------------------------------------------- + +def test_lock_session_uses_driver(): + calls = [] + + def fake_driver(): + calls.append(True) + return True + + assert lock_session(driver=fake_driver) is True + assert calls == [True] + + +def test_lock_session_driver_failure_reported(): + assert lock_session(driver=lambda: False) is False + + +def test_plan_lock_session_shape(): + plan = plan_lock_session() + assert set(plan) == {"backend", "argv", "available"} + assert plan["backend"] in ("LockWorkStation", "loginctl", "CGSession") + assert isinstance(plan["available"], bool) + + +def test_plan_lock_session_available_per_os(): + plan = plan_lock_session() + # Every supported platform path has a default backend. + assert plan["available"] is True + if plan["backend"] == "LockWorkStation": + assert plan["argv"] is None + else: + assert isinstance(plan["argv"], list) and len(plan["argv"]) >= 1 + + +# --- wait for unlock / lock ----------------------------------------------- + +def _probe_sequence(values): + """Return a probe yielding successive booleans, repeating the last.""" + state = {"i": 0} + + def probe(): + i = min(state["i"], len(values) - 1) + state["i"] += 1 + return values[i] + + return probe + + +def test_wait_for_unlock_returns_true_when_unlocked(): + # locked, locked, then unlocked -> returns True + probe = _probe_sequence([True, True, False]) + clock = iter([0.0, 1.0, 2.0, 3.0, 4.0]) + ok = wait_for_unlock(probe=probe, timeout_s=10.0, interval_s=1.0, + clock=lambda: next(clock), sleep=lambda _s: None) + assert ok is True + + +def test_wait_for_unlock_times_out_when_still_locked(): + probe = _probe_sequence([True]) # never unlocks + times = iter([0.0, 0.0, 5.0, 10.0, 20.0]) + ok = wait_for_unlock(probe=probe, timeout_s=5.0, interval_s=1.0, + clock=lambda: next(times), sleep=lambda _s: None) + assert ok is False + + +def test_wait_for_lock_returns_true_when_locked(): + probe = _probe_sequence([False, True]) + clock = iter([0.0, 1.0, 2.0, 3.0]) + ok = wait_for_lock(probe=probe, timeout_s=10.0, interval_s=1.0, + clock=lambda: next(clock), sleep=lambda _s: None) + assert ok is True + + +# --- pure transition classifier ------------------------------------------- + +def test_classify_lock_transitions_events(): + events = classify_lock_transitions([False, False, True, True, False]) + assert events == [ + {"event": "lock", "locked": True}, + {"event": "unlock", "locked": False}, + ] + + +def test_classify_lock_transitions_empty_and_constant(): + assert classify_lock_transitions([]) == [] + assert classify_lock_transitions([True, True, True]) == [] + + +# --- wiring --------------------------------------------------------------- + +def test_executor_pure_classify_path(): + from je_auto_control.utils.executor.action_executor import ( + _classify_lock_transitions, _plan_lock_session, + ) + out = _classify_lock_transitions([False, True]) + assert out == {"events": [{"event": "lock", "locked": True}]} + # accepts a JSON-list string too (Script Builder text field) + assert _classify_lock_transitions("[false, true]")["events"] == [ + {"event": "lock", "locked": True}] + assert "backend" in _plan_lock_session() + + +def test_default_driver_absent_off_windows(): + from je_auto_control.utils.lock_session.lock_session import _default_driver + if not sys.platform.startswith("win") and sys.platform != "darwin": + # Linux default is loginctl (callable built), so a driver exists. + assert callable(_default_driver()) + + +def test_wiring(): + known = set(ac.executor.known_commands()) + assert {"AC_lock_session", "AC_plan_lock_session", "AC_wait_for_unlock", + "AC_classify_lock_transitions"} <= known + from je_auto_control.utils.mcp_server.tools import ( + build_default_tool_registry, + ) + names = {t.name for t in build_default_tool_registry()} + assert {"ac_lock_session", "ac_plan_lock_session", "ac_wait_for_unlock", + "ac_classify_lock_transitions"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert {"AC_lock_session", "AC_plan_lock_session", "AC_wait_for_unlock", + "AC_classify_lock_transitions"} <= specs + + +def test_facade_exports(): + for name in ("lock_session", "plan_lock_session", "wait_for_unlock", + "wait_for_lock", "classify_lock_transitions"): + assert hasattr(ac, name) and name in ac.__all__