99 lines
4.9 KiB
Python
99 lines
4.9 KiB
Python
import uuid
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
from app.services import match_or_build as mob
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_match_wins_before_category_gate():
|
|
"""A strong published-flow match returns 'matched' even if category disabled."""
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
|
|
return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "VPN", "score": 0.9}])), \
|
|
patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=[])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "vpn down", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "matched"
|
|
assert res["session_kind"] == "flow"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suggest_band():
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
|
|
return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.66}])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "suggest"
|
|
assert res["near_miss"]["flow_name"] == "X"
|
|
assert "flow_id" in res["near_miss"] and isinstance(res["near_miss"]["flow_id"], str)
|
|
assert res["near_miss"]["score"] == 0.66
|
|
assert res["can_build"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_out_of_scope_when_category_disabled_on_build_path():
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(return_value=[])), \
|
|
patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
|
|
patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["vpn_connect"])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "printer jam", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "out_of_scope"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_when_enabled_and_no_match():
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(return_value=[])), \
|
|
patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
|
|
patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["printer"])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "printer jam", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "build"
|
|
assert res["session_kind"] == "ai_build"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_build_skips_match_but_still_gates():
|
|
fm = AsyncMock(return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.99}])
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=fm), \
|
|
patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
|
|
patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["printer"])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=True)
|
|
fm.assert_not_called()
|
|
assert res["outcome"] == "build"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_score_exactly_match_threshold_is_matched():
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
|
|
return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.75}])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "matched"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_score_exactly_suggest_threshold_is_suggest():
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
|
|
return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.60}])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "suggest"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_score_below_suggest_falls_through_to_build_path():
|
|
with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
|
|
return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.4}])), \
|
|
patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
|
|
patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["printer"])):
|
|
res = await mob.match_or_build(uuid.uuid4(), "printer", None, "t1", db=AsyncMock(), force_build=False)
|
|
assert res["outcome"] == "build"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classify_keyword_fallback_matches_word():
|
|
with patch.object(mob, "get_ai_provider", side_effect=RuntimeError("model down")):
|
|
cat = await mob.classify("the printer is jammed")
|
|
assert cat == "printer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classify_keyword_fallback_no_substring_false_match():
|
|
# "have" must NOT match teams_zoom_av via the 'av' token; no real category word present
|
|
with patch.object(mob, "get_ai_provider", side_effect=RuntimeError("model down")):
|
|
cat = await mob.classify("i have a general question")
|
|
assert cat == "unknown"
|