""" PyTest: script_handling/progress """ import time from unittest.mock import patch from pytest import CaptureFixture from corelibs.script_handling.progress import Progress class TestProgressInit: """Test suite for Progress initialization""" def test_default_initialization(self): """Test Progress initialization with default parameters""" prg = Progress() assert prg.verbose is False assert prg.precision == 1 assert prg.microtime == 0 assert prg.wide_time is False assert prg.prefix_lb is False assert prg.linecount == 0 assert prg.filesize == 0 assert prg.count == 0 assert prg.start is not None def test_initialization_with_verbose(self): """Test Progress initialization with verbose enabled""" prg = Progress(verbose=1) assert prg.verbose is True prg = Progress(verbose=5) assert prg.verbose is True prg = Progress(verbose=0) assert prg.verbose is False def test_initialization_with_precision(self): """Test Progress initialization with different precision values""" # Normal precision prg = Progress(precision=0) assert prg.precision == 0 assert prg.percent_print == 3 prg = Progress(precision=2) assert prg.precision == 2 assert prg.percent_print == 6 prg = Progress(precision=10) assert prg.precision == 10 assert prg.percent_print == 14 # Ten step precision prg = Progress(precision=-1) assert prg.precision == 0 assert prg.precision_ten_step == 10 assert prg.percent_print == 3 # Five step precision prg = Progress(precision=-2) assert prg.precision == 0 assert prg.precision_ten_step == 5 assert prg.percent_print == 3 def test_initialization_with_microtime(self): """Test Progress initialization with microtime settings""" prg = Progress(microtime=-1) assert prg.microtime == -1 prg = Progress(microtime=0) assert prg.microtime == 0 prg = Progress(microtime=1) assert prg.microtime == 1 def test_initialization_with_wide_time(self): """Test Progress initialization with wide_time flag""" prg = Progress(wide_time=True) assert prg.wide_time is True prg = Progress(wide_time=False) assert prg.wide_time is False def test_initialization_with_prefix_lb(self): """Test Progress initialization with prefix line break""" prg = Progress(prefix_lb=True) assert prg.prefix_lb is True prg = Progress(prefix_lb=False) assert prg.prefix_lb is False def test_initialization_combined_parameters(self): """Test Progress initialization with multiple parameters""" prg = Progress(verbose=1, precision=2, microtime=1, wide_time=True, prefix_lb=True) assert prg.verbose is True assert prg.precision == 2 assert prg.microtime == 1 assert prg.wide_time is True assert prg.prefix_lb is True class TestProgressSetters: """Test suite for Progress setter methods""" def test_set_verbose(self): """Test set_verbose method""" prg = Progress() assert prg.set_verbose(1) is True assert prg.verbose is True assert prg.set_verbose(10) is True assert prg.verbose is True assert prg.set_verbose(0) is False assert prg.verbose is False def test_set_precision(self): """Test set_precision method""" prg = Progress() # Valid precision values assert prg.set_precision(0) == 0 assert prg.precision == 0 assert prg.set_precision(5) == 5 assert prg.precision == 5 assert prg.set_precision(10) == 10 assert prg.precision == 10 # Ten step precision prg.set_precision(-1) assert prg.precision == 0 assert prg.precision_ten_step == 10 # Five step precision prg.set_precision(-2) assert prg.precision == 0 assert prg.precision_ten_step == 5 # Invalid precision (too low) assert prg.set_precision(-3) == 0 assert prg.precision == 0 # Invalid precision (too high) assert prg.set_precision(11) == 0 assert prg.precision == 0 def test_set_linecount(self): """Test set_linecount method""" prg = Progress() assert prg.set_linecount(100) == 100 assert prg.linecount == 100 assert prg.set_linecount(1000) == 1000 assert prg.linecount == 1000 # Zero or negative should set to 1 assert prg.set_linecount(0) == 1 assert prg.linecount == 1 assert prg.set_linecount(-10) == 1 assert prg.linecount == 1 def test_set_filesize(self): """Test set_filesize method""" prg = Progress() assert prg.set_filesize(1024) == 1024 assert prg.filesize == 1024 assert prg.set_filesize(1048576) == 1048576 assert prg.filesize == 1048576 # Zero or negative should set to 1 assert prg.set_filesize(0) == 1 assert prg.filesize == 1 assert prg.set_filesize(-100) == 1 assert prg.filesize == 1 def test_set_wide_time(self): """Test set_wide_time method""" prg = Progress() assert prg.set_wide_time(True) is True assert prg.wide_time is True assert prg.set_wide_time(False) is False assert prg.wide_time is False def test_set_micro_time(self): """Test set_micro_time method""" prg = Progress() assert prg.set_micro_time(-1) == -1 assert prg.microtime == -1 assert prg.set_micro_time(0) == 0 assert prg.microtime == 0 assert prg.set_micro_time(1) == 1 assert prg.microtime == 1 def test_set_prefix_lb(self): """Test set_prefix_lb method""" prg = Progress() assert prg.set_prefix_lb(True) is True assert prg.prefix_lb is True assert prg.set_prefix_lb(False) is False assert prg.prefix_lb is False def test_set_start_time(self): """Test set_start_time method""" prg = Progress() initial_start = prg.start # Wait a bit and set new start time time.sleep(0.01) new_time = time.time() prg.set_start_time(new_time) # Original start should not change assert prg.start == initial_start # But start_time and start_run should update assert prg.start_time == new_time assert prg.start_run == new_time def test_set_start_time_custom_value(self): """Test set_start_time with custom time value""" prg = Progress() custom_time = 1234567890.0 prg.start = None # Reset start to test first-time setting prg.set_start_time(custom_time) assert prg.start == custom_time assert prg.start_time == custom_time assert prg.start_run == custom_time def test_set_eta_start_time(self): """Test set_eta_start_time method""" prg = Progress() custom_time = time.time() + 100 prg.set_eta_start_time(custom_time) assert prg.start_time == custom_time assert prg.start_run == custom_time def test_set_end_time(self): """Test set_end_time method""" prg = Progress() start_time = time.time() prg.set_start_time(start_time) time.sleep(0.01) end_time = time.time() prg.set_end_time(end_time) assert prg.end == end_time assert prg.end_time == end_time assert prg.run_time is not None assert prg.run_time > 0 def test_set_end_time_with_none_start(self): """Test set_end_time when start is None""" prg = Progress() prg.start = None end_time = time.time() prg.set_end_time(end_time) assert prg.end == end_time assert prg.run_time == end_time class TestProgressReset: """Test suite for Progress reset method""" def test_reset_basic(self): """Test reset method resets counter variables""" prg = Progress() prg.set_linecount(1000) prg.set_filesize(10240) prg.count = 500 prg.current_count = 500 prg.lines_processed = 100 prg.reset() assert prg.count == 0 assert prg.current_count == 0 assert prg.linecount == 0 assert prg.lines_processed == 0 assert prg.filesize == 0 assert prg.last_percent == 0 def test_reset_preserves_start(self): """Test reset preserves the original start time""" prg = Progress() original_start = prg.start prg.reset() # Original start should still be set from initialization assert prg.start == original_start def test_reset_clears_runtime_data(self): """Test reset clears runtime calculation data""" prg = Progress() prg.eta = 100.5 prg.full_time_needed = 50.2 prg.last_group = 10.1 prg.lines_in_last_group = 5.5 prg.lines_in_global = 3.3 prg.reset() assert prg.eta == 0 assert prg.full_time_needed == 0 assert prg.last_group == 0 assert prg.lines_in_last_group == 0 assert prg.lines_in_global == 0 class TestProgressShowPosition: """Test suite for Progress show_position method""" def test_show_position_basic_linecount(self): """Test show_position with basic line count""" prg = Progress(verbose=0) prg.set_linecount(100) # Process some lines for _ in range(10): prg.show_position() assert prg.count == 10 assert prg.file_pos == 10 def test_show_position_with_filesize(self): """Test show_position with file size parameter""" prg = Progress(verbose=0) prg.set_filesize(1024) prg.show_position(512) assert prg.count == 1 assert prg.file_pos == 512 assert prg.count_size == 512 def test_show_position_percent_calculation(self): """Test show_position calculates percentage correctly""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # Process 50 lines for _ in range(50): prg.show_position() assert prg.last_percent == 50.0 def test_show_position_ten_step_precision(self): """Test show_position with ten step precision""" prg = Progress(verbose=0, precision=-1) prg.set_linecount(100) # Process lines, should only update at 10% intervals for _ in range(15): prg.show_position() # Should be at 10% (not 15%) assert prg.last_percent == 10 def test_show_position_five_step_precision(self): """Test show_position with five step precision""" prg = Progress(verbose=0, precision=-2) prg.set_linecount(100) # Process lines, should only update at 5% intervals for _ in range(7): prg.show_position() # Should be at 5% (not 7%) assert prg.last_percent == 5 def test_show_position_change_flag(self): """Test show_position sets change flag correctly""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # First call should trigger change (at 1%) prg.show_position() assert prg.change == 1 last_percent = prg.last_percent # Keep calling - each percent increment triggers change prg.show_position() # At precision=0, each 1% is a new change if prg.last_percent != last_percent: assert prg.change == 1 else: assert prg.change == 0 def test_show_position_with_verbose_output(self, capsys: CaptureFixture[str]): """Test show_position produces output when verbose is enabled""" prg = Progress(verbose=1, precision=0) prg.set_linecount(100) # Process until percent changes for _ in range(10): prg.show_position() captured = capsys.readouterr() assert "Processed" in captured.out assert "Lines" in captured.out def test_show_position_with_prefix_lb(self): """Test show_position with prefix line break""" prg = Progress(verbose=1, precision=0, prefix_lb=True) prg.set_linecount(100) # Process until percent changes for _ in range(10): prg.show_position() assert prg.string.startswith("\n") def test_show_position_lines_processed_calculation(self): """Test show_position calculates lines processed correctly""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # First call at 1% prg.show_position() first_lines_processed = prg.lines_processed assert first_lines_processed == 1 # Process to 2% (need to process 1 more line) prg.show_position() # lines_processed should be 1 (from 1 to 2) assert prg.lines_processed == 1 def test_show_position_eta_calculation(self): """Test show_position calculates ETA""" prg = Progress(verbose=0, precision=0) prg.set_linecount(1000) # We need to actually process lines for percent to change # Process 100 lines to get to ~10% for _ in range(100): prg.show_position() # ETA should be set after percent changes assert prg.eta is not None assert prg.eta >= 0 def test_show_position_with_filesize_output(self, capsys: CaptureFixture[str]): """Test show_position output with filesize information""" prg = Progress(verbose=1, precision=0) prg.set_filesize(10240) # Process with filesize for i in range(1, 1025): prg.show_position(i) captured = capsys.readouterr() # Should contain byte information assert "B" in captured.out or "KB" in captured.out def test_show_position_bytes_calculation(self): """Test show_position calculates bytes per second""" prg = Progress(verbose=0, precision=0) prg.set_filesize(10240) # Process enough bytes to trigger a percent change # Need to process ~102 bytes for 1% of 10240 prg.show_position(102) # After percent change, bytes stats should be set assert prg.bytes_in_last_group >= 0 assert prg.bytes_in_global >= 0 def test_show_position_current_count_tracking(self): """Test show_position tracks current count correctly""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) for _ in range(10): prg.show_position() # Current count should be updated to last change point assert prg.current_count == 10 assert prg.count == 10 def test_show_position_full_time_calculation(self): """Test show_position calculates full time needed""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # Process enough to trigger percent change for _ in range(10): prg.show_position() assert prg.full_time_needed is not None assert prg.full_time_needed >= 0 def test_show_position_last_group_time(self): """Test show_position tracks last group time""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # Process enough to trigger percent change for _ in range(10): prg.show_position() # last_group should be set after percent change assert prg.last_group >= 0 def test_show_position_zero_eta_edge_case(self): """Test show_position handles negative ETA gracefully""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # Process all lines for _ in range(100): prg.show_position() # ETA should not be negative assert prg.eta is not None assert prg.eta >= 0 def test_show_position_no_filesize_string_format(self): """Test show_position string format without filesize""" prg = Progress(verbose=1, precision=0) prg.set_linecount(100) for _ in range(10): prg.show_position() # String should not contain byte information assert "b/s" not in prg.string assert "Lines" in prg.string def test_show_position_wide_time_format(self): """Test show_position with wide time formatting""" prg = Progress(verbose=1, precision=0, wide_time=True) prg.set_linecount(100) for _ in range(10): prg.show_position() # With wide_time, time fields should be formatted with specific width assert prg.string != "" def test_show_position_microtime_on(self): """Test show_position with microtime enabled""" prg = Progress(verbose=0, precision=0, microtime=1) prg.set_linecount(100) with patch('time.time') as mock_time: mock_time.return_value = 1000.0 prg.set_start_time(1000.0) mock_time.return_value = 1000.5 for _ in range(10): prg.show_position() # Microtime should be enabled assert prg.microtime == 1 def test_show_position_microtime_off(self): """Test show_position with microtime disabled""" prg = Progress(verbose=0, precision=0, microtime=-1) prg.set_linecount(100) for _ in range(10): prg.show_position() assert prg.microtime == -1 def test_show_position_lines_per_second_global(self): """Test show_position calculates global lines per second""" prg = Progress(verbose=0, precision=0) prg.set_linecount(1000) # Process 100 lines to trigger percent changes for _ in range(100): prg.show_position() # After processing, lines_in_global should be calculated assert prg.lines_in_global >= 0 def test_show_position_lines_per_second_last_group(self): """Test show_position calculates last group lines per second""" prg = Progress(verbose=0, precision=0) prg.set_linecount(1000) # Process lines to trigger percent changes for _ in range(100): prg.show_position() # After processing, lines_in_last_group should be calculated assert prg.lines_in_last_group >= 0 def test_show_position_returns_string(self): """Test show_position returns the progress string""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) result = "" for _ in range(10): result = prg.show_position() # Should return string on percent change assert isinstance(result, str) class TestProgressEdgeCases: """Test suite for edge cases and error conditions""" def test_zero_linecount_protection(self): """Test Progress handles zero linecount gracefully""" prg = Progress(verbose=0) prg.set_filesize(1024) # Should not crash with zero linecount prg.show_position(512) assert prg.file_pos == 512 def test_zero_filesize_protection(self): """Test Progress handles zero filesize gracefully""" prg = Progress(verbose=0) prg.set_linecount(100) # Should not crash with zero filesize prg.show_position() assert isinstance(prg.string, str) def test_division_by_zero_protection_last_group(self): """Test Progress protects against division by zero in last_group""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) with patch('time.time') as mock_time: # Same time for start and end mock_time.return_value = 1000.0 prg.set_start_time(1000.0) for _ in range(10): prg.show_position() # Should handle zero time difference assert prg.lines_in_last_group >= 0 def test_division_by_zero_protection_full_time(self): """Test Progress protects against division by zero in full_time_needed""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # Process lines very quickly for _ in range(10): prg.show_position() # Should handle very small time differences without crashing # lines_in_global should be a valid number (>= 0) assert isinstance(prg.lines_in_global, (int, float)) def test_none_start_protection(self): """Test Progress handles None start time""" prg = Progress(verbose=0, precision=0) prg.start = None prg.set_linecount(100) # Should not crash prg.show_position() assert prg.start == 0 def test_none_start_time_protection(self): """Test Progress handles None start_time""" prg = Progress(verbose=0, precision=0) prg.start_time = None prg.set_linecount(100) # Should not crash and should set start_time during processing prg.show_position() # start_time will be set to 0 internally when None is encountered # But during percent calculation, it may be reset to current time assert prg.start_time is not None def test_precision_boundary_values(self): """Test precision at boundary values""" prg = Progress() # Minimum valid assert prg.set_precision(-2) == 0 # Maximum valid assert prg.set_precision(10) == 10 # Below minimum assert prg.set_precision(-3) == 0 # Above maximum assert prg.set_precision(11) == 0 def test_large_linecount_handling(self): """Test Progress handles large linecount values""" prg = Progress(verbose=0) large_count = 10_000_000 prg.set_linecount(large_count) assert prg.linecount == large_count # Should handle calculations without overflow prg.show_position() assert prg.count == 1 def test_large_filesize_handling(self): """Test Progress handles large filesize values""" prg = Progress(verbose=0) large_size = 10_737_418_240 # 10 GB prg.set_filesize(large_size) assert prg.filesize == large_size # Should handle calculations without overflow prg.show_position(1024) assert prg.file_pos == 1024 class TestProgressIntegration: """Integration tests for Progress class""" def test_complete_progress_workflow(self, capsys: CaptureFixture[str]): """Test complete progress workflow from start to finish""" prg = Progress(verbose=1, precision=0) prg.set_linecount(100) # Simulate processing for _ in range(100): prg.show_position() prg.set_end_time() assert prg.count == 100 assert prg.last_percent == 100.0 assert prg.run_time is not None captured = capsys.readouterr() assert "Processed" in captured.out def test_progress_with_filesize_workflow(self): """Test progress workflow with file size tracking""" prg = Progress(verbose=0, precision=0) prg.set_filesize(10240) # Simulate reading file in chunks for pos in range(0, 10240, 1024): prg.show_position(pos + 1024) assert prg.count == 10 assert prg.count_size == 10240 def test_reset_and_reuse(self): """Test resetting and reusing Progress instance""" prg = Progress(verbose=0, precision=0) # First run prg.set_linecount(100) for _ in range(100): prg.show_position() assert prg.count == 100 # Reset prg.reset() assert prg.count == 0 # Second run prg.set_linecount(50) for _ in range(50): prg.show_position() assert prg.count == 50 def test_multiple_precision_changes(self): """Test changing precision multiple times""" prg = Progress(verbose=0) prg.set_precision(0) assert prg.precision == 0 prg.set_precision(2) assert prg.precision == 2 prg.set_precision(-1) assert prg.precision == 0 assert prg.precision_ten_step == 10 def test_eta_start_time_adjustment(self): """Test adjusting ETA start time mid-processing""" prg = Progress(verbose=0, precision=0) prg.set_linecount(1000) # Process some lines for _ in range(100): prg.show_position() # Adjust ETA start time (simulating delay like DB query) new_time = time.time() prg.set_eta_start_time(new_time) # Continue processing for _ in range(100): prg.show_position() assert prg.start_run == new_time def test_verbose_toggle_during_processing(self): """Test toggling verbose flag during processing""" prg = Progress(verbose=0, precision=0) prg.set_linecount(100) # Process without output for _ in range(50): prg.show_position() # Enable verbose prg.set_verbose(1) assert prg.verbose is True # Continue with output for _ in range(50): prg.show_position() assert prg.count == 100