Michael Chihlas
ac89e7b2fa
Server-assigns a uuid4 id to every AI-generated node (Finding 1 showstopper:
nodes had no id but the advance protocol keys on node_id, so ai_build walks
never advanced past question 1). Replaces the hidden {"node_type":"meta"}
walked_path convention with real category/problem_text/pending_node columns on
l1_walk_sessions (migration 61dda4f615c6) — fixes junk proposals + off-by-one
depth cap (Findings 8,9), and pending_node replays the served node on re-mount
(no duplicate paid LLM call). Intake honors explicit flow_id and adhoc=True
(Findings 4,5); flow_proposals.l1_session_id FK -> CASCADE (Finding 6 time
bomb); L1 category GET is owner+admin like PATCH and require_account_owner_or_admin
delegates to User.can_manage_account (Finding 7); escalate falls back to default
recipients + filters deleted_at + warns when empty (Finding 10). Cleanups: dead
ticket_ref removed, IntakeResponse per-outcome validator, unused acknowledged
dropped, escalations partial index, restored a deleted audit assertion.
Full Phase 2A backend set: 110 passed / 0 failed.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
99 lines
4.8 KiB
Python
99 lines
4.8 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, 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, 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, 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, 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, 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, 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, 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, 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"
|