forked from genewildish/Mainline
David Gwilliam
c57617bb3d
- 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
225 lines
6.3 KiB
Python
225 lines
6.3 KiB
Python
"""
|
|
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
|