fenrir/tests/unit/test_pty_terminal_sequences.py

import threading
import time
from unittest.mock import Mock

import pytest

from fenrirscreenreader.core.eventData import FenrirEventType
from fenrirscreenreader.screenDriver.ptyDriver import PTYConstants
from fenrirscreenreader.screenDriver.ptyDriver import Terminal
from fenrirscreenreader.screenDriver.ptyDriver import driver as PtyDriver


class DummyProcessInput:
    def __init__(self):
        self.data = []

    def write(self, data):
        self.data.append(data)


@pytest.mark.unit
def test_csi_sequences_with_intermediate_characters_do_not_render_final_byte():
    terminal = Terminal(10, 3, DummyProcessInput())

    terminal.feed(b"\x1b[2026$p\x1b[2048$pX")
    screen = terminal.get_screen_content()

    assert screen["text"].splitlines()[0] == "X         "
    assert "p" not in screen["text"]


@pytest.mark.unit
def test_private_sgr_sequence_from_fullscreen_apps_does_not_crash():
    terminal = Terminal(10, 3, DummyProcessInput())

    terminal.feed(b"\x1b[?25;7mX")
    screen = terminal.get_screen_content()

    assert screen["text"].splitlines()[0] == "X         "


@pytest.mark.unit
def test_optional_float_setting_uses_default_when_missing():
    settings_manager = type(
        "SettingsManager",
        (),
        {"get_setting": lambda self, section, setting: ""},
    )()
    pty_driver = PtyDriver()

    assert (
        pty_driver._get_optional_float_setting(
            settings_manager,
            "screen",
            "ptyOutputTimeout",
            PTYConstants.OUTPUT_READ_TIMEOUT,
        )
        == PTYConstants.OUTPUT_READ_TIMEOUT
    )


@pytest.mark.unit
def test_pty_stdin_input_interrupts_output_when_all_keys_interrupt_enabled():
    pty_driver = PtyDriver()
    settings_manager = Mock()
    settings_manager.get_setting_as_bool.return_value = True
    settings_manager.get_setting.return_value = ""
    output_manager = Mock()
    pty_driver.env = {
        "runtime": {
            "SettingsManager": settings_manager,
            "OutputManager": output_manager,
        }
    }

    pty_driver.interrupt_output_on_stdin_input(b"a")
    pty_driver.stdin_interrupt_thread.join(timeout=1.0)

    output_manager.interrupt_output.assert_called_once_with()


@pytest.mark.unit
def test_pty_stdin_input_interrupt_does_not_block_input_injection():
    pty_driver = PtyDriver()
    settings_manager = Mock()
    settings_manager.get_setting_as_bool.return_value = True
    settings_manager.get_setting.return_value = ""
    interrupt_started = threading.Event()
    release_interrupt = threading.Event()

    def slow_interrupt():
        interrupt_started.set()
        release_interrupt.wait(timeout=1.0)

    output_manager = Mock(interrupt_output=Mock(side_effect=slow_interrupt))
    pty_driver.env = {
        "input": {"curr_input": []},
        "runtime": {
            "SettingsManager": settings_manager,
            "OutputManager": output_manager,
            "DebugManager": Mock(write_debug_out=Mock()),
        },
    }
    pty_driver.inject_text_to_screen = Mock()

    start_time = time.monotonic()
    pty_driver.handle_stdin_input(b"a", Mock())
    elapsed = time.monotonic() - start_time

    try:
        assert interrupt_started.wait(timeout=0.2)
        assert elapsed < 0.2
        pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
    finally:
        release_interrupt.set()
        pty_driver.stdin_interrupt_thread.join(timeout=1.0)


@pytest.mark.unit
def test_pty_raw_tab_records_recent_tab_keypress():
    pty_driver = PtyDriver()
    settings_manager = Mock()
    settings_manager.get_setting_as_bool.return_value = False
    input_manager = Mock()
    pty_driver.env = {
        "input": {"curr_input": []},
        "runtime": {
            "DebugManager": Mock(write_debug_out=Mock()),
            "InputManager": input_manager,
            "SettingsManager": settings_manager,
        },
    }
    pty_driver.inject_text_to_screen = Mock()

    pty_driver.handle_stdin_input(b"\t", Mock())

    input_manager.record_unmanaged_keypress.assert_called_once_with("KEY_TAB")
    pty_driver.inject_text_to_screen.assert_called_once_with(b"\t")


@pytest.mark.unit
def test_pty_plain_stdin_does_not_record_tab_keypress():
    pty_driver = PtyDriver()
    settings_manager = Mock()
    settings_manager.get_setting_as_bool.return_value = False
    input_manager = Mock()
    pty_driver.env = {
        "input": {"curr_input": []},
        "runtime": {
            "DebugManager": Mock(write_debug_out=Mock()),
            "InputManager": input_manager,
            "SettingsManager": settings_manager,
        },
    }
    pty_driver.inject_text_to_screen = Mock()

    pty_driver.handle_stdin_input(b"a", Mock())

    input_manager.record_unmanaged_keypress.assert_not_called()
    pty_driver.inject_text_to_screen.assert_called_once_with(b"a")


@pytest.mark.unit
def test_pty_stdin_input_honors_interrupt_disabled():
    pty_driver = PtyDriver()
    settings_manager = Mock()
    settings_manager.get_setting_as_bool.return_value = False
    output_manager = Mock()
    pty_driver.env = {
        "runtime": {
            "SettingsManager": settings_manager,
            "OutputManager": output_manager,
        }
    }

    pty_driver.interrupt_output_on_stdin_input(b"a")

    output_manager.interrupt_output.assert_not_called()


@pytest.mark.unit
def test_pty_stdin_input_leaves_filtered_interrupts_to_key_events():
    pty_driver = PtyDriver()
    settings_manager = Mock()
    settings_manager.get_setting_as_bool.return_value = True
    settings_manager.get_setting.return_value = "KEY_ENTER"
    output_manager = Mock()
    pty_driver.env = {
        "runtime": {
            "SettingsManager": settings_manager,
            "OutputManager": output_manager,
        }
    }

    pty_driver.interrupt_output_on_stdin_input(b"a")

    output_manager.interrupt_output.assert_not_called()


@pytest.mark.unit
def test_pty_backspace_with_fenrir_key_synthesizes_shortcut_events():
    pty_driver = PtyDriver()
    event_queue = Mock()
    pty_driver.env = {
        "input": {"curr_input": ["KEY_FENRIR"]},
    }

    handled = pty_driver.synthesize_backspace_shortcut(b"\x7f", event_queue)

    assert handled is True
    assert event_queue.put.call_count == 2
    first_event = event_queue.put.call_args_list[0].args[0]
    second_event = event_queue.put.call_args_list[1].args[0]
    assert first_event["Type"] == FenrirEventType.keyboard_input
    assert first_event["data"]["event_name"] == "KEY_BACKSPACE"
    assert first_event["data"]["event_state"] == 1
    assert second_event["data"]["event_state"] == 0


@pytest.mark.unit
def test_pty_plain_backspace_is_not_synthesized():
    pty_driver = PtyDriver()
    event_queue = Mock()
    pty_driver.env = {
        "input": {"curr_input": []},
    }

    handled = pty_driver.synthesize_backspace_shortcut(b"\x7f", event_queue)

    assert handled is False
    event_queue.put.assert_not_called()