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
6 changes: 3 additions & 3 deletions src/skillspector/input_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -813,7 +813,7 @@ def resolve(self, input_path: str) -> tuple[Path, str]:
if self._is_file_url(input_path):
return self._download_file(input_path), "url"
normalized_local_path = validate_local_input_path(Path(input_path))
if input_path.endswith(".zip"):
if input_path.lower().endswith(".zip"):
return self._extract_zip(normalized_local_path), "zip"
if input_path.endswith(".md"):
return self._wrap_single_file(normalized_local_path), "file"
Expand Down Expand Up @@ -1366,7 +1366,7 @@ def _download_file(self, url: str) -> Path:
download_path.unlink(missing_ok=True)
raise

is_zip = filename.endswith(".zip") or content_type.startswith("application/zip")
is_zip = filename.lower().endswith(".zip") or content_type.startswith("application/zip")
if is_zip:
zip_path = temp_dir / "download.zip"
download_path.replace(zip_path)
Expand All @@ -1389,7 +1389,7 @@ def _download_transitive_file(self, url: str) -> Path:
raise ValueError(f"Failed to download file: {exc}") from exc
temp_dir = self._get_temp_dir()
self._record_artifacts(1)
if filename.endswith(".zip") or headers.get("content-type", "").startswith(
if filename.lower().endswith(".zip") or headers.get("content-type", "").startswith(
"application/zip"
):
zip_path = temp_dir / "download.zip"
Expand Down
47 changes: 44 additions & 3 deletions tests/unit/test_input_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -374,19 +374,21 @@ def test_windows_no_follow_open_rejects_canonical_path_mismatch(
_open_regular_file_from_windows_handle(source)


def test_resolve_zip_file(tmp_path: Path) -> None:
"""Resolving a .zip file extracts and returns the extract dir."""
@pytest.mark.parametrize("extension", [".zip", ".ZIP", ".ZiP"])
def test_resolve_zip_file(tmp_path: Path, extension: str) -> None:
"""ZIP extensions are recognized regardless of case for local inputs."""
import zipfile

(tmp_path / "SKILL.md").write_text("# Skill", encoding="utf-8")
zip_path = tmp_path / "skill.zip"
zip_path = tmp_path / f"skill{extension}"
with zipfile.ZipFile(zip_path, "w") as zf:
zf.write(tmp_path / "SKILL.md", "SKILL.md")
handler = InputHandler()
try:
resolved, source_type = handler.resolve(str(zip_path))
assert resolved.is_dir()
assert source_type == "zip"
assert (resolved / "SKILL.md").read_text(encoding="utf-8") == "# Skill"
finally:
handler.cleanup()

Expand Down Expand Up @@ -601,6 +603,45 @@ def serve(request: httpx.Request) -> httpx.Response:
handler.cleanup()


@pytest.mark.parametrize("budgeted", [False, True], ids=["direct", "workflow-budget"])
@pytest.mark.parametrize("extension", [".zip", ".ZIP", ".ZiP"])
def test_file_url_download_extracts_zip_regardless_of_extension_case(
monkeypatch: pytest.MonkeyPatch, extension: str, budgeted: bool
) -> None:
"""ZIP filenames trigger extraction even without a ZIP content type."""
import io
import zipfile

skill = b"---\nname: demo\ndescription: demo\n---\n# Demo\n"
archive = io.BytesIO()
with zipfile.ZipFile(archive, "w") as zf:
zf.writestr("SKILL.md", skill)
url = f"https://raw.githubusercontent.com/org/repo/main/skill{extension}"
requested: list[str] = []

def serve(request: httpx.Request) -> httpx.Response:
requested.append(str(request.url))
return httpx.Response(
200, content=archive.getvalue(), headers={"content-type": "application/octet-stream"}
)

real_client = httpx.Client
monkeypatch.setattr(
"skillspector.input_handler.httpx.Client",
lambda *args, **kwargs: real_client(*args, transport=httpx.MockTransport(serve), **kwargs),
)
monkeypatch.setattr("skillspector.input_handler._is_private_ip", lambda _host: False)
handler = InputHandler(transitive_budget=WorkflowResourceBudget() if budgeted else None)
try:
resolved, source_type = handler.resolve(url)

assert source_type == "url"
assert requested == [url]
assert (resolved / "SKILL.md").read_bytes() == skill
finally:
handler.cleanup()


def test_validate_url_host_scp_extracts_github() -> None:
"""_validate_url_host extracts 'github.com' from an scp-style URL."""
with patch("skillspector.input_handler._is_private_ip", return_value=False):
Expand Down
Loading