cottongin/cursor-flasher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6610919a58661cd165c9c8b46527a2e0ac3b7536
cursor-flasher/tests/test_daemon.py
cottongin 6610919a58 Show overlay on all Spaces and alongside fullscreen apps 
Set canJoinAllSpaces + fullScreenAuxiliary on overlay windows so the
border renders regardless of which Space or fullscreen app is active.
In window mode, fall back to screen-edge border when Cursor isn't
frontmost to avoid a floating rectangle on other Spaces.

Made-with: Cursor
2026-03-11 03:07:12 -04:00

610 lines
26 KiB
Python
Raw Blame History

"""Tests for the daemon's message handling logic."""
import json
import time
from unittest.mock import patch, MagicMock
from cursor_flasher.config import Config, StyleConfig, ThemeStyles
from cursor_flasher.daemon import FlasherDaemon
PATCH_APPEARANCE = "cursor_flasher.daemon._get_system_appearance"
PATCH_FOCUSED = "cursor_flasher.daemon.get_focused_cursor_window"
class TestFlasherDaemon:
def _make_daemon(self, **config_overrides) -> FlasherDaemon:
config = Config(**config_overrides)
with patch("cursor_flasher.daemon.OverlayManager") as MockOverlay:
daemon = FlasherDaemon(config)
daemon.overlay.is_pulsing = False
daemon.overlay.active_tags = set()
daemon.overlay.has_tag = lambda tag: tag in daemon.overlay.active_tags
return daemon
# --- preToolUse / pending ---
def test_preToolUse_queues_pending(self):
daemon = self._make_daemon()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/path" in daemon._pending_approvals
assert daemon._pending_approvals["/path"].tool == "Shell"
daemon.overlay.add_pulse.assert_not_called()
def test_pending_promotes_after_delay(self):
daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen")
window = {"title": "my-project", "frame": ((0, 0), (800, 600))}
screen = ((0, 0), (1920, 1080))
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [screen], daemon.config.dark.running
)
assert "/path" in daemon._active_pulses
assert "/path" not in daemon._pending_approvals
def test_postToolUse_cancels_pending(self):
"""Auto-approved tools: postToolUse arrives before delay, cancels the pulse."""
daemon = self._make_daemon(approval_delay=10.0)
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/path" in daemon._pending_approvals
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode()
)
assert "/path" not in daemon._pending_approvals
daemon.overlay.add_pulse.assert_not_called()
def test_preToolUse_skips_non_approval_tool(self):
daemon = self._make_daemon()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Read"}).encode()
)
assert not daemon._pending_approvals
def test_preToolUse_respects_custom_tool_list(self):
daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen", approval_tools=["Shell", "MCP"])
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "MCP"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=((0, 0), (1920, 1080))), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.add_pulse.assert_called_once()
# --- stop / flash ---
def test_stop_flashes_briefly(self):
daemon = self._make_daemon(flash_mode="screen")
window = {"title": "my-project", "frame": ((0, 0), (800, 600))}
screen = ((0, 0), (1920, 1080))
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.add_flash.assert_called_once_with(
"/path", [screen], daemon.config.dark.completed
)
def test_stop_flashes_window_frame_when_window_mode(self):
daemon = self._make_daemon(flash_mode="window")
window = {"title": "my-project", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="light"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.add_flash.assert_called_once_with(
"/path", [((0, 0), (800, 600))], daemon.config.light.completed
)
def test_window_mode_falls_back_to_screen_when_cursor_not_frontmost(self):
"""Window mode falls back to screen frame when Cursor isn't frontmost."""
daemon = self._make_daemon(approval_delay=0.0, flash_mode="window")
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
screen = ((0, 0), (1920, 1080))
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [screen], daemon.config.dark.running
)
def test_stop_falls_back_to_screen_when_cursor_not_frontmost(self):
"""Stop flash in window mode falls back to screen when Cursor isn't frontmost."""
daemon = self._make_daemon(flash_mode="window")
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
screen = ((0, 0), (1920, 1080))
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.add_flash.assert_called_once_with(
"/path", [screen], daemon.config.dark.completed
)
def test_allscreens_mode_uses_all_screens(self):
daemon = self._make_daemon(flash_mode="allscreens", approval_delay=0.0)
window = {"title": "my-project", "frame": ((0, 0), (800, 600))}
screens = [((0, 0), (1920, 1080)), ((-1920, 0), (1920, 1080))]
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.all_screen_frames", return_value=screens), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.add_pulse.assert_called_once_with(
"/path", screens, daemon.config.dark.running
)
# --- stop interactions with active pulse ---
def test_stop_dismisses_active_pulse_for_workspace(self):
daemon = self._make_daemon()
daemon._active_pulses["/path"] = MagicMock()
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=None), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_with("/path")
assert "/path" not in daemon._active_pulses
def test_stop_clears_pending(self):
daemon = self._make_daemon(approval_delay=10.0)
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/path" in daemon._pending_approvals
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
assert "/path" not in daemon._pending_approvals
# --- postToolUse / dismiss ---
def test_postToolUse_dismisses_active_pulse(self):
daemon = self._make_daemon()
daemon._active_pulses["/path"] = MagicMock()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_with("/path")
assert "/path" not in daemon._active_pulses
def test_postToolUseFailure_dismisses_pulse(self):
daemon = self._make_daemon()
daemon._active_pulses["/path"] = MagicMock()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUseFailure", "tool": "Shell"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_with("/path")
# --- cooldown ---
def test_cooldown_prevents_rapid_triggers(self):
daemon = self._make_daemon(cooldown=5.0)
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
daemon._last_flash["/path"] = time.monotonic()
daemon._pending_approvals.clear()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/path" not in daemon._pending_approvals
def test_cooldown_is_per_workspace(self):
daemon = self._make_daemon(cooldown=5.0)
daemon._last_flash["/pathA"] = time.monotonic()
daemon._handle_message(
json.dumps({"workspace": "/pathB", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/pathB" in daemon._pending_approvals
# --- misc ---
def test_invalid_json_ignored(self):
daemon = self._make_daemon()
daemon._handle_message(b"not json")
daemon.overlay.add_pulse.assert_not_called()
daemon.overlay.add_flash.assert_not_called()
def test_no_window_found(self):
daemon = self._make_daemon(approval_delay=0.0)
daemon._handle_message(
json.dumps({"workspace": "/nope", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=None):
daemon._check_pending()
daemon.overlay.add_pulse.assert_not_called()
# --- focus dismiss ---
def test_focus_transition_dismisses_focused_workspace(self):
"""Pulse dismisses when user switches TO Cursor, only for the focused window."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/pathA"] = _ActivePulse("/pathA", "project-a", time.monotonic())
daemon._active_pulses["/pathB"] = _ActivePulse("/pathB", "project-b", time.monotonic())
daemon._cursor_was_frontmost = False
focused = {"title": "project-a", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch(PATCH_FOCUSED, return_value=focused):
daemon._check_focus()
daemon.overlay.dismiss_tag.assert_called_once_with("/pathA")
assert "/pathA" not in daemon._active_pulses
assert "/pathB" in daemon._active_pulses
def test_focus_no_dismiss_when_cursor_already_frontmost(self):
"""No dismiss if Cursor was already frontmost (no transition)."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic())
daemon._cursor_was_frontmost = True
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch(PATCH_FOCUSED, return_value={"title": "proj", "frame": ((0, 0), (800, 600))}):
daemon._check_focus()
daemon.overlay.dismiss_tag.assert_not_called()
def test_focus_tracks_state_changes(self):
"""_cursor_was_frontmost updates each tick."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic())
daemon._cursor_was_frontmost = True
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False):
daemon._check_focus()
assert daemon._cursor_was_frontmost is False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch(PATCH_FOCUSED, return_value={"title": "proj", "frame": ((0, 0), (800, 600))}):
daemon._check_focus()
daemon.overlay.dismiss_tag.assert_called_once_with("/path")
def test_focus_no_dismiss_when_not_pulsing(self):
daemon = self._make_daemon()
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True):
daemon._check_focus()
daemon.overlay.dismiss_tag.assert_not_called()
# --- input dismiss ---
def test_input_dismiss_targets_focused_workspace(self):
"""Recent input + Cursor frontmost + past grace period = dismiss focused workspace only."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/pathA"] = _ActivePulse("/pathA", "project-a", time.monotonic() - 2.0)
daemon._active_pulses["/pathB"] = _ActivePulse("/pathB", "project-b", time.monotonic() - 2.0)
focused = {"title": "project-a", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.2), \
patch(PATCH_FOCUSED, return_value=focused):
daemon._check_input_dismiss()
daemon.overlay.dismiss_tag.assert_called_once_with("/pathA")
assert "/pathA" not in daemon._active_pulses
assert "/pathB" in daemon._active_pulses
def test_input_dismiss_skipped_during_grace_period(self):
"""No dismiss if pulse just started (within grace period)."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 0.1)
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.05):
daemon._check_input_dismiss()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_skipped_when_not_pulsing(self):
daemon = self._make_daemon()
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.1):
daemon._check_input_dismiss()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_skipped_when_cursor_not_frontmost(self):
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 2.0)
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False):
daemon._check_input_dismiss()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_ignores_old_input(self):
"""Input from before the pulse started should not trigger dismiss."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 2.0)
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=5.0):
daemon._check_input_dismiss()
daemon.overlay.dismiss_tag.assert_not_called()
# --- sound ---
def test_running_style_sound_plays_on_approval(self):
"""Running style with sound configured plays on approval pulse."""
running = StyleConfig(sound="Glass", volume=0.5)
dark = ThemeStyles(running=running)
daemon = self._make_daemon(approval_delay=0.0, flash_mode="window", dark=dark)
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert") as mock_alert, \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
mock_alert.assert_called_once_with(running)
def test_completed_style_sound_plays_on_stop(self):
"""Completed style with sound configured plays on stop flash."""
completed = StyleConfig(sound="Ping", volume=0.7)
dark = ThemeStyles(completed=completed)
daemon = self._make_daemon(flash_mode="window", dark=dark)
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert") as mock_alert, \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
mock_alert.assert_called_once_with(completed)
def test_no_sound_when_style_sound_empty(self):
"""No sound plays when the style has sound="" (the completed default)."""
completed = StyleConfig(sound="", volume=0.0)
dark = ThemeStyles(completed=completed)
daemon = self._make_daemon(flash_mode="window", dark=dark)
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert") as mock_alert, \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
mock_alert.assert_called_once_with(completed)
def test_custom_colors_per_mode(self):
"""Different colors for running vs completed are passed through."""
running = StyleConfig(color="#FF0000")
completed = StyleConfig(color="#00FF00")
dark = ThemeStyles(running=running, completed=completed)
daemon = self._make_daemon(
approval_delay=0.0, flash_mode="window", dark=dark,
)
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [((0, 0), (800, 600))], running
)
daemon.overlay.add_pulse.reset_mock()
daemon._last_flash.clear()
daemon._active_pulses.clear()
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.add_flash.assert_called_once_with(
"/path", [((0, 0), (800, 600))], completed
)
def test_theme_auto_uses_light_when_system_light(self):
"""Auto theme resolves to light styles when system is in light mode."""
dark_running = StyleConfig(color="#111111")
light_running = StyleConfig(color="#EEEEEE")
daemon = self._make_daemon(
approval_delay=0.0, flash_mode="window",
dark=ThemeStyles(running=dark_running),
light=ThemeStyles(running=light_running),
theme="auto",
)
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="light"):
daemon._check_pending()
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [((0, 0), (800, 600))], light_running
)
# --- multi-workspace scenarios ---
def test_two_workspaces_pulse_simultaneously(self):
"""Two workspaces can pulse at the same time."""
daemon = self._make_daemon(approval_delay=0.0, flash_mode="window")
win_a = {"title": "project-a", "frame": ((0, 0), (800, 600))}
win_b = {"title": "project-b", "frame": ((900, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "preToolUse", "tool": "Shell"}).encode()
)
daemon._handle_message(
json.dumps({"workspace": "/b", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", side_effect=[win_a, win_b]), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
assert daemon.overlay.add_pulse.call_count == 2
assert "/a" in daemon._active_pulses
assert "/b" in daemon._active_pulses
def test_dismiss_one_workspace_keeps_other_pulsing(self):
"""postToolUse for workspace A only dismisses A, B keeps pulsing."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/a"] = _ActivePulse("/a", "project-a", time.monotonic())
daemon._active_pulses["/b"] = _ActivePulse("/b", "project-b", time.monotonic())
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "postToolUse", "tool": "Shell"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_once_with("/a")
assert "/a" not in daemon._active_pulses
assert "/b" in daemon._active_pulses
def test_stop_only_affects_its_workspace(self):
"""Stop for workspace A dismisses A's pulse and flashes A, B keeps pulsing."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon(flash_mode="window")
daemon._active_pulses["/a"] = _ActivePulse("/a", "project-a", time.monotonic())
daemon._active_pulses["/b"] = _ActivePulse("/b", "project-b", time.monotonic())
window = {"title": "project-a", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "stop"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_with("/a")
daemon.overlay.add_flash.assert_called_once()
assert "/a" not in daemon._active_pulses
assert "/b" in daemon._active_pulses
def test_postToolUse_only_cancels_matching_workspace_pending(self):
"""postToolUse for workspace A doesn't cancel workspace B's pending."""
daemon = self._make_daemon(approval_delay=10.0)
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "preToolUse", "tool": "Shell"}).encode()
)
daemon._handle_message(
json.dumps({"workspace": "/b", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/a" in daemon._pending_approvals
assert "/b" in daemon._pending_approvals
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "postToolUse", "tool": "Shell"}).encode()
)
assert "/a" not in daemon._pending_approvals
assert "/b" in daemon._pending_approvals
Reference in New Issue View Git Blame Copy Permalink