mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-12 14:15:14 +00:00
1 parent
2b4d68bea9
commit
b09cbc089d
3 files changed
+238
-2
No files matched your search
@@ -25,6 +25,7 @@ from solidlsp.ls_utils import PathUtils
|
||||
from solidlsp.lsp_protocol_handler.lsp_types import InitializeParams
|
||||
from solidlsp.lsp_protocol_handler.server import ProcessLaunchInfo
|
||||
from solidlsp.settings import SolidLSPSettings
|
||||
from solidlsp.util.zip import SafeZipExtractor
|
||||
|
||||
from .common import RuntimeDependency
|
||||
|
||||
@@ -386,8 +387,13 @@ class CSharpLanguageServer(SolidLanguageServer):
|
||||
package_extract_dir = temp_dir / f"{package_name}.{package_version}"
|
||||
package_extract_dir.mkdir(exist_ok=True)
|
||||
|
||||
with zipfile.ZipFile(nupkg_file, "r") as zip_ref:
|
||||
zip_ref.extractall(package_extract_dir)
|
||||
# Use SafeZipExtractor to handle long paths and skip errors
|
||||
extractor = SafeZipExtractor(
|
||||
archive_path=nupkg_file,
|
||||
extract_dir=package_extract_dir,
|
||||
verbose=False
|
||||
)
|
||||
extractor.extract_all()
|
||||
|
||||
# Clean up the nupkg file
|
||||
nupkg_file.unlink()
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Optional, List
|
||||
import fnmatch
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
class SafeZipExtractor:
|
||||
"""
|
||||
A utility class for extracting ZIP archives safely.
|
||||
|
||||
Features:
|
||||
- Handles long file paths on Windows
|
||||
- Skips files that fail to extract, continuing with the rest
|
||||
- Creates necessary directories automatically
|
||||
- Optional include/exclude pattern filters
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
archive_path: Path,
|
||||
extract_dir: Path,
|
||||
verbose: bool = True,
|
||||
include_patterns: Optional[List[str]] = None,
|
||||
exclude_patterns: Optional[List[str]] = None
|
||||
) -> None:
|
||||
"""
|
||||
Initialize the SafeZipExtractor.
|
||||
|
||||
:param archive_path: Path to the ZIP archive file
|
||||
:param extract_dir: Directory where files will be extracted
|
||||
:param verbose: Whether to log status messages
|
||||
:param include_patterns: List of glob patterns for files to extract (None = all files)
|
||||
:param exclude_patterns: List of glob patterns for files to skip
|
||||
"""
|
||||
self.archive_path = Path(archive_path)
|
||||
self.extract_dir = Path(extract_dir)
|
||||
self.verbose = verbose
|
||||
self.include_patterns = include_patterns or []
|
||||
self.exclude_patterns = exclude_patterns or []
|
||||
|
||||
def extract_all(self) -> None:
|
||||
"""
|
||||
Extract all files from the archive, skipping any that fail.
|
||||
"""
|
||||
if not self.archive_path.exists():
|
||||
raise FileNotFoundError(f"Archive not found: {self.archive_path}")
|
||||
|
||||
if self.verbose:
|
||||
log.info(f"Extracting from: {self.archive_path} to {self.extract_dir}")
|
||||
|
||||
with zipfile.ZipFile(self.archive_path, "r") as zip_ref:
|
||||
for member in zip_ref.infolist():
|
||||
if self._should_extract(member.filename):
|
||||
self._extract_member(zip_ref, member)
|
||||
elif self.verbose:
|
||||
log.info(f"Skipped: {member.filename}")
|
||||
|
||||
def _should_extract(self, filename: str) -> bool:
|
||||
"""
|
||||
Determine whether a file should be extracted based on include/exclude patterns.
|
||||
|
||||
:param filename: The file name from the archive
|
||||
:return: True if the file should be extracted
|
||||
"""
|
||||
# If include_patterns is set, only extract if it matches at least one pattern
|
||||
if self.include_patterns:
|
||||
if not any(fnmatch.fnmatch(filename, pattern) for pattern in self.include_patterns):
|
||||
return False
|
||||
|
||||
# If exclude_patterns is set, skip if it matches any pattern
|
||||
if self.exclude_patterns:
|
||||
if any(fnmatch.fnmatch(filename, pattern) for pattern in self.exclude_patterns):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _extract_member(self, zip_ref: zipfile.ZipFile, member: zipfile.ZipInfo) -> None:
|
||||
"""
|
||||
Extract a single member from the archive with error handling.
|
||||
|
||||
:param zip_ref: Open ZipFile object
|
||||
:param member: ZipInfo object representing the file
|
||||
"""
|
||||
try:
|
||||
target_path = self.extract_dir / member.filename
|
||||
|
||||
# Ensure directory structure exists
|
||||
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Handle long paths on Windows
|
||||
final_path = self._normalize_path(target_path)
|
||||
|
||||
# Extract file
|
||||
with zip_ref.open(member) as source, open(final_path, "wb") as target:
|
||||
target.write(source.read())
|
||||
|
||||
if self.verbose:
|
||||
log.info(f"Extracted: {member.filename}")
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Failed to extract {member.filename}: {e}")
|
||||
|
||||
@staticmethod
|
||||
def _normalize_path(path: Path) -> Path:
|
||||
"""
|
||||
Adjust path to handle long paths on Windows.
|
||||
|
||||
:param path: Original path
|
||||
:return: Normalized path
|
||||
"""
|
||||
if sys.platform.startswith("win"):
|
||||
return Path(r"\\?\{}".format(os.path.abspath(path)))
|
||||
return path
|
||||
|
||||
|
||||
# Example usage:
|
||||
# extractor = SafeZipExtractor(
|
||||
# archive_path=Path("file.nupkg"),
|
||||
# extract_dir=Path("extract_dir"),
|
||||
# include_patterns=["*.dll", "*.xml"],
|
||||
# exclude_patterns=["*.pdb"]
|
||||
# )
|
||||
# extractor.extract_all()
|
||||
@@ -0,0 +1,103 @@
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
|
||||
from solidlsp.util.zip import SafeZipExtractor
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_zip_file(tmp_path: Path) -> Path:
|
||||
"""Create a temporary ZIP file for testing."""
|
||||
zip_path = tmp_path / "test.zip"
|
||||
with zipfile.ZipFile(zip_path, "w") as zipf:
|
||||
zipf.writestr("file1.txt", "Hello World 1")
|
||||
zipf.writestr("file2.txt", "Hello World 2")
|
||||
zipf.writestr("folder/file3.txt", "Hello World 3")
|
||||
return zip_path
|
||||
|
||||
|
||||
def test_extract_all_success(temp_zip_file: Path, tmp_path: Path) -> None:
|
||||
"""All files should extract without error."""
|
||||
dest_dir = tmp_path / "extracted"
|
||||
extractor = SafeZipExtractor(temp_zip_file, dest_dir, verbose=False)
|
||||
extractor.extract_all()
|
||||
|
||||
assert (dest_dir / "file1.txt").read_text() == "Hello World 1"
|
||||
assert (dest_dir / "file2.txt").read_text() == "Hello World 2"
|
||||
assert (dest_dir / "folder" / "file3.txt").read_text() == "Hello World 3"
|
||||
|
||||
|
||||
def test_include_patterns(temp_zip_file: Path, tmp_path: Path) -> None:
|
||||
"""Only files matching include_patterns should be extracted."""
|
||||
dest_dir = tmp_path / "extracted"
|
||||
extractor = SafeZipExtractor(
|
||||
temp_zip_file, dest_dir, verbose=False, include_patterns=["*.txt"]
|
||||
)
|
||||
extractor.extract_all()
|
||||
|
||||
assert (dest_dir / "file1.txt").exists()
|
||||
assert (dest_dir / "file2.txt").exists()
|
||||
assert (dest_dir / "folder" / "file3.txt").exists()
|
||||
|
||||
|
||||
def test_exclude_patterns(temp_zip_file: Path, tmp_path: Path) -> None:
|
||||
"""Files matching exclude_patterns should be skipped."""
|
||||
dest_dir = tmp_path / "extracted"
|
||||
extractor = SafeZipExtractor(
|
||||
temp_zip_file, dest_dir, verbose=False, exclude_patterns=["file2.txt"]
|
||||
)
|
||||
extractor.extract_all()
|
||||
|
||||
assert (dest_dir / "file1.txt").exists()
|
||||
assert not (dest_dir / "file2.txt").exists()
|
||||
assert (dest_dir / "folder" / "file3.txt").exists()
|
||||
|
||||
|
||||
def test_include_and_exclude_patterns(temp_zip_file: Path, tmp_path: Path) -> None:
|
||||
"""Exclude should override include if both match."""
|
||||
dest_dir = tmp_path / "extracted"
|
||||
extractor = SafeZipExtractor(
|
||||
temp_zip_file,
|
||||
dest_dir,
|
||||
verbose=False,
|
||||
include_patterns=["*.txt"],
|
||||
exclude_patterns=["file1.txt"],
|
||||
)
|
||||
extractor.extract_all()
|
||||
|
||||
assert not (dest_dir / "file1.txt").exists()
|
||||
assert (dest_dir / "file2.txt").exists()
|
||||
assert (dest_dir / "folder" / "file3.txt").exists()
|
||||
|
||||
|
||||
def test_skip_on_error(monkeypatch, temp_zip_file: Path, tmp_path: Path) -> None:
|
||||
"""Should skip a file that raises an error and continue extracting others."""
|
||||
dest_dir = tmp_path / "extracted"
|
||||
|
||||
original_open = zipfile.ZipFile.open
|
||||
|
||||
def failing_open(self, member, *args, **kwargs):
|
||||
if member.filename == "file2.txt":
|
||||
raise IOError("Simulated failure")
|
||||
return original_open(self, member, *args, **kwargs)
|
||||
|
||||
# Patch the method on the class, not on an instance
|
||||
monkeypatch.setattr(zipfile.ZipFile, "open", failing_open)
|
||||
|
||||
extractor = SafeZipExtractor(temp_zip_file, dest_dir, verbose=False)
|
||||
extractor.extract_all()
|
||||
|
||||
assert (dest_dir / "file1.txt").exists()
|
||||
assert not (dest_dir / "file2.txt").exists()
|
||||
assert (dest_dir / "folder" / "file3.txt").exists()
|
||||
|
||||
@pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows-only test")
|
||||
def test_long_path_normalization(temp_zip_file: Path, tmp_path: Path) -> None:
|
||||
"""Ensure _normalize_path adds \\?\ prefix on Windows."""
|
||||
dest_dir = tmp_path / ("a" * 250) # Simulate long path
|
||||
extractor = SafeZipExtractor(temp_zip_file, dest_dir, verbose=False)
|
||||
norm_path = extractor._normalize_path(dest_dir / "file.txt")
|
||||
assert str(norm_path).startswith("\\\\?\\")
|
||||
Reference in new issue
Block a user