Two session models shared one process. `/v1/*` and `/v1/stream` are the live one; `/api/sessions*` and `/api/compare_live_vs_still` were the earlier model, still constructed on every `create_app()` through `build_legacy_scans` even though nothing called them. The web client and `stream_client.py` speak only `/v1`. Deleted: `observe.py` (ScanService/PerceptionService/PerceptionAdapter), `yolo_adapter.py` (imported only by the legacy builder), `pipeline.py` and `test_suite.py` (the old single-image CLI, reachable only from each other), `test_observe.py` (its Session* tests), the eight legacy route handlers, `build_legacy_scans` with its two facades, `create_fastapi_app`, and the `scans` parameter on `create_app`. `IdentityRankingTests` moved to `test_identity.py` before `test_observe.py` was deleted: those five ranking rules are the live path's contract, not the legacy model's. `test_api.py` keeps every `/v1` and WebSocket test. Verified: 246 tests OK (2 skipped), and a route dump of a freshly built app shows no `/api/*` path with all nine `/v1` routes and the WebSocket intact.
52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""Ranking contract for decide_identity: what a score set is allowed to name.
|
|
|
|
Moved out of test_observe.py when the legacy session model was deleted; the
|
|
ranking rules are the live path's, not the legacy one's.
|
|
"""
|
|
|
|
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 identity import ( # noqa: E402
|
|
IDENTITY_ACCEPTED,
|
|
IDENTITY_AMBIGUOUS,
|
|
IDENTITY_REJECTED,
|
|
IDENTITY_UNKNOWN,
|
|
decide_identity,
|
|
)
|
|
|
|
|
|
class IdentityRankingTests(unittest.TestCase):
|
|
def test_zero_scores_are_rejected_not_a_sku(self):
|
|
ranked = decide_identity({"sku_a": 0.0, "sku_b": 0.0})
|
|
self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED)
|
|
self.assertIsNone(ranked["best_sku"])
|
|
self.assertEqual(ranked["status"], IDENTITY_REJECTED)
|
|
|
|
def test_empty_scores_are_rejected(self):
|
|
ranked = decide_identity({})
|
|
self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED)
|
|
self.assertIsNone(ranked["best_sku"])
|
|
|
|
def test_weak_unique_score_is_unknown(self):
|
|
ranked = decide_identity({"sku_a": 0.4, "sku_b": 0.0})
|
|
self.assertEqual(ranked["identity_level"], IDENTITY_UNKNOWN)
|
|
self.assertIsNone(ranked["best_sku"])
|
|
|
|
def test_close_scores_are_ambiguous(self):
|
|
ranked = decide_identity({"sku_a": 20.0, "sku_b": 19.5})
|
|
self.assertEqual(ranked["identity_level"], IDENTITY_AMBIGUOUS)
|
|
self.assertIsNone(ranked["best_sku"])
|
|
|
|
def test_unique_strong_score_is_accepted(self):
|
|
ranked = decide_identity({"sku_a": 40.0, "sku_b": 5.0})
|
|
self.assertEqual(ranked["identity_level"], IDENTITY_ACCEPTED)
|
|
self.assertEqual(ranked["best_sku"], "sku_a")
|
|
self.assertEqual(ranked["status"], "success")
|