Storm Dragon
293 lines
10 KiB
Python
293 lines
10 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
|
|
def _load_progress_module():
|
|
module_path = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "src"
|
|
/ "fenrirscreenreader"
|
|
/ "commands"
|
|
/ "onScreenUpdate"
|
|
/ "65000-progress_detector.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"fenrir_progress_detector", module_path
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_skips_typing_delta():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
command.env = {
|
|
"commandBuffer": {"progress_monitoring": True},
|
|
"runtime": {"DebugManager": Mock(write_debug_out=Mock())},
|
|
"screen": {"new_delta": "x", "new_delta_is_typing": True},
|
|
}
|
|
command.is_current_line_prompt = Mock(return_value=False)
|
|
command.is_real_progress_update = Mock(return_value=True)
|
|
command.detect_progress = Mock()
|
|
|
|
command.run()
|
|
|
|
command.is_real_progress_update.assert_not_called()
|
|
command.detect_progress.assert_not_called()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_allows_long_tqdm_transfer_delta():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
sample = (
|
|
"88%|"
|
|
"████████████████████████████████████████████████████████████████"
|
|
"████████████████████████████████████████████████████████████████"
|
|
"████████████████████████████████████████████████████████████████"
|
|
"█████████████████████████▊ "
|
|
"| 843M/954M [00:54<00:07, 15.2MB/s]"
|
|
)
|
|
command.env = {
|
|
"commandBuffer": {"progress_monitoring": True},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
|
|
assert len(sample) > 200
|
|
assert command.is_real_progress_update()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_beeps_for_long_tqdm_transfer_delta():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
sample = (
|
|
"90%|"
|
|
"████████████████████████████████████████████████████████████████"
|
|
"████████████████████████████████████████████████████████████████"
|
|
"████████████████████████████████████████████████████████████████"
|
|
"█████████████████████████████████████▍ "
|
|
"| 856M/954M [00:56<00:14, 6.78MB/s]"
|
|
)
|
|
command.env = {
|
|
"commandBuffer": {
|
|
"progress_monitoring": True,
|
|
"lastProgressValue": -1,
|
|
"lastProgressTime": 0,
|
|
},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_delta_is_typing": False,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
command.play_progress_tone = Mock()
|
|
|
|
command.run()
|
|
|
|
command.play_progress_tone.assert_called_once_with(90.0)
|
|
assert command.env["commandBuffer"]["lastProgressValue"] == 90.0
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_beeps_for_interruptible_status_without_ellipsis():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
sample = "◦ Files: (1m 04s • esc to interrupt)"
|
|
command.env = {
|
|
"commandBuffer": {
|
|
"progress_monitoring": True,
|
|
"lastProgressValue": -1,
|
|
"lastProgressTime": 0,
|
|
},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_delta_is_typing": False,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
command.play_activity_beep = Mock()
|
|
|
|
command.run()
|
|
|
|
command.play_activity_beep.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_allows_long_key_value_status_delta():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
sample = (
|
|
"frame=33480 fps=429 q=28.0 size= 195328KiB "
|
|
"time=00:23:16.26 bitrate=1146.0kbits/s dup=0 drop=1 "
|
|
"speed=17.9x elapsed=0:01:18.01 "
|
|
"filter=scale,format output=archive-video-final-render-pass.mkv "
|
|
"metadata=preserve-chapters-and-stream-layout"
|
|
)
|
|
command.env = {
|
|
"commandBuffer": {"progress_monitoring": True},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
|
|
assert len(sample) > 200
|
|
assert command.is_real_progress_update()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_beeps_for_ffmpeg_status_line():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
sample = (
|
|
"frame=33480 fps=429 q=28.0 size= 195328KiB "
|
|
"time=00:23:16.26 bitrate=1146.0kbits/s dup=0 drop=1 "
|
|
"speed=17.9x elapsed=0:01:18.01"
|
|
)
|
|
command.env = {
|
|
"commandBuffer": {
|
|
"progress_monitoring": True,
|
|
"lastProgressValue": -1,
|
|
"lastProgressTime": 0,
|
|
},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_delta_is_typing": False,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
command.play_activity_beep = Mock()
|
|
|
|
command.run()
|
|
|
|
command.play_activity_beep.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_ignores_generic_key_value_text():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
sample = "host=server mode=debug user=Username retry=3 status=ready"
|
|
command.env = {
|
|
"commandBuffer": {
|
|
"progress_monitoring": True,
|
|
"lastProgressValue": -1,
|
|
"lastProgressTime": 0,
|
|
},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_delta_is_typing": False,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
command.play_activity_beep = Mock()
|
|
command.play_progress_tone = Mock()
|
|
|
|
command.run()
|
|
|
|
command.play_activity_beep.assert_not_called()
|
|
command.play_progress_tone.assert_not_called()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_progress_detector_beeps_for_pacman_total_in_multiline_delta():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
total_line = (
|
|
"Total ( 3/749) "
|
|
"830.7 MiB 4.70 MiB/s 04:05 "
|
|
"[#################################################"
|
|
"-----------------------------------------------------------------------] 41%"
|
|
)
|
|
sample = (
|
|
"package-one 125.0 MiB 2.10 MiB/s 00:12 "
|
|
"[####################--------------------] 50%\n"
|
|
+ total_line
|
|
)
|
|
command.env = {
|
|
"commandBuffer": {
|
|
"progress_monitoring": True,
|
|
"lastProgressValue": -1,
|
|
"lastProgressTime": 0,
|
|
},
|
|
"runtime": {
|
|
"DebugManager": Mock(write_debug_out=Mock()),
|
|
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
|
|
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
|
|
},
|
|
"screen": {
|
|
"new_delta": sample,
|
|
"new_delta_is_typing": False,
|
|
"new_content_text": sample,
|
|
"old_cursor": {"x": 0, "y": 0},
|
|
"new_cursor": {"x": 0, "y": 0},
|
|
},
|
|
}
|
|
command.play_progress_tone = Mock()
|
|
|
|
command.run()
|
|
|
|
command.play_progress_tone.assert_called_once_with(41.0)
|
|
assert command.env["commandBuffer"]["lastProgressValue"] == 41.0
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_explicit_progress_delta_ignores_percentage_in_bracketed_prose():
|
|
progress_module = _load_progress_module()
|
|
command = progress_module.command()
|
|
|
|
assert not command.is_explicit_progress_delta(
|
|
"Release notes [issue #749] are 41% complete"
|
|
)
|