46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
from mka.application.progress_timing import ProcessingTimer
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self, value: float = 0.0) -> None:
|
|
self.value = value
|
|
|
|
def __call__(self) -> float:
|
|
return self.value
|
|
|
|
|
|
def test_completed_duration_is_retained_while_active_stage_increases() -> None:
|
|
clock = FakeClock(10.0)
|
|
timer = ProcessingTimer(clock)
|
|
timer.start_run()
|
|
timer.start_stage("preparing")
|
|
clock.value = 18.0
|
|
timer.finish_stage("preparing")
|
|
timer.start_stage("transcription")
|
|
|
|
clock.value = 20.0
|
|
first = timer.snapshot()
|
|
clock.value = 25.5
|
|
second = timer.snapshot()
|
|
|
|
assert first.stage_durations == {"preparing": 8.0, "transcription": 2.0}
|
|
assert second.stage_durations == {"preparing": 8.0, "transcription": 7.5}
|
|
assert second.total_duration == 15.5
|
|
|
|
|
|
def test_failed_active_stage_and_total_duration_are_frozen() -> None:
|
|
clock = FakeClock(2.0)
|
|
timer = ProcessingTimer(clock)
|
|
timer.start_run()
|
|
clock.value = 5.0
|
|
timer.start_stage("transcription")
|
|
clock.value = 14.0
|
|
timer.finish_run()
|
|
clock.value = 99.0
|
|
|
|
snapshot = timer.snapshot()
|
|
|
|
assert snapshot.stage_durations["transcription"] == 9.0
|
|
assert snapshot.total_duration == 12.0
|
|
assert snapshot.running is False
|