Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions src/copilot_usage/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import click
from rich.console import Console
from rich.table import Table
from watchdog.events import FileSystemEventHandler # type: ignore[import-untyped]
from watchdog.observers import Observer # type: ignore[import-untyped]

from copilot_usage.models import SessionSummary
Expand Down Expand Up @@ -114,14 +115,15 @@ def _read_line_nonblocking(timeout: float = 0.5) -> str | None:
return None


class _FileChangeHandler:
class _FileChangeHandler(FileSystemEventHandler): # type: ignore[misc]
"""Watchdog handler that triggers refresh on any session-state change."""

def __init__(self, change_event: threading.Event) -> None:
super().__init__()
self._change_event = change_event
self._last_trigger = 0.0

def dispatch(self, event: object) -> None: # noqa: ANN001
def dispatch(self, event: object) -> None:
now = time.monotonic()
if now - self._last_trigger > 2.0: # debounce 2s
self._last_trigger = now
Expand All @@ -132,7 +134,7 @@ def _start_observer(session_path: Path, change_event: threading.Event) -> object
"""Start a watchdog observer monitoring *session_path* for changes."""
handler = _FileChangeHandler(change_event)
observer = Observer()
observer.schedule(handler, str(session_path), recursive=True) # type: ignore[arg-type]
observer.schedule(handler, str(session_path), recursive=True)
observer.daemon = True
observer.start()
return observer
Expand Down
16 changes: 16 additions & 0 deletions tests/copilot_usage/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -586,3 +586,19 @@ def test_dispatch_fires_again_after_debounce_gap(self) -> None:
handler._last_trigger = _time.monotonic() - 3.0 # pyright: ignore[reportPrivateUsage]
handler.dispatch(object())
assert event.is_set()

def test_inherits_from_filesystemeventhandler(self) -> None:
"""_FileChangeHandler inherits from watchdog FileSystemEventHandler."""
from watchdog.events import (
FileSystemEventHandler, # type: ignore[import-untyped]
)

from copilot_usage.cli import (
_FileChangeHandler, # pyright: ignore[reportPrivateUsage]
)

event = threading.Event()
handler = _FileChangeHandler(event)
assert isinstance(handler, FileSystemEventHandler)
handler.dispatch(object())
assert event.is_set()
Loading