Files
zk-data-agent/tests/test_jupyter_runtime.py
2026-06-15 17:14:54 +08:00

72 lines
2.4 KiB
Python

from __future__ import annotations
import unittest
from src.jupyter_runtime import (
build_terminal_script_launcher,
clean_terminal_output,
extract_terminal_command_output,
)
class TestJupyterRuntimeOutputFiltering(unittest.TestCase):
def test_extract_terminal_command_output_drops_wrapper_echo(self) -> None:
start_marker = '__ZK_AGENT_START_test__'
raw = (
'root@host:/workspace# mkdir -p /workspace && cd /workspace && '
'printf "\\n__ZK_AGENT_START_test__\\n"; '
'( setsid bash -lc \'export A=1; echo real-output\' ) &\r\n'
f'{start_marker}\r\n'
'real-output\r\n'
)
output = extract_terminal_command_output(
clean_terminal_output(raw),
start_marker=start_marker,
)
self.assertEqual(output, 'real-output')
def test_extract_terminal_command_output_uses_last_marker(self) -> None:
start_marker = '__ZK_AGENT_START_test__'
output = extract_terminal_command_output(
f'echo text containing {start_marker}\n{start_marker}\nactual',
start_marker=start_marker,
)
self.assertEqual(output, 'actual')
def test_clean_terminal_output_drops_wrapper_job_control_line(self) -> None:
cleaned = clean_terminal_output(
'real-output\r\n'
'[1] + Done (setsid bash -lc "export A=1; lscpu")\r\n'
)
self.assertEqual(cleaned, 'real-output')
def test_clean_terminal_output_drops_continuation_prompt(self) -> None:
cleaned = clean_terminal_output('real-output\r\n# >\r\n')
self.assertEqual(cleaned, 'real-output')
def test_terminal_script_launcher_hides_complex_wrapper_from_shell(self) -> None:
launcher = build_terminal_script_launcher(
script_path='/workspace/scratchpad/.run_scripts/run.sh',
marker='__ZK_AGENT_EXIT_test__',
)
self.assertNotIn('setsid bash -lc', launcher)
self.assertEqual(
launcher,
'bash /workspace/scratchpad/.run_scripts/run.sh; '
'__zk_outer_status=$?; '
'rm -f /workspace/scratchpad/.run_scripts/run.sh; '
'if [ "$__zk_outer_status" -ne 0 ]; then '
'printf "\\n__ZK_AGENT_EXIT_test__:%s\\n" "$__zk_outer_status"; '
'fi',
)
if __name__ == '__main__':
unittest.main()