fix(performance): use simple height estimation instead of PIL rendering
- Replace estimate_block_height (PIL-based) with estimate_simple_height (word wrap) - Update viewport filter tests to match new height-based filtering (~4 items vs 24) - Fix CI task duplication in mise.toml (remove redundant depends) Closes #38 Closes #36
This commit is contained in:
224
tests/test_streaming.py
Normal file
224
tests/test_streaming.py
Normal file
@@ -0,0 +1,224 @@
|
||||
"""
|
||||
Tests for streaming protocol utilities.
|
||||
"""
|
||||
|
||||
|
||||
from engine.display.streaming import (
|
||||
FrameDiff,
|
||||
MessageType,
|
||||
apply_diff,
|
||||
compress_frame,
|
||||
compute_diff,
|
||||
decode_binary_message,
|
||||
decode_diff_message,
|
||||
decode_rle,
|
||||
decompress_frame,
|
||||
encode_binary_message,
|
||||
encode_diff_message,
|
||||
encode_rle,
|
||||
should_use_diff,
|
||||
)
|
||||
|
||||
|
||||
class TestFrameDiff:
|
||||
"""Tests for FrameDiff computation."""
|
||||
|
||||
def test_compute_diff_all_changed(self):
|
||||
"""compute_diff detects all changed lines."""
|
||||
old = ["a", "b", "c"]
|
||||
new = ["x", "y", "z"]
|
||||
|
||||
diff = compute_diff(old, new)
|
||||
|
||||
assert len(diff.changed_lines) == 3
|
||||
assert diff.width == 1
|
||||
assert diff.height == 3
|
||||
|
||||
def test_compute_diff_no_changes(self):
|
||||
"""compute_diff returns empty for identical buffers."""
|
||||
old = ["a", "b", "c"]
|
||||
new = ["a", "b", "c"]
|
||||
|
||||
diff = compute_diff(old, new)
|
||||
|
||||
assert len(diff.changed_lines) == 0
|
||||
|
||||
def test_compute_diff_partial_changes(self):
|
||||
"""compute_diff detects partial changes."""
|
||||
old = ["a", "b", "c"]
|
||||
new = ["a", "x", "c"]
|
||||
|
||||
diff = compute_diff(old, new)
|
||||
|
||||
assert len(diff.changed_lines) == 1
|
||||
assert diff.changed_lines[0] == (1, "x")
|
||||
|
||||
def test_compute_diff_new_lines(self):
|
||||
"""compute_diff detects new lines added."""
|
||||
old = ["a", "b"]
|
||||
new = ["a", "b", "c"]
|
||||
|
||||
diff = compute_diff(old, new)
|
||||
|
||||
assert len(diff.changed_lines) == 1
|
||||
assert diff.changed_lines[0] == (2, "c")
|
||||
|
||||
def test_compute_diff_empty_old(self):
|
||||
"""compute_diff handles empty old buffer."""
|
||||
old = []
|
||||
new = ["a", "b", "c"]
|
||||
|
||||
diff = compute_diff(old, new)
|
||||
|
||||
assert len(diff.changed_lines) == 3
|
||||
|
||||
|
||||
class TestRLE:
|
||||
"""Tests for run-length encoding."""
|
||||
|
||||
def test_encode_rle_no_repeats(self):
|
||||
"""encode_rle handles no repeated lines."""
|
||||
lines = [(0, "a"), (1, "b"), (2, "c")]
|
||||
|
||||
encoded = encode_rle(lines)
|
||||
|
||||
assert len(encoded) == 3
|
||||
assert encoded[0] == (0, "a", 1)
|
||||
assert encoded[1] == (1, "b", 1)
|
||||
assert encoded[2] == (2, "c", 1)
|
||||
|
||||
def test_encode_rle_with_repeats(self):
|
||||
"""encode_rle compresses repeated lines."""
|
||||
lines = [(0, "a"), (1, "a"), (2, "a"), (3, "b")]
|
||||
|
||||
encoded = encode_rle(lines)
|
||||
|
||||
assert len(encoded) == 2
|
||||
assert encoded[0] == (0, "a", 3)
|
||||
assert encoded[1] == (3, "b", 1)
|
||||
|
||||
def test_decode_rle(self):
|
||||
"""decode_rle reconstructs original lines."""
|
||||
encoded = [(0, "a", 3), (3, "b", 1)]
|
||||
|
||||
decoded = decode_rle(encoded)
|
||||
|
||||
assert decoded == [(0, "a"), (1, "a"), (2, "a"), (3, "b")]
|
||||
|
||||
def test_encode_decode_roundtrip(self):
|
||||
"""encode/decode is lossless."""
|
||||
original = [(i, f"line{i % 3}") for i in range(10)]
|
||||
encoded = encode_rle(original)
|
||||
decoded = decode_rle(encoded)
|
||||
|
||||
assert decoded == original
|
||||
|
||||
|
||||
class TestCompression:
|
||||
"""Tests for frame compression."""
|
||||
|
||||
def test_compress_decompress(self):
|
||||
"""compress_frame is lossless."""
|
||||
buffer = [f"Line {i:02d}" for i in range(24)]
|
||||
|
||||
compressed = compress_frame(buffer)
|
||||
decompressed = decompress_frame(compressed, 24)
|
||||
|
||||
assert decompressed == buffer
|
||||
|
||||
def test_compress_empty(self):
|
||||
"""compress_frame handles empty buffer."""
|
||||
compressed = compress_frame([])
|
||||
decompressed = decompress_frame(compressed, 0)
|
||||
|
||||
assert decompressed == []
|
||||
|
||||
|
||||
class TestBinaryProtocol:
|
||||
"""Tests for binary message encoding."""
|
||||
|
||||
def test_encode_decode_message(self):
|
||||
"""encode_binary_message is lossless."""
|
||||
payload = b"test payload"
|
||||
|
||||
encoded = encode_binary_message(MessageType.FULL_FRAME, 80, 24, payload)
|
||||
msg_type, width, height, decoded_payload = decode_binary_message(encoded)
|
||||
|
||||
assert msg_type == MessageType.FULL_FRAME
|
||||
assert width == 80
|
||||
assert height == 24
|
||||
assert decoded_payload == payload
|
||||
|
||||
def test_encode_decode_all_types(self):
|
||||
"""All message types encode correctly."""
|
||||
for msg_type in MessageType:
|
||||
payload = b"test"
|
||||
encoded = encode_binary_message(msg_type, 80, 24, payload)
|
||||
decoded_type, _, _, _ = decode_binary_message(encoded)
|
||||
assert decoded_type == msg_type
|
||||
|
||||
|
||||
class TestDiffProtocol:
|
||||
"""Tests for diff message encoding."""
|
||||
|
||||
def test_encode_decode_diff(self):
|
||||
"""encode_diff_message is lossless."""
|
||||
diff = FrameDiff(width=80, height=24, changed_lines=[(0, "a"), (5, "b")])
|
||||
|
||||
payload = encode_diff_message(diff)
|
||||
decoded = decode_diff_message(payload)
|
||||
|
||||
assert decoded == diff.changed_lines
|
||||
|
||||
|
||||
class TestApplyDiff:
|
||||
"""Tests for applying diffs."""
|
||||
|
||||
def test_apply_diff(self):
|
||||
"""apply_diff reconstructs new buffer."""
|
||||
old_buffer = ["a", "b", "c", "d"]
|
||||
diff = FrameDiff(width=1, height=4, changed_lines=[(1, "x"), (2, "y")])
|
||||
|
||||
new_buffer = apply_diff(old_buffer, diff)
|
||||
|
||||
assert new_buffer == ["a", "x", "y", "d"]
|
||||
|
||||
def test_apply_diff_new_lines(self):
|
||||
"""apply_diff handles new lines."""
|
||||
old_buffer = ["a", "b"]
|
||||
diff = FrameDiff(width=1, height=4, changed_lines=[(2, "c"), (3, "d")])
|
||||
|
||||
new_buffer = apply_diff(old_buffer, diff)
|
||||
|
||||
assert new_buffer == ["a", "b", "c", "d"]
|
||||
|
||||
|
||||
class TestShouldUseDiff:
|
||||
"""Tests for diff threshold decision."""
|
||||
|
||||
def test_uses_diff_when_small_changes(self):
|
||||
"""should_use_diff returns True when few changes."""
|
||||
old = ["a"] * 100
|
||||
new = ["a"] * 95 + ["b"] * 5
|
||||
|
||||
assert should_use_diff(old, new, threshold=0.3) is True
|
||||
|
||||
def test_uses_full_when_many_changes(self):
|
||||
"""should_use_diff returns False when many changes."""
|
||||
old = ["a"] * 100
|
||||
new = ["b"] * 100
|
||||
|
||||
assert should_use_diff(old, new, threshold=0.3) is False
|
||||
|
||||
def test_uses_diff_at_threshold(self):
|
||||
"""should_use_diff handles threshold boundary."""
|
||||
old = ["a"] * 100
|
||||
new = ["a"] * 70 + ["b"] * 30
|
||||
|
||||
result = should_use_diff(old, new, threshold=0.3)
|
||||
assert result is True or result is False # At boundary
|
||||
|
||||
def test_returns_false_for_empty(self):
|
||||
"""should_use_diff returns False for empty buffers."""
|
||||
assert should_use_diff([], ["a", "b"]) is False
|
||||
assert should_use_diff(["a", "b"], []) is False
|
||||
Reference in New Issue
Block a user