Skip to content
Closed
Changes from 3 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
53 changes: 51 additions & 2 deletions tests/copilot_usage/test_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@
from collections.abc import Iterator
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 +629,55 @@ 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],
) -> Iterator[os.DirEntry[str]]:
if str(path) != str(tmp_path):
return original_scandir(path) # type: ignore[return-value]

class _WrappedCtx:
"""Context manager that wraps scandir entries."""

def __enter__(self) -> Iterator[os.DirEntry[str]]:
with original_scandir(path) as it:
entries: list[os.DirEntry[str]] = list(it)
wrapped: list[os.DirEntry[str]] = []
for e in entries:
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")
wrapped.append(cast(os.DirEntry[str], m))
else:
wrapped.append(e)
return iter(wrapped)

def __exit__(self, *a: object) -> None:
pass
Comment thread
microsasa marked this conversation as resolved.
Outdated

return _WrappedCtx() # type: ignore[return-value]
Comment thread
microsasa marked this conversation as resolved.
Outdated

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