Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
33aae43b40 | ||
|
|
148b3b3780 | ||
|
|
8a15c2d7db | ||
|
|
75447562cd | ||
|
|
502b3a6b78 | ||
|
|
e552fca0b2 | ||
|
|
698eb9cf09 | ||
|
|
8e5e1e3271 | ||
|
|
15e27f6aa6 | ||
|
|
a2ef8dd115 |
No files matched your search
+20
-15
@@ -8,20 +8,23 @@ on:
|
||||
branches: [main, develop]
|
||||
tags:
|
||||
- "v*.*.*"
|
||||
paths-ignore:
|
||||
- "*.md"
|
||||
- "LICENSE"
|
||||
- "docs/**"
|
||||
- ".gitignore"
|
||||
paths:
|
||||
- "src/**"
|
||||
- "tests/**"
|
||||
- ".github/workflows/ci.yml"
|
||||
- "pyproject.toml"
|
||||
- "src/main.py"
|
||||
pull_request:
|
||||
branches: [main]
|
||||
paths-ignore:
|
||||
- "*.md"
|
||||
- "LICENSE"
|
||||
- "docs/**"
|
||||
- ".gitignore"
|
||||
paths:
|
||||
- "src/**"
|
||||
- "tests/**"
|
||||
- ".github/workflows/ci.yml"
|
||||
- "pyproject.toml"
|
||||
- "src/main.py"
|
||||
release:
|
||||
types: [published]
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
test:
|
||||
@@ -31,7 +34,7 @@ jobs:
|
||||
strategy:
|
||||
matrix:
|
||||
os: [ubuntu-latest, windows-latest, macos-latest]
|
||||
python-version: ["3.12", "3.13"]
|
||||
python-version: ["3.12", "3.13", "3.14"]
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
@@ -59,8 +62,8 @@ jobs:
|
||||
pytest --cov=tzst --cov-branch --cov-report=xml
|
||||
|
||||
- name: Upload coverage to Codecov
|
||||
if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12'
|
||||
uses: codecov/codecov-action@v5
|
||||
if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12' && github.event_name == 'push' && github.ref == 'refs/heads/main'
|
||||
uses: codecov/codecov-action@v6
|
||||
with:
|
||||
token: ${{ secrets.CODECOV_TOKEN }}
|
||||
slug: xixu-me/tzst
|
||||
@@ -167,7 +170,7 @@ jobs:
|
||||
shell: bash
|
||||
run: |
|
||||
VERSION=${GITHUB_REF#refs/tags/v}
|
||||
echo "version=$VERSION" >> $GITHUB_OUTPUT
|
||||
echo "version=$VERSION" >> "$GITHUB_OUTPUT"
|
||||
echo "Version: $VERSION"
|
||||
|
||||
- name: Install dependencies
|
||||
@@ -190,6 +193,7 @@ jobs:
|
||||
|
||||
- name: Verify binary (Windows)
|
||||
if: matrix.os == 'windows-latest'
|
||||
shell: pwsh
|
||||
run: |
|
||||
if (-not (Test-Path "dist/tzst.exe")) {
|
||||
Write-Host "Binary build failed!"
|
||||
@@ -243,6 +247,7 @@ jobs:
|
||||
|
||||
- name: Create zip archive (Windows)
|
||||
if: matrix.os == 'windows-latest'
|
||||
shell: pwsh
|
||||
run: |
|
||||
cd archive
|
||||
Compress-Archive -Path * -DestinationPath ../tzst-${{ steps.version.outputs.version }}-${{ matrix.os_name }}-${{ matrix.arch }}.zip
|
||||
@@ -267,7 +272,7 @@ jobs:
|
||||
id: version
|
||||
run: |
|
||||
VERSION=${GITHUB_REF#refs/tags/v}
|
||||
echo "version=$VERSION" >> $GITHUB_OUTPUT
|
||||
echo "version=$VERSION" >> "$GITHUB_OUTPUT"
|
||||
|
||||
- name: Download all binary artifacts
|
||||
uses: actions/download-artifact@v8
|
||||
|
||||
+3
-3
@@ -23,7 +23,7 @@ This project follows the principles of respectful collaboration. Please be kind,
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- Python 3.12 or higher
|
||||
- Python 3.12 or higher (tested on 3.12-3.14)
|
||||
- Git
|
||||
- Basic knowledge of tar archives and compression
|
||||
|
||||
@@ -198,7 +198,7 @@ This project uses [Ruff](https://docs.astral.sh/ruff/) for linting and formattin
|
||||
Settings are defined in `pyproject.toml`:
|
||||
|
||||
- Line length: 88 characters
|
||||
- Target Python version: 3.12+
|
||||
- Target Python version: 3.12+ (tested on 3.12-3.14)
|
||||
- Import sorting with isort
|
||||
- Quote style: double quotes
|
||||
|
||||
@@ -356,7 +356,7 @@ When suggesting features:
|
||||
|
||||
### Compatibility
|
||||
|
||||
- Support Python 3.12+
|
||||
- Support Python 3.12+ with CI coverage for 3.12-3.14
|
||||
- Test on multiple platforms (Windows, macOS, Linux)
|
||||
- Consider different filesystem behaviors
|
||||
- Maintain backwards compatibility when possible
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
> [!TIP]
|
||||
> 欢迎加入“Xget 开源与 AI 交流群”,一起交流开源项目、AI 应用、工程实践、效率工具和独立开发;如果你也在做产品、写代码、折腾项目或者对开源和 AI 感兴趣,欢迎[**进群**](https://file.xi-xu.me/QR%20Codes/%E7%BE%A4%E4%BA%8C%E7%BB%B4%E7%A0%81.png)认识更多认真做事、乐于分享的朋友。
|
||||
|
||||
<h1 align="center">
|
||||
<img src="https://raw.githubusercontent.com/xixu-me/tzst/refs/heads/main/docs/_static/tzst-logo.png" width="300">
|
||||
</h1><br>
|
||||
@@ -450,7 +453,7 @@ except KeyboardInterrupt:
|
||||
|
||||
## 📋 Requirements
|
||||
|
||||
- 🐍 Python 3.12 or higher
|
||||
- 🐍 Python 3.12 or higher (tested on 3.12-3.14)
|
||||
- 📦 zstandard >= 0.19.0
|
||||
|
||||
## 🛠️ Development
|
||||
|
||||
+4
-1
@@ -1,3 +1,6 @@
|
||||
> [!TIP]
|
||||
> 欢迎加入“Xget 开源与 AI 交流群”,一起交流开源项目、AI 应用、工程实践、效率工具和独立开发;如果你也在做产品、写代码、折腾项目或者对开源和 AI 感兴趣,欢迎[**进群**](https://file.xi-xu.me/QR%20Codes/%E7%BE%A4%E4%BA%8C%E7%BB%B4%E7%A0%81.png)认识更多认真做事、乐于分享的朋友。
|
||||
|
||||
<h1 align="center">
|
||||
<img src="https://raw.githubusercontent.com/xixu-me/tzst/refs/heads/main/docs/_static/tzst-logo.png" width="300">
|
||||
</h1><br>
|
||||
@@ -448,7 +451,7 @@ except KeyboardInterrupt:
|
||||
|
||||
## 📋 系统要求
|
||||
|
||||
- 🐍 Python 3.12 或更高版本
|
||||
- 🐍 Python 3.12 或更高版本(已测试 3.12-3.14)
|
||||
- 📦 zstandard >= 0.19.0
|
||||
|
||||
## 🛠️ 开发指南
|
||||
|
||||
+2
-2
@@ -116,7 +116,7 @@ ruff format --check src tests
|
||||
Settings are defined in `pyproject.toml`:
|
||||
|
||||
- Line length: 88 characters
|
||||
- Target Python version: 3.12+
|
||||
- Target Python version: 3.12+ (tested on 3.12-3.14)
|
||||
- Import sorting with isort
|
||||
- Quote style: double quotes
|
||||
|
||||
@@ -315,7 +315,7 @@ docs(readme): update installation instructions
|
||||
|
||||
### Compatibility
|
||||
|
||||
- Support Python 3.12+
|
||||
- Support Python 3.12+ with CI coverage for 3.12-3.14
|
||||
- Test on multiple platforms (Windows, macOS, Linux)
|
||||
- Consider different filesystem behaviors
|
||||
- Maintain backwards compatibility when possible
|
||||
|
||||
+1
-1
@@ -183,5 +183,5 @@ Licensed under the [BSD 3-Clause](https://github.com/xixu-me/tzst/blob/main/LICE
|
||||
|
||||
## Requirements
|
||||
|
||||
- Python 3.12 or higher
|
||||
- Python 3.12 or higher (tested on 3.12-3.14)
|
||||
- zstandard >= 0.19.0
|
||||
@@ -18,6 +18,7 @@ classifiers = [
|
||||
"Operating System :: OS Independent",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Programming Language :: Python :: 3.14",
|
||||
"Topic :: System :: Archiving :: Compression",
|
||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||
]
|
||||
|
||||
@@ -4,7 +4,7 @@ Leveraging cutting-edge Zstandard compression to deliver superior performance,
|
||||
security, and reliability.
|
||||
"""
|
||||
|
||||
__version__ = "1.3.2"
|
||||
__version__ = "1.3.3"
|
||||
|
||||
from .core import (
|
||||
TzstArchive,
|
||||
|
||||
@@ -0,0 +1,246 @@
|
||||
"""Targeted tests for remaining uncovered CLI branches."""
|
||||
|
||||
import json
|
||||
import runpy
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from tzst import create_archive
|
||||
from tzst.cli import (
|
||||
_interactive_conflict_callback,
|
||||
_is_extreme_compression_level_in_argv,
|
||||
_normalize_archive_path,
|
||||
_validate_command_in_argv,
|
||||
_validate_compression_level_in_argv,
|
||||
_validate_filter_in_argv,
|
||||
cmd_extract_flat,
|
||||
cmd_extract_full,
|
||||
cmd_version,
|
||||
main,
|
||||
)
|
||||
|
||||
|
||||
class TestCliCoverageGaps:
|
||||
"""Exercise the remaining CLI coverage hotspots."""
|
||||
|
||||
def test_normalize_archive_path_handles_tar_and_custom_extensions(self):
|
||||
assert _normalize_archive_path(Path("bundle.tar")) == Path("bundle.tar.zst")
|
||||
assert _normalize_archive_path(Path("bundle.backup")) == Path(
|
||||
"bundle.backup.tzst"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("choice", "expected"),
|
||||
[
|
||||
("A", "replace_all"),
|
||||
("S", "skip_all"),
|
||||
("U", "auto_rename_all"),
|
||||
],
|
||||
)
|
||||
def test_interactive_conflict_callback_covers_remaining_choices(
|
||||
self, monkeypatch, temp_dir, choice, expected
|
||||
):
|
||||
monkeypatch.setattr("builtins.input", lambda _prompt: choice)
|
||||
|
||||
result = _interactive_conflict_callback(temp_dir / "existing.txt")
|
||||
|
||||
assert result.value == expected
|
||||
|
||||
def test_extract_full_rejects_json_interactive_conflicts(self, temp_dir):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
|
||||
args = SimpleNamespace(
|
||||
archive=str(archive_path),
|
||||
output=None,
|
||||
files=[],
|
||||
streaming=False,
|
||||
filter="data",
|
||||
conflict_resolution="replace",
|
||||
interactive=True,
|
||||
json_output=True,
|
||||
no_banner=True,
|
||||
)
|
||||
|
||||
assert cmd_extract_full(args) == 1
|
||||
|
||||
def test_extract_full_json_success_payload(self, temp_dir, capsys):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
output_dir = temp_dir / "extract"
|
||||
|
||||
result = main(
|
||||
[
|
||||
"--json",
|
||||
"x",
|
||||
str(archive_path),
|
||||
"-o",
|
||||
str(output_dir),
|
||||
"--conflict-resolution",
|
||||
"replace",
|
||||
]
|
||||
)
|
||||
|
||||
assert result == 0
|
||||
payload = json.loads(capsys.readouterr().out)
|
||||
assert payload["command"] == "extract"
|
||||
assert payload["flatten"] is False
|
||||
|
||||
def test_extract_full_handles_file_not_found_from_backend(
|
||||
self, temp_dir, monkeypatch
|
||||
):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
|
||||
def fail_extract(*args, **kwargs):
|
||||
raise FileNotFoundError("backend missing file")
|
||||
|
||||
monkeypatch.setattr("tzst.cli.extract_archive", fail_extract)
|
||||
|
||||
assert main(["x", str(archive_path)]) == 1
|
||||
|
||||
def test_extract_flat_rejects_json_interactive_conflicts(self, temp_dir):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
|
||||
args = SimpleNamespace(
|
||||
archive=str(archive_path),
|
||||
output=None,
|
||||
files=[],
|
||||
streaming=False,
|
||||
filter="data",
|
||||
conflict_resolution="replace",
|
||||
interactive=True,
|
||||
json_output=True,
|
||||
no_banner=True,
|
||||
)
|
||||
|
||||
assert cmd_extract_flat(args) == 1
|
||||
|
||||
def test_extract_flat_json_success_payload(self, temp_dir, capsys):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
output_dir = temp_dir / "extract"
|
||||
|
||||
result = main(
|
||||
[
|
||||
"--json",
|
||||
"e",
|
||||
str(archive_path),
|
||||
"-o",
|
||||
str(output_dir),
|
||||
"--conflict-resolution",
|
||||
"replace",
|
||||
]
|
||||
)
|
||||
|
||||
assert result == 0
|
||||
payload = json.loads(capsys.readouterr().out)
|
||||
assert payload["command"] == "extract-flat"
|
||||
assert payload["flatten"] is True
|
||||
|
||||
def test_extract_flat_handles_file_not_found_from_backend(
|
||||
self, temp_dir, monkeypatch
|
||||
):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
|
||||
def fail_extract(*args, **kwargs):
|
||||
raise FileNotFoundError("backend missing file")
|
||||
|
||||
monkeypatch.setattr("tzst.cli.extract_archive", fail_extract)
|
||||
|
||||
assert main(["e", str(archive_path)]) == 1
|
||||
|
||||
def test_list_handles_file_not_found_from_backend(self, temp_dir, monkeypatch):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
|
||||
def fail_list(*args, **kwargs):
|
||||
raise FileNotFoundError("backend missing file")
|
||||
|
||||
monkeypatch.setattr("tzst.cli.list_archive", fail_list)
|
||||
|
||||
assert main(["l", str(archive_path)]) == 1
|
||||
|
||||
def test_test_command_json_success_and_file_not_found_backend(
|
||||
self, temp_dir, capsys, monkeypatch
|
||||
):
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("content")
|
||||
create_archive(archive_path, [source_file], use_temp_file=False)
|
||||
|
||||
result = main(["--json", "t", str(archive_path)])
|
||||
assert result == 0
|
||||
payload = json.loads(capsys.readouterr().out)
|
||||
assert payload["command"] == "test"
|
||||
assert payload["healthy"] is True
|
||||
|
||||
def fail_test(*args, **kwargs):
|
||||
raise FileNotFoundError("backend missing file")
|
||||
|
||||
monkeypatch.setattr("tzst.cli.test_archive", fail_test)
|
||||
assert main(["t", str(archive_path)]) == 1
|
||||
|
||||
def test_cmd_version_prints_when_no_banner_requested(self, capsys):
|
||||
args = SimpleNamespace(json_output=False, no_banner=True)
|
||||
|
||||
assert cmd_version(args) == 0
|
||||
assert capsys.readouterr().out.startswith("tzst ")
|
||||
|
||||
def test_validate_compression_level_in_argv_covers_short_flag_and_index_error(
|
||||
self, capsys
|
||||
):
|
||||
assert _validate_compression_level_in_argv(["-c", "0"]) is True
|
||||
assert "Invalid compression level: 0" in capsys.readouterr().err
|
||||
assert _validate_compression_level_in_argv(["-c"]) is False
|
||||
|
||||
def test_validate_filter_in_argv_handles_missing_value(self):
|
||||
assert _validate_filter_in_argv(["--filter"]) is False
|
||||
|
||||
def test_validate_command_in_argv_handles_empty_and_invalid_input(self, capsys):
|
||||
assert _validate_command_in_argv([]) is False
|
||||
assert _validate_command_in_argv(["bogus"]) is True
|
||||
assert "Invalid command: 'bogus'" in capsys.readouterr().err
|
||||
|
||||
def test_is_extreme_compression_level_in_argv_covers_all_remaining_paths(self):
|
||||
assert _is_extreme_compression_level_in_argv([]) is False
|
||||
assert _is_extreme_compression_level_in_argv(["-c", "1000"]) is True
|
||||
assert _is_extreme_compression_level_in_argv(["-l", "abc"]) is False
|
||||
assert _is_extreme_compression_level_in_argv(["--level"]) is False
|
||||
|
||||
def test_cli_module_main_guard_executes(self, monkeypatch):
|
||||
monkeypatch.delitem(sys.modules, "tzst.cli", raising=False)
|
||||
monkeypatch.setattr(sys, "argv", ["tzst", "--version"])
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
runpy.run_module("tzst.cli", run_name="__main__")
|
||||
|
||||
assert exc_info.value.code == 0
|
||||
|
||||
def test_package_main_module_executes(self, monkeypatch):
|
||||
monkeypatch.delitem(sys.modules, "tzst.__main__", raising=False)
|
||||
monkeypatch.setattr(sys, "argv", ["python", "--version"])
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
runpy.run_module("tzst.__main__", run_name="__main__")
|
||||
|
||||
assert exc_info.value.code == 0
|
||||
@@ -0,0 +1,440 @@
|
||||
"""Targeted tests for remaining uncovered core branches."""
|
||||
|
||||
import os
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
import tzst.core as core_module
|
||||
from tzst.core import (
|
||||
ConflictResolution,
|
||||
ConflictResolutionState,
|
||||
TzstArchive,
|
||||
_get_unique_filename,
|
||||
_move_file_cross_platform,
|
||||
create_archive,
|
||||
extract_archive,
|
||||
)
|
||||
from tzst.exceptions import TzstArchiveError, TzstDecompressionError
|
||||
|
||||
|
||||
def _make_read_archive() -> TzstArchive:
|
||||
archive = TzstArchive("dummy.tzst", "r")
|
||||
archive._tarfile = Mock()
|
||||
archive.mode = "r"
|
||||
return archive
|
||||
|
||||
|
||||
def _make_write_archive() -> TzstArchive:
|
||||
archive = TzstArchive("dummy.tzst", "w")
|
||||
archive._tarfile = Mock()
|
||||
archive.mode = "w"
|
||||
return archive
|
||||
|
||||
|
||||
class TestCoreCoverageGaps:
|
||||
"""Exercise the remaining core coverage hotspots."""
|
||||
|
||||
def test_conflict_resolution_state_apply_to_all(self):
|
||||
state = ConflictResolutionState(ConflictResolution.AUTO_RENAME_ALL)
|
||||
assert state.apply_to_all is True
|
||||
|
||||
state = ConflictResolutionState(ConflictResolution.REPLACE)
|
||||
assert state.apply_to_all is False
|
||||
|
||||
def test_get_unique_filename_returns_original_for_missing_path(self, temp_dir):
|
||||
missing_path = temp_dir / "missing.txt"
|
||||
assert _get_unique_filename(missing_path) == missing_path
|
||||
|
||||
def test_move_file_cross_platform_falls_back_to_copy_and_delete(
|
||||
self, temp_dir, monkeypatch
|
||||
):
|
||||
src = temp_dir / "source.txt"
|
||||
dst = temp_dir / "target.txt"
|
||||
src.write_text("cross-drive content")
|
||||
|
||||
def fail_rename(self, target):
|
||||
raise OSError("cross-device link")
|
||||
|
||||
monkeypatch.setattr(Path, "rename", fail_rename)
|
||||
|
||||
_move_file_cross_platform(src, dst)
|
||||
|
||||
assert dst.read_text() == "cross-drive content"
|
||||
assert not src.exists()
|
||||
|
||||
def test_move_file_cross_platform_reraises_original_rename_error(
|
||||
self, temp_dir, monkeypatch
|
||||
):
|
||||
src = temp_dir / "source.txt"
|
||||
dst = temp_dir / "target.txt"
|
||||
src.write_text("cross-drive content")
|
||||
|
||||
def fail_rename(self, target):
|
||||
raise OSError("rename failed")
|
||||
|
||||
def fail_copy(*args, **kwargs):
|
||||
raise RuntimeError("copy failed")
|
||||
|
||||
monkeypatch.setattr(Path, "rename", fail_rename)
|
||||
monkeypatch.setattr("shutil.copy2", fail_copy)
|
||||
|
||||
with pytest.raises(OSError, match="rename failed"):
|
||||
_move_file_cross_platform(src, dst)
|
||||
|
||||
def test_exit_suppresses_close_errors(self, monkeypatch):
|
||||
archive = TzstArchive("dummy.tzst", "w")
|
||||
monkeypatch.setattr(archive, "close", Mock(side_effect=RuntimeError("boom")))
|
||||
|
||||
archive.__exit__(None, None, None)
|
||||
|
||||
def test_open_wraps_zstd_failures(self, temp_dir, monkeypatch):
|
||||
archive_path = temp_dir / "broken.tzst"
|
||||
archive_path.write_bytes(b"not-a-valid-archive")
|
||||
|
||||
class BrokenDecompressor:
|
||||
def stream_reader(self, fileobj):
|
||||
raise RuntimeError("zstd decoder exploded")
|
||||
|
||||
monkeypatch.setattr(core_module.zstd, "ZstdDecompressor", BrokenDecompressor)
|
||||
|
||||
with pytest.raises(TzstDecompressionError, match="zstd decoder exploded"):
|
||||
TzstArchive(archive_path, "r").open()
|
||||
|
||||
def test_open_wraps_append_mode_when_mode_changes_after_init(self):
|
||||
archive = TzstArchive("dummy.tzst", "w")
|
||||
archive.mode = "a"
|
||||
|
||||
with pytest.raises(
|
||||
TzstArchiveError, match="Append mode is not currently supported"
|
||||
):
|
||||
archive.open()
|
||||
|
||||
def test_open_wraps_invalid_mode_when_mode_changes_after_init(self):
|
||||
archive = TzstArchive("dummy.tzst", "w")
|
||||
archive.mode = "invalid"
|
||||
|
||||
with pytest.raises(TzstArchiveError, match="Invalid mode: invalid"):
|
||||
archive.open()
|
||||
|
||||
def test_add_requires_open_archive(self, temp_dir):
|
||||
archive = TzstArchive(temp_dir / "archive.tzst", "w")
|
||||
|
||||
with pytest.raises(RuntimeError, match="Archive not open"):
|
||||
archive.add(temp_dir / "file.txt")
|
||||
|
||||
def test_add_requires_write_mode(self, temp_dir):
|
||||
archive = _make_write_archive()
|
||||
archive.mode = "r"
|
||||
file_path = temp_dir / "file.txt"
|
||||
file_path.write_text("content")
|
||||
|
||||
with pytest.raises(RuntimeError, match="Archive not open for writing"):
|
||||
archive.add(file_path)
|
||||
|
||||
def test_add_raises_file_not_found_for_missing_input(self, temp_dir):
|
||||
archive = _make_write_archive()
|
||||
|
||||
with pytest.raises(FileNotFoundError, match="File not found"):
|
||||
archive.add(temp_dir / "missing.txt")
|
||||
|
||||
def test_add_wraps_permission_errors(self, temp_dir):
|
||||
archive = _make_write_archive()
|
||||
archive._tarfile.add.side_effect = PermissionError("denied")
|
||||
file_path = temp_dir / "file.txt"
|
||||
file_path.write_text("content")
|
||||
|
||||
with pytest.raises(TzstArchiveError, match="Failed to add"):
|
||||
archive.add(file_path)
|
||||
|
||||
def test_extract_validates_open_state_and_mode(self, temp_dir):
|
||||
closed_archive = TzstArchive(temp_dir / "archive.tzst", "r")
|
||||
with pytest.raises(RuntimeError, match="Archive not open"):
|
||||
closed_archive.extract()
|
||||
|
||||
wrong_mode_archive = _make_write_archive()
|
||||
with pytest.raises(RuntimeError, match="Archive not open for reading"):
|
||||
wrong_mode_archive.extract()
|
||||
|
||||
def test_extract_rejects_specific_members_in_streaming_mode(self):
|
||||
archive = _make_read_archive()
|
||||
archive.streaming = True
|
||||
|
||||
with pytest.raises(RuntimeError, match="specific members is not supported"):
|
||||
archive.extract("file.txt")
|
||||
|
||||
def test_extract_passes_member_specific_kwargs(self, temp_dir):
|
||||
archive = _make_read_archive()
|
||||
output_dir = temp_dir / "extract"
|
||||
|
||||
archive.extract(
|
||||
"nested/file.txt",
|
||||
output_dir,
|
||||
set_attrs=False,
|
||||
numeric_owner=True,
|
||||
filter="tar",
|
||||
)
|
||||
|
||||
archive._tarfile.extract.assert_called_once_with(
|
||||
"nested/file.txt",
|
||||
path=output_dir,
|
||||
set_attrs=False,
|
||||
numeric_owner=True,
|
||||
filter="tar",
|
||||
)
|
||||
|
||||
def test_extract_wraps_streaming_structure_errors(self, temp_dir):
|
||||
archive = _make_read_archive()
|
||||
archive.streaming = True
|
||||
archive._tarfile.extractall.side_effect = tarfile.StreamError(
|
||||
"stream seeking failed"
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="Extraction failed in streaming mode"):
|
||||
archive.extract(path=temp_dir / "extract")
|
||||
|
||||
def test_extract_reraises_non_streaming_errors(self, temp_dir):
|
||||
archive = _make_read_archive()
|
||||
archive._tarfile.extractall.side_effect = OSError("disk full")
|
||||
|
||||
with pytest.raises(OSError, match="disk full"):
|
||||
archive.extract(path=temp_dir / "extract")
|
||||
|
||||
def test_extractall_validates_open_state_and_members_behavior(self, temp_dir):
|
||||
closed_archive = TzstArchive(temp_dir / "archive.tzst", "r")
|
||||
with pytest.raises(RuntimeError, match="Archive not open"):
|
||||
closed_archive.extractall()
|
||||
|
||||
wrong_mode_archive = _make_write_archive()
|
||||
with pytest.raises(RuntimeError, match="Archive not open for reading"):
|
||||
wrong_mode_archive.extractall()
|
||||
|
||||
streaming_archive = _make_read_archive()
|
||||
streaming_archive.streaming = True
|
||||
with pytest.raises(RuntimeError, match="specific members is not supported"):
|
||||
streaming_archive.extractall(members=[Mock()])
|
||||
|
||||
def test_extractall_passes_members_and_reraises_other_errors(self, temp_dir):
|
||||
archive = _make_read_archive()
|
||||
members = [SimpleNamespace(name="file.txt")]
|
||||
output_dir = temp_dir / "extract"
|
||||
|
||||
archive.extractall(
|
||||
output_dir, members=members, numeric_owner=True, filter="tar"
|
||||
)
|
||||
|
||||
archive._tarfile.extractall.assert_called_once_with(
|
||||
path=output_dir,
|
||||
numeric_owner=True,
|
||||
filter="tar",
|
||||
members=members,
|
||||
)
|
||||
|
||||
archive = _make_read_archive()
|
||||
archive._tarfile.extractall.side_effect = OSError("permission denied")
|
||||
with pytest.raises(OSError, match="permission denied"):
|
||||
archive.extractall(output_dir)
|
||||
|
||||
def test_getnames_list_and_test_runtime_paths(self):
|
||||
archive = _make_read_archive()
|
||||
archive._tarfile.getnames.return_value = ["a.txt"]
|
||||
assert archive.getnames() == ["a.txt"]
|
||||
|
||||
closed_archive = TzstArchive("dummy.tzst", "r")
|
||||
with pytest.raises(RuntimeError, match="Archive not open"):
|
||||
closed_archive.list()
|
||||
|
||||
wrong_mode_archive = _make_write_archive()
|
||||
with pytest.raises(RuntimeError, match="Archive not open for reading"):
|
||||
wrong_mode_archive.list()
|
||||
|
||||
failing_archive = _make_read_archive()
|
||||
failing_archive.getmembers = Mock(side_effect=RuntimeError("boom"))
|
||||
assert failing_archive.test() is False
|
||||
|
||||
def test_create_archive_supports_tar_extension(self, temp_dir):
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("archive me")
|
||||
|
||||
create_archive(temp_dir / "bundle.tar", [source_file], use_temp_file=False)
|
||||
|
||||
assert (temp_dir / "bundle.tar.zst").exists()
|
||||
|
||||
def test_create_archive_ignores_unlink_cleanup_failures(
|
||||
self, temp_dir, monkeypatch
|
||||
):
|
||||
source_file = temp_dir / "source.txt"
|
||||
source_file.write_text("archive me")
|
||||
archive_path = temp_dir / "broken.tzst"
|
||||
temp_path = temp_dir / ".broken.tzst.tmp"
|
||||
|
||||
def fake_mkstemp(*args, **kwargs):
|
||||
fd = os.open(temp_path, os.O_CREAT | os.O_RDWR)
|
||||
return fd, str(temp_path)
|
||||
|
||||
def fail_unlink(self):
|
||||
raise OSError("locked")
|
||||
|
||||
monkeypatch.setattr(core_module.tempfile, "mkstemp", fake_mkstemp)
|
||||
monkeypatch.setattr(
|
||||
core_module, "_create_archive_impl", Mock(side_effect=RuntimeError("boom"))
|
||||
)
|
||||
monkeypatch.setattr(Path, "unlink", fail_unlink)
|
||||
|
||||
with pytest.raises(RuntimeError, match="boom"):
|
||||
create_archive(archive_path, [source_file])
|
||||
|
||||
assert temp_path.exists()
|
||||
|
||||
def test_extract_archive_handles_flatten_members_and_invalid_resolution(
|
||||
self, temp_dir
|
||||
):
|
||||
file_path = temp_dir / "source.txt"
|
||||
file_path.write_text("content")
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
create_archive(archive_path, [file_path], use_temp_file=False)
|
||||
|
||||
output_dir = temp_dir / "extract"
|
||||
extract_archive(
|
||||
archive_path,
|
||||
output_dir,
|
||||
members=["source.txt"],
|
||||
flatten=True,
|
||||
conflict_resolution="definitely-invalid",
|
||||
)
|
||||
|
||||
assert (output_dir / "source.txt").read_text() == "content"
|
||||
|
||||
def test_extract_archive_flatten_respects_initial_exit_resolution(self, temp_dir):
|
||||
file_path = temp_dir / "source.txt"
|
||||
file_path.write_text("content")
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
create_archive(archive_path, [file_path], use_temp_file=False)
|
||||
|
||||
output_dir = temp_dir / "extract"
|
||||
extract_archive(
|
||||
archive_path,
|
||||
output_dir,
|
||||
flatten=True,
|
||||
conflict_resolution=ConflictResolution.EXIT,
|
||||
)
|
||||
|
||||
assert not (output_dir / "source.txt").exists()
|
||||
|
||||
def test_extract_archive_flatten_conflict_skip_exit_and_auto_rename(self, temp_dir):
|
||||
first = temp_dir / "first.txt"
|
||||
second = temp_dir / "second.txt"
|
||||
first.write_text("first")
|
||||
second.write_text("second")
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
create_archive(archive_path, [first, second], use_temp_file=False)
|
||||
|
||||
skip_output = temp_dir / "skip"
|
||||
skip_output.mkdir()
|
||||
(skip_output / "first.txt").write_text("existing")
|
||||
extract_archive(
|
||||
archive_path,
|
||||
skip_output,
|
||||
members=["first.txt"],
|
||||
flatten=True,
|
||||
conflict_resolution=ConflictResolution.SKIP,
|
||||
)
|
||||
assert (skip_output / "first.txt").read_text() == "existing"
|
||||
|
||||
rename_output = temp_dir / "rename"
|
||||
rename_output.mkdir()
|
||||
(rename_output / "first.txt").write_text("existing")
|
||||
extract_archive(
|
||||
archive_path,
|
||||
rename_output,
|
||||
members=["first.txt"],
|
||||
flatten=True,
|
||||
conflict_resolution=ConflictResolution.AUTO_RENAME,
|
||||
)
|
||||
assert (rename_output / "first.txt").read_text() == "existing"
|
||||
assert (rename_output / "first_1.txt").read_text() == "first"
|
||||
|
||||
exit_output = temp_dir / "exit"
|
||||
exit_output.mkdir()
|
||||
(exit_output / "first.txt").write_text("existing")
|
||||
extract_archive(
|
||||
archive_path,
|
||||
exit_output,
|
||||
members=["first.txt", "second.txt"],
|
||||
flatten=True,
|
||||
conflict_resolution=ConflictResolution.ASK,
|
||||
interactive_callback=lambda _path: ConflictResolution.EXIT,
|
||||
)
|
||||
assert (exit_output / "first.txt").read_text() == "existing"
|
||||
assert not (exit_output / "second.txt").exists()
|
||||
|
||||
def test_extract_archive_members_support_auto_rename_and_plain_extract(
|
||||
self, temp_dir
|
||||
):
|
||||
source_root = temp_dir / "source"
|
||||
nested_dir = source_root / "nested"
|
||||
nested_dir.mkdir(parents=True)
|
||||
conflict_file = nested_dir / "conflict.txt"
|
||||
plain_file = source_root / "plain.txt"
|
||||
conflict_file.write_text("conflict-content")
|
||||
plain_file.write_text("plain-content")
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
create_archive(archive_path, [conflict_file, plain_file], use_temp_file=False)
|
||||
|
||||
output_dir = temp_dir / "extract"
|
||||
output_dir.mkdir()
|
||||
target_conflict = output_dir / "nested" / "conflict.txt"
|
||||
target_conflict.parent.mkdir(parents=True, exist_ok=True)
|
||||
target_conflict.write_text("existing")
|
||||
|
||||
extract_archive(
|
||||
archive_path,
|
||||
output_dir,
|
||||
members=["nested/conflict.txt", "plain.txt"],
|
||||
flatten=False,
|
||||
conflict_resolution=ConflictResolution.AUTO_RENAME,
|
||||
)
|
||||
|
||||
assert target_conflict.read_text() == "existing"
|
||||
assert (
|
||||
output_dir / "nested" / "conflict_1.txt"
|
||||
).read_text() == "conflict-content"
|
||||
assert (output_dir / "plain.txt").read_text() == "plain-content"
|
||||
|
||||
def test_extract_archive_members_honor_initial_exit_state(self, temp_dir):
|
||||
file_path = temp_dir / "source.txt"
|
||||
file_path.write_text("content")
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
create_archive(archive_path, [file_path], use_temp_file=False)
|
||||
|
||||
output_dir = temp_dir / "extract"
|
||||
extract_archive(
|
||||
archive_path,
|
||||
output_dir,
|
||||
members=["source.txt"],
|
||||
conflict_resolution=ConflictResolution.EXIT,
|
||||
)
|
||||
|
||||
assert not (output_dir / "source.txt").exists()
|
||||
|
||||
def test_extract_archive_extractall_breaks_on_exit_conflict(self, temp_dir):
|
||||
file_path = temp_dir / "source.txt"
|
||||
file_path.write_text("content")
|
||||
archive_path = temp_dir / "archive.tzst"
|
||||
create_archive(archive_path, [file_path], use_temp_file=False)
|
||||
|
||||
output_dir = temp_dir / "extract"
|
||||
output_dir.mkdir()
|
||||
(output_dir / "source.txt").write_text("existing")
|
||||
|
||||
extract_archive(
|
||||
archive_path,
|
||||
output_dir,
|
||||
conflict_resolution=ConflictResolution.ASK,
|
||||
interactive_callback=lambda _path: ConflictResolution.EXIT,
|
||||
)
|
||||
|
||||
assert (output_dir / "source.txt").read_text() == "existing"
|
||||
@@ -0,0 +1,22 @@
|
||||
"""Project metadata and CI support tests."""
|
||||
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def test_pyproject_declares_python_314_classifier():
|
||||
"""Package metadata should advertise Python 3.14 support."""
|
||||
pyproject = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text("utf-8"))
|
||||
|
||||
classifiers = pyproject["project"]["classifiers"]
|
||||
|
||||
assert "Programming Language :: Python :: 3.14" in classifiers
|
||||
|
||||
|
||||
def test_ci_matrix_includes_python_314():
|
||||
"""Main test workflow should run on Python 3.14."""
|
||||
workflow = (REPO_ROOT / ".github" / "workflows" / "ci.yml").read_text("utf-8")
|
||||
|
||||
assert 'python-version: ["3.12", "3.13", "3.14"]' in workflow
|
||||
Reference in new issue
Block a user