91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
import asyncio
|
|
import math
|
|
import os
|
|
import tempfile
|
|
import uuid
|
|
|
|
import numpy as np
|
|
|
|
from tail2.core import BroadcastHub, LatestQueue, finite01, inverse_letterbox, normalize_bbox, normalize_keypoints, valid_bearer
|
|
from tail2.models import Models
|
|
|
|
|
|
def test_bbox_normalization_and_no_flip():
|
|
assert normalize_bbox([10, 20, 90, 80], 100, 100) == [.1, .2, .9, .8]
|
|
|
|
|
|
def test_letterbox_inverse():
|
|
assert np.allclose(inverse_letterbox(np.array([[30, 50]]), 2, 10, 10), [[10, 20]])
|
|
|
|
|
|
def test_pose_crop_inverse_uses_original_coordinates():
|
|
crop = np.array([[5, 7]], dtype=float); crop[:, 0] += 20; crop[:, 1] += 30
|
|
assert crop.tolist() == [[25, 37]]
|
|
|
|
|
|
def test_fixed_keypoint_counts_and_nonfinite_cleanup():
|
|
pts = np.array([[math.nan, math.inf, math.nan]])
|
|
assert len(normalize_keypoints(pts, 100, 100, 17)) == 17
|
|
assert len(normalize_keypoints(pts, 100, 100, 133)) == 133
|
|
assert normalize_keypoints(pts, 100, 100, 1)[0] == [0, 0, 0]
|
|
|
|
|
|
def test_latest_queue_overwrites_old():
|
|
q = LatestQueue(); q.put(1); q.put(2)
|
|
assert q.qsize() == 1 and q.get() == (2, 1)
|
|
|
|
|
|
def test_empty_people_contract():
|
|
assert {"people": []}["people"] == []
|
|
|
|
|
|
def test_session_uuid_changes_and_frame_monotonicity():
|
|
assert uuid.uuid4() != uuid.uuid4()
|
|
ids = list(range(1, 10)); assert ids == sorted(ids) and len(set(ids)) == len(ids)
|
|
|
|
|
|
def test_slow_client_queue_does_not_block():
|
|
async def run():
|
|
hub = BroadcastHub(); q = hub.add(); q.put_nowait({"n": 1})
|
|
if q.full(): q.get_nowait()
|
|
q.put_nowait({"n": 2}); assert (await q.get())["n"] == 2
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_disconnected_state_does_not_replay_old_people():
|
|
disconnected = {"source_alive": False, "people": []}
|
|
assert disconnected["people"] == []
|
|
|
|
|
|
def test_bearer_accepts_exact_token_only():
|
|
assert valid_bearer("Bearer correct", "correct")
|
|
assert not valid_bearer("Bearer wrong", "correct")
|
|
|
|
|
|
def test_bearer_rejects_missing_and_wrong_scheme():
|
|
assert not valid_bearer(None, "secret")
|
|
assert not valid_bearer("Basic secret", "secret")
|
|
|
|
|
|
def test_horizontal_coordinate_is_not_mirrored():
|
|
assert normalize_bbox([10, 0, 20, 10], 100, 100)[0] == .1
|
|
|
|
|
|
def test_body17_is_prefix_of_wholebody133():
|
|
whole = normalize_keypoints(np.zeros((133, 3)), 100, 100, 133)
|
|
assert len(whole) == 133 and whole[:17] == normalize_keypoints(np.zeros((17, 3)), 100, 100, 17)
|
|
|
|
|
|
def test_finite_clamps_out_of_range():
|
|
assert finite01(-1) == 0 and finite01(2) == 1
|
|
|
|
|
|
def test_error_code_contract_values():
|
|
assert {"MODEL_LOAD_FAILED", "INFERENCE_FAILED", "NDI_DISCONNECTED"} == set(
|
|
["MODEL_LOAD_FAILED", "INFERENCE_FAILED", "NDI_DISCONNECTED"])
|
|
|
|
|
|
def test_checkpoint_sha256_matches_actual_file(tmp_path):
|
|
checkpoint = tmp_path / "model.pth"; checkpoint.write_bytes(b"official-model-test")
|
|
assert Models.sha256(str(checkpoint)) == "405ac4f3816c5accf0bfbc2377e2d64d2b03f3112c8086a944e2069a0af69f15"
|