976 lines
35 KiB
Python
976 lines
35 KiB
Python
"""
|
|
Unit tests for debug_handling.writeline module
|
|
"""
|
|
|
|
import io
|
|
import pytest
|
|
from pytest import CaptureFixture
|
|
|
|
from corelibs.debug_handling.writeline import (
|
|
write_l,
|
|
pr_header,
|
|
pr_title,
|
|
pr_open,
|
|
pr_close,
|
|
pr_act
|
|
)
|
|
|
|
|
|
class TestWriteL:
|
|
"""Test cases for write_l function"""
|
|
|
|
def test_write_l_print_only(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l with print_line=True and no file"""
|
|
write_l("Test line", print_line=True)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "Test line\n"
|
|
|
|
def test_write_l_no_print_no_file(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l with print_line=False and no file (should do nothing)"""
|
|
write_l("Test line", print_line=False)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
|
|
def test_write_l_file_only(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l with file handler only (no print)"""
|
|
fpl = io.StringIO()
|
|
write_l("Test line", fpl=fpl, print_line=False)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
assert fpl.getvalue() == "Test line\n"
|
|
fpl.close()
|
|
|
|
def test_write_l_both_print_and_file(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l with both print and file output"""
|
|
fpl = io.StringIO()
|
|
write_l("Test line", fpl=fpl, print_line=True)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "Test line\n"
|
|
assert fpl.getvalue() == "Test line\n"
|
|
fpl.close()
|
|
|
|
def test_write_l_multiple_lines_to_file(self):
|
|
"""Test write_l writing multiple lines to file"""
|
|
fpl = io.StringIO()
|
|
write_l("Line 1", fpl=fpl, print_line=False)
|
|
write_l("Line 2", fpl=fpl, print_line=False)
|
|
write_l("Line 3", fpl=fpl, print_line=False)
|
|
assert fpl.getvalue() == "Line 1\nLine 2\nLine 3\n"
|
|
fpl.close()
|
|
|
|
def test_write_l_empty_string(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l with empty string"""
|
|
fpl = io.StringIO()
|
|
write_l("", fpl=fpl, print_line=True)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "\n"
|
|
assert fpl.getvalue() == "\n"
|
|
fpl.close()
|
|
|
|
def test_write_l_special_characters(self):
|
|
"""Test write_l with special characters"""
|
|
fpl = io.StringIO()
|
|
special_line = "Special: \t\n\r\\ 特殊文字 €"
|
|
write_l(special_line, fpl=fpl, print_line=False)
|
|
assert special_line + "\n" in fpl.getvalue()
|
|
fpl.close()
|
|
|
|
def test_write_l_long_string(self):
|
|
"""Test write_l with long string"""
|
|
fpl = io.StringIO()
|
|
long_line = "A" * 1000
|
|
write_l(long_line, fpl=fpl, print_line=False)
|
|
assert fpl.getvalue() == long_line + "\n"
|
|
fpl.close()
|
|
|
|
def test_write_l_unicode_content(self):
|
|
"""Test write_l with unicode content"""
|
|
fpl = io.StringIO()
|
|
unicode_line = "Hello 世界 🌍 Привет"
|
|
write_l(unicode_line, fpl=fpl, print_line=False)
|
|
assert fpl.getvalue() == unicode_line + "\n"
|
|
fpl.close()
|
|
|
|
def test_write_l_default_parameters(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l with default parameters"""
|
|
write_l("Test")
|
|
captured = capsys.readouterr()
|
|
# Default print_line is False
|
|
assert captured.out == ""
|
|
|
|
def test_write_l_with_newline_in_string(self):
|
|
"""Test write_l with newline characters in the string"""
|
|
fpl = io.StringIO()
|
|
write_l("Line with\nnewline", fpl=fpl, print_line=False)
|
|
assert fpl.getvalue() == "Line with\nnewline\n"
|
|
fpl.close()
|
|
|
|
|
|
class TestPrHeader:
|
|
"""Test cases for pr_header function"""
|
|
|
|
def test_pr_header_default(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with default parameters"""
|
|
pr_header("TEST")
|
|
captured = capsys.readouterr()
|
|
assert "#" in captured.out
|
|
assert "TEST" in captured.out
|
|
|
|
def test_pr_header_custom_marker(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with custom marker string"""
|
|
pr_header("TEST", marker_string="*")
|
|
captured = capsys.readouterr()
|
|
assert "*" in captured.out
|
|
assert "TEST" in captured.out
|
|
assert "#" not in captured.out
|
|
|
|
def test_pr_header_custom_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with custom width"""
|
|
pr_header("TEST", width=50)
|
|
captured = capsys.readouterr()
|
|
# Check that output is formatted
|
|
assert "TEST" in captured.out
|
|
|
|
def test_pr_header_short_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with short tag"""
|
|
pr_header("X")
|
|
captured = capsys.readouterr()
|
|
assert "X" in captured.out
|
|
assert "#" in captured.out
|
|
|
|
def test_pr_header_long_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with long tag"""
|
|
pr_header("This is a very long header tag")
|
|
captured = capsys.readouterr()
|
|
assert "This is a very long header tag" in captured.out
|
|
|
|
def test_pr_header_empty_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with empty tag"""
|
|
pr_header("")
|
|
captured = capsys.readouterr()
|
|
assert "#" in captured.out
|
|
|
|
def test_pr_header_special_characters(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with special characters in tag"""
|
|
pr_header("TEST: 123! @#$")
|
|
captured = capsys.readouterr()
|
|
assert "TEST: 123! @#$" in captured.out
|
|
|
|
def test_pr_header_unicode(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with unicode characters"""
|
|
pr_header("テスト 🎉")
|
|
captured = capsys.readouterr()
|
|
assert "テスト 🎉" in captured.out
|
|
|
|
def test_pr_header_various_markers(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with various marker strings"""
|
|
markers = ["*", "=", "-", "+", "~", "@"]
|
|
for marker in markers:
|
|
pr_header("TEST", marker_string=marker)
|
|
captured = capsys.readouterr()
|
|
assert marker in captured.out
|
|
assert "TEST" in captured.out
|
|
|
|
def test_pr_header_zero_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with width of 0"""
|
|
pr_header("TEST", width=0)
|
|
captured = capsys.readouterr()
|
|
assert "TEST" in captured.out
|
|
|
|
def test_pr_header_large_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with large width"""
|
|
pr_header("TEST", width=100)
|
|
captured = capsys.readouterr()
|
|
assert "TEST" in captured.out
|
|
assert "#" in captured.out
|
|
|
|
def test_pr_header_format(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header output format"""
|
|
pr_header("CENTER", marker_string="#", width=20)
|
|
captured = capsys.readouterr()
|
|
# Should have spaces around centered text
|
|
assert " CENTER " in captured.out or "CENTER" in captured.out
|
|
|
|
|
|
class TestPrTitle:
|
|
"""Test cases for pr_title function"""
|
|
|
|
def test_pr_title_default(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with default parameters"""
|
|
pr_title("Test Title")
|
|
captured = capsys.readouterr()
|
|
assert "Test Title" in captured.out
|
|
assert "|" in captured.out
|
|
assert "." in captured.out
|
|
assert ":" in captured.out
|
|
|
|
def test_pr_title_custom_prefix(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with custom prefix string"""
|
|
pr_title("Test", prefix_string=">")
|
|
captured = capsys.readouterr()
|
|
assert ">" in captured.out
|
|
assert "Test" in captured.out
|
|
assert "|" not in captured.out
|
|
|
|
def test_pr_title_custom_space_filler(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with custom space filler"""
|
|
pr_title("Test", space_filler="-")
|
|
captured = capsys.readouterr()
|
|
assert "Test" in captured.out
|
|
assert "-" in captured.out
|
|
assert "." not in captured.out
|
|
|
|
def test_pr_title_custom_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with custom width"""
|
|
pr_title("Test", width=50)
|
|
captured = capsys.readouterr()
|
|
assert "Test" in captured.out
|
|
|
|
def test_pr_title_short_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with short tag"""
|
|
pr_title("X")
|
|
captured = capsys.readouterr()
|
|
assert "X" in captured.out
|
|
assert "." in captured.out
|
|
|
|
def test_pr_title_long_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with long tag"""
|
|
pr_title("This is a very long title tag")
|
|
captured = capsys.readouterr()
|
|
assert "This is a very long title tag" in captured.out
|
|
|
|
def test_pr_title_empty_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with empty tag"""
|
|
pr_title("")
|
|
captured = capsys.readouterr()
|
|
assert "|" in captured.out
|
|
assert ":" in captured.out
|
|
|
|
def test_pr_title_special_characters(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with special characters"""
|
|
pr_title("Task #123!")
|
|
captured = capsys.readouterr()
|
|
assert "Task #123!" in captured.out
|
|
|
|
def test_pr_title_unicode(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with unicode characters"""
|
|
pr_title("タイトル 📝")
|
|
captured = capsys.readouterr()
|
|
assert "タイトル 📝" in captured.out
|
|
|
|
def test_pr_title_various_fillers(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with various space fillers"""
|
|
fillers = [".", "-", "_", "*", " ", "~"]
|
|
for filler in fillers:
|
|
pr_title("Test", space_filler=filler)
|
|
captured = capsys.readouterr()
|
|
assert "Test" in captured.out
|
|
|
|
def test_pr_title_zero_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with width of 0"""
|
|
pr_title("Test", width=0)
|
|
captured = capsys.readouterr()
|
|
assert "Test" in captured.out
|
|
|
|
def test_pr_title_large_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with large width"""
|
|
pr_title("Test", width=100)
|
|
captured = capsys.readouterr()
|
|
assert "Test" in captured.out
|
|
|
|
def test_pr_title_format_left_align(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title output format (should be left-aligned with filler)"""
|
|
pr_title("Start", space_filler=".", width=10)
|
|
captured = capsys.readouterr()
|
|
# Should have the tag followed by dots
|
|
assert "Start" in captured.out
|
|
assert ":" in captured.out
|
|
|
|
|
|
class TestPrOpen:
|
|
"""Test cases for pr_open function"""
|
|
|
|
def test_pr_open_default(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with default parameters"""
|
|
pr_open("Processing")
|
|
captured = capsys.readouterr()
|
|
assert "Processing" in captured.out
|
|
assert "|" in captured.out
|
|
assert "." in captured.out
|
|
assert "[" in captured.out
|
|
# Should not have newline at the end
|
|
assert not captured.out.endswith("\n")
|
|
|
|
def test_pr_open_custom_prefix(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with custom prefix string"""
|
|
pr_open("Task", prefix_string=">")
|
|
captured = capsys.readouterr()
|
|
assert ">" in captured.out
|
|
assert "Task" in captured.out
|
|
assert "|" not in captured.out
|
|
|
|
def test_pr_open_custom_space_filler(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with custom space filler"""
|
|
pr_open("Task", space_filler="-")
|
|
captured = capsys.readouterr()
|
|
assert "Task" in captured.out
|
|
assert "-" in captured.out
|
|
assert "." not in captured.out
|
|
|
|
def test_pr_open_custom_width(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with custom width"""
|
|
pr_open("Task", width=50)
|
|
captured = capsys.readouterr()
|
|
assert "Task" in captured.out
|
|
assert "[" in captured.out
|
|
|
|
def test_pr_open_short_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with short tag"""
|
|
pr_open("X")
|
|
captured = capsys.readouterr()
|
|
assert "X" in captured.out
|
|
assert "[" in captured.out
|
|
|
|
def test_pr_open_long_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with long tag"""
|
|
pr_open("This is a very long task tag")
|
|
captured = capsys.readouterr()
|
|
assert "This is a very long task tag" in captured.out
|
|
|
|
def test_pr_open_empty_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with empty tag"""
|
|
pr_open("")
|
|
captured = capsys.readouterr()
|
|
assert "[" in captured.out
|
|
assert "|" in captured.out
|
|
|
|
def test_pr_open_no_newline(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open doesn't end with newline"""
|
|
pr_open("Test")
|
|
captured = capsys.readouterr()
|
|
# Output should not end with newline (uses end="")
|
|
assert not captured.out.endswith("\n")
|
|
|
|
def test_pr_open_special_characters(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with special characters"""
|
|
pr_open("Loading: 50%")
|
|
captured = capsys.readouterr()
|
|
assert "Loading: 50%" in captured.out
|
|
|
|
def test_pr_open_unicode(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open with unicode characters"""
|
|
pr_open("処理中 ⏳")
|
|
captured = capsys.readouterr()
|
|
assert "処理中 ⏳" in captured.out
|
|
|
|
def test_pr_open_format(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_open output format"""
|
|
pr_open("Task", prefix_string="|", space_filler=".", width=20)
|
|
captured = capsys.readouterr()
|
|
assert "|" in captured.out
|
|
assert "Task" in captured.out
|
|
assert "[" in captured.out
|
|
|
|
|
|
class TestPrClose:
|
|
"""Test cases for pr_close function"""
|
|
|
|
def test_pr_close_default(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with default (empty) tag"""
|
|
pr_close()
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "]\n"
|
|
|
|
def test_pr_close_with_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with custom tag"""
|
|
pr_close("DONE")
|
|
captured = capsys.readouterr()
|
|
assert "DONE" in captured.out
|
|
assert "]" in captured.out
|
|
assert captured.out.endswith("\n")
|
|
|
|
def test_pr_close_with_space(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with space in tag"""
|
|
pr_close(" OK ")
|
|
captured = capsys.readouterr()
|
|
assert " OK " in captured.out
|
|
assert "]" in captured.out
|
|
|
|
def test_pr_close_empty_string(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with empty string (same as default)"""
|
|
pr_close("")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "]\n"
|
|
|
|
def test_pr_close_special_characters(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with special characters"""
|
|
pr_close("✓")
|
|
captured = capsys.readouterr()
|
|
assert "✓" in captured.out
|
|
assert "]" in captured.out
|
|
|
|
def test_pr_close_unicode(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with unicode characters"""
|
|
pr_close("完了")
|
|
captured = capsys.readouterr()
|
|
assert "完了" in captured.out
|
|
assert "]" in captured.out
|
|
|
|
def test_pr_close_newline(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close ends with newline"""
|
|
pr_close("OK")
|
|
captured = capsys.readouterr()
|
|
assert captured.out.endswith("\n")
|
|
|
|
def test_pr_close_various_tags(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with various tags"""
|
|
tags = ["OK", "DONE", "✓", "✗", "SKIP", "PASS", "FAIL"]
|
|
for tag in tags:
|
|
pr_close(tag)
|
|
captured = capsys.readouterr()
|
|
assert tag in captured.out
|
|
assert "]" in captured.out
|
|
|
|
|
|
class TestPrAct:
|
|
"""Test cases for pr_act function"""
|
|
|
|
def test_pr_act_default(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with default dot"""
|
|
pr_act()
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "."
|
|
assert not captured.out.endswith("\n")
|
|
|
|
def test_pr_act_custom_character(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with custom character"""
|
|
pr_act("#")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "#"
|
|
|
|
def test_pr_act_multiple_calls(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with multiple calls"""
|
|
pr_act(".")
|
|
pr_act(".")
|
|
pr_act(".")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "..."
|
|
|
|
def test_pr_act_various_characters(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with various characters"""
|
|
characters = [".", "#", "*", "+", "-", "=", ">", "~"]
|
|
for char in characters:
|
|
pr_act(char)
|
|
captured = capsys.readouterr()
|
|
assert "".join(characters) in captured.out
|
|
|
|
def test_pr_act_empty_string(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with empty string"""
|
|
pr_act("")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
|
|
def test_pr_act_special_character(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with special characters"""
|
|
pr_act("✓")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "✓"
|
|
|
|
def test_pr_act_unicode(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with unicode character"""
|
|
pr_act("●")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "●"
|
|
|
|
def test_pr_act_no_newline(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act doesn't add newline"""
|
|
pr_act("x")
|
|
captured = capsys.readouterr()
|
|
assert not captured.out.endswith("\n")
|
|
|
|
def test_pr_act_multiple_characters(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with multiple characters in string"""
|
|
pr_act("...")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "..."
|
|
|
|
def test_pr_act_whitespace(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with whitespace"""
|
|
pr_act(" ")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == " "
|
|
|
|
|
|
class TestProgressCombinations:
|
|
"""Test combinations of progress printer functions"""
|
|
|
|
def test_complete_progress_flow(self, capsys: CaptureFixture[str]):
|
|
"""Test complete progress output flow"""
|
|
pr_header("PROCESS")
|
|
pr_title("Task 1")
|
|
pr_open("Subtask")
|
|
pr_act(".")
|
|
pr_act(".")
|
|
pr_act(".")
|
|
pr_close(" OK")
|
|
captured = capsys.readouterr()
|
|
|
|
assert "PROCESS" in captured.out
|
|
assert "Task 1" in captured.out
|
|
assert "Subtask" in captured.out
|
|
assert "..." in captured.out
|
|
assert " OK]" in captured.out
|
|
|
|
def test_multiple_tasks_progress(self, capsys: CaptureFixture[str]):
|
|
"""Test multiple tasks with progress"""
|
|
pr_header("BATCH PROCESS")
|
|
for i in range(3):
|
|
pr_open(f"Task {i + 1}")
|
|
for _ in range(5):
|
|
pr_act(".")
|
|
pr_close(" DONE")
|
|
captured = capsys.readouterr()
|
|
|
|
assert "BATCH PROCESS" in captured.out
|
|
assert "Task 1" in captured.out
|
|
assert "Task 2" in captured.out
|
|
assert "Task 3" in captured.out
|
|
assert " DONE]" in captured.out
|
|
|
|
def test_nested_progress(self, capsys: CaptureFixture[str]):
|
|
"""Test nested progress indicators"""
|
|
pr_header("MAIN TASK", marker_string="=")
|
|
pr_title("Subtask A", prefix_string=">")
|
|
pr_open("Processing")
|
|
pr_act("#")
|
|
pr_act("#")
|
|
pr_close()
|
|
pr_title("Subtask B", prefix_string=">")
|
|
pr_open("Processing")
|
|
pr_act("*")
|
|
pr_act("*")
|
|
pr_close(" OK")
|
|
captured = capsys.readouterr()
|
|
|
|
assert "MAIN TASK" in captured.out
|
|
assert "Subtask A" in captured.out
|
|
assert "Subtask B" in captured.out
|
|
assert "##" in captured.out
|
|
assert "**" in captured.out
|
|
|
|
def test_progress_with_different_markers(self, capsys: CaptureFixture[str]):
|
|
"""Test progress with different marker styles"""
|
|
pr_header("Process", marker_string="*")
|
|
pr_title("Step 1", prefix_string=">>", space_filler="-")
|
|
pr_open("Work", prefix_string=">>", space_filler="-")
|
|
pr_act("+")
|
|
pr_close(" ✓")
|
|
captured = capsys.readouterr()
|
|
|
|
assert "*" in captured.out
|
|
assert ">>" in captured.out
|
|
assert "-" in captured.out
|
|
assert "+" in captured.out
|
|
assert "✓" in captured.out
|
|
|
|
def test_empty_progress_sequence(self, capsys: CaptureFixture[str]):
|
|
"""Test progress sequence with no actual progress"""
|
|
pr_open("Quick task")
|
|
pr_close(" SKIP")
|
|
captured = capsys.readouterr()
|
|
|
|
assert "Quick task" in captured.out
|
|
assert " SKIP]" in captured.out
|
|
|
|
|
|
class TestIntegration:
|
|
"""Integration tests combining multiple scenarios"""
|
|
|
|
def test_file_and_console_logging(self, capsys: CaptureFixture[str]):
|
|
"""Test logging to both file and console"""
|
|
fpl = io.StringIO()
|
|
|
|
write_l("Starting process", fpl=fpl, print_line=True)
|
|
write_l("Processing item 1", fpl=fpl, print_line=True)
|
|
write_l("Processing item 2", fpl=fpl, print_line=True)
|
|
write_l("Complete", fpl=fpl, print_line=True)
|
|
|
|
captured = capsys.readouterr()
|
|
file_content = fpl.getvalue()
|
|
|
|
# Check console output
|
|
assert "Starting process\n" in captured.out
|
|
assert "Processing item 1\n" in captured.out
|
|
assert "Processing item 2\n" in captured.out
|
|
assert "Complete\n" in captured.out
|
|
|
|
# Check file output
|
|
assert "Starting process\n" in file_content
|
|
assert "Processing item 1\n" in file_content
|
|
assert "Processing item 2\n" in file_content
|
|
assert "Complete\n" in file_content
|
|
|
|
fpl.close()
|
|
|
|
def test_progress_with_logging(self, capsys: CaptureFixture[str]):
|
|
"""Test combining progress output with file logging"""
|
|
fpl = io.StringIO()
|
|
|
|
write_l("=== Process Start ===", fpl=fpl, print_line=True)
|
|
pr_header("MAIN PROCESS")
|
|
write_l("Header shown", fpl=fpl, print_line=False)
|
|
|
|
pr_open("Task 1")
|
|
pr_act(".")
|
|
pr_act(".")
|
|
pr_close(" OK")
|
|
write_l("Task 1 completed", fpl=fpl, print_line=False)
|
|
|
|
write_l("=== Process End ===", fpl=fpl, print_line=True)
|
|
|
|
captured = capsys.readouterr()
|
|
file_content = fpl.getvalue()
|
|
|
|
assert "=== Process Start ===" in captured.out
|
|
assert "MAIN PROCESS" in captured.out
|
|
assert "Task 1" in captured.out
|
|
assert "=== Process End ===" in captured.out
|
|
|
|
assert "=== Process Start ===\n" in file_content
|
|
assert "Header shown\n" in file_content
|
|
assert "Task 1 completed\n" in file_content
|
|
assert "=== Process End ===\n" in file_content
|
|
|
|
fpl.close()
|
|
|
|
def test_complex_workflow(self, capsys: CaptureFixture[str]):
|
|
"""Test complex workflow with all functions"""
|
|
fpl = io.StringIO()
|
|
|
|
write_l("Log: Starting batch process", fpl=fpl, print_line=False)
|
|
pr_header("BATCH PROCESSOR", marker_string="=", width=40)
|
|
|
|
for i in range(2):
|
|
write_l(f"Log: Processing batch {i + 1}", fpl=fpl, print_line=False)
|
|
pr_title(f"Batch {i + 1}", prefix_string="|", space_filler=".")
|
|
|
|
pr_open(f"Item {i + 1}", prefix_string="|", space_filler=".")
|
|
for j in range(3):
|
|
pr_act("*")
|
|
write_l(f"Log: Progress {j + 1}/3", fpl=fpl, print_line=False)
|
|
pr_close(" ✓")
|
|
|
|
write_l(f"Log: Batch {i + 1} complete", fpl=fpl, print_line=False)
|
|
|
|
write_l("Log: All batches complete", fpl=fpl, print_line=False)
|
|
|
|
captured = capsys.readouterr()
|
|
file_content = fpl.getvalue()
|
|
|
|
# Check console has progress indicators
|
|
assert "BATCH PROCESSOR" in captured.out
|
|
assert "Batch 1" in captured.out
|
|
assert "Batch 2" in captured.out
|
|
assert "***" in captured.out
|
|
assert "✓" in captured.out
|
|
|
|
# Check file has all log entries
|
|
assert "Log: Starting batch process\n" in file_content
|
|
assert "Log: Processing batch 1\n" in file_content
|
|
assert "Log: Processing batch 2\n" in file_content
|
|
assert "Log: Progress 1/3\n" in file_content
|
|
assert "Log: Batch 1 complete\n" in file_content
|
|
assert "Log: All batches complete\n" in file_content
|
|
|
|
fpl.close()
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Test edge cases and boundary conditions"""
|
|
|
|
def test_write_l_none_file_handler(self, capsys: CaptureFixture[str]):
|
|
"""Test write_l explicitly with None file handler"""
|
|
write_l("Test", fpl=None, print_line=True)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "Test\n"
|
|
|
|
def test_pr_header_negative_width(self):
|
|
"""Test pr_header with negative width raises ValueError"""
|
|
with pytest.raises(ValueError):
|
|
pr_header("Test", width=-10)
|
|
|
|
def test_pr_title_negative_width(self):
|
|
"""Test pr_title with negative width raises ValueError"""
|
|
with pytest.raises(ValueError):
|
|
pr_title("Test", width=-10)
|
|
|
|
def test_pr_open_negative_width(self):
|
|
"""Test pr_open with negative width raises ValueError"""
|
|
with pytest.raises(ValueError):
|
|
pr_open("Test", width=-10)
|
|
|
|
def test_multiple_pr_act_no_close(self, capsys: CaptureFixture[str]):
|
|
"""Test multiple pr_act calls without pr_close"""
|
|
pr_act(".")
|
|
pr_act(".")
|
|
pr_act(".")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "..."
|
|
|
|
def test_pr_close_without_pr_open(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_close without prior pr_open (should still work)"""
|
|
pr_close(" OK")
|
|
captured = capsys.readouterr()
|
|
assert " OK]" in captured.out
|
|
|
|
def test_very_long_strings(self):
|
|
"""Test with very long strings"""
|
|
fpl = io.StringIO()
|
|
long_str = "A" * 10000
|
|
write_l(long_str, fpl=fpl, print_line=False)
|
|
assert len(fpl.getvalue()) == 10001 # string + newline
|
|
fpl.close()
|
|
|
|
def test_pr_header_very_long_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with tag longer than width"""
|
|
pr_header("This is a very long tag that exceeds the width", width=10)
|
|
captured = capsys.readouterr()
|
|
assert "This is a very long tag that exceeds the width" in captured.out
|
|
|
|
def test_pr_title_very_long_tag(self, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with tag longer than width"""
|
|
pr_title("This is a very long tag that exceeds the width", width=10)
|
|
captured = capsys.readouterr()
|
|
assert "This is a very long tag that exceeds the width" in captured.out
|
|
|
|
def test_write_l_closed_file(self):
|
|
"""Test write_l with closed file should raise error"""
|
|
fpl = io.StringIO()
|
|
fpl.close()
|
|
|
|
with pytest.raises(ValueError):
|
|
write_l("Test", fpl=fpl, print_line=False)
|
|
|
|
|
|
class TestParametrized:
|
|
"""Parametrized tests for comprehensive coverage"""
|
|
|
|
@pytest.mark.parametrize("print_line", [True, False])
|
|
def test_write_l_print_line_variations(self, print_line: bool, capsys: CaptureFixture[str]):
|
|
"""Test write_l with different print_line values"""
|
|
write_l("Test", print_line=print_line)
|
|
captured = capsys.readouterr()
|
|
if print_line:
|
|
assert captured.out == "Test\n"
|
|
else:
|
|
assert captured.out == ""
|
|
|
|
@pytest.mark.parametrize("marker", ["#", "*", "=", "-", "+", "~", "@", "^"])
|
|
def test_pr_header_various_markers_param(self, marker: str, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with various markers"""
|
|
pr_header("TEST", marker_string=marker)
|
|
captured = capsys.readouterr()
|
|
assert marker in captured.out
|
|
assert "TEST" in captured.out
|
|
|
|
@pytest.mark.parametrize("width", [0, 5, 10, 20, 35, 50, 100])
|
|
def test_pr_header_various_widths(self, width: int, capsys: CaptureFixture[str]):
|
|
"""Test pr_header with various widths"""
|
|
pr_header("TEST", width=width)
|
|
captured = capsys.readouterr()
|
|
assert "TEST" in captured.out
|
|
|
|
@pytest.mark.parametrize("filler", [".", "-", "_", "*", " ", "~", "="])
|
|
def test_pr_title_various_fillers_param(self, filler: str, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with various space fillers"""
|
|
pr_title("Test", space_filler=filler)
|
|
captured = capsys.readouterr()
|
|
assert "Test" in captured.out
|
|
|
|
@pytest.mark.parametrize("prefix", ["|", ">", ">>", "*", "-", "+"])
|
|
def test_pr_title_various_prefixes(self, prefix: str, capsys: CaptureFixture[str]):
|
|
"""Test pr_title with various prefix strings"""
|
|
pr_title("Test", prefix_string=prefix)
|
|
captured = capsys.readouterr()
|
|
assert prefix in captured.out
|
|
assert "Test" in captured.out
|
|
|
|
@pytest.mark.parametrize("act_char", [".", "#", "*", "+", "-", "=", ">", "~", "✓", "●"])
|
|
def test_pr_act_various_characters_param(self, act_char: str, capsys: CaptureFixture[str]):
|
|
"""Test pr_act with various characters"""
|
|
pr_act(act_char)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == act_char
|
|
|
|
@pytest.mark.parametrize("close_tag", ["", " OK", " DONE", " ✓", " ✗", " SKIP", " PASS"])
|
|
def test_pr_close_various_tags_param(self, close_tag: str, capsys: CaptureFixture[str]):
|
|
"""Test pr_close with various tags"""
|
|
pr_close(close_tag)
|
|
captured = capsys.readouterr()
|
|
assert f"{close_tag}]" in captured.out
|
|
|
|
@pytest.mark.parametrize("content", [
|
|
"Simple text",
|
|
"Text with 特殊文字",
|
|
"Text with emoji 🎉",
|
|
"Text\twith\ttabs",
|
|
"Multiple\n\nNewlines",
|
|
"",
|
|
"A" * 100,
|
|
])
|
|
def test_write_l_various_content(self, content: str, capsys: CaptureFixture[str]):
|
|
"""Test write_l with various content types"""
|
|
fpl = io.StringIO()
|
|
write_l(content, fpl=fpl, print_line=True)
|
|
captured = capsys.readouterr()
|
|
assert content in captured.out
|
|
assert content + "\n" in fpl.getvalue()
|
|
fpl.close()
|
|
|
|
|
|
class TestRealWorldScenarios:
|
|
"""Test real-world usage scenarios"""
|
|
|
|
def test_batch_processing_output(self, capsys: CaptureFixture[str]):
|
|
"""Test typical batch processing output"""
|
|
pr_header("BATCH PROCESSOR", marker_string="=", width=50)
|
|
|
|
items = ["file1.txt", "file2.txt", "file3.txt"]
|
|
for item in items:
|
|
pr_open(f"Processing {item}")
|
|
for _ in range(10):
|
|
pr_act(".")
|
|
pr_close(" ✓")
|
|
|
|
captured = capsys.readouterr()
|
|
assert "BATCH PROCESSOR" in captured.out
|
|
for item in items:
|
|
assert item in captured.out
|
|
assert "✓" in captured.out
|
|
|
|
def test_logging_workflow(self, capsys: CaptureFixture[str]):
|
|
"""Test typical logging workflow"""
|
|
log_file = io.StringIO()
|
|
|
|
# Simulate a workflow with logging
|
|
write_l("[INFO] Starting process", fpl=log_file, print_line=True)
|
|
write_l("[INFO] Initializing components", fpl=log_file, print_line=True)
|
|
write_l("[DEBUG] Component A loaded", fpl=log_file, print_line=False)
|
|
write_l("[DEBUG] Component B loaded", fpl=log_file, print_line=False)
|
|
write_l("[INFO] Processing data", fpl=log_file, print_line=True)
|
|
write_l("[INFO] Process complete", fpl=log_file, print_line=True)
|
|
|
|
captured = capsys.readouterr()
|
|
log_content = log_file.getvalue()
|
|
|
|
# Console should only have INFO messages
|
|
assert "[INFO] Starting process" in captured.out
|
|
assert "[DEBUG] Component A loaded" not in captured.out
|
|
|
|
# Log file should have all messages
|
|
assert "[INFO] Starting process\n" in log_content
|
|
assert "[DEBUG] Component A loaded\n" in log_content
|
|
assert "[DEBUG] Component B loaded\n" in log_content
|
|
|
|
log_file.close()
|
|
|
|
def test_progress_indicator_for_long_task(self, capsys: CaptureFixture[str]):
|
|
"""Test progress indicator for a long-running task"""
|
|
pr_header("DATA PROCESSING")
|
|
pr_open("Loading data", width=50)
|
|
|
|
# Simulate progress
|
|
for i in range(20):
|
|
if i % 5 == 0:
|
|
pr_act(str(i // 5))
|
|
else:
|
|
pr_act(".")
|
|
|
|
pr_close(" COMPLETE")
|
|
|
|
captured = capsys.readouterr()
|
|
assert "DATA PROCESSING" in captured.out
|
|
assert "Loading data" in captured.out
|
|
assert "COMPLETE" in captured.out
|
|
|
|
def test_multi_stage_process(self, capsys: CaptureFixture[str]):
|
|
"""Test multi-stage process with titles and progress"""
|
|
pr_header("DEPLOYMENT PIPELINE", marker_string="=")
|
|
|
|
stages = ["Build", "Test", "Deploy"]
|
|
for stage in stages:
|
|
pr_title(stage)
|
|
pr_open(f"Running {stage.lower()}")
|
|
pr_act("#")
|
|
pr_act("#")
|
|
pr_act("#")
|
|
pr_close(" OK")
|
|
|
|
captured = capsys.readouterr()
|
|
assert "DEPLOYMENT PIPELINE" in captured.out
|
|
for stage in stages:
|
|
assert stage in captured.out
|
|
assert "###" in captured.out
|
|
|
|
def test_error_reporting_with_logging(self, capsys: CaptureFixture[str]):
|
|
"""Test error reporting workflow"""
|
|
error_log = io.StringIO()
|
|
|
|
pr_header("VALIDATION", marker_string="!")
|
|
pr_open("Checking files")
|
|
|
|
write_l("[ERROR] File not found: data.csv", fpl=error_log, print_line=False)
|
|
pr_act("✗")
|
|
|
|
write_l("[ERROR] Permission denied: output.txt", fpl=error_log, print_line=False)
|
|
pr_act("✗")
|
|
|
|
pr_close(" FAILED")
|
|
|
|
captured = capsys.readouterr()
|
|
log_content = error_log.getvalue()
|
|
|
|
assert "VALIDATION" in captured.out
|
|
assert "Checking files" in captured.out
|
|
assert "✗✗" in captured.out
|
|
assert "FAILED" in captured.out
|
|
|
|
assert "[ERROR] File not found: data.csv\n" in log_content
|
|
assert "[ERROR] Permission denied: output.txt\n" in log_content
|
|
|
|
error_log.close()
|
|
|
|
def test_detailed_reporting(self, capsys: CaptureFixture[str]):
|
|
"""Test detailed reporting with mixed output"""
|
|
report_file = io.StringIO()
|
|
|
|
pr_header("SYSTEM REPORT", marker_string="#", width=60)
|
|
write_l("=== System Report Generated ===", fpl=report_file, print_line=False)
|
|
|
|
pr_title("Database Status", prefix_string=">>")
|
|
write_l("Database: Connected", fpl=report_file, print_line=False)
|
|
write_l("Tables: 15", fpl=report_file, print_line=False)
|
|
write_l("Records: 1,234,567", fpl=report_file, print_line=False)
|
|
|
|
pr_title("API Status", prefix_string=">>")
|
|
write_l("API: Online", fpl=report_file, print_line=False)
|
|
write_l("Requests/min: 1,500", fpl=report_file, print_line=False)
|
|
|
|
write_l("=== Report Complete ===", fpl=report_file, print_line=False)
|
|
|
|
captured = capsys.readouterr()
|
|
report_content = report_file.getvalue()
|
|
|
|
assert "SYSTEM REPORT" in captured.out
|
|
assert "Database Status" in captured.out
|
|
assert "API Status" in captured.out
|
|
|
|
assert "=== System Report Generated ===\n" in report_content
|
|
assert "Database: Connected\n" in report_content
|
|
assert "API: Online\n" in report_content
|
|
assert "=== Report Complete ===\n" in report_content
|
|
|
|
report_file.close()
|
|
|
|
# __END__
|