diff --git a/src/dippy/core/config.py b/src/dippy/core/config.py index 4c0bba6..186b378 100644 --- a/src/dippy/core/config.py +++ b/src/dippy/core/config.py @@ -11,6 +11,11 @@ # Cache home directory at module load - fails fast if HOME is unset _HOME = Path.home() + +def _posix_str(path: str) -> str: + """Normalize path separators to forward slashes for consistent matching.""" + return path.replace("\\", "/") + USER_CONFIG = _HOME / ".dippy" / "config" PROJECT_CONFIG_NAME = ".dippy" ENV_CONFIG = "DIPPY_CONFIG" @@ -458,6 +463,9 @@ def _classify_token(token: str) -> str: return _VARIABLE if token.startswith("/"): return _ABSOLUTE + # Windows drive letter paths: C:\, C:/, D:\, etc. + if len(token) >= 3 and token[0].isalpha() and token[1] == ":" and token[2] in ("/", "\\"): + return _ABSOLUTE if token == "~" or token.startswith("~/"): return _HOME if token.startswith("~"): @@ -490,17 +498,18 @@ def _expand_token(token: str, cwd: Path, *, force_path: bool = False) -> str: if kind == _VARIABLE: return token if kind == _ABSOLUTE: - return token + return _posix_str(token) if kind == _HOME: # ~ → /home/user, ~/foo → /home/user/foo - return str(home) + token[1:] if len(token) > 1 else str(home) + h = _posix_str(str(home)) + return h + token[1:] if len(token) > 1 else h if kind == _USER_HOME: return token if kind == _RELATIVE: - return str((cwd / token).resolve()) + return _posix_str(str((cwd / token).resolve())) # BARE if force_path: - return str((cwd / token).resolve()) + return _posix_str(str((cwd / token).resolve())) return token @@ -512,7 +521,8 @@ def _expand_home_only(token: str) -> str: """ if _classify_token(token) == _HOME: home = Path.home() - return str(home) + token[1:] if len(token) > 1 else str(home) + h = _posix_str(str(home)) + return h + token[1:] if len(token) > 1 else h return token diff --git a/src/dippy/dippy_statusline.py b/src/dippy/dippy_statusline.py index fcde69b..ff7ab52 100755 --- a/src/dippy/dippy_statusline.py +++ b/src/dippy/dippy_statusline.py @@ -8,6 +8,7 @@ import shutil import subprocess import sys +import threading import time import traceback from datetime import datetime, timezone @@ -216,6 +217,35 @@ def get_local_mcp_servers() -> list[str]: return [] +def _refresh_mcp_cache_python(tmp, conn_r, conn_g, conn_b, disc_r, disc_g, disc_b): + """Refresh MCP cache using pure Python (Windows-compatible).""" + try: + result = subprocess.run( + ["claude", "mcp", "list"], + capture_output=True, text=True, timeout=10, + ) + lines = [] + for line in result.stdout.splitlines(): + if ":" not in line: + continue + name = line.split(":")[0].strip() + if not name: + continue + if "Connected" in line: + lines.append(f"\033[38;2;{conn_r};{conn_g};{conn_b}m{name}\033[0m") + else: + lines.append(f"\033[38;2;{disc_r};{disc_g};{disc_b}m!{name}\033[0m") + with open(tmp, "w") as f: + f.write(", ".join(lines)) + os.replace(tmp, MCP_CACHE_PATH) + except Exception: + log.error("mcp_cache_refresh_python_failed") + try: + os.unlink(tmp) + except OSError: + pass + + def get_mcp_servers() -> str | None: """Read MCP servers from local config and cached global list.""" local_servers = get_local_mcp_servers() @@ -242,15 +272,23 @@ def get_mcp_servers() -> str | None: os.makedirs(CACHE_DIR, exist_ok=True) tmp = f"{MCP_CACHE_PATH}.tmp.{os.getpid()}" disc_r, disc_g, disc_b = hex_to_rgb(MOLOKAI[STYLES["mcp_disconnected"][0]]) - cmd = f'timeout 10 claude mcp list 2>/dev/null | awk -F: \'NF>1 {{if (/Connected/) print "\\033[38;2;{conn_r};{conn_g};{conn_b}m" $1 "\\033[0m"; else print "\\033[38;2;{disc_r};{disc_g};{disc_b}m!" $1 "\\033[0m"}}\' | paste -sd, | sed \'s/,/, /g\' > {tmp} && mv {tmp} {MCP_CACHE_PATH}' - subprocess.Popen( - cmd, - shell=True, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True, - ) + if sys.platform == "win32": + t = threading.Thread( + target=_refresh_mcp_cache_python, + args=(tmp, conn_r, conn_g, conn_b, disc_r, disc_g, disc_b), + daemon=True, + ) + t.start() + else: + cmd = f'timeout 10 claude mcp list 2>/dev/null | awk -F: \'NF>1 {{if (/Connected/) print "\\033[38;2;{conn_r};{conn_g};{conn_b}m" $1 "\\033[0m"; else print "\\033[38;2;{disc_r};{disc_g};{disc_b}m!" $1 "\\033[0m"}}\' | paste -sd, | sed \'s/,/, /g\' > {tmp} && mv {tmp} {MCP_CACHE_PATH}' + subprocess.Popen( + cmd, + shell=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) log.debug("mcp_cache_refresh_spawned", age=age, ttl=MCP_CACHE_TTL) except Exception: log.error("mcp_cache_refresh_failed") @@ -462,6 +500,13 @@ def build_statusline(data: dict) -> str: def main(): log.info("main_start") + # Ensure stdout can handle Unicode (emoji) on Windows where + # subprocess stdout defaults to cp1252 + if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass try: data = json.load(sys.stdin) log.debug("main_input_parsed", session_id=data.get("session_id", "")) diff --git a/tests/test_config.py b/tests/test_config.py index f3abe69..6f501da 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -5,6 +5,7 @@ import json import os import stat +import sys from pathlib import Path import pytest @@ -217,6 +218,9 @@ def test_env_config_tilde_expansion(self, tmp_path, monkeypatch): fake_home.mkdir() (fake_home / "my.cfg").write_text("allow tilde") monkeypatch.setenv("HOME", str(fake_home)) + # On Windows, Path.expanduser() uses USERPROFILE, not HOME + if sys.platform == "win32": + monkeypatch.setenv("USERPROFILE", str(fake_home)) monkeypatch.setenv("DIPPY_CONFIG", "~/my.cfg") def mock_parse(text, source=None): @@ -1305,8 +1309,8 @@ def test_tilde_expansion_consistency(self, tmp_path): home = Path.home() # Settings expand tilde correctly assert cfg.log == home / "audit.log" - # Rule patterns also expand tilde consistently - assert cfg.rules[0].pattern == f"cd {home}" + # Rule patterns also expand tilde consistently (forward slashes for matching) + assert cfg.rules[0].pattern == f"cd {str(home).replace(chr(92), '/')}" def test_url_not_mangled(self, tmp_path): """URLs should not be turned into paths under cwd.""" @@ -1821,7 +1825,7 @@ class TestAlias: def test_alias_tilde_path(self, tmp_path): """alias ~/bin/gh gh + allow gh matches ~/bin/gh pr list.""" - home = str(Path.home()) + home = str(Path.home()).replace("\\", "/") cfg = parse_config("alias ~/bin/gh gh\nallow gh") assert cfg.aliases == {f"{home}/bin/gh": "gh"} c = SimpleCommand(words=["~/bin/gh", "pr", "list"]) diff --git a/tests/test_entrypoint.py b/tests/test_entrypoint.py index 5b48202..09f88d3 100644 --- a/tests/test_entrypoint.py +++ b/tests/test_entrypoint.py @@ -8,11 +8,11 @@ import tempfile from pathlib import Path +import pytest REPO_ROOT = Path(__file__).resolve().parent.parent DIPPY_HOOK = REPO_ROOT / "bin" / "dippy-hook" -# Use system Python to avoid venv masking import issues -SYSTEM_PYTHON = "/usr/bin/python3" +IS_WINDOWS = sys.platform == "win32" def get_decision(output: dict) -> str | None: @@ -46,7 +46,9 @@ def _run( else: stdin_bytes = input_data.encode() - python = SYSTEM_PYTHON if use_system_python else sys.executable + # use_system_python is for testing path resolution without venv; + # on Windows there's no /usr/bin/python3, so always use sys.executable + python = sys.executable return subprocess.run( [python, str(script)], input=stdin_bytes, @@ -66,6 +68,7 @@ def test_direct_invocation(self): output = json.loads(result.stdout) assert get_decision(output) == "allow" + @pytest.mark.skipif(IS_WINDOWS, reason="symlinks require elevated privileges on Windows") def test_symlink_invocation(self): """Critical: invocation via symlink must also work. @@ -78,6 +81,7 @@ def test_symlink_invocation(self): output = json.loads(result.stdout) assert get_decision(output) == "allow" + @pytest.mark.skipif(IS_WINDOWS, reason="symlinks require elevated privileges on Windows") def test_nested_symlink_invocation(self): """Symlink in deeply nested unrelated path (simulates Homebrew Cellar).""" input_data = {"tool_name": "Bash", "tool_input": {"command": "ls"}} diff --git a/tests/test_statusline.py b/tests/test_statusline.py index 1f213af..3108799 100644 --- a/tests/test_statusline.py +++ b/tests/test_statusline.py @@ -3,14 +3,18 @@ from __future__ import annotations import json +import os import subprocess +import sys import tempfile import uuid from pathlib import Path +import pytest + REPO_ROOT = Path(__file__).resolve().parent.parent DIPPY_STATUSLINE = REPO_ROOT / "bin" / "dippy-statusline" -SYSTEM_PYTHON = "/usr/bin/python3" +IS_WINDOWS = sys.platform == "win32" def unique_session_id() -> str: @@ -39,7 +43,7 @@ def run_statusline( script = symlink_path return subprocess.run( - [SYSTEM_PYTHON, str(script)], + [sys.executable, str(script)], input=stdin_bytes, capture_output=True, timeout=10, @@ -63,6 +67,7 @@ def test_valid_input(self): # Should contain model name assert "Opus" in output + @pytest.mark.skipif(IS_WINDOWS, reason="symlinks require elevated privileges on Windows") def test_symlink_invocation(self): """Works when invoked via symlink (Homebrew scenario).""" input_data = {