solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for observe.see utilities."""
5
6import pytest
7from PIL import Image
8
9from solstone.observe.see import (
10 decode_frames,
11 draw_bounding_box,
12 image_to_jpeg_bytes,
13)
14
15
16def test_draw_bounding_box():
17 """Test drawing bounding box on image."""
18 img = Image.new("RGB", (100, 100), color="white")
19
20 # Draw a box - should not crash
21 box_2d = [10, 20, 30, 40] # y_min, x_min, y_max, x_max
22 draw_bounding_box(img, box_2d, color="red", width=3)
23
24 # Image should still be same size (mutated in place)
25 assert img.size == (100, 100)
26
27
28def test_image_to_jpeg_bytes():
29 """Test converting image to JPEG bytes."""
30 img = Image.new("RGB", (50, 50), color="green")
31
32 # Convert to JPEG
33 jpeg_bytes = image_to_jpeg_bytes(img, quality=85)
34
35 assert isinstance(jpeg_bytes, bytes)
36 assert len(jpeg_bytes) > 0
37 # JPEG files start with FF D8 magic bytes
38 assert jpeg_bytes[:2] == b"\xff\xd8"
39
40
41def test_decode_frames_empty_list():
42 """Test decode_frames with empty frame list."""
43 result = decode_frames("dummy.mp4", [])
44 assert result == []
45
46
47def test_decode_frames_missing_frame_id():
48 """Test decode_frames raises error when frame_id is missing."""
49 frames = [{"timestamp": 1.0}]
50
51 with pytest.raises(ValueError, match="must have 'frame_id' field"):
52 decode_frames("dummy.mp4", frames)
53
54
55def test_decode_frames_uses_one_based_frame_ids(monkeypatch):
56 """Test decode_frames maps 1-based frame_id values to decoded frames."""
57 import numpy as np
58
59 class FakeFrame:
60 def __init__(self, color: int):
61 self.pts = 1
62 self._color = color
63
64 def to_ndarray(self, format: str):
65 assert format == "rgb24"
66 return np.full((2, 2, 3), self._color, dtype=np.uint8)
67
68 class FakeContainer:
69 def __init__(self):
70 self.streams = type("Streams", (), {"video": [object()]})
71 self._frames = [FakeFrame(10), FakeFrame(20)]
72
73 def decode(self, stream):
74 return iter(self._frames)
75
76 def __enter__(self):
77 return self
78
79 def __exit__(self, exc_type, exc, tb):
80 return False
81
82 class FakeAv:
83 @staticmethod
84 def open(path):
85 return FakeContainer()
86
87 monkeypatch.setitem(__import__("sys").modules, "av", FakeAv)
88
89 frames = [{"frame_id": 1}, {"frame_id": 2}]
90 images = decode_frames("dummy.mp4", frames, annotate_boxes=False)
91
92 assert images[0].getpixel((0, 0)) == (10, 10, 10)
93 assert images[1].getpixel((0, 0)) == (20, 20, 20)
94
95
96def test_decode_frames_stops_after_highest_requested_frame(monkeypatch):
97 """Test decode_frames exits once it has filled the highest requested frame."""
98 import numpy as np
99
100 seen = {"count": 0}
101
102 class FakeFrame:
103 def __init__(self, color: int):
104 self.pts = 1
105 self._color = color
106
107 def to_ndarray(self, format: str):
108 assert format == "rgb24"
109 return np.full((2, 2, 3), self._color, dtype=np.uint8)
110
111 class FakeContainer:
112 def __init__(self):
113 self.streams = type("Streams", (), {"video": [object()]})
114
115 def decode(self, stream):
116 for index in range(30):
117 seen["count"] += 1
118 yield FakeFrame(index)
119
120 def __enter__(self):
121 return self
122
123 def __exit__(self, exc_type, exc, tb):
124 return False
125
126 class FakeAv:
127 @staticmethod
128 def open(path):
129 return FakeContainer()
130
131 monkeypatch.setitem(__import__("sys").modules, "av", FakeAv)
132
133 frames = [{"frame_id": 7}, {"frame_id": 12}, {"frame_id": 23}]
134 images = decode_frames("dummy.mp4", frames, annotate_boxes=False)
135
136 assert seen["count"] == 23
137 assert all(image is not None for image in images)