Skip to content
Open
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
61 changes: 59 additions & 2 deletions tests/copilot_usage/test_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,11 @@
import os
import time
from collections.abc import Iterator
from contextlib import AbstractContextManager
from datetime import UTC, datetime
from pathlib import Path
from typing import SupportsIndex, overload
from unittest.mock import patch
from typing import SupportsIndex, cast, overload
from unittest.mock import MagicMock, patch

import pytest
from pydantic import ValidationError
Expand Down Expand Up @@ -629,6 +630,62 @@ def _bomb(path: str | os.PathLike[str]) -> Iterator[os.DirEntry[str]]:
assert len(result) == 1
assert result[0][0].parent.name == "sess-good"

def test_full_scandir_is_dir_oserror_skips_entry(self, tmp_path: Path) -> None:
"""Skip a root-level entry whose ``is_dir()`` raises ``OSError``.

Simulates a broken symlink or ``EACCES`` on ``lstat`` by wrapping
``os.scandir`` so that one entry's ``is_dir()`` raises ``OSError``.
The faulting entry must be silently skipped, not crash discovery.
"""
good = tmp_path / "sess-good"
_write_events(good / "events.jsonl", _START_EVENT)
bad = tmp_path / "sess-bad"
_write_events(bad / "events.jsonl", _START_EVENT)

original_scandir = os.scandir

def _patched_scandir(
path: str | os.PathLike[str],
) -> AbstractContextManager[Iterator[os.DirEntry[str]]]:
if str(path) != str(tmp_path):
return original_scandir(path)

class _WrappedCtx:
"""Context manager keeping the scandir iterator open."""

def __init__(self) -> None:
self._it: Iterator[os.DirEntry[str]] | None = None

def _iter_wrapped_entries(self) -> Iterator[os.DirEntry[str]]:
if self._it is None:
return
for e in self._it:
if e.name == "sess-bad":
m = MagicMock(spec=os.DirEntry)
m.name = e.name
m.path = e.path
m.is_dir.side_effect = OSError("lstat failed")
yield cast(os.DirEntry[str], m)
else:
yield e

def __enter__(self) -> Iterator[os.DirEntry[str]]:
self._it = original_scandir(path)
return self._iter_wrapped_entries()

def __exit__(self, *a: object) -> None:
if self._it is not None:
self._it.close() # type: ignore[union-attr]
self._it = None
Comment on lines +656 to +679
Copy link

Copilot AI Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self._it is typed as Iterator[...] | None, but it’s always the concrete iterator returned by os.scandir(...), which provides .close(). Typing it as the concrete scandir iterator type (or a small Protocol exposing __iter__ + close()) would let you call .close() without # type: ignore[union-attr], keeping the test strictly type-safe.

Copilot uses AI. Check for mistakes.

return _WrappedCtx()

with patch("copilot_usage.parser.os.scandir", side_effect=_patched_scandir):
_, result = _discover_with_identity(tmp_path)

assert len(result) == 1
assert result[0][0].parent.name == "sess-good"


# ---------------------------------------------------------------------------
# _discover_with_identity — linear scan (issue #773)
Expand Down
Loading