test: defend engine and tracking contracts

test_engine: detection normalization, one-caller-at-a-time detection, method
validation, undecodable payloads, crop isolation, classify_boxes accepting the
detection dicts the API actually hands it, summary counting, empty-slot
normalization across frame sizes, the still-image store (roundtrip, TTL, bound),
catalog/stock behaviour, and template-image path resolution that refuses
anything outside the catalog.

test_tracking: per-session id numbering, the case that matters (constructing a
second tracker must not reset a running session's ids), independent sessions,
pan continuity, buffer expiry producing a new id, detection index mapping, and
reset scoping.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kiyo9w 2026-09-24 12:43:32 +09:00
parent 6902067322
commit 90cce64cc8
2 changed files with 394 additions and 0 deletions

View file

@ -0,0 +1,110 @@
"""
Contract tests for session tracking.
The important one is ID isolation: ultralytics mints track ids from a
process-global counter and resets it when a tracker is constructed, so two live
clients would otherwise be able to hand each other's cached labels to the wrong
product.
"""
from __future__ import annotations
import os
import sys
import unittest
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
from tracking import SessionTracker # noqa: E402
def pan(tracker, start_x, frames=8, step=10, width=60, height=80, y=100):
ids = []
for index in range(frames):
x = start_x + index * step
tracks = tracker.update([[x, y, x + width, y + height]], [0.9])
ids.append(tracks[0].track_id if tracks else None)
return ids
class IdIsolationTests(unittest.TestCase):
def test_sessions_do_not_share_ids(self):
first = SessionTracker()
second = SessionTracker()
first_ids = pan(first, 100)
second_ids = pan(second, 100)
self.assertEqual(len(set(first_ids)), 1, "a panning box keeps one id")
self.assertEqual(len(set(second_ids)), 1, "a panning box keeps one id")
self.assertEqual(first_ids[0], 1)
self.assertEqual(second_ids[0], 1, "each session starts its own numbering")
def test_constructing_a_tracker_does_not_reset_a_running_session(self):
"""ultralytics resets the global id counter in BYTETracker.__init__; a
second client connecting mid-scan must not renumber the first client's
tracks (its cached labels are keyed by id)."""
first = SessionTracker()
ids = pan(first, 100, frames=6, step=5)
self.assertEqual(len(set(ids)), 1)
second = SessionTracker() # a second client connects
continued = pan(first, 100 + 6 * 5, frames=6, step=5)
self.assertEqual(set(continued), {ids[0]}, "ids must keep counting, not restart")
self.assertEqual(pan(second, 400)[0], 1, "the new session still starts at 1")
def test_two_sessions_track_the_same_product_independently(self):
first = SessionTracker()
second = SessionTracker()
first_ids = pan(first, 100)
second_ids = pan(second, 100)
self.assertEqual(first_ids, second_ids, "same motion, same ids per session")
# ... but they are separate objects: dropping one does not touch the other
for _ in range(10):
first.update([], [])
self.assertEqual(len(set(pan(second, 100 + 8 * 10, frames=3, step=5))), 1)
def test_two_boxes_get_distinct_ids(self):
tracker = SessionTracker()
tracks = tracker.update([[10, 10, 70, 90], [200, 10, 260, 90]], [0.9, 0.8])
self.assertEqual(len(tracks), 2)
self.assertEqual(len({track.track_id for track in tracks}), 2)
class TrackLifecycleTests(unittest.TestCase):
def test_pan_continuity(self):
tracker = SessionTracker()
ids = pan(tracker, 50, frames=10, step=12)
self.assertEqual(len(set(ids)), 1)
def test_absent_track_is_dropped_and_gets_a_new_id(self):
tracker = SessionTracker(buffer=5)
first_ids = pan(tracker, 50, frames=2)
for _ in range(8):
tracker.update([], [])
for _ in range(3): # a re-detected object is unconfirmed for a frame or two
tracks = tracker.update([[50, 100, 110, 180]], [0.9])
self.assertTrue(tracks, "the object is tracked again")
self.assertNotEqual(tracks[0].track_id, first_ids[0], "a dropped track must not resume its old id")
def test_detection_index_maps_back_to_the_input_box(self):
tracker = SessionTracker()
tracks = tracker.update(
[[10, 10, 70, 90], [200, 10, 260, 90], [400, 10, 460, 90]], [0.9, 0.85, 0.95]
)
indices = sorted(track.det_index for track in tracks)
self.assertEqual(indices, [0, 1, 2])
def test_reset_starts_numbering_over_without_touching_others(self):
other = SessionTracker()
other_ids = pan(other, 100, frames=6, step=5)
tracker = SessionTracker()
pan(tracker, 100, frames=6, step=5)
tracker.reset()
after = pan(tracker, 300)
self.assertEqual(set(after), {1}, "reset restarts this session's numbering")
continued = pan(other, 100 + 6 * 5, frames=4, step=5)
self.assertEqual(set(continued), {other_ids[0]}, "the other session keeps counting")
if __name__ == "__main__":
unittest.main()