xixu-me
4e0ec638cc
Introduced `_is_extreme_compression_level_in_argv` to detect extreme compression levels (>=1000) and updated `_handle_parsing_errors` to return exit code 1 for such cases. Adjusted related tests to validate the new behavior and fixed inconsistencies in test assertions for compression level and filter validation.
1783 lines
65 KiB
Python
1783 lines
65 KiB
Python
"""Comprehensive tests for tzst CLI interface.
|
|
|
|
This file consolidates all CLI-related tests to eliminate duplication
|
|
and provide a single source of truth for CLI testing.
|
|
"""
|
|
|
|
import argparse
|
|
|
|
import pytest
|
|
|
|
from tzst.cli import (
|
|
create_parser,
|
|
format_size,
|
|
main,
|
|
print_banner,
|
|
validate_compression_level,
|
|
)
|
|
|
|
|
|
class TestUtilityFunctions:
|
|
"""Test CLI utility functions."""
|
|
|
|
def test_format_size_bytes(self):
|
|
"""Test format_size with byte values."""
|
|
assert format_size(0) == " 0.0 B"
|
|
assert format_size(1) == " 1.0 B"
|
|
assert format_size(999) == " 999.0 B"
|
|
|
|
def test_format_size_kilobytes(self):
|
|
"""Test format_size with kilobyte values."""
|
|
assert format_size(1024) == " 1.0 KB"
|
|
assert format_size(1536) == " 1.5 KB"
|
|
assert format_size(2048) == " 2.0 KB"
|
|
|
|
def test_format_size_megabytes(self):
|
|
"""Test format_size with megabyte values."""
|
|
assert format_size(1024 * 1024) == " 1.0 MB"
|
|
mb_1_5 = int(1.5 * 1024 * 1024)
|
|
assert format_size(mb_1_5) == " 1.5 MB"
|
|
|
|
def test_format_size_gigabytes(self):
|
|
"""Test format_size with gigabyte values."""
|
|
assert format_size(1024 * 1024 * 1024) == " 1.0 GB"
|
|
gb_2_5 = int(2.5 * 1024 * 1024 * 1024)
|
|
assert format_size(gb_2_5) == " 2.5 GB"
|
|
|
|
def test_format_size_terabytes(self):
|
|
"""Test format_size with terabyte values."""
|
|
tb_size = 1024 * 1024 * 1024 * 1024
|
|
assert format_size(tb_size) == " 1.0 TB"
|
|
|
|
def test_format_size_petabytes(self):
|
|
"""Test format_size with very large values (petabytes)."""
|
|
huge_size = 1024 * 1024 * 1024 * 1024 * 1024 * 2
|
|
result = format_size(huge_size)
|
|
assert result.endswith(" PB")
|
|
assert "2.0" in result
|
|
|
|
def test_print_banner_output(self, capsys):
|
|
"""Test print_banner function output."""
|
|
print_banner()
|
|
captured = capsys.readouterr()
|
|
|
|
# Should print empty line, version line, empty line
|
|
lines = captured.out.split("\n")
|
|
assert len(lines) >= 3
|
|
assert lines[0] == "" # Empty line
|
|
assert "tzst" in lines[1]
|
|
assert "Copyright" in lines[1]
|
|
assert lines[2] == "" # Empty line
|
|
|
|
def test_validate_compression_level_valid(self):
|
|
"""Test validate_compression_level with valid levels."""
|
|
assert validate_compression_level("1") == 1
|
|
assert validate_compression_level("11") == 11
|
|
assert validate_compression_level("22") == 22
|
|
|
|
def test_validate_compression_level_invalid_range(self):
|
|
"""Test validate_compression_level with out-of-range values."""
|
|
with pytest.raises(
|
|
argparse.ArgumentTypeError, match="Invalid compression level: 0"
|
|
):
|
|
validate_compression_level("0")
|
|
|
|
with pytest.raises(
|
|
argparse.ArgumentTypeError, match="Invalid compression level: 23"
|
|
):
|
|
validate_compression_level("23")
|
|
|
|
def test_validate_compression_level_invalid_format(self):
|
|
"""Test validate_compression_level with non-numeric values."""
|
|
with pytest.raises(
|
|
argparse.ArgumentTypeError, match="Invalid compression level: 'abc'"
|
|
):
|
|
validate_compression_level("abc")
|
|
|
|
with pytest.raises(
|
|
argparse.ArgumentTypeError, match="Invalid compression level: '1.5'"
|
|
):
|
|
validate_compression_level("1.5")
|
|
|
|
|
|
class TestCLIParser:
|
|
"""Test CLI argument parsing."""
|
|
|
|
def test_parser_creation(self):
|
|
"""Test that parser can be created."""
|
|
parser = create_parser()
|
|
assert parser is not None
|
|
|
|
def test_add_command_parsing(self):
|
|
"""Test parsing of add command."""
|
|
parser = create_parser()
|
|
args = parser.parse_args(["a", "test.tzst", "file1.txt", "file2.txt"])
|
|
|
|
assert args.command == "a"
|
|
assert args.archive == "test.tzst"
|
|
assert args.files == ["file1.txt", "file2.txt"]
|
|
assert args.compression_level == 3 # default
|
|
|
|
def test_extract_command_parsing(self):
|
|
"""Test parsing of extract command."""
|
|
parser = create_parser()
|
|
args = parser.parse_args(["x", "test.tzst", "-o", "output"])
|
|
|
|
assert args.command == "x"
|
|
assert args.archive == "test.tzst"
|
|
assert args.output == "output"
|
|
|
|
def test_list_command_parsing(self):
|
|
"""Test parsing of list command."""
|
|
parser = create_parser()
|
|
args = parser.parse_args(["l", "test.tzst", "-v"])
|
|
|
|
assert args.command == "l"
|
|
assert args.archive == "test.tzst"
|
|
assert args.verbose is True
|
|
|
|
def test_test_command_parsing(self):
|
|
"""Test parsing of test command."""
|
|
parser = create_parser()
|
|
args = parser.parse_args(["t", "test.tzst"])
|
|
|
|
assert args.command == "t"
|
|
assert args.archive == "test.tzst"
|
|
|
|
|
|
class TestCLICommands:
|
|
"""Test CLI command execution."""
|
|
|
|
def test_add_command(self, sample_files, temp_dir):
|
|
"""Test add command functionality."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Run add command
|
|
result = main(["a", str(archive_path), *file_paths])
|
|
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
def test_list_command(self, sample_files, temp_dir):
|
|
"""Test list command functionality."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Create archive first
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Run list command
|
|
result = main(["l", str(archive_path)])
|
|
|
|
assert result == 0
|
|
|
|
def test_extract_command(self, sample_files, temp_dir):
|
|
"""Test extract command functionality."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
extract_dir = temp_dir / "extracted"
|
|
|
|
# Create archive first
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Run extract command
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
|
|
assert result == 0
|
|
assert extract_dir.exists()
|
|
|
|
def test_test_command(self, sample_files, temp_dir):
|
|
"""Test test command functionality."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Create archive first
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Run test command
|
|
result = main(["t", str(archive_path)])
|
|
|
|
assert result == 0
|
|
|
|
|
|
class TestCLIErrorHandling:
|
|
"""Test CLI error handling."""
|
|
|
|
def test_missing_archive(self, temp_dir):
|
|
"""Test handling of missing archive file."""
|
|
fake_archive = temp_dir / "fake.tzst"
|
|
|
|
result = main(["l", str(fake_archive)])
|
|
assert result == 1
|
|
|
|
def test_missing_files_to_add(self, temp_dir):
|
|
"""Test handling of missing files to add."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
fake_file = temp_dir / "fake.txt"
|
|
|
|
result = main(["a", str(archive_path), str(fake_file)])
|
|
assert result == 1
|
|
|
|
def test_no_command(self):
|
|
"""Test handling of no command provided."""
|
|
result = main([])
|
|
assert result == 1
|
|
|
|
def test_keyboard_interrupt_handling(self, temp_dir):
|
|
"""Test that KeyboardInterrupt is handled properly."""
|
|
# This test is more conceptual since we can't easily simulate KeyboardInterrupt
|
|
# in a unit test, but we can verify the error handling structure exists
|
|
from tzst.cli import cmd_add
|
|
|
|
# Create a mock args object
|
|
class MockArgs:
|
|
def __init__(self):
|
|
self.archive = str(temp_dir / "interrupt_test.tzst")
|
|
self.files = ["non_existent_file.txt"]
|
|
self.compression_level = 3
|
|
|
|
# Test that the function handles FileNotFoundError properly
|
|
result = cmd_add(MockArgs())
|
|
assert result == 1 # Should return error code for missing files
|
|
|
|
|
|
class TestCLIAliases:
|
|
"""Test CLI command aliases."""
|
|
|
|
def test_add_aliases(self, sample_files, temp_dir):
|
|
"""Test add command aliases."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Test 'add' alias
|
|
result = main(["add", str(archive_path), *file_paths])
|
|
assert result == 0
|
|
|
|
# Test 'create' alias
|
|
archive_path2 = temp_dir / "test2.tzst"
|
|
result = main(["create", str(archive_path2), *file_paths])
|
|
assert result == 0
|
|
|
|
def test_extract_aliases(self, sample_files, temp_dir):
|
|
"""Test extract command aliases."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
extract_dir = temp_dir / "extracted"
|
|
|
|
# Create archive first
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test 'extract' alias
|
|
result = main(["extract", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
|
|
def test_list_aliases(self, sample_files, temp_dir):
|
|
"""Test list command aliases."""
|
|
archive_path = temp_dir / "test.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Create archive first
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test 'list' alias
|
|
result = main(["list", str(archive_path)])
|
|
assert result == 0
|
|
|
|
|
|
class TestCLIHelp:
|
|
"""Test CLI help and documentation."""
|
|
|
|
def test_help_command(self):
|
|
"""Test help command execution."""
|
|
# Test main help
|
|
result = main(["--help"]) # Help should exit successfully
|
|
assert result == 0
|
|
|
|
def test_help_contains_security_info(self, capsys):
|
|
"""Test that help output contains security information."""
|
|
try:
|
|
main(["--help"])
|
|
except SystemExit:
|
|
pass # Help exits with SystemExit
|
|
|
|
captured = capsys.readouterr()
|
|
help_output = captured.out + captured.err
|
|
|
|
# Should mention security filters
|
|
assert "filter" in help_output.lower() or "security" in help_output.lower()
|
|
|
|
def test_command_specific_help(self):
|
|
"""Test command-specific help if available."""
|
|
# Some CLI implementations support command-specific help
|
|
# This test can be extended based on actual implementation
|
|
pass
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestCLIIntegration:
|
|
"""Integration tests for CLI."""
|
|
|
|
def test_full_workflow(self, sample_files, temp_dir):
|
|
"""Test complete workflow: create, list, test, extract."""
|
|
archive_path = temp_dir / "workflow.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
extract_dir = temp_dir / "workflow_extracted"
|
|
|
|
# Create archive
|
|
result = main(["a", str(archive_path), *file_paths])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
# List contents
|
|
result = main(["l", str(archive_path)])
|
|
assert result == 0
|
|
|
|
# Test integrity
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 0
|
|
|
|
# Extract
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
assert extract_dir.exists()
|
|
|
|
# Verify extracted files exist and have correct content
|
|
for file_path in sample_files:
|
|
if file_path.is_file():
|
|
relative_path = file_path.relative_to(sample_files[0].parent)
|
|
extracted_file = extract_dir / relative_path
|
|
assert extracted_file.exists()
|
|
|
|
# Compare content
|
|
original_content = file_path.read_bytes()
|
|
extracted_content = extracted_file.read_bytes()
|
|
assert original_content == extracted_content
|
|
|
|
|
|
class TestCLIStreamingOptions:
|
|
"""Test CLI streaming options."""
|
|
|
|
def test_extract_streaming_flag(self, sample_files, temp_dir):
|
|
"""Test extract command with streaming flag."""
|
|
archive_path = temp_dir / "cli_streaming_test.tzst"
|
|
file_paths = [f for f in sample_files if f.is_file()]
|
|
|
|
# Create archive first
|
|
from tzst import create_archive
|
|
|
|
create_archive(archive_path, file_paths)
|
|
|
|
# Test extract with streaming
|
|
parser = create_parser()
|
|
args = parser.parse_args(
|
|
["x", str(archive_path), "--streaming", "-o", str(temp_dir / "extracted")]
|
|
)
|
|
|
|
assert args.command == "x"
|
|
assert hasattr(args, "streaming")
|
|
assert args.streaming is True
|
|
|
|
def test_list_streaming_flag(self, sample_files, temp_dir):
|
|
"""Test list command with streaming flag."""
|
|
archive_path = temp_dir / "cli_list_streaming.tzst"
|
|
file_paths = [f for f in sample_files if f.is_file()]
|
|
|
|
# Create archive first
|
|
from tzst import create_archive
|
|
|
|
create_archive(archive_path, file_paths)
|
|
|
|
# Test list with streaming
|
|
parser = create_parser()
|
|
args = parser.parse_args(["l", str(archive_path), "--streaming"])
|
|
|
|
assert args.command == "l"
|
|
assert hasattr(args, "streaming")
|
|
assert args.streaming is True
|
|
|
|
def test_test_streaming_flag(self, sample_files, temp_dir):
|
|
"""Test test command with streaming flag."""
|
|
archive_path = temp_dir / "cli_test_streaming.tzst"
|
|
file_paths = [f for f in sample_files if f.is_file()]
|
|
|
|
# Create archive first
|
|
from tzst import create_archive
|
|
|
|
create_archive(archive_path, file_paths)
|
|
|
|
# Test test command with streaming
|
|
parser = create_parser()
|
|
args = parser.parse_args(["t", str(archive_path), "--streaming"])
|
|
|
|
assert args.command == "t"
|
|
assert hasattr(args, "streaming")
|
|
assert args.streaming is True
|
|
|
|
|
|
class TestCLICompressionLevels:
|
|
"""Test CLI compression level handling."""
|
|
|
|
def test_valid_compression_levels(self, sample_files, temp_dir):
|
|
"""Test various valid compression levels."""
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Test compression levels 1, 3, 10, 22
|
|
for level in [1, 3, 10, 22]:
|
|
archive_path = temp_dir / f"level_{level}.tzst"
|
|
result = main(["a", str(archive_path), *file_paths, "-l", str(level)])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
|
|
class TestCLIAtomicOperations:
|
|
"""Test CLI atomic and non-atomic operations."""
|
|
|
|
def test_no_atomic_flag(self, sample_files, temp_dir):
|
|
"""Test --no-atomic flag functionality."""
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "no_atomic.tzst"
|
|
|
|
# Test --no-atomic flag
|
|
result = main(["a", str(archive_path), *file_paths, "--no-atomic"])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
# Verify archive integrity
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 0
|
|
|
|
def test_atomic_vs_no_atomic_results(self, sample_files, temp_dir):
|
|
"""Test that atomic and non-atomic modes produce equivalent archives."""
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
|
|
# Create archive with atomic mode (default)
|
|
atomic_archive = temp_dir / "atomic.tzst"
|
|
result = main(["a", str(atomic_archive), *file_paths])
|
|
assert result == 0
|
|
|
|
# Create archive with non-atomic mode
|
|
non_atomic_archive = temp_dir / "non_atomic.tzst"
|
|
result = main(["a", str(non_atomic_archive), *file_paths, "--no-atomic"])
|
|
assert result == 0
|
|
|
|
# Both archives should be valid and contain same files
|
|
result_atomic = main(["t", str(atomic_archive)])
|
|
result_non_atomic = main(["t", str(non_atomic_archive)])
|
|
assert result_atomic == 0
|
|
assert result_non_atomic == 0
|
|
|
|
|
|
class TestCLISecurityFilters:
|
|
"""Test CLI security filtering options."""
|
|
|
|
def test_tar_filter(self, sample_files, temp_dir):
|
|
"""Test --filter tar option."""
|
|
# First create an archive
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "filter_test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test extraction with tar filter
|
|
extract_dir = temp_dir / "tar_filtered"
|
|
result = main(
|
|
["x", str(archive_path), "-o", str(extract_dir), "--filter", "tar"]
|
|
)
|
|
assert result == 0
|
|
assert extract_dir.exists()
|
|
|
|
def test_data_filter(self, sample_files, temp_dir):
|
|
"""Test --filter data option."""
|
|
# First create an archive
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "filter_test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test extraction with data filter
|
|
extract_dir = temp_dir / "data_filtered"
|
|
result = main(
|
|
["x", str(archive_path), "-o", str(extract_dir), "--filter", "data"]
|
|
)
|
|
assert result == 0
|
|
assert extract_dir.exists()
|
|
|
|
def test_invalid_filter(self, sample_files, temp_dir):
|
|
"""Test invalid filter option returns error."""
|
|
# First create an archive
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "filter_test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test extraction with invalid filter
|
|
extract_dir = temp_dir / "invalid_filtered"
|
|
result = main(
|
|
["x", str(archive_path), "-o", str(extract_dir), "--filter", "invalid"]
|
|
)
|
|
assert result == 1 # Should fail
|
|
|
|
|
|
class TestCLIStreamingOperations:
|
|
"""Test CLI streaming operations."""
|
|
|
|
def test_list_streaming(self, sample_files, temp_dir):
|
|
"""Test list command with --streaming flag."""
|
|
# First create an archive
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "streaming_test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test list with streaming
|
|
result = main(["l", str(archive_path), "--streaming"])
|
|
assert result == 0
|
|
|
|
def test_test_streaming(self, sample_files, temp_dir):
|
|
"""Test test command with --streaming flag."""
|
|
# First create an archive
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "streaming_test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test archive integrity with streaming
|
|
result = main(["t", str(archive_path), "--streaming"])
|
|
assert result == 0
|
|
|
|
def test_extract_streaming(self, sample_files, temp_dir):
|
|
"""Test extract command with --streaming flag."""
|
|
# First create an archive
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "streaming_test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Test extraction with streaming
|
|
extract_dir = temp_dir / "streaming_extracted"
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir), "--streaming"])
|
|
assert result == 0
|
|
assert extract_dir.exists()
|
|
|
|
|
|
class TestCLISpecialFiles:
|
|
"""Test CLI with special file types and names."""
|
|
|
|
def test_empty_files(self, temp_dir):
|
|
"""Test archiving empty files."""
|
|
# Create empty file
|
|
empty_file = temp_dir / "empty.txt"
|
|
empty_file.touch()
|
|
|
|
archive_path = temp_dir / "empty_test.tzst"
|
|
result = main(["a", str(archive_path), str(empty_file)])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
def test_files_with_spaces_and_special_chars(self, temp_dir):
|
|
"""Test files with spaces and special characters."""
|
|
# Create file with special characters
|
|
special_file = temp_dir / "file with spaces & special chars!@#$%.txt"
|
|
special_file.write_text("Special file content")
|
|
|
|
archive_path = temp_dir / "special_chars.tzst"
|
|
result = main(["a", str(archive_path), str(special_file)])
|
|
assert result == 0
|
|
|
|
def test_very_long_filenames(self, temp_dir):
|
|
"""Test files with very long names."""
|
|
# Create file with very long name
|
|
long_name = "long_" + "x" * 200 + ".txt"
|
|
long_file = temp_dir / long_name
|
|
long_file.write_text("Long filename content")
|
|
|
|
archive_path = temp_dir / "long_filename.tzst"
|
|
result = main(["a", str(archive_path), str(long_file)])
|
|
assert result == 0
|
|
|
|
def test_files_with_dots(self, temp_dir):
|
|
"""Test files with multiple dots in name."""
|
|
# Create file with multiple dots
|
|
dotted_file = temp_dir / "multiple.dots.in.name.txt"
|
|
dotted_file.write_text("Dotted file content")
|
|
|
|
archive_path = temp_dir / "dotted.tzst"
|
|
result = main(["a", str(archive_path), str(dotted_file)])
|
|
assert result == 0
|
|
|
|
|
|
class TestCLILargeFiles:
|
|
"""Test CLI with large files."""
|
|
|
|
def test_large_text_file(self, temp_dir):
|
|
"""Test archiving large text files."""
|
|
# Create large text file (1MB+)
|
|
large_file = temp_dir / "large.txt"
|
|
content = "This is a large file content.\n" * 50000
|
|
large_file.write_text(content)
|
|
|
|
archive_path = temp_dir / "large_file.tzst"
|
|
result = main(["a", str(archive_path), str(large_file), "-l", "22"])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
def test_binary_file(self, temp_dir):
|
|
"""Test archiving binary files."""
|
|
# Create binary file
|
|
binary_file = temp_dir / "binary.bin"
|
|
binary_content = bytes(range(256)) * 1000 # 256KB of binary data
|
|
binary_file.write_bytes(binary_content)
|
|
|
|
archive_path = temp_dir / "binary.tzst"
|
|
result = main(["a", str(archive_path), str(binary_file)])
|
|
assert result == 0
|
|
|
|
|
|
class TestCLIUnicodeHandling:
|
|
"""Test CLI with Unicode and international characters."""
|
|
|
|
def test_unicode_filenames(self, temp_dir):
|
|
"""Test files with Unicode characters in names."""
|
|
# Create file with Unicode characters (may fail on some platforms)
|
|
try:
|
|
unicode_file = temp_dir / "测试文件_🌟.txt"
|
|
unicode_file.write_text("Unicode content", encoding="utf-8")
|
|
|
|
archive_path = temp_dir / "unicode.tzst"
|
|
result = main(["a", str(archive_path), str(unicode_file)])
|
|
# Result may be 0 (success) or 1 (failure) depending on platform
|
|
assert result in [0, 1]
|
|
except (OSError, UnicodeError):
|
|
# Skip if platform doesn't support Unicode filenames
|
|
pytest.skip("Platform doesn't support Unicode filenames")
|
|
|
|
def test_unicode_content(self, temp_dir):
|
|
"""Test files with Unicode content."""
|
|
# Create file with Unicode content
|
|
unicode_content_file = temp_dir / "unicode_content.txt"
|
|
unicode_content = "Hello 世界! 🌍 Здравствуй мир!"
|
|
unicode_content_file.write_text(unicode_content, encoding="utf-8")
|
|
|
|
archive_path = temp_dir / "unicode_content.tzst"
|
|
result = main(["a", str(archive_path), str(unicode_content_file)])
|
|
assert result == 0
|
|
|
|
# Test extraction and content verification
|
|
extract_dir = temp_dir / "unicode_extracted"
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
|
|
extracted_file = extract_dir / "unicode_content.txt"
|
|
assert extracted_file.exists()
|
|
extracted_content = extracted_file.read_text(encoding="utf-8")
|
|
assert extracted_content == unicode_content
|
|
|
|
|
|
class TestCLIExitCodes:
|
|
"""Test CLI exit codes for proper script integration."""
|
|
|
|
def test_successful_operations_return_zero(self, sample_files, temp_dir):
|
|
"""Test that successful operations return exit code 0."""
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "success.tzst"
|
|
|
|
# Create archive
|
|
result = main(["a", str(archive_path), *file_paths])
|
|
assert result == 0
|
|
|
|
# List archive
|
|
result = main(["l", str(archive_path)])
|
|
assert result == 0
|
|
|
|
# Test archive
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 0
|
|
|
|
# Extract archive
|
|
extract_dir = temp_dir / "extracted"
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
|
|
def test_error_conditions_return_nonzero(self, temp_dir):
|
|
"""Test that error conditions return non-zero exit codes."""
|
|
# Non-existent archive
|
|
result = main(["l", str(temp_dir / "nonexistent.tzst")])
|
|
assert result != 0
|
|
|
|
# Non-existent file to archive
|
|
result = main(["a", str(temp_dir / "test.tzst"), "nonexistent.txt"])
|
|
assert result != 0
|
|
|
|
# Invalid compression level
|
|
result = main(["a", str(temp_dir / "test.tzst"), __file__, "-l", "0"])
|
|
assert result != 0
|
|
|
|
# Invalid filter
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), __file__]) # Create valid archive first
|
|
result = main(["x", str(archive_path), "--filter", "invalid"])
|
|
assert result != 0
|
|
|
|
def test_help_and_version_return_proper_codes(self):
|
|
"""Test help and version commands return appropriate codes."""
|
|
# Help should return non-zero (standard for help)
|
|
result = main([])
|
|
assert result == 1
|
|
|
|
# Version should return zero
|
|
result = main(["--version"])
|
|
assert result == 0
|
|
|
|
|
|
class TestCLIPerformance:
|
|
"""Test CLI performance with various scenarios."""
|
|
|
|
def test_large_number_of_small_files(self, temp_dir):
|
|
"""Test archiving many small files."""
|
|
# Create multiple small files
|
|
files = []
|
|
for i in range(50):
|
|
small_file = temp_dir / f"small_{i:03d}.txt"
|
|
small_file.write_text(f"Small file {i} content")
|
|
files.append(str(small_file))
|
|
|
|
archive_path = temp_dir / "many_files.tzst"
|
|
result = main(["a", str(archive_path), *files])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
def test_nested_directory_structure(self, temp_dir):
|
|
"""Test archiving nested directory structures."""
|
|
# Create nested structure
|
|
for i in range(5):
|
|
nested_dir = temp_dir / f"level_{i}"
|
|
for j in range(i + 1):
|
|
nested_dir = nested_dir / f"sub_{j}"
|
|
nested_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
nested_file = nested_dir / f"file_{i}.txt"
|
|
nested_file.write_text(f"Nested file at level {i}")
|
|
|
|
archive_path = temp_dir / "nested.tzst"
|
|
result = main(["a", str(archive_path), str(temp_dir / "level_0")])
|
|
assert result == 0
|
|
|
|
|
|
class TestCLIRealWorldScenarios:
|
|
"""Test scenarios discovered during comprehensive real-world testing."""
|
|
|
|
def test_no_atomic_flag_with_path_resolution(self, temp_dir):
|
|
"""Test --no-atomic flag with various path scenarios.
|
|
|
|
This tests the critical bug fix where --no-atomic failed due to
|
|
path resolution issues when changing working directory.
|
|
"""
|
|
# Create test file
|
|
test_file = temp_dir / "test_no_atomic.txt"
|
|
test_file.write_text("No atomic test content")
|
|
|
|
# Test no-atomic in current directory
|
|
archive_path = temp_dir / "no_atomic_test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file), "--no-atomic"])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
# Verify archive integrity
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 0 # Test extraction works
|
|
extract_dir = temp_dir / "no_atomic_extracted"
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
extracted_file = extract_dir / "test_no_atomic.txt"
|
|
assert extracted_file.exists()
|
|
|
|
def test_no_atomic_flag_in_subdirectory(self, temp_dir):
|
|
"""Test --no-atomic flag when archive is in subdirectory."""
|
|
# Create subdirectory and test file
|
|
subdir = temp_dir / "subdir"
|
|
subdir.mkdir()
|
|
test_file = temp_dir / "test_sub.txt"
|
|
test_file.write_text("Subdirectory test content")
|
|
|
|
# Create archive in subdirectory with --no-atomic
|
|
archive_path = subdir / "sub_no_atomic.tzst"
|
|
result = main(["a", str(archive_path), str(test_file), "--no-atomic"])
|
|
assert result == 0
|
|
assert archive_path.exists()
|
|
|
|
def test_compression_level_clamping(self, temp_dir):
|
|
"""Test that extreme compression levels are handled properly."""
|
|
test_file = temp_dir / "clamp_test.txt"
|
|
test_file.write_text("Compression clamping test")
|
|
|
|
# Test compression level higher than maximum
|
|
archive_path = temp_dir / "extreme_compression.tzst"
|
|
result = main(["a", str(archive_path), str(test_file), "-l", "50"])
|
|
assert result == 2 # argparse error
|
|
|
|
# Test negative compression level
|
|
result = main(["a", str(archive_path), str(test_file), "-l", "-1"])
|
|
assert result == 2 # argparse error
|
|
|
|
def test_whitespace_only_file(self, temp_dir):
|
|
"""Test archiving files with only whitespace content."""
|
|
whitespace_file = temp_dir / "whitespace.txt"
|
|
whitespace_file.write_text(" \n\t\n \n")
|
|
|
|
archive_path = temp_dir / "whitespace.tzst"
|
|
result = main(["a", str(archive_path), str(whitespace_file)])
|
|
assert result == 0
|
|
|
|
# Extract and verify content preserved
|
|
extract_dir = temp_dir / "whitespace_extracted"
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
|
|
extracted_file = extract_dir / "whitespace.txt"
|
|
assert extracted_file.exists()
|
|
assert extracted_file.read_text() == " \n\t\n \n"
|
|
|
|
def test_null_bytes_binary_file(self, temp_dir):
|
|
"""Test archiving binary files with null bytes."""
|
|
null_file = temp_dir / "null_bytes.bin"
|
|
null_content = b"\x00\x00\x00\x01\x00\x02\x00\x00\x03"
|
|
null_file.write_bytes(null_content)
|
|
|
|
archive_path = temp_dir / "null_bytes.tzst"
|
|
result = main(["a", str(archive_path), str(null_file)])
|
|
assert result == 0
|
|
|
|
# Extract and verify binary content preserved
|
|
extract_dir = temp_dir / "null_extracted"
|
|
result = main(["x", str(archive_path), "-o", str(extract_dir)])
|
|
assert result == 0
|
|
|
|
extracted_file = extract_dir / "null_bytes.bin"
|
|
assert extracted_file.exists()
|
|
assert extracted_file.read_bytes() == null_content
|
|
|
|
def test_streaming_with_filters_combination(self, temp_dir):
|
|
"""Test streaming flag combined with security filters."""
|
|
# Create test archive
|
|
test_file = temp_dir / "stream_filter.txt"
|
|
test_file.write_text("Streaming with filter test")
|
|
|
|
archive_path = temp_dir / "stream_filter.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 0
|
|
|
|
# Test list with streaming
|
|
result = main(["l", str(archive_path), "--streaming"])
|
|
assert result == 0
|
|
|
|
# Test extract with streaming and filter
|
|
extract_dir = temp_dir / "stream_filter_extracted"
|
|
result = main(
|
|
[
|
|
"x",
|
|
str(archive_path),
|
|
"-o",
|
|
str(extract_dir),
|
|
"--streaming",
|
|
"--filter",
|
|
"data",
|
|
]
|
|
)
|
|
assert result == 0
|
|
|
|
def test_exit_code_propagation_fix(self, temp_dir):
|
|
"""Test that exit codes are properly propagated (fixed issue)."""
|
|
# Test successful operation returns 0
|
|
test_file = temp_dir / "exit_test.txt"
|
|
test_file.write_text("Exit code test")
|
|
|
|
archive_path = temp_dir / "exit_test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 0 # Success should return 0
|
|
|
|
# Test error operation returns non-zero
|
|
result = main(["l", "nonexistent_archive.tzst"])
|
|
assert result != 0 # Error should return non-zero
|
|
|
|
|
|
class TestCLISecurityFilterParsing:
|
|
"""Test CLI security filter argument parsing."""
|
|
|
|
def test_extract_filter_argument_parsing(self):
|
|
"""Test that --filter argument is parsed correctly for extract commands."""
|
|
parser = create_parser()
|
|
|
|
# Test x/extract command with filter
|
|
args = parser.parse_args(["x", "test.tzst", "--filter", "data"])
|
|
assert args.command == "x"
|
|
assert args.filter == "data"
|
|
|
|
args = parser.parse_args(["extract", "test.tzst", "--filter", "tar"])
|
|
assert args.command == "extract"
|
|
assert args.filter == "tar"
|
|
|
|
# Test e/extract-flat command with filter
|
|
args = parser.parse_args(["e", "test.tzst", "--filter", "data"])
|
|
assert args.command == "e"
|
|
assert args.filter == "data"
|
|
|
|
def test_filter_default_value(self):
|
|
"""Test that filter defaults to 'data'."""
|
|
parser = create_parser()
|
|
|
|
# Test default for x command
|
|
args = parser.parse_args(["x", "test.tzst"])
|
|
assert args.filter == "data"
|
|
|
|
# Test default for e command
|
|
args = parser.parse_args(["e", "test.tzst"])
|
|
assert args.filter == "data"
|
|
|
|
def test_filter_invalid_choices(self):
|
|
"""Test that invalid filter choices are rejected."""
|
|
parser = create_parser()
|
|
|
|
# Invalid filter should raise SystemExit (argparse error)
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(["x", "test.tzst", "--filter", "invalid"])
|
|
|
|
def test_extract_with_filter(self, sample_files, temp_dir):
|
|
"""Test extract command with security filters."""
|
|
archive_path = temp_dir / "cli_extract_filter.tzst"
|
|
file_paths = [str(f) for f in sample_files if f.is_file()][:1]
|
|
|
|
# Create archive first
|
|
result = main(["a", str(archive_path), *file_paths])
|
|
assert result == 0
|
|
|
|
# Test extraction with different filters
|
|
for filter_type in ["data", "tar", "fully_trusted"]:
|
|
output_dir = temp_dir / f"extracted_{filter_type}"
|
|
result = main(
|
|
["x", str(archive_path), "--filter", filter_type, "-o", str(output_dir)]
|
|
)
|
|
assert result == 0
|
|
|
|
def test_filter_with_other_options(self):
|
|
"""Test filter option combined with other options."""
|
|
parser = create_parser()
|
|
|
|
# Test with streaming and output directory
|
|
args = parser.parse_args(
|
|
[
|
|
"x",
|
|
"test.tzst",
|
|
"file1.txt",
|
|
"file2.txt",
|
|
"--filter",
|
|
"tar",
|
|
"--streaming",
|
|
"-o",
|
|
"output_dir",
|
|
]
|
|
)
|
|
assert args.filter == "tar"
|
|
assert args.streaming is True
|
|
assert args.output == "output_dir"
|
|
assert args.files == ["file1.txt", "file2.txt"]
|
|
|
|
def test_filter_not_available_for_non_extract_commands(self):
|
|
"""Test that --filter is only available for extract commands."""
|
|
parser = create_parser()
|
|
|
|
# Should work for extract commands
|
|
parser.parse_args(["x", "test.tzst", "--filter", "data"])
|
|
parser.parse_args(["e", "test.tzst", "--filter", "tar"])
|
|
|
|
# Should fail for non-extract commands
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(["a", "test.tzst", "file.txt", "--filter", "data"])
|
|
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(["l", "test.tzst", "--filter", "data"])
|
|
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(["t", "test.tzst", "--filter", "data"])
|
|
|
|
|
|
class TestCLIBoundaryConditions:
|
|
"""Test boundary conditions and edge cases."""
|
|
|
|
def test_compression_level_boundaries(self, temp_dir):
|
|
"""Test exact boundary values for compression levels."""
|
|
test_file = temp_dir / "boundary_test.txt"
|
|
test_file.write_text("Boundary compression test")
|
|
|
|
# Test minimum valid level (1)
|
|
archive_min = temp_dir / "min_compression.tzst"
|
|
result = main(["a", str(archive_min), str(test_file), "-l", "1"])
|
|
assert result == 0
|
|
|
|
# Test maximum valid level (22)
|
|
archive_max = temp_dir / "max_compression.tzst"
|
|
result = main(["a", str(archive_max), str(test_file), "-l", "22"])
|
|
assert result == 0
|
|
|
|
# Test just below minimum (should fail with argparse error)
|
|
result = main(["a", "temp.tzst", str(test_file), "-l", "0"])
|
|
assert result == 2 # argparse error for invalid choice
|
|
|
|
# Test just above maximum (should fail or clamp)
|
|
result = main(["a", "temp.tzst", str(test_file), "-l", "23"])
|
|
assert result in [0, 1, 2] # May clamp or error
|
|
|
|
def test_edge_case_scenarios(self, temp_dir):
|
|
"""Test edge cases and boundary conditions."""
|
|
# Test with special characters in paths
|
|
special_file = temp_dir / "file with spaces.txt"
|
|
special_file.write_text("content")
|
|
special_archive = temp_dir / "special archive.tzst"
|
|
|
|
result = main(["a", str(special_archive), str(special_file)])
|
|
# Should handle special characters gracefully
|
|
|
|
if result == 0:
|
|
# If creation succeeded, test other operations
|
|
result = main(["l", str(special_archive)])
|
|
assert result == 0
|
|
|
|
result = main(["t", str(special_archive)])
|
|
assert result == 0
|
|
|
|
def test_maximum_filename_length(self, temp_dir):
|
|
"""Test files with maximum allowable filename length."""
|
|
# Create file with very long name (close to filesystem limit)
|
|
long_name = "a" * 200 + ".txt"
|
|
try:
|
|
long_file = temp_dir / long_name
|
|
long_file.write_text("Long filename test")
|
|
|
|
archive_path = temp_dir / "long_filename.tzst"
|
|
result = main(["a", str(archive_path), str(long_file)])
|
|
# Should succeed or fail gracefully
|
|
assert result in [0, 1]
|
|
except OSError:
|
|
# Skip if filesystem doesn't support such long names
|
|
pytest.skip("Filesystem doesn't support long filenames")
|
|
|
|
|
|
class TestCompressionLevelValidation:
|
|
"""Test compression level validation in CLI."""
|
|
|
|
def test_valid_compression_levels(self, temp_dir):
|
|
"""Test that valid compression levels (1-22) work correctly."""
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("Test content")
|
|
|
|
# Test boundary values and some middle values
|
|
valid_levels = [1, 2, 10, 22]
|
|
|
|
for level in valid_levels:
|
|
archive_path = temp_dir / f"test_level_{level}.tzst"
|
|
result = main(["a", str(archive_path), str(test_file), "-l", str(level)])
|
|
assert result == 0, f"Compression level {level} should be valid"
|
|
|
|
def test_invalid_compression_levels(self, temp_dir):
|
|
"""Test that invalid compression levels return proper error codes."""
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("Test content")
|
|
archive_path = temp_dir / "test.tzst"
|
|
|
|
# Test invalid levels that should return exit code 2 (argparse error)
|
|
invalid_levels = [0, 23, 50, 100, -1, -10]
|
|
|
|
for level in invalid_levels:
|
|
result = main(["a", str(archive_path), str(test_file), "-l", str(level)])
|
|
assert result == 2, (
|
|
f"Invalid compression level {level} should return exit code 2"
|
|
)
|
|
|
|
def test_non_numeric_compression_levels(self, temp_dir):
|
|
"""Test that non-numeric compression levels return proper error codes."""
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("Test content")
|
|
archive_path = temp_dir / "test.tzst"
|
|
|
|
# Test non-numeric values
|
|
invalid_values = ["abc", "1.5", "high", "max", ""]
|
|
|
|
for value in invalid_values:
|
|
result = main(["a", str(archive_path), str(test_file), "-l", value])
|
|
assert result == 2, (
|
|
f"Non-numeric compression level '{value}' should return exit code 2"
|
|
)
|
|
|
|
def test_compression_level_extremes(self, temp_dir):
|
|
"""Test compression level validation with extreme values."""
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("Test content")
|
|
archive_path = temp_dir / "test.tzst"
|
|
|
|
# Test level 50 - should return argparse error code 2
|
|
result = main(["a", str(archive_path), str(test_file), "-l", "50"])
|
|
assert result == 2, "Compression level 50 should return exit code 2"
|
|
|
|
|
|
class TestCLIVersionCommand:
|
|
"""Test version command functionality."""
|
|
|
|
def test_version_command_direct(self):
|
|
"""Test version command handler directly."""
|
|
from tzst.cli import cmd_version
|
|
|
|
class MockArgs:
|
|
pass
|
|
|
|
args = MockArgs()
|
|
result = cmd_version(args)
|
|
assert result == 0
|
|
|
|
def test_version_command_via_main(self):
|
|
"""Test version command via main function."""
|
|
result = main(["--version"])
|
|
assert result == 0
|
|
|
|
def test_version_flag_parsing(self):
|
|
"""Test that version flag is properly parsed."""
|
|
parser = create_parser()
|
|
args = parser.parse_args(["--version"])
|
|
assert hasattr(args, "version")
|
|
assert args.version is True
|
|
|
|
|
|
class TestCLIValidationFunctions:
|
|
"""Test internal CLI validation functions."""
|
|
|
|
def test_validate_compression_level_in_argv_valid(self):
|
|
"""Test compression level validation with valid levels."""
|
|
from tzst.cli import _validate_compression_level_in_argv
|
|
|
|
# Test valid levels
|
|
assert not _validate_compression_level_in_argv(["a", "test.tzst", "-l", "3"])
|
|
assert not _validate_compression_level_in_argv(
|
|
["a", "test.tzst", "--level", "22"]
|
|
)
|
|
assert not _validate_compression_level_in_argv(
|
|
["a", "test.tzst", "--level", "1"]
|
|
)
|
|
|
|
def test_validate_compression_level_in_argv_invalid_range(self, capsys):
|
|
"""Test compression level validation with invalid range."""
|
|
from tzst.cli import _validate_compression_level_in_argv
|
|
|
|
# Test invalid range
|
|
assert _validate_compression_level_in_argv(["a", "test.tzst", "-l", "0"])
|
|
captured = capsys.readouterr()
|
|
assert "Invalid compression level: 0" in captured.err
|
|
assert "Must be between 1 and 22" in captured.err
|
|
|
|
assert _validate_compression_level_in_argv(["a", "test.tzst", "--level", "23"])
|
|
captured = capsys.readouterr()
|
|
assert "Invalid compression level: 23" in captured.err
|
|
|
|
def test_validate_compression_level_in_argv_invalid_format(self, capsys):
|
|
"""Test compression level validation with invalid format."""
|
|
from tzst.cli import _validate_compression_level_in_argv
|
|
|
|
# Test invalid format
|
|
assert _validate_compression_level_in_argv(["a", "test.tzst", "-l", "abc"])
|
|
captured = capsys.readouterr()
|
|
assert "Invalid compression level: 'abc'" in captured.err
|
|
assert "Must be an integer" in captured.err
|
|
|
|
def test_validate_compression_level_in_argv_no_level(self):
|
|
"""Test compression level validation when no level specified."""
|
|
from tzst.cli import _validate_compression_level_in_argv
|
|
|
|
# Test no level specified
|
|
assert not _validate_compression_level_in_argv(["a", "test.tzst", "file.txt"])
|
|
|
|
def test_validate_filter_in_argv_valid(self):
|
|
"""Test filter validation with valid filters."""
|
|
from tzst.cli import _validate_filter_in_argv
|
|
|
|
# Test valid filters
|
|
assert not _validate_filter_in_argv(["x", "test.tzst", "--filter", "data"])
|
|
assert not _validate_filter_in_argv(["x", "test.tzst", "--filter", "tar"])
|
|
assert not _validate_filter_in_argv(
|
|
["x", "test.tzst", "--filter", "fully_trusted"]
|
|
)
|
|
|
|
def test_validate_filter_in_argv_invalid(self, capsys):
|
|
"""Test filter validation with invalid filter."""
|
|
from tzst.cli import _validate_filter_in_argv
|
|
|
|
# Test invalid filter
|
|
assert _validate_filter_in_argv(["x", "test.tzst", "--filter", "invalid"])
|
|
captured = capsys.readouterr()
|
|
assert "Invalid filter specified: invalid" in captured.err
|
|
assert "Must be one of: data, tar, fully_trusted" in captured.err
|
|
|
|
def test_validate_filter_in_argv_no_filter(self):
|
|
"""Test filter validation when no filter specified."""
|
|
from tzst.cli import _validate_filter_in_argv
|
|
|
|
# Test no filter specified
|
|
assert not _validate_filter_in_argv(["x", "test.tzst"])
|
|
|
|
|
|
class TestCLIErrorHandlingInternal:
|
|
def test_handle_parsing_errors_help_requested(self):
|
|
"""Test error handling when help is requested."""
|
|
from tzst.cli import _handle_parsing_errors
|
|
|
|
# Simulate help request (exit code 0)
|
|
e = SystemExit(0)
|
|
result = _handle_parsing_errors(e, ["--help"])
|
|
assert result == 0
|
|
|
|
def test_handle_parsing_errors_invalid_filter(self, capsys):
|
|
"""Test error handling for invalid filter."""
|
|
from tzst.cli import _handle_parsing_errors
|
|
|
|
# Simulate parsing error (exit code 2) with invalid filter
|
|
e = SystemExit(2)
|
|
argv = ["x", "test.tzst", "--filter", "invalid"]
|
|
result = _handle_parsing_errors(e, argv)
|
|
assert result == 1 # Should convert to exit code 1
|
|
captured = capsys.readouterr()
|
|
assert "Invalid filter specified" in captured.err
|
|
|
|
def test_handle_parsing_errors_other_errors(self):
|
|
"""Test error handling for other parsing errors."""
|
|
from tzst.cli import _handle_parsing_errors
|
|
|
|
# Test with exit code 2 but no special validation errors
|
|
e = SystemExit(2)
|
|
result = _handle_parsing_errors(e, ["invalid", "command"])
|
|
assert result == 2
|
|
|
|
# Test with None exit code
|
|
e = SystemExit(None)
|
|
result = _handle_parsing_errors(e, [])
|
|
assert result == 1
|
|
|
|
def test_parse_arguments_success(self):
|
|
"""Test successful argument parsing."""
|
|
from tzst.cli import _parse_arguments
|
|
|
|
parser = create_parser()
|
|
args, error_code = _parse_arguments(parser, ["a", "test.tzst", "file.txt"])
|
|
assert args is not None
|
|
assert error_code is None
|
|
assert args.command == "a"
|
|
|
|
def test_parse_arguments_error(self):
|
|
"""Test argument parsing with errors."""
|
|
from tzst.cli import _parse_arguments
|
|
|
|
parser = create_parser()
|
|
args, error_code = _parse_arguments(parser, ["invalid"])
|
|
assert args is None
|
|
assert error_code == 2
|
|
|
|
def test_execute_command_no_func(self, capsys):
|
|
"""Test command execution when no function is set."""
|
|
from tzst.cli import _execute_command
|
|
|
|
class MockArgs:
|
|
pass
|
|
|
|
args = MockArgs()
|
|
parser = create_parser()
|
|
result = _execute_command(args, parser)
|
|
assert result == 1
|
|
# Should print help when no function is available
|
|
|
|
def test_execute_command_version_flag(self):
|
|
"""Test command execution with version flag."""
|
|
from tzst.cli import _execute_command
|
|
|
|
class MockArgs:
|
|
version = True
|
|
|
|
args = MockArgs()
|
|
parser = create_parser()
|
|
result = _execute_command(args, parser)
|
|
assert result == 0
|
|
|
|
def test_invalid_arguments_main(self):
|
|
"""Test main function with invalid arguments."""
|
|
result = main(["invalid-command"])
|
|
assert result == 2 # Argument parsing error
|
|
|
|
|
|
class TestCLIStreamingAndAdvancedOptions:
|
|
"""Test CLI streaming and advanced options."""
|
|
|
|
def test_streaming_option_parsing(self):
|
|
"""Test that streaming options are properly parsed."""
|
|
parser = create_parser()
|
|
|
|
# Test list command with streaming
|
|
args = parser.parse_args(["l", "test.tzst", "--streaming"])
|
|
assert hasattr(args, "streaming")
|
|
assert args.streaming is True
|
|
|
|
# Test test command with streaming
|
|
args = parser.parse_args(["t", "test.tzst", "--streaming"])
|
|
assert hasattr(args, "streaming")
|
|
assert args.streaming is True
|
|
|
|
def test_atomic_options_parsing(self):
|
|
"""Test atomic operation options."""
|
|
parser = create_parser()
|
|
|
|
# Test with atomic (default)
|
|
args = parser.parse_args(["a", "test.tzst", "file.txt"])
|
|
assert not hasattr(args, "no_atomic") or not args.no_atomic
|
|
|
|
# Test with no-atomic
|
|
args = parser.parse_args(["a", "test.tzst", "file.txt", "--no-atomic"])
|
|
assert hasattr(args, "no_atomic")
|
|
assert args.no_atomic is True
|
|
|
|
def test_verbose_options_parsing(self):
|
|
"""Test verbose option parsing."""
|
|
parser = create_parser()
|
|
|
|
# Test list command with verbose
|
|
args = parser.parse_args(["l", "test.tzst", "-v"])
|
|
assert hasattr(args, "verbose")
|
|
assert args.verbose is True # Test list command with --verbose
|
|
args = parser.parse_args(["l", "test.tzst", "--verbose"])
|
|
assert hasattr(args, "verbose")
|
|
assert args.verbose is True
|
|
|
|
|
|
class TestCLIEdgeCases:
|
|
"""Test CLI edge cases and boundary conditions."""
|
|
|
|
def test_empty_arguments(self):
|
|
"""Test main function with no arguments."""
|
|
result = main([])
|
|
# Should show help and exit with error code
|
|
assert result in [1, 2]
|
|
|
|
def test_help_with_subcommands(self):
|
|
"""Test help display for specific subcommands."""
|
|
# Test that help can be requested for specific commands
|
|
result = main(["a", "--help"])
|
|
assert result == 0
|
|
|
|
result = main(["x", "--help"])
|
|
assert result == 0
|
|
|
|
def test_compression_level_edge_values(self):
|
|
"""Test compression level with edge values."""
|
|
parser = create_parser()
|
|
|
|
# Test minimum valid level
|
|
args = parser.parse_args(["a", "test.tzst", "file.txt", "-l", "1"])
|
|
assert args.compression_level == 1
|
|
|
|
# Test maximum valid level
|
|
args = parser.parse_args(["a", "test.tzst", "file.txt", "--level", "22"])
|
|
assert args.compression_level == 22
|
|
|
|
def test_filter_options_all_values(self):
|
|
"""Test all valid filter option values."""
|
|
parser = create_parser()
|
|
|
|
for filter_val in ["data", "tar", "fully_trusted"]:
|
|
args = parser.parse_args(["x", "test.tzst", "--filter", filter_val])
|
|
assert args.filter == filter_val
|
|
|
|
def test_output_directory_parsing(self):
|
|
"""Test output directory parsing for extract commands."""
|
|
parser = create_parser()
|
|
|
|
# Test extract with output directory
|
|
args = parser.parse_args(["x", "test.tzst", "-o", "/tmp/output"])
|
|
assert args.output == "/tmp/output"
|
|
|
|
# Test extract-flat with output directory
|
|
args = parser.parse_args(["e", "test.tzst", "--output", "/tmp/output"])
|
|
assert args.output == "/tmp/output"
|
|
|
|
|
|
class TestCLIFileOperations:
|
|
"""Test CLI file operation error scenarios."""
|
|
|
|
def test_test_command_missing_file(self, temp_dir):
|
|
"""Test test command with missing archive file."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["t", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
def test_list_command_missing_file(self, temp_dir):
|
|
"""Test list command with missing archive file."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["l", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
def test_extract_command_missing_file(self, temp_dir):
|
|
"""Test extract command with missing archive file."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["x", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
|
|
class TestCLIExceptionHandling:
|
|
"""Test CLI exception handling for comprehensive coverage."""
|
|
|
|
def test_oserror_in_validate_files(self, temp_dir, monkeypatch):
|
|
"""Test OSError exception handling in _validate_files function."""
|
|
from pathlib import Path
|
|
|
|
from tzst.cli import _validate_files
|
|
|
|
# Create a path that will trigger OSError when checking existence
|
|
invalid_path = Path(temp_dir / "invalid")
|
|
|
|
# Mock Path.exists to raise OSError at the class level
|
|
def mock_exists(self):
|
|
if str(self).endswith("invalid"):
|
|
raise OSError("Invalid path")
|
|
return True
|
|
|
|
monkeypatch.setattr(Path, "exists", mock_exists)
|
|
|
|
with pytest.raises(OSError):
|
|
_validate_files([invalid_path])
|
|
|
|
def test_file_not_found_error_in_extract(self, temp_dir):
|
|
"""Test FileNotFoundError handling in extract command."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["x", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
def test_file_not_found_error_in_extract_flat(self, temp_dir):
|
|
"""Test FileNotFoundError handling in extract-flat command."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["e", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
def test_file_not_found_error_in_list(self, temp_dir):
|
|
"""Test FileNotFoundError handling in list command."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["l", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
def test_file_not_found_error_in_test(self, temp_dir):
|
|
"""Test FileNotFoundError handling in test command."""
|
|
missing_archive = temp_dir / "missing.tzst"
|
|
result = main(["t", str(missing_archive)])
|
|
assert result == 1
|
|
|
|
def test_tzst_archive_error_in_add(self, temp_dir, monkeypatch):
|
|
"""Test TzstArchiveError handling in add command."""
|
|
from tzst.exceptions import TzstArchiveError
|
|
|
|
# Create a file to add
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("test content")
|
|
|
|
# Mock create_archive to raise TzstArchiveError
|
|
def mock_create_archive(*args, **kwargs):
|
|
raise TzstArchiveError("Mock archive error")
|
|
|
|
monkeypatch.setattr("tzst.cli.create_archive", mock_create_archive)
|
|
|
|
archive_path = temp_dir / "test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 1
|
|
|
|
def test_tzst_archive_error_in_extract(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test TzstArchiveError handling in extract command."""
|
|
from tzst.exceptions import TzstArchiveError
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock extract_archive to raise TzstArchiveError
|
|
def mock_extract_archive(*args, **kwargs):
|
|
raise TzstArchiveError("Mock extraction error")
|
|
|
|
monkeypatch.setattr("tzst.cli.extract_archive", mock_extract_archive)
|
|
|
|
result = main(["x", str(archive_path)])
|
|
assert result == 1
|
|
|
|
def test_tzst_archive_error_in_list(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test TzstArchiveError handling in list command."""
|
|
from tzst.exceptions import TzstArchiveError
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock list_archive to raise TzstArchiveError
|
|
def mock_list_archive(*args, **kwargs):
|
|
raise TzstArchiveError("Mock list error")
|
|
|
|
monkeypatch.setattr("tzst.cli.list_archive", mock_list_archive)
|
|
|
|
result = main(["l", str(archive_path)])
|
|
assert result == 1
|
|
|
|
def test_tzst_archive_error_in_test(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test TzstArchiveError handling in test command."""
|
|
from tzst.exceptions import TzstArchiveError
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock test_archive to raise TzstArchiveError
|
|
def mock_test_archive(*args, **kwargs):
|
|
raise TzstArchiveError("Mock test error")
|
|
|
|
monkeypatch.setattr("tzst.cli.test_archive", mock_test_archive)
|
|
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 1
|
|
|
|
def test_value_error_in_add(self, temp_dir, monkeypatch):
|
|
"""Test ValueError handling in add command."""
|
|
|
|
# Create a file to add
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("test content")
|
|
|
|
# Mock create_archive to raise ValueError
|
|
def mock_create_archive(*args, **kwargs):
|
|
raise ValueError("Invalid compression level")
|
|
|
|
monkeypatch.setattr("tzst.cli.create_archive", mock_create_archive)
|
|
|
|
archive_path = temp_dir / "test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 1
|
|
|
|
def test_oserror_in_add(self, temp_dir, monkeypatch):
|
|
"""Test OSError handling in add command."""
|
|
|
|
# Create a file to add
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("test content")
|
|
|
|
# Mock create_archive to raise OSError
|
|
def mock_create_archive(*args, **kwargs):
|
|
raise OSError("Permission denied")
|
|
|
|
monkeypatch.setattr("tzst.cli.create_archive", mock_create_archive)
|
|
|
|
archive_path = temp_dir / "test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 1
|
|
|
|
def test_keyboard_interrupt_in_add(self, temp_dir, monkeypatch):
|
|
"""Test KeyboardInterrupt handling in add command."""
|
|
|
|
# Create a file to add
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("test content")
|
|
|
|
# Mock create_archive to raise KeyboardInterrupt
|
|
def mock_create_archive(*args, **kwargs):
|
|
raise KeyboardInterrupt()
|
|
|
|
monkeypatch.setattr("tzst.cli.create_archive", mock_create_archive)
|
|
|
|
archive_path = temp_dir / "test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 130 # SIGINT exit code
|
|
|
|
def test_generic_exception_in_add(self, temp_dir, monkeypatch):
|
|
"""Test generic Exception handling in add command."""
|
|
|
|
# Create a file to add
|
|
test_file = temp_dir / "test.txt"
|
|
test_file.write_text("test content")
|
|
|
|
# Mock create_archive to raise generic Exception
|
|
def mock_create_archive(*args, **kwargs):
|
|
raise Exception("Unexpected error")
|
|
|
|
monkeypatch.setattr("tzst.cli.create_archive", mock_create_archive)
|
|
|
|
archive_path = temp_dir / "test.tzst"
|
|
result = main(["a", str(archive_path), str(test_file)])
|
|
assert result == 1
|
|
|
|
def test_generic_exception_in_extract(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test generic Exception handling in extract command."""
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock extract_archive to raise generic Exception
|
|
def mock_extract_archive(*args, **kwargs):
|
|
raise Exception("Unexpected extraction error")
|
|
|
|
monkeypatch.setattr("tzst.cli.extract_archive", mock_extract_archive)
|
|
|
|
result = main(["x", str(archive_path)])
|
|
assert result == 1
|
|
|
|
def test_generic_exception_in_list(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test generic Exception handling in list command."""
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock list_archive to raise generic Exception
|
|
def mock_list_archive(*args, **kwargs):
|
|
raise Exception("Unexpected list error")
|
|
|
|
monkeypatch.setattr("tzst.cli.list_archive", mock_list_archive)
|
|
|
|
result = main(["l", str(archive_path)])
|
|
assert result == 1
|
|
|
|
def test_generic_exception_in_test(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test generic Exception handling in test command."""
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock test_archive to raise generic Exception
|
|
def mock_test_archive(*args, **kwargs):
|
|
raise Exception("Unexpected test error")
|
|
|
|
monkeypatch.setattr("tzst.cli.test_archive", mock_test_archive)
|
|
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 1
|
|
|
|
def test_test_archive_failure(self, sample_files, temp_dir, monkeypatch):
|
|
"""Test archive test failure detection."""
|
|
|
|
# Create a valid archive first
|
|
file_paths = [str(f) for f in sample_files if f.is_file()]
|
|
archive_path = temp_dir / "test.tzst"
|
|
main(["a", str(archive_path), *file_paths])
|
|
|
|
# Mock test_archive to return False (test failed)
|
|
def mock_test_archive(*args, **kwargs):
|
|
return False
|
|
|
|
monkeypatch.setattr("tzst.cli.test_archive", mock_test_archive)
|
|
|
|
result = main(["t", str(archive_path)])
|
|
assert result == 1
|
|
|
|
|
|
class TestCLIArgumentParsingEdgeCases:
|
|
"""Test advanced argument parsing edge cases."""
|
|
|
|
def test_validation_functions_edge_cases(self):
|
|
"""Test validation functions with edge cases."""
|
|
from tzst.cli import (
|
|
_validate_compression_level_in_argv,
|
|
_validate_filter_in_argv,
|
|
)
|
|
|
|
# Test compression level validation with edge cases
|
|
assert (
|
|
_validate_compression_level_in_argv(["x", "test.tzst", "-l", "1"]) is False
|
|
) # Valid level returns False
|
|
assert (
|
|
_validate_compression_level_in_argv(["x", "test.tzst", "-l", "22"]) is False
|
|
) # Valid level returns False
|
|
assert (
|
|
_validate_compression_level_in_argv(["x", "test.tzst", "-l", "0"]) is True
|
|
) # Invalid level returns True
|
|
assert (
|
|
_validate_compression_level_in_argv(["x", "test.tzst", "-l", "23"]) is True
|
|
) # Invalid level returns True
|
|
assert (
|
|
_validate_compression_level_in_argv(["x", "test.tzst", "-l", "abc"]) is True
|
|
) # Invalid level returns True
|
|
assert (
|
|
_validate_compression_level_in_argv(["x", "test.tzst"]) is False
|
|
) # No compression level returns False
|
|
|
|
# Test filter validation with edge cases
|
|
assert (
|
|
_validate_filter_in_argv(["x", "test.tzst", "--filter", "data"]) is False
|
|
) # Valid filter returns False
|
|
assert (
|
|
_validate_filter_in_argv(["x", "test.tzst", "--filter", "tar"]) is False
|
|
) # Valid filter returns False
|
|
assert (
|
|
_validate_filter_in_argv(["x", "test.tzst", "--filter", "invalid"]) is True
|
|
) # Invalid filter returns True
|
|
assert (
|
|
_validate_filter_in_argv(["x", "test.tzst"]) is False
|
|
) # No filter returns False
|
|
|
|
def test_handle_parsing_errors_edge_cases(self):
|
|
"""Test _handle_parsing_errors with various edge cases."""
|
|
from tzst.cli import _handle_parsing_errors
|
|
|
|
# Test with different exit codes
|
|
e = SystemExit(0)
|
|
result = _handle_parsing_errors(e, ["--help"])
|
|
assert result == 0
|
|
|
|
e = SystemExit(1)
|
|
result = _handle_parsing_errors(e, ["some", "args"])
|
|
assert result == 1
|
|
|
|
e = SystemExit(None)
|
|
result = _handle_parsing_errors(e, [])
|
|
assert result == 1 # Test with compression level error
|
|
e = SystemExit(2)
|
|
result = _handle_parsing_errors(e, ["a", "test.tzst", "file.txt", "-l", "1000"])
|
|
assert result == 1
|
|
|
|
# Test with filter error
|
|
e = SystemExit(2)
|
|
result = _handle_parsing_errors(e, ["x", "test.tzst", "--filter", "badfilter"])
|
|
assert result == 1
|
|
|
|
def test_parse_arguments_edge_cases(self):
|
|
"""Test _parse_arguments with edge cases."""
|
|
from tzst.cli import _parse_arguments, create_parser
|
|
|
|
parser = create_parser()
|
|
|
|
# Test successful parsing
|
|
args, error_code = _parse_arguments(parser, ["a", "test.tzst", "file.txt"])
|
|
assert args is not None
|
|
assert error_code is None
|
|
|
|
# Test parsing error
|
|
args, error_code = _parse_arguments(parser, ["invalid"])
|
|
assert args is None
|
|
assert error_code == 2
|
|
|
|
def test_execute_command_edge_cases(self):
|
|
"""Test _execute_command with edge cases."""
|
|
from tzst.cli import _execute_command, create_parser
|
|
|
|
parser = create_parser()
|
|
|
|
# Test with version flag
|
|
class MockArgsVersion:
|
|
version = True
|
|
|
|
args = MockArgsVersion()
|
|
result = _execute_command(args, parser)
|
|
assert result == 0
|
|
|
|
# Test with no function attribute
|
|
class MockArgsNoFunc:
|
|
version = False
|
|
|
|
args = MockArgsNoFunc()
|
|
result = _execute_command(args, parser)
|
|
assert result == 1
|