841 lines
25 KiB
Python
841 lines
25 KiB
Python
"""
|
|
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
|