107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Tests for BrandResolver stage (discovery mode)."""
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from core.detect.models import BoundingBox, Frame, TextCandidate
|
|
from core.detect.stages.models import ResolverConfig
|
|
from core.detect.stages.brand_resolver import resolve_brands, _normalize, _match_session
|
|
|
|
|
|
CONFIG = ResolverConfig(fuzzy_threshold=75)
|
|
|
|
|
|
def _make_candidate(text: str, confidence: float = 0.9) -> TextCandidate:
|
|
dummy_frame = Frame(sequence=0, chunk_id=0, timestamp=1.0,
|
|
image=np.zeros((10, 10, 3), dtype=np.uint8))
|
|
dummy_box = BoundingBox(x=0, y=0, w=10, h=10, confidence=0.8, label="text")
|
|
return TextCandidate(frame=dummy_frame, bbox=dummy_box, text=text, ocr_confidence=confidence)
|
|
|
|
|
|
def test_session_match():
|
|
session = {"nike": "Nike", "fly emirates": "Emirates"}
|
|
assert _match_session("Nike", session) == "Nike"
|
|
assert _match_session("nike", session) == "Nike"
|
|
assert _match_session("FLY EMIRATES", session) == "Emirates"
|
|
assert _match_session("unknown", session) is None
|
|
|
|
|
|
def test_resolve_with_session(monkeypatch):
|
|
events = []
|
|
monkeypatch.setattr("core.detect.emit.push_detect_event",
|
|
lambda job_id, etype, data: events.append((etype, data)))
|
|
|
|
session = {"nike": "Nike", "emirates": "Emirates"}
|
|
candidates = [_make_candidate("Nike"), _make_candidate("EMIRATES")]
|
|
|
|
matched, unresolved = resolve_brands(
|
|
candidates, CONFIG, session_brands=session,
|
|
)
|
|
|
|
assert len(matched) == 2
|
|
assert len(unresolved) == 0
|
|
assert matched[0].brand == "Nike"
|
|
assert matched[1].brand == "Emirates"
|
|
|
|
|
|
def test_resolve_unresolved_without_db(monkeypatch):
|
|
events = []
|
|
monkeypatch.setattr("core.detect.emit.push_detect_event",
|
|
lambda job_id, etype, data: events.append((etype, data)))
|
|
|
|
candidates = [_make_candidate("random garbage text")]
|
|
|
|
matched, unresolved = resolve_brands(
|
|
candidates, CONFIG, session_brands={},
|
|
)
|
|
|
|
assert len(matched) == 0
|
|
assert len(unresolved) == 1
|
|
|
|
|
|
def test_resolve_empty(monkeypatch):
|
|
events = []
|
|
monkeypatch.setattr("core.detect.emit.push_detect_event",
|
|
lambda job_id, etype, data: events.append((etype, data)))
|
|
|
|
matched, unresolved = resolve_brands([], CONFIG, session_brands={})
|
|
|
|
assert len(matched) == 0
|
|
assert len(unresolved) == 0
|
|
|
|
|
|
def test_resolve_builds_session_during_run(monkeypatch):
|
|
"""Session brands accumulate during a single run — second candidate benefits."""
|
|
events = []
|
|
monkeypatch.setattr("core.detect.emit.push_detect_event",
|
|
lambda job_id, etype, data: events.append((etype, data)))
|
|
|
|
session = {"nike": "Nike"}
|
|
candidates = [
|
|
_make_candidate("Nike"), # hits session
|
|
_make_candidate("unknown"), # misses everything
|
|
]
|
|
|
|
matched, unresolved = resolve_brands(
|
|
candidates, CONFIG, session_brands=session,
|
|
)
|
|
|
|
assert len(matched) == 1
|
|
assert matched[0].brand == "Nike"
|
|
assert len(unresolved) == 1
|
|
|
|
|
|
def test_events_emitted(monkeypatch):
|
|
events = []
|
|
monkeypatch.setattr("core.detect.emit.push_detect_event",
|
|
lambda job_id, etype, data: events.append((etype, data)))
|
|
|
|
session = {"nike": "Nike"}
|
|
candidates = [_make_candidate("Nike")]
|
|
|
|
resolve_brands(candidates, CONFIG, session_brands=session, job_id="test-job")
|
|
|
|
event_types = [e[0] for e in events]
|
|
assert "log" in event_types
|
|
assert "detection" in event_types
|