Add native Windows host daemon (bleak/WinRT)
Adds a Windows port of the host daemon so the Clawdmeter stays connected on Windows independent of WSL. Mirrors the macOS daemon and speaks the existing GATT data service unchanged — no firmware changes. - claude_usage_daemon_windows.py: Windows-local OAuth token read + Anthropic poll + BLE scan/connect/write, with auto-reconnect (connect-retry wrapper, zombie-link break, split fast/slow backoff) - tray_windows.py: pystray login-startup tray app (status icon + Quit) - autostart_windows.py: winreg HKCU\Run autostart via pythonw.exe - icon_assets.py: per-state tray icons composited from logo.h - install-windows.ps1 + daemon/README-windows.md: turnkey setup - pytest suite: token / poll / reconnect / tray / autostart / no-WSL guard
This commit is contained in:
kvenanzi
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"accessToken": "sk-ant-test-5678"
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"claudeAiOauth": {
|
||||
"accessToken": "sk-ant-test-1234",
|
||||
"refreshToken": "sk-ant-ort-test-5678",
|
||||
"expiresAt": 9999999999000,
|
||||
"scopes": ["user:inference", "user:profile"]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for daemon/autostart_windows.py — APP-01.
|
||||
|
||||
Covers the winreg HKCU\\Run enable/disable/is_enabled login-autostart toggle.
|
||||
winreg is NOT importable off-Windows; these tests patch it via
|
||||
patch("daemon.autostart_windows.winreg", ...) so they run on any platform.
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_autostart.py -x -q
|
||||
"""
|
||||
from unittest.mock import MagicMock, patch, call
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers — build a fake winreg module with the attributes autostart_windows
|
||||
# references. Using a MagicMock as the module means all attribute accesses
|
||||
# on it (HKEY_CURRENT_USER, KEY_SET_VALUE, etc.) automatically produce child
|
||||
# MagicMocks, which is exactly what we want.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_winreg_mock(*, query_raises=False):
|
||||
"""Return a configured MagicMock that stands in for the winreg module."""
|
||||
winreg = MagicMock()
|
||||
|
||||
# Constants — assign simple sentinel values so equality checks work.
|
||||
winreg.HKEY_CURRENT_USER = "HKEY_CURRENT_USER"
|
||||
winreg.KEY_SET_VALUE = 0x0002
|
||||
winreg.KEY_QUERY_VALUE = 0x0001
|
||||
winreg.REG_SZ = 1
|
||||
|
||||
# OpenKey is used as a context manager; return a MagicMock key handle that
|
||||
# supports __enter__ / __exit__.
|
||||
key_handle = MagicMock()
|
||||
key_handle.__enter__ = MagicMock(return_value=key_handle)
|
||||
key_handle.__exit__ = MagicMock(return_value=False)
|
||||
winreg.OpenKey = MagicMock(return_value=key_handle)
|
||||
|
||||
# QueryValueEx behaviour is configured by the caller.
|
||||
if query_raises:
|
||||
winreg.QueryValueEx = MagicMock(side_effect=FileNotFoundError("not found"))
|
||||
else:
|
||||
winreg.QueryValueEx = MagicMock(return_value=("some_command", 1))
|
||||
|
||||
return winreg, key_handle
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# test_enable_writes_run_value
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_enable_writes_run_value():
|
||||
"""enable() opens HKCU Run key with KEY_SET_VALUE and calls SetValueEx with
|
||||
value name 'Clawdmeter' and type REG_SZ."""
|
||||
winreg, key_handle = _make_winreg_mock()
|
||||
|
||||
with patch("daemon.autostart_windows.winreg", winreg):
|
||||
import daemon.autostart_windows as mod
|
||||
mod.enable()
|
||||
|
||||
# OpenKey must have been called with HKCU and the Run key path
|
||||
winreg.OpenKey.assert_called_once_with(
|
||||
winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Run",
|
||||
0,
|
||||
winreg.KEY_SET_VALUE,
|
||||
)
|
||||
|
||||
# SetValueEx must have been called with the correct value name and type
|
||||
winreg.SetValueEx.assert_called_once()
|
||||
args = winreg.SetValueEx.call_args[0]
|
||||
assert args[0] is key_handle, "SetValueEx first arg must be the opened key handle"
|
||||
assert args[1] == "Clawdmeter", "Value name must be 'Clawdmeter'"
|
||||
assert args[3] == winreg.REG_SZ, "Value type must be REG_SZ"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# test_command_uses_pythonw
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_command_uses_pythonw():
|
||||
"""The command string written by enable() contains 'pythonw.exe', does NOT
|
||||
contain a bare 'python.exe' token (D-08, no console), and is quoted (starts
|
||||
with a double-quote character)."""
|
||||
winreg, key_handle = _make_winreg_mock()
|
||||
|
||||
captured_commands = []
|
||||
|
||||
def capture_set_value_ex(key, name, reserved, reg_type, value):
|
||||
captured_commands.append(value)
|
||||
|
||||
winreg.SetValueEx = MagicMock(side_effect=capture_set_value_ex)
|
||||
|
||||
with patch("daemon.autostart_windows.winreg", winreg):
|
||||
import daemon.autostart_windows as mod
|
||||
mod.enable()
|
||||
|
||||
assert len(captured_commands) == 1, "SetValueEx must have been called exactly once"
|
||||
cmd = captured_commands[0]
|
||||
|
||||
# Must reference pythonw.exe (D-08)
|
||||
assert "pythonw.exe" in cmd, f"Command must contain 'pythonw.exe'; got: {cmd!r}"
|
||||
|
||||
# Must NOT contain a bare 'python.exe' (without the 'w') as a standalone token
|
||||
# A command like '"...pythonw.exe" ...' is fine; '"...python.exe" ...' is not.
|
||||
import re
|
||||
assert not re.search(r'(?<![a-z])python\.exe', cmd), (
|
||||
f"Command must not reference a bare 'python.exe'; got: {cmd!r}"
|
||||
)
|
||||
|
||||
# Must start with a double-quote (paths are quoted for space safety)
|
||||
assert cmd.startswith('"'), (
|
||||
f"Command must start with '\"' (quoted path); got: {cmd!r}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# test_disable_idempotent
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_disable_idempotent():
|
||||
"""disable() calls DeleteValue; when DeleteValue raises FileNotFoundError,
|
||||
disable() swallows it and returns without raising (idempotent-on-missing)."""
|
||||
winreg, key_handle = _make_winreg_mock()
|
||||
winreg.DeleteValue = MagicMock(side_effect=FileNotFoundError("not found"))
|
||||
|
||||
with patch("daemon.autostart_windows.winreg", winreg):
|
||||
import daemon.autostart_windows as mod
|
||||
# Must not raise even though DeleteValue raises FileNotFoundError
|
||||
mod.disable() # no exception expected
|
||||
|
||||
winreg.DeleteValue.assert_called_once()
|
||||
|
||||
|
||||
def test_disable_calls_delete_value_with_correct_name():
|
||||
"""disable() calls DeleteValue with the value name 'Clawdmeter'."""
|
||||
winreg, key_handle = _make_winreg_mock()
|
||||
winreg.DeleteValue = MagicMock()
|
||||
|
||||
with patch("daemon.autostart_windows.winreg", winreg):
|
||||
import daemon.autostart_windows as mod
|
||||
mod.disable()
|
||||
|
||||
winreg.DeleteValue.assert_called_once()
|
||||
args = winreg.DeleteValue.call_args[0]
|
||||
assert args[1] == "Clawdmeter", f"DeleteValue must target 'Clawdmeter'; got {args[1]!r}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# test_is_enabled
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_is_enabled_true_when_value_present():
|
||||
"""is_enabled() returns True when QueryValueEx succeeds (value is present)."""
|
||||
winreg, key_handle = _make_winreg_mock(query_raises=False)
|
||||
|
||||
with patch("daemon.autostart_windows.winreg", winreg):
|
||||
import daemon.autostart_windows as mod
|
||||
result = mod.is_enabled()
|
||||
|
||||
assert result is True, "is_enabled() must return True when QueryValueEx succeeds"
|
||||
|
||||
|
||||
def test_is_enabled_false_when_value_absent():
|
||||
"""is_enabled() returns False when QueryValueEx raises FileNotFoundError."""
|
||||
winreg, key_handle = _make_winreg_mock(query_raises=True)
|
||||
|
||||
with patch("daemon.autostart_windows.winreg", winreg):
|
||||
import daemon.autostart_windows as mod
|
||||
result = mod.is_enabled()
|
||||
|
||||
assert result is False, "is_enabled() must return False when QueryValueEx raises FileNotFoundError"
|
||||
@@ -0,0 +1,162 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for daemon/icon_assets.py — APP-01 tray icon asset layer.
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_icon.py -x -q
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from daemon.icon_assets import _expand565, load_logo_rgba
|
||||
|
||||
# The "fixture" is the real in-repo firmware logo header — trusted asset.
|
||||
LOGO_H = Path(__file__).parent.parent.parent / "firmware" / "src" / "logo.h"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 1: logo parse + RGB565->RGB888 expand
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_logo_parse():
|
||||
"""load_logo_rgba returns an 80x80 RGBA image; dominant opaque color is #DE7552."""
|
||||
img = load_logo_rgba(str(LOGO_H))
|
||||
assert img.mode == "RGBA", f"Expected RGBA, got {img.mode}"
|
||||
assert img.size == (80, 80), f"Expected (80,80), got {img.size}"
|
||||
|
||||
# Collect all fully-opaque pixels and find the dominant RGB.
|
||||
from collections import Counter
|
||||
px = img.load()
|
||||
opaque = [
|
||||
(px[x, y][0], px[x, y][1], px[x, y][2])
|
||||
for y in range(80)
|
||||
for x in range(80)
|
||||
if px[x, y][3] == 255
|
||||
]
|
||||
assert opaque, "No fully-opaque pixels found in logo"
|
||||
dominant = Counter(opaque).most_common(1)[0][0]
|
||||
# Brand hex #DE7552 = (222, 117, 82)
|
||||
assert dominant == (222, 117, 82), (
|
||||
f"Expected dominant opaque color (222,117,82), got {dominant}"
|
||||
)
|
||||
|
||||
|
||||
def test_rgb565_expand():
|
||||
"""_expand565 uses proper rounding, not a *8 bit-shift."""
|
||||
assert _expand565(0xDBAA) == (222, 117, 82), (
|
||||
f"0xDBAA should be (222,117,82), got {_expand565(0xDBAA)}"
|
||||
)
|
||||
assert _expand565(0x0000) == (0, 0, 0), (
|
||||
f"0x0000 should be (0,0,0), got {_expand565(0x0000)}"
|
||||
)
|
||||
assert _expand565(0xFFFF) == (255, 255, 255), (
|
||||
f"0xFFFF should be (255,255,255), got {_expand565(0xFFFF)}"
|
||||
)
|
||||
|
||||
|
||||
def test_logo_parse_bounds_check():
|
||||
"""load_logo_rgba raises ValueError if the data array length != W*H*3."""
|
||||
import tempfile, os
|
||||
|
||||
# Write a malformed header with fewer bytes than expected
|
||||
malformed = (
|
||||
"#pragma once\n"
|
||||
"static const uint8_t logo_data[100] = {\n"
|
||||
" 0x00, 0x01, 0x02\n"
|
||||
"};\n"
|
||||
)
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".h", delete=False) as f:
|
||||
f.write(malformed)
|
||||
path = f.name
|
||||
try:
|
||||
with pytest.raises(ValueError):
|
||||
load_logo_rgba(path)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 2: state->image corner-bubble compositor
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_state_icon_bubble():
|
||||
"""state_icon returns distinct 32x32 RGBA images per state with correct bubble color."""
|
||||
from daemon.icon_assets import state_icon
|
||||
|
||||
base = load_logo_rgba(str(LOGO_H))
|
||||
connected = state_icon(base, "connected", 32)
|
||||
scanning = state_icon(base, "scanning", 32)
|
||||
error = state_icon(base, "error", 32)
|
||||
|
||||
# All are 32x32 RGBA
|
||||
for name, img in [("connected", connected), ("scanning", scanning), ("error", error)]:
|
||||
assert img.mode == "RGBA", f"{name}: expected RGBA, got {img.mode}"
|
||||
assert img.size == (32, 32), f"{name}: expected (32,32), got {img.size}"
|
||||
|
||||
# The three images must be pixel-distinct from each other
|
||||
def img_bytes(img):
|
||||
return img.tobytes()
|
||||
|
||||
assert img_bytes(connected) != img_bytes(scanning), "connected and scanning should differ"
|
||||
assert img_bytes(connected) != img_bytes(error), "connected and error should differ"
|
||||
assert img_bytes(scanning) != img_bytes(error), "scanning and error should differ"
|
||||
|
||||
# Sample the bottom-right corner region; the closest bubble color should match the state.
|
||||
# Bubble colors: connected (60,200,90), scanning (240,180,40), error (220,60,60)
|
||||
BUBBLE_COLORS = {
|
||||
"connected": (60, 200, 90),
|
||||
"scanning": (240, 180, 40),
|
||||
"error": (220, 60, 60),
|
||||
}
|
||||
|
||||
def color_distance(c1, c2):
|
||||
return sum((a - b) ** 2 for a, b in zip(c1, c2)) ** 0.5
|
||||
|
||||
def nearest_bubble(pixel_rgb):
|
||||
return min(BUBBLE_COLORS.items(), key=lambda kv: color_distance(pixel_rgb, kv[1]))[0]
|
||||
|
||||
size = 32
|
||||
r = size // 3
|
||||
# Sample the center of the expected bubble region (bottom-right corner)
|
||||
bx = size - r // 2 - 2
|
||||
by = size - r // 2 - 2
|
||||
bx = max(0, min(bx, size - 1))
|
||||
by = max(0, min(by, size - 1))
|
||||
|
||||
for state_name, img in [("connected", connected), ("scanning", scanning), ("error", error)]:
|
||||
px = img.load()
|
||||
pixel = px[bx, by][:3] # RGB only
|
||||
nearest = nearest_bubble(pixel)
|
||||
assert nearest == state_name, (
|
||||
f"State '{state_name}': bottom-right corner pixel {pixel} is nearest to "
|
||||
f"'{nearest}' bubble, expected '{state_name}'"
|
||||
)
|
||||
|
||||
|
||||
def test_build_icons_once():
|
||||
"""build_state_icons returns a dict with connected/scanning/error as distinct Images."""
|
||||
from daemon.icon_assets import build_state_icons
|
||||
|
||||
base = load_logo_rgba(str(LOGO_H))
|
||||
icons = build_state_icons(base)
|
||||
|
||||
assert set(icons.keys()) == {"connected", "scanning", "error"}, (
|
||||
f"Expected keys connected/scanning/error, got {set(icons.keys())}"
|
||||
)
|
||||
|
||||
# All distinct
|
||||
def img_bytes(img):
|
||||
return img.tobytes()
|
||||
|
||||
imgs = list(icons.values())
|
||||
assert img_bytes(imgs[0]) != img_bytes(imgs[1])
|
||||
assert img_bytes(imgs[0]) != img_bytes(imgs[2])
|
||||
assert img_bytes(imgs[1]) != img_bytes(imgs[2])
|
||||
|
||||
|
||||
def test_state_icon_unknown_state():
|
||||
"""state_icon raises KeyError/ValueError on an unknown state string."""
|
||||
from daemon.icon_assets import state_icon
|
||||
|
||||
base = load_logo_rgba(str(LOGO_H))
|
||||
with pytest.raises((KeyError, ValueError)):
|
||||
state_icon(base, "unknown_state", 32)
|
||||
@@ -0,0 +1,55 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Static no-WSL-paths regression guard — APP-02 / D-10.
|
||||
|
||||
Asserts that the daemon, tray, and autostart sources reference no WSL-specific
|
||||
paths. This is a CI-surviving regression lock that needs no hardware: if a
|
||||
future edit accidentally introduces a ``\\wsl$``, ``wsl.exe``, ``/home/``, or
|
||||
``/mnt/`` reference into any of the three core Windows daemon source files, this
|
||||
test will fail with a message that names the offending pattern and file.
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_no_wsl.py -x -q
|
||||
"""
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
# The four WSL-path patterns that must never appear in the daemon sources.
|
||||
FORBIDDEN = [r"\\wsl\$", r"wsl\.exe", r"/home/", r"/mnt/"]
|
||||
|
||||
# The three Windows-daemon source files covered by the guard.
|
||||
SOURCES = [
|
||||
Path("daemon/claude_usage_daemon_windows.py"),
|
||||
Path("daemon/tray_windows.py"),
|
||||
Path("daemon/autostart_windows.py"),
|
||||
]
|
||||
|
||||
|
||||
def test_no_wsl_paths_in_daemon():
|
||||
"""daemon/claude_usage_daemon_windows.py references no WSL paths."""
|
||||
_assert_clean(Path("daemon/claude_usage_daemon_windows.py"))
|
||||
|
||||
|
||||
def test_no_wsl_paths_in_tray():
|
||||
"""daemon/tray_windows.py references no WSL paths."""
|
||||
_assert_clean(Path("daemon/tray_windows.py"))
|
||||
|
||||
|
||||
def test_no_wsl_paths_in_autostart():
|
||||
"""daemon/autostart_windows.py references no WSL paths."""
|
||||
_assert_clean(Path("daemon/autostart_windows.py"))
|
||||
|
||||
|
||||
def _assert_clean(source: Path) -> None:
|
||||
"""Assert that none of the FORBIDDEN patterns appear in the given source file.
|
||||
|
||||
Reads the file relative to the repository root (the cwd pytest is invoked
|
||||
from). Fails with a descriptive message naming the leaked pattern and file
|
||||
so the regression is immediately actionable.
|
||||
"""
|
||||
text = source.read_text(encoding="utf-8")
|
||||
for pat in FORBIDDEN:
|
||||
match = re.search(pat, text)
|
||||
assert match is None, (
|
||||
f"WSL path leaked into {source}: pattern {pat!r} found at "
|
||||
f"position {match.start()} — "
|
||||
f"context: {text[max(0, match.start()-20):match.end()+20]!r}"
|
||||
)
|
||||
@@ -0,0 +1,455 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for poll_api / pct / reset_minutes / JSON-shape — POLL-01.
|
||||
|
||||
These tests cover the Anthropic API polling logic ported from the macOS daemon.
|
||||
All tests mock httpx so no real network calls are made.
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_poll.py -x -q
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from daemon.claude_usage_daemon_windows import AuthError, poll_api
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_mock_response(status_code=200, headers=None):
|
||||
"""Build a mock httpx.Response-like object with controllable headers."""
|
||||
resp = MagicMock()
|
||||
resp.status_code = status_code
|
||||
resp.text = "mocked"
|
||||
# httpx headers are case-insensitive; MagicMock .get() must behave the same
|
||||
header_data = headers or {}
|
||||
resp.headers = MagicMock()
|
||||
resp.headers.get = lambda name, default=None: header_data.get(name.lower(), default)
|
||||
return resp
|
||||
|
||||
|
||||
def _run(coro):
|
||||
"""Run a coroutine synchronously for synchronous test functions."""
|
||||
return asyncio.get_event_loop().run_until_complete(coro)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: full poll_api with realistic ratelimit headers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_poll_api_nominal(monkeypatch):
|
||||
"""poll_api with a 200 response + ratelimit headers produces the correct payload."""
|
||||
now = time.time()
|
||||
reset_5h = str(now + 3600) # 60 minutes from now
|
||||
reset_7d = str(now + 86400) # 1440 minutes from now
|
||||
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.42",
|
||||
"anthropic-ratelimit-unified-5h-reset": reset_5h,
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.10",
|
||||
"anthropic-ratelimit-unified-7d-reset": reset_7d,
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
async def fake_post(*args, **kwargs):
|
||||
return mock_resp
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = fake_post
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload is not None
|
||||
assert payload["s"] == 42
|
||||
assert payload["w"] == 10
|
||||
assert payload["st"] == "allowed"
|
||||
assert payload["ok"] is True
|
||||
# reset_minutes allows ±1 minute tolerance
|
||||
assert abs(payload["sr"] - 60) <= 1, f"Expected ~60, got {payload['sr']}"
|
||||
assert abs(payload["wr"] - 1440) <= 1, f"Expected ~1440, got {payload['wr']}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: pct() correctness — exercised through poll_api output
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_pct_42_percent(monkeypatch):
|
||||
"""pct('0.42') -> 42."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.42",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.10",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["s"] == 42
|
||||
|
||||
|
||||
def test_pct_100_percent(monkeypatch):
|
||||
"""pct('1.0') -> 100."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "1.0",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "1.0",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["s"] == 100
|
||||
assert payload["w"] == 100
|
||||
|
||||
|
||||
def test_pct_empty_string_defaults_to_zero(monkeypatch):
|
||||
"""pct('') -> 0 (missing header defaults to '0', but empty string -> 0)."""
|
||||
now = time.time()
|
||||
# Override default so utilization header returns "" explicitly
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["s"] == 0
|
||||
assert payload["w"] == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: reset_minutes() — exercised through poll_api output
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_reset_minutes_60_minutes(monkeypatch):
|
||||
"""reset_minutes(now+3600) -> ~60."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert abs(payload["sr"] - 60) <= 1, f"Expected ~60, got {payload['sr']}"
|
||||
|
||||
|
||||
def test_reset_minutes_negative_clamps_to_zero(monkeypatch):
|
||||
"""reset_minutes for a past timestamp clamps to 0."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now - 100), # 100s in the past
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now - 100), # also in the past
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["sr"] == 0
|
||||
assert payload["wr"] == 0
|
||||
|
||||
|
||||
def test_reset_minutes_invalid_string_returns_zero(monkeypatch):
|
||||
"""reset_minutes('notanumber') -> 0 (ValueError-safe)."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-5h-reset": "notanumber",
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-7d-reset": "notanumber",
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["sr"] == 0
|
||||
assert payload["wr"] == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: missing headers default gracefully
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_missing_utilization_headers_default_to_zero(monkeypatch):
|
||||
"""Missing utilization headers produce 0 (hdr default '0' -> pct('0') = 0)."""
|
||||
now = time.time()
|
||||
# No utilization or status headers — only reset headers present
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["s"] == 0
|
||||
assert payload["w"] == 0
|
||||
|
||||
|
||||
def test_missing_status_header_defaults_to_unknown(monkeypatch):
|
||||
"""Missing 5h-status header defaults to 'unknown'."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
# NOTE: no 5h-status header
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload["st"] == "unknown"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: poll_api returns None on HTTP >= 400
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_poll_api_returns_none_on_4xx(monkeypatch):
|
||||
"""poll_api returns None when response status code is >= 400."""
|
||||
mock_resp = _make_mock_response(status_code=429, headers={})
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
result = _run(poll_api("fake-token"))
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_poll_api_returns_none_on_5xx(monkeypatch):
|
||||
"""poll_api returns None when response status code is >= 500."""
|
||||
mock_resp = _make_mock_response(status_code=500, headers={})
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
result = _run(poll_api("fake-token"))
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: poll_api raises AuthError ONLY on a genuine 401/403 (SC#5 fix)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("status", [401, 403])
|
||||
def test_poll_api_raises_autherror_on_401_403(status):
|
||||
"""A real auth rejection must raise AuthError — the only signal that warrants
|
||||
the actionable 'token expired — run claude login' toast. Transient failures
|
||||
(5xx, 429, network) return None instead and must NOT trigger that toast."""
|
||||
mock_resp = _make_mock_response(status_code=status, headers={})
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
with pytest.raises(AuthError):
|
||||
_run(poll_api("fake-token"))
|
||||
|
||||
|
||||
def test_poll_api_returns_none_not_autherror_on_429(monkeypatch):
|
||||
"""Rate-limit (429) is transient — None, NOT AuthError (regression guard for
|
||||
the 401/403-vs-other-4xx split)."""
|
||||
mock_resp = _make_mock_response(status_code=429, headers={})
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
result = _run(poll_api("fake-token"))
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: poll_api returns None on httpx.HTTPError
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_poll_api_returns_none_on_http_error(monkeypatch):
|
||||
"""poll_api returns None when httpx.HTTPError is raised (network failure)."""
|
||||
import httpx
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
result = _run(poll_api("fake-token"))
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: compact JSON wire shape (no spaces after ':' or ',')
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_wire_bytes_compact_json_shape(monkeypatch):
|
||||
"""The JSON-encoded payload uses compact separators (',':') — no spaces."""
|
||||
now = time.time()
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.42",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.10",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
payload = _run(poll_api("fake-token"))
|
||||
|
||||
assert payload is not None
|
||||
# Encode exactly as the wire layer will (Session.write_payload uses this form)
|
||||
wire_bytes = json.dumps(payload, separators=(",", ":")).encode()
|
||||
wire_str = wire_bytes.decode()
|
||||
|
||||
# Compact form: no space after ':' or ','
|
||||
assert ": " not in wire_str, f"Non-compact JSON detected: {wire_str!r}"
|
||||
assert ", " not in wire_str, f"Non-compact JSON detected: {wire_str!r}"
|
||||
|
||||
# Must start with '{' and contain all required keys
|
||||
assert wire_str.startswith("{")
|
||||
for key in ("s", "sr", "w", "wr", "st", "ok"):
|
||||
assert f'"{key}"' in wire_str, f"Missing key {key!r} in wire bytes: {wire_str!r}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test: token is NOT logged (T-02-01 threat mitigation)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_poll_api_does_not_log_token(monkeypatch, capsys):
|
||||
"""poll_api must not print the bearer token (T-02-01: token never logged)."""
|
||||
now = time.time()
|
||||
secret_token = "sk-ant-secret-token-12345"
|
||||
|
||||
mock_resp = _make_mock_response(
|
||||
status_code=200,
|
||||
headers={
|
||||
"anthropic-ratelimit-unified-5h-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
|
||||
"anthropic-ratelimit-unified-7d-utilization": "0.5",
|
||||
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
|
||||
"anthropic-ratelimit-unified-5h-status": "allowed",
|
||||
},
|
||||
)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch("httpx.AsyncClient", return_value=mock_client):
|
||||
_run(poll_api(secret_token))
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert secret_token not in captured.out, "Token leaked to stdout (T-02-01 violation)"
|
||||
assert secret_token not in captured.err, "Token leaked to stderr (T-02-01 violation)"
|
||||
@@ -0,0 +1,723 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for connect_and_run reconnect hardening — BLE-03.
|
||||
|
||||
Covers:
|
||||
D-01: connect-retry wrapper (post-wake WinRT failure modes)
|
||||
D-03: zombie-link consecutive-failure break (stale is_connected)
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_reconnect.py -x -q
|
||||
"""
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from bleak.exc import BleakError
|
||||
|
||||
from daemon.claude_usage_daemon_windows import (
|
||||
AuthError,
|
||||
Session,
|
||||
_wait_first,
|
||||
connect_and_run,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _run(coro):
|
||||
"""Run a coroutine synchronously for synchronous test functions."""
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
def _make_device(address="AA:BB:CC:DD:EE:FF"):
|
||||
"""Build a minimal fake BLEDevice."""
|
||||
device = MagicMock()
|
||||
device.address = address
|
||||
return device
|
||||
|
||||
|
||||
async def _make_event(set_):
|
||||
ev = asyncio.Event()
|
||||
if set_:
|
||||
ev.set()
|
||||
return ev
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# D-01: connect-retry wrapper tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_connect_retry_exhaustion_on_bleak_error(monkeypatch, capsys):
|
||||
"""BleakError on every connect attempt exhausts CONNECT_RETRIES then returns False."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(side_effect=BleakError("Unreachable"))
|
||||
mock_client.is_connected = False
|
||||
mock_client.disconnect = AsyncMock()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
|
||||
result = _run(connect_and_run(device, stop_event))
|
||||
|
||||
assert result is False
|
||||
assert mock_client.connect.call_count == mod.CONNECT_RETRIES
|
||||
|
||||
|
||||
def test_connect_retry_exhaustion_on_timeout_error(monkeypatch, capsys):
|
||||
"""asyncio.TimeoutError on every connect attempt is treated same as BleakError."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(side_effect=asyncio.TimeoutError())
|
||||
mock_client.is_connected = False
|
||||
mock_client.disconnect = AsyncMock()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
|
||||
result = _run(connect_and_run(device, stop_event))
|
||||
|
||||
assert result is False
|
||||
assert mock_client.connect.call_count == mod.CONNECT_RETRIES
|
||||
|
||||
|
||||
def test_connect_retry_calls_disconnect_between_attempts(monkeypatch):
|
||||
"""Guarded disconnect() is called between failed connect attempts."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(side_effect=BleakError("Unreachable"))
|
||||
mock_client.is_connected = False
|
||||
mock_client.disconnect = AsyncMock()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
|
||||
_run(connect_and_run(device, stop_event))
|
||||
|
||||
# disconnect is called between attempts (at least CONNECT_RETRIES - 1 times)
|
||||
assert mock_client.disconnect.call_count >= mod.CONNECT_RETRIES - 1
|
||||
|
||||
|
||||
def test_connect_success_on_first_attempt_no_extra_retries(monkeypatch):
|
||||
"""First-attempt success consumes exactly 1 connect call and proceeds past connect block."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
# stop_event is set so the loop exits immediately after connecting
|
||||
stop_event = asyncio.run(_make_event(True))
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(return_value=None) # success
|
||||
mock_client.is_connected = True
|
||||
mock_client.disconnect = AsyncMock()
|
||||
mock_client.start_notify = AsyncMock()
|
||||
mock_client.write_gatt_char = AsyncMock(return_value=None)
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api", new=AsyncMock(return_value={"ok": True})):
|
||||
_run(connect_and_run(device, stop_event))
|
||||
|
||||
assert mock_client.connect.call_count == 1
|
||||
|
||||
|
||||
def test_connect_retry_exhaustion_does_not_log_token(monkeypatch, capsys):
|
||||
"""On exhaustion, no log line contains the patched token sentinel (T-03-01)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
TOKEN_SENTINEL = "sk-ant-SUPERSECRET-DO-NOT-LOG-12345"
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(side_effect=BleakError("Unreachable"))
|
||||
mock_client.is_connected = False
|
||||
mock_client.disconnect = AsyncMock()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value=TOKEN_SENTINEL), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
|
||||
_run(connect_and_run(device, stop_event))
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert TOKEN_SENTINEL not in captured.out, "Token sentinel leaked to stdout (T-03-01)"
|
||||
assert TOKEN_SENTINEL not in captured.err, "Token sentinel leaked to stderr (T-03-01)"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# D-03: zombie-link consecutive-failure break tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_zombie_client():
|
||||
"""Build a mock BleakClient that connects successfully but has is_connected stuck True."""
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(return_value=None)
|
||||
mock_client.is_connected = True # stale flag — never goes False
|
||||
mock_client.disconnect = AsyncMock()
|
||||
mock_client.start_notify = AsyncMock()
|
||||
return mock_client
|
||||
|
||||
|
||||
def test_zombie_link_break_after_limit_consecutive_failures(monkeypatch):
|
||||
"""Loop breaks after exactly ZOMBIE_BREAK_LIMIT consecutive False writes (default 1)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
mock_client = _make_zombie_client()
|
||||
|
||||
write_call_count = [0]
|
||||
|
||||
async def fake_write_payload(payload):
|
||||
write_call_count[0] += 1
|
||||
return False # always fail — zombie link
|
||||
|
||||
fake_session = AsyncMock()
|
||||
fake_session.write_payload = fake_write_payload
|
||||
fake_session.refresh_requested = MagicMock()
|
||||
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
|
||||
fake_session.refresh_requested.clear = MagicMock()
|
||||
fake_session.refresh_requested.wait = AsyncMock()
|
||||
|
||||
# Force elapsed >= POLL_INTERVAL immediately
|
||||
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
|
||||
|
||||
async def fast_wait_for(coro, timeout):
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api",
|
||||
new=AsyncMock(return_value={"ok": True})), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
|
||||
side_effect=fast_wait_for):
|
||||
result = _run(connect_and_run(device, stop_event))
|
||||
|
||||
# With ZOMBIE_BREAK_LIMIT=1, one False write should break the loop
|
||||
assert write_call_count[0] == mod.ZOMBIE_BREAK_LIMIT
|
||||
# Should return used_successfully=False (no successful write)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_zombie_counter_resets_on_success_with_raised_limit(monkeypatch):
|
||||
"""A failed write followed by success resets counter (limit raised to 2 to exercise reset)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
mock_client = _make_zombie_client()
|
||||
|
||||
# Sequence: False (counter=1), True (counter reset to 0), False (counter=1 again), break
|
||||
write_results = iter([False, True, False])
|
||||
write_call_count = [0]
|
||||
|
||||
async def fake_write_payload(payload):
|
||||
write_call_count[0] += 1
|
||||
try:
|
||||
return next(write_results)
|
||||
except StopIteration:
|
||||
return False
|
||||
|
||||
# After success, subsequent False write breaks at limit=2 (requires 2 consecutive)
|
||||
# With limit=2: False (1), True (reset to 0), False (1), False (2 -> break)
|
||||
# But we only have 3 items in write_results; after StopIteration returns False.
|
||||
# Let's use a longer sequence to ensure reset-then-2-failures trip the break.
|
||||
write_results2 = [False, True, False, False]
|
||||
write_call_count2 = [0]
|
||||
|
||||
async def fake_write_payload2(payload):
|
||||
write_call_count2[0] += 1
|
||||
if write_call_count2[0] - 1 < len(write_results2):
|
||||
return write_results2[write_call_count2[0] - 1]
|
||||
return False
|
||||
|
||||
fake_session = AsyncMock()
|
||||
fake_session.write_payload = fake_write_payload2
|
||||
fake_session.refresh_requested = MagicMock()
|
||||
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
|
||||
fake_session.refresh_requested.clear = MagicMock()
|
||||
fake_session.refresh_requested.wait = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
|
||||
monkeypatch.setattr(mod, "ZOMBIE_BREAK_LIMIT", 2) # raise limit to test reset logic
|
||||
|
||||
async def fast_wait_for(coro, timeout):
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api",
|
||||
new=AsyncMock(return_value={"ok": True})), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
|
||||
side_effect=fast_wait_for):
|
||||
result = _run(connect_and_run(device, stop_event))
|
||||
|
||||
# With limit=2 and sequence [False, True, False, False]:
|
||||
# cycle 1: False -> consecutive_failures=1 (no break, limit=2)
|
||||
# cycle 2: True -> consecutive_failures=0 (reset)
|
||||
# cycle 3: False -> consecutive_failures=1 (no break)
|
||||
# cycle 4: False -> consecutive_failures=2 -> break
|
||||
assert write_call_count2[0] == 4, (
|
||||
f"Expected 4 write calls (reset-on-success logic), got {write_call_count2[0]}"
|
||||
)
|
||||
# used_successfully=True because cycle 2 succeeded
|
||||
assert result is True
|
||||
|
||||
|
||||
def test_zombie_break_disconnect_called_in_finally(monkeypatch):
|
||||
"""The finally block calls client.disconnect() exactly once on the zombie-break path."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
mock_client = _make_zombie_client()
|
||||
|
||||
async def fake_write_payload(payload):
|
||||
return False # always fail
|
||||
|
||||
fake_session = AsyncMock()
|
||||
fake_session.write_payload = fake_write_payload
|
||||
fake_session.refresh_requested = MagicMock()
|
||||
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
|
||||
fake_session.refresh_requested.clear = MagicMock()
|
||||
fake_session.refresh_requested.wait = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
|
||||
|
||||
async def fast_wait_for(coro, timeout):
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api",
|
||||
new=AsyncMock(return_value={"ok": True})), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
|
||||
side_effect=fast_wait_for):
|
||||
_run(connect_and_run(device, stop_event))
|
||||
|
||||
# The finally block calls disconnect() exactly once
|
||||
assert mock_client.disconnect.call_count == 1
|
||||
|
||||
|
||||
def test_zombie_break_returns_used_successfully_false(monkeypatch):
|
||||
"""connect_and_run returns used_successfully=False after zombie break with no writes."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
mock_client = _make_zombie_client()
|
||||
|
||||
async def fake_write_payload(payload):
|
||||
return False
|
||||
|
||||
fake_session = AsyncMock()
|
||||
fake_session.write_payload = fake_write_payload
|
||||
fake_session.refresh_requested = MagicMock()
|
||||
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
|
||||
fake_session.refresh_requested.clear = MagicMock()
|
||||
fake_session.refresh_requested.wait = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
|
||||
|
||||
async def fast_wait_for(coro, timeout):
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api",
|
||||
new=AsyncMock(return_value={"ok": True})), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
|
||||
side_effect=fast_wait_for):
|
||||
result = _run(connect_and_run(device, stop_event))
|
||||
|
||||
# main() uses this return value to route into reconnect branch
|
||||
assert result is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# D-05: split fast-reconnect vs slow-search backoff in main()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_next_backoff_slow_search_doubles_to_60():
|
||||
"""_next_backoff doubles correctly and never exceeds 60 (slow-search cap)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
values = []
|
||||
b = 1
|
||||
for _ in range(10):
|
||||
b = mod._next_backoff(b, 60)
|
||||
values.append(b)
|
||||
|
||||
assert values == [2, 4, 8, 16, 32, 60, 60, 60, 60, 60]
|
||||
assert max(values) <= 60
|
||||
|
||||
|
||||
def test_next_backoff_fast_reconnect_doubles_to_cap():
|
||||
"""_next_backoff doubles correctly and never exceeds RECONNECT_BACKOFF_CAP (default 8)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
cap = mod.RECONNECT_BACKOFF_CAP
|
||||
assert cap < 60, "Fast cap must be strictly lower than search cap"
|
||||
|
||||
values = []
|
||||
b = 1
|
||||
for _ in range(8):
|
||||
b = mod._next_backoff(b, cap)
|
||||
values.append(b)
|
||||
|
||||
# Should double until hitting the cap, then stay there
|
||||
assert max(values) <= cap
|
||||
# Should reach the cap (not just stay at 1)
|
||||
assert values[-1] == cap
|
||||
|
||||
|
||||
def test_next_backoff_one_to_two():
|
||||
"""_next_backoff(1, 60) == 2 (basic sanity)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
assert mod._next_backoff(1, 60) == 2
|
||||
|
||||
|
||||
def test_next_backoff_at_cap_stays():
|
||||
"""_next_backoff(cap, cap) == cap (does not overflow)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
assert mod._next_backoff(mod.RECONNECT_BACKOFF_CAP, mod.RECONNECT_BACKOFF_CAP) == mod.RECONNECT_BACKOFF_CAP
|
||||
|
||||
|
||||
def test_main_scan_miss_uses_search_backoff():
|
||||
"""When scan_for_device returns None, asyncio.wait_for receives search_backoff timeout values."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
# Capture main()'s internal stop_event by intercepting asyncio.Event()
|
||||
internal_stop_event = [None]
|
||||
real_Event = asyncio.Event
|
||||
|
||||
def capturing_Event():
|
||||
ev = real_Event()
|
||||
internal_stop_event[0] = ev
|
||||
return ev
|
||||
|
||||
recorded_timeouts = []
|
||||
call_count = [0]
|
||||
MAX_CALLS = 3
|
||||
|
||||
async def fake_scan():
|
||||
return None # always miss -> slow-search regime
|
||||
|
||||
async def fake_wait_for(coro, timeout):
|
||||
recorded_timeouts.append(timeout)
|
||||
call_count[0] += 1
|
||||
if call_count[0] >= MAX_CALLS and internal_stop_event[0] is not None:
|
||||
internal_stop_event[0].set() # terminate main()'s outer while loop
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.asyncio.Event", side_effect=capturing_Event), \
|
||||
patch("daemon.claude_usage_daemon_windows.scan_for_device", side_effect=fake_scan), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for", side_effect=fake_wait_for):
|
||||
_run(mod.main())
|
||||
|
||||
# Should have recorded timeouts from search_backoff sequence: 1, 2, 4 (then stop)
|
||||
assert len(recorded_timeouts) >= 2
|
||||
# Timeouts should be doubling (search_backoff sequence)
|
||||
assert recorded_timeouts[0] == 1
|
||||
assert recorded_timeouts[1] == 2
|
||||
# None should exceed the search cap (60)
|
||||
assert all(t <= 60 for t in recorded_timeouts)
|
||||
|
||||
|
||||
def test_main_connect_fail_uses_reconnect_backoff():
|
||||
"""When connect_and_run returns False, asyncio.wait_for receives reconnect_backoff timeouts (fast cap)."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
# Capture main()'s internal stop_event
|
||||
internal_stop_event = [None]
|
||||
real_Event = asyncio.Event
|
||||
|
||||
def capturing_Event():
|
||||
ev = real_Event()
|
||||
internal_stop_event[0] = ev
|
||||
return ev
|
||||
|
||||
fake_device = _make_device()
|
||||
recorded_timeouts = []
|
||||
call_count = [0]
|
||||
MAX_CALLS = 3
|
||||
|
||||
async def fake_scan():
|
||||
return fake_device # always finds device
|
||||
|
||||
async def fake_connect_and_run(device, event, tray_state=None):
|
||||
return False # always fails -> fast-reconnect regime
|
||||
|
||||
async def fake_wait_for(coro, timeout):
|
||||
recorded_timeouts.append(timeout)
|
||||
call_count[0] += 1
|
||||
if call_count[0] >= MAX_CALLS and internal_stop_event[0] is not None:
|
||||
internal_stop_event[0].set()
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.asyncio.Event", side_effect=capturing_Event), \
|
||||
patch("daemon.claude_usage_daemon_windows.scan_for_device", side_effect=fake_scan), \
|
||||
patch("daemon.claude_usage_daemon_windows.connect_and_run", side_effect=fake_connect_and_run), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for", side_effect=fake_wait_for):
|
||||
_run(mod.main())
|
||||
|
||||
# Should have recorded timeouts from reconnect_backoff sequence: 1, 2, 4 (then stop)
|
||||
assert len(recorded_timeouts) >= 2
|
||||
assert recorded_timeouts[0] == 1
|
||||
assert recorded_timeouts[1] == 2
|
||||
# All timeouts must be at or below RECONNECT_BACKOFF_CAP (fast cap, < 60)
|
||||
assert all(t <= mod.RECONNECT_BACKOFF_CAP for t in recorded_timeouts)
|
||||
|
||||
|
||||
def test_main_reconnect_backoff_reset_on_success():
|
||||
"""A successful connect_and_run (returns True) resets reconnect_backoff to 1."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
# Capture main()'s internal stop_event
|
||||
internal_stop_event = [None]
|
||||
real_Event = asyncio.Event
|
||||
|
||||
def capturing_Event():
|
||||
ev = real_Event()
|
||||
internal_stop_event[0] = ev
|
||||
return ev
|
||||
|
||||
fake_device = _make_device()
|
||||
recorded_timeouts = []
|
||||
call_count = [0]
|
||||
|
||||
# Sequence: fail (reconnect_backoff=1), succeed (reset), fail (reconnect_backoff=1 again)
|
||||
connect_results = [False, True, False]
|
||||
connect_idx = [0]
|
||||
|
||||
async def fake_scan():
|
||||
return fake_device
|
||||
|
||||
async def fake_connect_and_run(device, event, tray_state=None):
|
||||
idx = connect_idx[0]
|
||||
connect_idx[0] += 1
|
||||
if idx < len(connect_results):
|
||||
return connect_results[idx]
|
||||
return False
|
||||
|
||||
async def fake_wait_for(coro, timeout):
|
||||
recorded_timeouts.append(timeout)
|
||||
call_count[0] += 1
|
||||
if call_count[0] >= 2 and internal_stop_event[0] is not None:
|
||||
internal_stop_event[0].set() # stop after 2 waits (first fail + post-success fail)
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.asyncio.Event", side_effect=capturing_Event), \
|
||||
patch("daemon.claude_usage_daemon_windows.scan_for_device", side_effect=fake_scan), \
|
||||
patch("daemon.claude_usage_daemon_windows.connect_and_run", side_effect=fake_connect_and_run), \
|
||||
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for", side_effect=fake_wait_for):
|
||||
_run(mod.main())
|
||||
|
||||
# First wait: reconnect_backoff=1 (initial failure)
|
||||
# Second wait: reconnect_backoff=1 again (reset by success, then another failure)
|
||||
assert len(recorded_timeouts) >= 2
|
||||
assert recorded_timeouts[0] == 1, f"Expected 1 on first fail, got {recorded_timeouts[0]}"
|
||||
assert recorded_timeouts[1] == 1, f"Expected 1 after success reset, got {recorded_timeouts[1]}"
|
||||
|
||||
|
||||
def test_main_no_saved_addr_file_or_skip_addr():
|
||||
"""main() does not reference SAVED_ADDR_FILE or skip_addr (Windows is stateless - D-04)."""
|
||||
import inspect
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
source = inspect.getsource(mod.main)
|
||||
assert "SAVED_ADDR_FILE" not in source, "main() must not reference SAVED_ADDR_FILE (D-04)"
|
||||
assert "skip_addr" not in source, "main() must not reference skip_addr (macOS-only)"
|
||||
assert "retrieve_connected" not in source.lower(), \
|
||||
"main() must not reference retrieve_connected (macOS HID path)"
|
||||
|
||||
|
||||
def test_requirements_windows_contains_required_deps():
|
||||
"""requirements-windows.txt must contain the expected deps.
|
||||
|
||||
Phase 3 (reconnect) added no new deps; Phase 4 (tray) adds pystray + Pillow.
|
||||
This test asserts the final expected state: bleak, httpx, pystray, Pillow
|
||||
must be present; winreg must NOT be listed (it is stdlib — no install needed).
|
||||
"""
|
||||
req_path = Path(__file__).parent.parent / "requirements-windows.txt"
|
||||
content = req_path.read_text()
|
||||
lines = {line.strip().lower() for line in content.splitlines()
|
||||
if line.strip() and not line.strip().startswith("#")}
|
||||
|
||||
assert "bleak" in lines, "bleak must be in requirements-windows.txt"
|
||||
assert "httpx" in lines, "httpx must be in requirements-windows.txt"
|
||||
assert "pystray" in lines, "pystray must be in requirements-windows.txt (Phase 4)"
|
||||
assert "pillow" in lines, "Pillow must be in requirements-windows.txt (Phase 4)"
|
||||
assert "winreg" not in lines, "winreg is stdlib — must NOT be in requirements-windows.txt"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# G-03-01: start_notify() OSError must not crash the daemon (SC#3 power-cycle)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_start_notify_oserror_does_not_crash_connect_and_run():
|
||||
"""G-03-01 regression: on post-power-cycle reconnect, WinRT's start_notify()
|
||||
CCCD write can raise a raw OSError/WinError when the just-rebooted peer GATT
|
||||
server is not yet ready. The optional refresh subscription must degrade
|
||||
gracefully — connect_and_run must NOT propagate the OSError and must proceed
|
||||
into the poll loop (returning normally), so the daemon never restarts (SC#3/SC#4).
|
||||
"""
|
||||
device = _make_device()
|
||||
# stop_event set so the poll loop exits immediately after subscription setup
|
||||
stop_event = asyncio.run(_make_event(True))
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.connect = AsyncMock(return_value=None) # connect succeeds
|
||||
mock_client.is_connected = True
|
||||
mock_client.disconnect = AsyncMock()
|
||||
# The exact failure observed on hardware (SC#3, 2026-06-02):
|
||||
# OSError: [WinError -2147023673] The operation was canceled by the user.
|
||||
mock_client.start_notify = AsyncMock(
|
||||
side_effect=OSError(-2147023673, "The operation was canceled by the user.")
|
||||
)
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api", new=AsyncMock(return_value={"ok": True})):
|
||||
# Must NOT raise OSError — graceful degradation into the poll loop.
|
||||
result = _run(connect_and_run(device, stop_event))
|
||||
|
||||
# start_notify was actually attempted (and raised), but was swallowed.
|
||||
assert mock_client.start_notify.call_count == 1
|
||||
# Function returned normally instead of propagating the OSError.
|
||||
assert result is False
|
||||
# The link was cleaned up via the finally block.
|
||||
assert mock_client.disconnect.call_count >= 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC#2 field report: write_payload() OSError must not crash the daemon thread
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_write_payload_oserror_returns_false_not_raises():
|
||||
"""SC#2 regression: write_gatt_char can raise a raw OSError/WinError (NOT a
|
||||
BleakError) when the peer GATT server goes transiently unavailable mid-write.
|
||||
write_payload must catch it and return False — tripping the zombie-link break
|
||||
for a clean reconnect — instead of propagating an uncaught exception that
|
||||
silently kills the daemon=True background thread and freezes the tray.
|
||||
"""
|
||||
mock_client = AsyncMock()
|
||||
mock_client.write_gatt_char = AsyncMock(
|
||||
side_effect=OSError(-2147023673, "The operation was canceled by the user.")
|
||||
)
|
||||
session = Session(mock_client)
|
||||
|
||||
result = _run(session.write_payload({"ok": True}))
|
||||
|
||||
assert result is False # caught and reported, not raised
|
||||
assert mock_client.write_gatt_char.call_count == 1
|
||||
|
||||
|
||||
def test_write_payload_bleak_error_still_returns_false():
|
||||
"""The pre-existing BleakError path must keep returning False (no regression
|
||||
from widening the except to also cover OSError)."""
|
||||
mock_client = AsyncMock()
|
||||
mock_client.write_gatt_char = AsyncMock(side_effect=BleakError("disconnected"))
|
||||
session = Session(mock_client)
|
||||
|
||||
assert _run(session.write_payload({"ok": True})) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC#3 graceful Quit: _wait_first wakes immediately on stop (clean disconnect)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_wait_first_returns_immediately_when_an_event_is_set():
|
||||
"""The poll loop's TICK wait must break the instant stop_event is set, so the
|
||||
finally: client.disconnect() runs before the process exits (SC#3). The outer
|
||||
wait_for(2s) fails fast if _wait_first wrongly blocks for the full 30s timeout."""
|
||||
async def go():
|
||||
refresh = asyncio.Event()
|
||||
stop = asyncio.Event()
|
||||
stop.set() # stop signalled
|
||||
await asyncio.wait_for(_wait_first(refresh, stop, timeout=30.0), timeout=2.0)
|
||||
assert not refresh.is_set() # loser waiter drained, refresh untouched
|
||||
_run(go())
|
||||
|
||||
|
||||
def test_wait_first_returns_after_timeout_when_no_event_set():
|
||||
"""With neither event set, _wait_first returns after `timeout` (the normal
|
||||
poll-tick path) rather than hanging."""
|
||||
async def go():
|
||||
await asyncio.wait_for(
|
||||
_wait_first(asyncio.Event(), asyncio.Event(), timeout=0.05), timeout=2.0
|
||||
)
|
||||
_run(go())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC#5: transient poll failure must NOT toast "token expired"; only a real
|
||||
# 401/403 (AuthError) should. A boot-time DNS blip returns None, not AuthError.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _connected_mock_client():
|
||||
client = AsyncMock()
|
||||
client.connect = AsyncMock(return_value=None)
|
||||
client.is_connected = True
|
||||
client.disconnect = AsyncMock()
|
||||
client.start_notify = AsyncMock()
|
||||
return client
|
||||
|
||||
|
||||
def test_transient_poll_failure_does_not_set_error():
|
||||
"""poll_api returning None (network/DNS, timeout, 5xx, 429) is transient and
|
||||
must leave the tray state untouched — not flip it to 'token expired' (SC#5
|
||||
field report: `getaddrinfo failed` at boot wrongly fired the toast)."""
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
tray_state = MagicMock()
|
||||
client = _connected_mock_client()
|
||||
|
||||
async def fake_poll(_token):
|
||||
stop_event.set() # end the loop after this single transient failure
|
||||
return None
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=client), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="tok"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api", new=fake_poll):
|
||||
_run(connect_and_run(device, stop_event, tray_state))
|
||||
|
||||
tray_state.set_error.assert_not_called()
|
||||
tray_state.set_connected.assert_not_called()
|
||||
|
||||
|
||||
def test_auth_error_sets_token_expired():
|
||||
"""A genuine 401/403 surfaces as AuthError and DOES flip the tray to the
|
||||
actionable 'token expired — run claude login' error state."""
|
||||
device = _make_device()
|
||||
stop_event = asyncio.run(_make_event(False))
|
||||
tray_state = MagicMock()
|
||||
client = _connected_mock_client()
|
||||
|
||||
async def fake_poll(_token):
|
||||
stop_event.set()
|
||||
raise AuthError(401)
|
||||
|
||||
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=client), \
|
||||
patch("daemon.claude_usage_daemon_windows.read_token", return_value="tok"), \
|
||||
patch("daemon.claude_usage_daemon_windows.poll_api", new=fake_poll):
|
||||
_run(connect_and_run(device, stop_event, tray_state))
|
||||
|
||||
tray_state.set_error.assert_called_once_with("token expired — run claude login")
|
||||
@@ -0,0 +1,198 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for daemon/claude_usage_daemon_windows.py — TOKEN-01.
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_token.py -x -q
|
||||
"""
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from daemon.claude_usage_daemon_windows import _extract_access_token, read_token, _windows_credential_candidates, _read_expiry
|
||||
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def test_extract_nested_shape():
|
||||
"""_extract_access_token handles the real Windows claudeAiOauth nested shape."""
|
||||
blob = (FIXTURES / "credentials_nested.json").read_text()
|
||||
assert _extract_access_token(blob) == "sk-ant-test-1234"
|
||||
|
||||
|
||||
def test_extract_direct_shape():
|
||||
"""_extract_access_token handles the legacy direct accessToken shape."""
|
||||
blob = (FIXTURES / "credentials_direct.json").read_text()
|
||||
assert _extract_access_token(blob) == "sk-ant-test-5678"
|
||||
|
||||
|
||||
def test_read_token_env_override(tmp_path, monkeypatch):
|
||||
"""read_token() honours CLAUDE_CREDENTIALS_PATH env override (D-03)."""
|
||||
creds = tmp_path / ".credentials.json"
|
||||
creds.write_text(json.dumps({"accessToken": "sk-ant-test-ENV"}))
|
||||
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(creds))
|
||||
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
||||
assert read_token() == "sk-ant-test-ENV"
|
||||
|
||||
|
||||
def test_read_token_primary_path(tmp_path, monkeypatch):
|
||||
"""read_token() reads from the primary candidate path (first hit wins)."""
|
||||
creds = tmp_path / ".claude" / ".credentials.json"
|
||||
creds.parent.mkdir(parents=True)
|
||||
creds.write_text(json.dumps({"claudeAiOauth": {"accessToken": "sk-ant-test-PRIMARY"}}))
|
||||
monkeypatch.delenv("CLAUDE_CREDENTIALS_PATH", raising=False)
|
||||
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
||||
# Monkeypatch _windows_credential_candidates to return only our tmp path
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
monkeypatch.setattr(mod, "_windows_credential_candidates", lambda: [creds])
|
||||
assert read_token() == "sk-ant-test-PRIMARY"
|
||||
|
||||
|
||||
def test_read_token_localappdata_fallback(tmp_path, monkeypatch):
|
||||
"""read_token() falls back to %LOCALAPPDATA%/Claude/.credentials.json when primary is absent."""
|
||||
missing_primary = tmp_path / "nonexistent_primary" / ".credentials.json"
|
||||
present_localappdata = tmp_path / "localappdata" / ".credentials.json"
|
||||
missing_appdata = tmp_path / "nonexistent_appdata" / ".credentials.json"
|
||||
|
||||
present_localappdata.parent.mkdir(parents=True)
|
||||
present_localappdata.write_text(json.dumps({"accessToken": "sk-ant-test-LA"}))
|
||||
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
monkeypatch.setattr(
|
||||
mod,
|
||||
"_windows_credential_candidates",
|
||||
lambda: [missing_primary, present_localappdata, missing_appdata],
|
||||
)
|
||||
assert read_token() == "sk-ant-test-LA"
|
||||
|
||||
|
||||
def test_read_token_appdata_fallback(tmp_path, monkeypatch):
|
||||
"""read_token() falls back to %APPDATA%/Claude/.credentials.json when primary and LOCALAPPDATA are absent."""
|
||||
missing_primary = tmp_path / "nonexistent_primary" / ".credentials.json"
|
||||
missing_localappdata = tmp_path / "nonexistent_localappdata" / ".credentials.json"
|
||||
present_appdata = tmp_path / "appdata" / ".credentials.json"
|
||||
|
||||
present_appdata.parent.mkdir(parents=True)
|
||||
present_appdata.write_text(json.dumps({"accessToken": "sk-ant-test-APP"}))
|
||||
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
monkeypatch.setattr(
|
||||
mod,
|
||||
"_windows_credential_candidates",
|
||||
lambda: [missing_primary, missing_localappdata, present_appdata],
|
||||
)
|
||||
assert read_token() == "sk-ant-test-APP"
|
||||
|
||||
|
||||
def test_read_token_no_file(tmp_path, monkeypatch):
|
||||
"""read_token() returns None when no credential file can be found."""
|
||||
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(tmp_path / "nonexistent.json"))
|
||||
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
||||
assert read_token() is None
|
||||
|
||||
|
||||
def test_read_token_config_dir_override(tmp_path, monkeypatch):
|
||||
"""read_token() honours the official CLAUDE_CONFIG_DIR env override."""
|
||||
creds = tmp_path / ".credentials.json"
|
||||
creds.write_text(json.dumps({"accessToken": "sk-ant-test-CFGDIR"}))
|
||||
monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path))
|
||||
monkeypatch.delenv("CLAUDE_CREDENTIALS_PATH", raising=False)
|
||||
assert read_token() == "sk-ant-test-CFGDIR"
|
||||
|
||||
|
||||
def test_read_expiry_decodes_milliseconds(monkeypatch):
|
||||
"""_read_expiry() divides expiresAt by 1000 (ms -> s); fixture 9999999999000 -> year 2286."""
|
||||
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(FIXTURES / "credentials_nested.json"))
|
||||
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
||||
result = _read_expiry()
|
||||
assert result.startswith("2286-"), f"Expected year 2286, got: {result}"
|
||||
|
||||
|
||||
# --- WR-03: regression guard for CR-01 (empty/blank token must not be accepted) ---
|
||||
|
||||
def test_extract_empty_token_is_none():
|
||||
"""_extract_access_token returns None for empty accessToken (CR-01 regression guard)."""
|
||||
assert _extract_access_token('{"accessToken": ""}') is None
|
||||
assert _extract_access_token('{}') is None
|
||||
|
||||
|
||||
|
||||
def test_read_token_empty_credential_file_returns_none(tmp_path, monkeypatch):
|
||||
"""read_token() returns None (not empty string) when credential file has empty accessToken."""
|
||||
creds = tmp_path / ".credentials.json"
|
||||
creds.write_text(json.dumps({"accessToken": ""}))
|
||||
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(creds))
|
||||
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
||||
assert read_token() is None
|
||||
|
||||
|
||||
# --- WR-01: regression guard for _read_expiry with non-dict top-level JSON ---
|
||||
|
||||
def test_read_expiry_non_dict_json_returns_unknown(tmp_path, monkeypatch):
|
||||
"""_read_expiry() returns 'expiry unknown' (not crash) for non-dict top-level JSON (WR-01)."""
|
||||
creds = tmp_path / ".credentials.json"
|
||||
creds.write_text("[1, 2, 3]")
|
||||
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(creds))
|
||||
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
|
||||
assert _read_expiry() == "expiry unknown"
|
||||
|
||||
|
||||
# --- WR-02: D-06 redaction requirement must be tested ---
|
||||
|
||||
def test_main_emits_linux_warning(monkeypatch):
|
||||
"""__main__ prints a non-fatal stderr warning on non-Windows platforms (new async runner).
|
||||
|
||||
Phase 2 replaced the Phase 1 token-printing __main__ with asyncio.run(main()). The
|
||||
new contract:
|
||||
- On non-Windows: emits "WinRT BLE will not be available" to stderr before the loop.
|
||||
- Enters the async scan/connect/poll loop (no longer prints token/expiry).
|
||||
This test interrupts the process after 3s to capture the warning without hanging.
|
||||
"""
|
||||
env = {**__import__("os").environ, "CLAUDE_CREDENTIALS_PATH": str(FIXTURES / "credentials_nested.json")}
|
||||
env.pop("CLAUDE_CONFIG_DIR", None)
|
||||
module = str(Path(__file__).parent.parent / "claude_usage_daemon_windows.py")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[sys.executable, module],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=3,
|
||||
)
|
||||
# If it exits cleanly, verify warning was emitted
|
||||
assert "WinRT BLE will not be available" in result.stderr
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
# Process is hanging in the scan loop — expected behavior on Linux.
|
||||
# The warning should appear in the partial stderr captured so far.
|
||||
partial_stderr = (exc.stderr or b"").decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
|
||||
assert "WinRT BLE will not be available" in partial_stderr, (
|
||||
f"Expected Linux/WSL warning in stderr before scan loop, got: {partial_stderr!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_main_emits_linux_warning_before_loop(monkeypatch):
|
||||
"""__main__ stderr warning appears before the async scan loop starts on Linux/WSL."""
|
||||
import signal as _signal
|
||||
env = {**__import__("os").environ}
|
||||
env.pop("CLAUDE_CONFIG_DIR", None)
|
||||
env.pop("CLAUDE_CREDENTIALS_PATH", None)
|
||||
module = str(Path(__file__).parent.parent / "claude_usage_daemon_windows.py")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[sys.executable, module],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=3,
|
||||
)
|
||||
# If it exits cleanly (KeyboardInterrupt path), check warning
|
||||
assert "WinRT BLE will not be available" in result.stderr
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
# Process is hanging in the scan loop — expected behavior on Linux.
|
||||
# The warning should appear in the partial stderr captured so far.
|
||||
partial_stderr = (exc.stderr or b"").decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
|
||||
assert "WinRT BLE will not be available" in partial_stderr, (
|
||||
f"Expected Linux/WSL warning in stderr before scan loop, got: {partial_stderr!r}"
|
||||
)
|
||||
@@ -0,0 +1,328 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for daemon/tray_windows.py — APP-01.
|
||||
|
||||
Covers:
|
||||
TrayState scalar setters and initial state
|
||||
header_text() for all three states including last_sync=None
|
||||
daemon main() accepts tray_state and populates ts.loop / ts.stop_event
|
||||
Quit routes through loop.call_soon_threadsafe (not stop_event.set directly)
|
||||
Error toast fires only on transition INTO error state (D-04)
|
||||
|
||||
All pystray usage is inside tray_windows.main() (deferred import), so these
|
||||
tests can import the pure helpers (TrayState, header_text) and test Quit/toast
|
||||
handlers with mocked icons without importing the GTK-less top-level pystray.
|
||||
|
||||
Run: python -m pytest daemon/tests/test_windows_tray.py -x -q
|
||||
"""
|
||||
import asyncio
|
||||
import time
|
||||
from unittest.mock import AsyncMock, MagicMock, patch, call
|
||||
|
||||
import pytest
|
||||
|
||||
from daemon.tray_windows import TrayState, header_text, _acquire_single_instance, _ERROR_ALREADY_EXISTS
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TrayState — initial state and setters
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_tray_state_initial():
|
||||
"""TrayState initialises to scanning state with no last_sync."""
|
||||
ts = TrayState()
|
||||
assert ts.state == "scanning"
|
||||
assert ts.reason == ""
|
||||
assert ts.last_sync is None
|
||||
assert ts.loop is None
|
||||
assert ts.stop_event is None
|
||||
|
||||
|
||||
def test_set_connected():
|
||||
"""set_connected(ts_float) sets state='connected', clears reason, records last_sync."""
|
||||
ts = TrayState()
|
||||
now = time.time()
|
||||
ts.set_connected(now)
|
||||
assert ts.state == "connected"
|
||||
assert ts.reason == ""
|
||||
assert ts.last_sync == now
|
||||
|
||||
|
||||
def test_set_scanning():
|
||||
"""set_scanning() sets state='scanning', clears reason."""
|
||||
ts = TrayState()
|
||||
ts.set_error("something bad") # put it in error first
|
||||
ts.set_scanning()
|
||||
assert ts.state == "scanning"
|
||||
assert ts.reason == ""
|
||||
|
||||
|
||||
def test_set_error():
|
||||
"""set_error(why) sets state='error' and stores the reason string."""
|
||||
ts = TrayState()
|
||||
ts.set_error("token expired — run claude login")
|
||||
assert ts.state == "error"
|
||||
assert ts.reason == "token expired — run claude login"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# header_text — D-05 string shapes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_header_text_scanning():
|
||||
"""header_text returns 'Scanning…' in scanning state."""
|
||||
ts = TrayState()
|
||||
ts.set_scanning()
|
||||
assert header_text(ts) == "Scanning…"
|
||||
|
||||
|
||||
def test_header_text_error():
|
||||
"""header_text returns 'Error: {reason}' in error state."""
|
||||
ts = TrayState()
|
||||
ts.set_error("token expired — run claude login")
|
||||
result = header_text(ts)
|
||||
assert result == "Error: token expired — run claude login"
|
||||
|
||||
|
||||
def test_header_text_connected_with_last_sync():
|
||||
"""header_text returns 'Connected · last update HH:MM' when last_sync is set."""
|
||||
ts = TrayState()
|
||||
# Use a known timestamp so we can predict the HH:MM string.
|
||||
known_ts = time.mktime(time.strptime("2026-06-01 14:32:00", "%Y-%m-%d %H:%M:%S"))
|
||||
ts.set_connected(known_ts)
|
||||
result = header_text(ts)
|
||||
# Extract the HH:MM portion from the actual local time expansion.
|
||||
expected_when = time.strftime("%H:%M", time.localtime(known_ts))
|
||||
assert result == f"Connected · last update {expected_when}"
|
||||
|
||||
|
||||
def test_header_text_connected_never_when_last_sync_none():
|
||||
"""header_text returns 'Connected · last update never' when last_sync is None."""
|
||||
ts = TrayState()
|
||||
# Manually set state without using set_connected so last_sync stays None.
|
||||
ts.state = "connected"
|
||||
ts.last_sync = None
|
||||
result = header_text(ts)
|
||||
assert result == "Connected · last update never"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# daemon main() populates ts.loop and ts.stop_event
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_main_populates_tray_state_loop_and_stop_event():
|
||||
"""daemon main(tray_state=ts) sets ts.loop and ts.stop_event before the loop body."""
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
ts = TrayState()
|
||||
populated = {}
|
||||
|
||||
async def _fake_scan():
|
||||
# Record the state of ts at first scan entry (after main() startup lines).
|
||||
populated["loop"] = ts.loop
|
||||
populated["stop_event"] = ts.stop_event
|
||||
# Signal stop so the loop exits cleanly.
|
||||
ts.stop_event.set()
|
||||
return None # no device found
|
||||
|
||||
with patch.object(mod, "scan_for_device", side_effect=_fake_scan):
|
||||
asyncio.run(mod.main(tray_state=ts))
|
||||
|
||||
assert populated.get("loop") is not None, "ts.loop must be set by daemon main()"
|
||||
assert populated.get("stop_event") is not None, "ts.stop_event must be set by daemon main()"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Quit handler routes through call_soon_threadsafe (not stop_event.set directly)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_quit_uses_call_soon_threadsafe():
|
||||
"""The Quit menu handler calls loop.call_soon_threadsafe(stop_event.set) and icon.stop().
|
||||
|
||||
It must NOT call stop_event.set() directly from the tray thread
|
||||
(RESEARCH Pitfall 2 / T-04-06 mitigation).
|
||||
"""
|
||||
# Build a TrayState with a mocked loop and stop_event.
|
||||
ts = TrayState()
|
||||
mock_loop = MagicMock()
|
||||
mock_stop_event = MagicMock()
|
||||
ts.loop = mock_loop
|
||||
ts.stop_event = mock_stop_event
|
||||
|
||||
# Build the Quit handler the same way tray_windows.main() does, without
|
||||
# importing pystray at the module level. We construct a local closure
|
||||
# that mirrors the on_quit body.
|
||||
mock_icon = MagicMock()
|
||||
|
||||
def _on_quit(icon_ref, _item):
|
||||
# This is the exact body from tray_windows.main() — keep in sync.
|
||||
ts.loop.call_soon_threadsafe(ts.stop_event.set)
|
||||
icon_ref.stop()
|
||||
|
||||
_on_quit(mock_icon, None)
|
||||
|
||||
# call_soon_threadsafe must have been called with stop_event.set as the arg.
|
||||
mock_loop.call_soon_threadsafe.assert_called_once_with(mock_stop_event.set)
|
||||
# icon.stop() must have been called.
|
||||
mock_icon.stop.assert_called_once()
|
||||
# stop_event.set() must NOT have been called directly.
|
||||
mock_stop_event.set.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Error toast fires only on transition INTO error (D-04)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_error_toast_on_entry_only():
|
||||
"""The tray refresh loop fires icon.notify() only on transition INTO error.
|
||||
|
||||
Sequence: scanning -> error -> error
|
||||
Expected: notify called exactly once (on the scanning->error transition).
|
||||
"""
|
||||
ts = TrayState()
|
||||
ts.set_scanning()
|
||||
|
||||
mock_icon = MagicMock()
|
||||
mock_icon._running = True
|
||||
|
||||
# Simulate the _refresh loop's state-change detection logic from tray_windows.main().
|
||||
# We run two transitions manually:
|
||||
# 1. scanning -> error (should call notify once)
|
||||
# 2. error -> error (no change — notify must NOT fire again)
|
||||
prev_state: dict = {"state": None}
|
||||
|
||||
def _process_state_change(new_state: str, reason: str = "") -> None:
|
||||
"""Mirror the relevant part of the _refresh loop body."""
|
||||
ts.state = new_state
|
||||
ts.reason = reason
|
||||
current = ts.state
|
||||
if current != prev_state["state"]:
|
||||
if current == "error" and prev_state["state"] != "error":
|
||||
mock_icon.notify(ts.reason or "Clawdmeter error", "Clawdmeter")
|
||||
prev_state["state"] = current
|
||||
|
||||
# Transition 1: scanning -> error (notify should fire)
|
||||
_process_state_change("scanning")
|
||||
_process_state_change("error", "token expired — run claude login")
|
||||
# Transition 2: error -> error (same state — no call)
|
||||
_process_state_change("error", "token expired — run claude login")
|
||||
|
||||
mock_icon.notify.assert_called_once_with(
|
||||
"token expired — run claude login", "Clawdmeter"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Single-instance guard (named mutex) — duplicate-launch / ARSO collision
|
||||
# ---------------------------------------------------------------------------
|
||||
# Field bug: Windows "restart apps after sign-in" (ARSO) restored a console
|
||||
# `python.exe tray_windows.py` instance while the headless `pythonw` autostart
|
||||
# also fired — two trays fighting over the one BLE link. The guard makes a
|
||||
# second instance exit before it touches BLE.
|
||||
|
||||
def test_single_instance_noop_off_windows():
|
||||
"""Off-Windows the guard is a no-op that returns a truthy sentinel (never None)."""
|
||||
with patch("daemon.tray_windows.sys") as mock_sys:
|
||||
mock_sys.platform = "linux"
|
||||
assert _acquire_single_instance() is not None
|
||||
|
||||
|
||||
def _fake_kernel32(last_error: int, handle: int):
|
||||
"""Build a fake ctypes module tree whose CreateMutexW returns `handle` and
|
||||
whose get_last_error() returns `last_error`."""
|
||||
fake_kernel32 = MagicMock()
|
||||
fake_kernel32.CreateMutexW.return_value = handle
|
||||
fake_ctypes = MagicMock()
|
||||
fake_ctypes.WinDLL.return_value = fake_kernel32
|
||||
fake_ctypes.get_last_error.return_value = last_error
|
||||
return fake_ctypes
|
||||
|
||||
|
||||
def test_single_instance_first_instance_gets_handle():
|
||||
"""First instance: CreateMutexW succeeds, no prior owner → returns the handle."""
|
||||
fake_ctypes = _fake_kernel32(last_error=0, handle=0xABCD)
|
||||
with patch("daemon.tray_windows.sys") as mock_sys, \
|
||||
patch.dict("sys.modules", {"ctypes": fake_ctypes, "ctypes.wintypes": MagicMock()}):
|
||||
mock_sys.platform = "win32"
|
||||
assert _acquire_single_instance() == 0xABCD
|
||||
|
||||
|
||||
def test_single_instance_second_instance_gets_none():
|
||||
"""Second instance: mutex already exists → returns None so caller exits."""
|
||||
fake_ctypes = _fake_kernel32(last_error=_ERROR_ALREADY_EXISTS, handle=0xABCD)
|
||||
with patch("daemon.tray_windows.sys") as mock_sys, \
|
||||
patch.dict("sys.modules", {"ctypes": fake_ctypes, "ctypes.wintypes": MagicMock()}):
|
||||
mock_sys.platform = "win32"
|
||||
assert _acquire_single_instance() is None
|
||||
|
||||
|
||||
def test_single_instance_fails_open_on_null_handle():
|
||||
"""If CreateMutexW returns NULL, fail OPEN (truthy) — never block tray startup."""
|
||||
fake_ctypes = _fake_kernel32(last_error=_ERROR_ALREADY_EXISTS, handle=0)
|
||||
with patch("daemon.tray_windows.sys") as mock_sys, \
|
||||
patch.dict("sys.modules", {"ctypes": fake_ctypes, "ctypes.wintypes": MagicMock()}):
|
||||
mock_sys.platform = "win32"
|
||||
result = _acquire_single_instance()
|
||||
assert result is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regression: cwd-independent package + asset resolution (SC#1 logon autostart)
|
||||
# ---------------------------------------------------------------------------
|
||||
# Field bug: launching `pythonw.exe daemon\tray_windows.py` at logon starts with
|
||||
# cwd = System32, so `import daemon.*` raised ModuleNotFoundError and the relative
|
||||
# logo path failed — the tray crashed silently with no icon. tray_windows must
|
||||
# self-locate the repo root from __file__ so it works from any working directory.
|
||||
|
||||
def test_repo_root_is_parent_of_daemon_package():
|
||||
"""_REPO_ROOT points at the dir that CONTAINS the daemon package."""
|
||||
import os
|
||||
import daemon.tray_windows as tw
|
||||
|
||||
assert os.path.isdir(os.path.join(tw._REPO_ROOT, "daemon"))
|
||||
assert os.path.isfile(
|
||||
os.path.join(tw._REPO_ROOT, "firmware", "src", "logo.h")
|
||||
), "brand logo must resolve from _REPO_ROOT, not the current working directory"
|
||||
|
||||
|
||||
def test_repo_root_on_sys_path_after_import():
|
||||
"""Importing tray_windows puts the repo root on sys.path so `daemon.*` resolves regardless of cwd."""
|
||||
import sys
|
||||
import daemon.tray_windows as tw
|
||||
|
||||
assert tw._REPO_ROOT in sys.path
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regression: daemon main() must run in a BACKGROUND thread (SC#1 tray launch)
|
||||
# ---------------------------------------------------------------------------
|
||||
# Field bug: under the tray the loop runs in threading.Thread (pystray owns the
|
||||
# main thread). OS signal-handler installation (loop.add_signal_handler /
|
||||
# signal.signal) only works on the main thread, so main() raised
|
||||
# "signal only works in main thread" and the daemon thread died on startup.
|
||||
# main() must guard signal setup to the main thread; the tray owns shutdown.
|
||||
|
||||
def test_main_runs_in_background_thread_without_signal_error():
|
||||
"""main(tray_state=ts) started from a non-main thread must not raise on signal setup."""
|
||||
import threading as _threading
|
||||
import daemon.claude_usage_daemon_windows as mod
|
||||
|
||||
ts = TrayState()
|
||||
errors: list = []
|
||||
|
||||
async def _fake_scan():
|
||||
ts.stop_event.set() # exit the loop immediately
|
||||
return None
|
||||
|
||||
def _run() -> None:
|
||||
try:
|
||||
with patch.object(mod, "scan_for_device", side_effect=_fake_scan):
|
||||
asyncio.run(mod.main(tray_state=ts))
|
||||
except Exception as exc: # noqa: BLE001 — capture for the assertion
|
||||
errors.append(exc)
|
||||
|
||||
t = _threading.Thread(target=_run)
|
||||
t.start()
|
||||
t.join(timeout=10)
|
||||
|
||||
assert not t.is_alive(), "daemon main() hung in background thread"
|
||||
assert not errors, f"main() raised in a background thread: {errors!r}"
|
||||
Reference in New Issue
Block a user