708 lines
27 KiB
Python
708 lines
27 KiB
Python
|
"""Tests for AI search (Step 3).
|
||
|
|
|
||
|
|
All LLM calls are mocked — CI never touches the network.
|
||
|
|
|
||
|
|
Coverage areas:
|
||
|
|
- expand_query JSON output parsing (valid, fenced, prose, bad JSON, timeout)
|
||
|
|
- Output contract enforcement (strict JSON array only)
|
||
|
|
- Expansion term count cap and length cap
|
||
|
|
- ai_search seam function
|
||
|
|
- GET /search with ai=1 trigger
|
||
|
|
- AI button visibility on search page
|
||
|
|
- Graceful degradation on failure
|
||
|
|
- ai_search_extra_hints appended to prompt
|
||
|
|
- ai_search_enabled toggle
|
||
|
|
"""
|
||
|
|
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.llm import (
|
||
|
|
_MAX_EXPANSION_TERMS,
|
||
|
|
_MAX_TERM_LENGTH,
|
||
|
|
ExpansionResult,
|
||
|
|
LLMResult,
|
||
|
|
_parse_json_string_array,
|
||
|
|
expand_query,
|
||
|
|
is_configured,
|
||
|
|
)
|
||
|
|
from app.main import _ai_search, _build_search_results
|
||
|
|
from app.models import AppSetting, Box, Item, SubItem
|
||
|
|
from app.settings_store import LLMConfig, get_app_settings, save_app_settings
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Helper: configure AI search for route tests
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
_AI_CFG = LLMConfig(
|
||
|
|
enabled=True,
|
||
|
|
base_url="https://api.example.com/v1",
|
||
|
|
model="gpt-4o-mini",
|
||
|
|
api_key="sk-test-key",
|
||
|
|
ai_search_enabled=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _enable_ai_search(client, db_session):
|
||
|
|
"""Persist a fully-configured AI search setup via the settings route."""
|
||
|
|
client.post(
|
||
|
|
"/settings",
|
||
|
|
data={
|
||
|
|
"enabled": "on",
|
||
|
|
"base_url": "https://api.example.com/v1",
|
||
|
|
"model": "gpt-4o-mini",
|
||
|
|
"api_key": "sk-test-key",
|
||
|
|
"ai_search_enabled": "on",
|
||
|
|
},
|
||
|
|
follow_redirects=False,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# _parse_json_string_array: strict JSON contract enforcement
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestParseJsonStringArray:
|
||
|
|
def test_valid_json_array(self):
|
||
|
|
result = _parse_json_string_array('["炒锅","平底锅","汤锅"]')
|
||
|
|
assert result == ["炒锅", "平底锅", "汤锅"]
|
||
|
|
|
||
|
|
def test_json_array_with_code_fence(self):
|
||
|
|
result = _parse_json_string_array('```json\n["锅","铲子"]\n```')
|
||
|
|
assert result == ["锅", "铲子"]
|
||
|
|
|
||
|
|
def test_json_array_with_code_fence_no_lang(self):
|
||
|
|
result = _parse_json_string_array('```\n["锅","铲子"]\n```')
|
||
|
|
assert result == ["锅", "铲子"]
|
||
|
|
|
||
|
|
def test_empty_string_returns_empty(self):
|
||
|
|
assert _parse_json_string_array("") == []
|
||
|
|
assert _parse_json_string_array(" ") == []
|
||
|
|
|
||
|
|
def test_prose_returns_empty(self):
|
||
|
|
"""Prose text does NOT become expansion terms — strict contract."""
|
||
|
|
assert _parse_json_string_array("I cannot help with that.") == []
|
||
|
|
|
||
|
|
def test_prose_newlines_returns_empty(self):
|
||
|
|
"""Line-separated prose does NOT become expansion terms."""
|
||
|
|
assert _parse_json_string_array("炒锅\n平底锅\n汤锅") == []
|
||
|
|
|
||
|
|
def test_prose_commas_returns_empty(self):
|
||
|
|
"""Comma-separated prose does NOT become expansion terms."""
|
||
|
|
assert _parse_json_string_array("炒锅, 平底锅, 汤锅") == []
|
||
|
|
|
||
|
|
def test_bad_json_returns_empty(self):
|
||
|
|
"""Invalid JSON returns empty — no fallback."""
|
||
|
|
assert _parse_json_string_array("{invalid json") == []
|
||
|
|
|
||
|
|
def test_json_object_returns_empty(self):
|
||
|
|
"""JSON object (non-array) returns empty."""
|
||
|
|
assert _parse_json_string_array('{"terms":["锅","厨具"]}') == []
|
||
|
|
|
||
|
|
def test_json_array_with_numbers_returns_empty(self):
|
||
|
|
"""Non-string items in array cause rejection — strict contract."""
|
||
|
|
assert _parse_json_string_array('[1, 2, 3]') == []
|
||
|
|
|
||
|
|
def test_json_array_with_mixed_types_returns_empty(self):
|
||
|
|
"""Mixed string/number array is rejected."""
|
||
|
|
assert _parse_json_string_array('["锅", 1]') == []
|
||
|
|
|
||
|
|
def test_empty_json_array(self):
|
||
|
|
result = _parse_json_string_array('[]')
|
||
|
|
assert result == []
|
||
|
|
|
||
|
|
def test_capped_at_max_terms(self):
|
||
|
|
"""More than _MAX_EXPANSION_TERMS items are truncated."""
|
||
|
|
terms = [f"词{i}" for i in range(20)]
|
||
|
|
json_str = "[" + ",".join(f'"{t}"' for t in terms) + "]"
|
||
|
|
result = _parse_json_string_array(json_str)
|
||
|
|
assert len(result) == _MAX_EXPANSION_TERMS
|
||
|
|
|
||
|
|
def test_long_terms_filtered_out(self):
|
||
|
|
"""Terms exceeding _MAX_TERM_LENGTH are silently dropped."""
|
||
|
|
short = "锅"
|
||
|
|
long_term = "A" * (_MAX_TERM_LENGTH + 1)
|
||
|
|
json_str = f'["{short}", "{long_term}"]'
|
||
|
|
result = _parse_json_string_array(json_str)
|
||
|
|
assert result == ["锅"]
|
||
|
|
|
||
|
|
def test_whitespace_stripped(self):
|
||
|
|
result = _parse_json_string_array('[" 锅 ", " 平底锅 "]')
|
||
|
|
assert result == ["锅", "平底锅"]
|
||
|
|
|
||
|
|
def test_empty_strings_filtered(self):
|
||
|
|
result = _parse_json_string_array('["锅", "", " ", "平底锅"]')
|
||
|
|
assert result == ["锅", "平底锅"]
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# expand_query: prompt, hints, graceful degradation
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestExpandQueryNew:
|
||
|
|
def test_returns_empty_when_not_configured(self):
|
||
|
|
cfg = LLMConfig(enabled=False)
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is None
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_parses_valid_json_response(self, mock_call):
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["炒锅","平底锅","汤锅","厨具"]'}}]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert "炒锅" in result.terms
|
||
|
|
assert "平底锅" in result.terms
|
||
|
|
assert "厨具" in result.terms
|
||
|
|
assert result.error is None
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_handles_json_with_code_fence(self, mock_call):
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [
|
||
|
|
{"message": {"content": '```json\n["炒锅","平底锅"]\n```'}}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert "炒锅" in result.terms
|
||
|
|
assert result.error is None
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_prose_response_returns_empty_no_error(self, mock_call):
|
||
|
|
"""Prose from model → empty terms, no error (successful call, unparseable output)."""
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": "I cannot help with that."}}]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is None
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_json_object_response_returns_empty_no_error(self, mock_call):
|
||
|
|
"""JSON object (non-array) → empty terms, no error."""
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '{"terms":["锅","厨具"]}'}}]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is None
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_timeout_returns_error(self, mock_call):
|
||
|
|
"""Timeout → empty terms + error message."""
|
||
|
|
mock_call.side_effect = httpx.TimeoutException("timeout")
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is not None
|
||
|
|
assert "超时" in result.error
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_network_error_returns_error(self, mock_call):
|
||
|
|
"""Network error → empty terms + error message."""
|
||
|
|
mock_call.side_effect = httpx.ConnectError("refused")
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is not None
|
||
|
|
assert "无法连接" in result.error
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_http_error_returns_error(self, mock_call):
|
||
|
|
"""HTTP error → empty terms + error message."""
|
||
|
|
mock_call.side_effect = httpx.HTTPStatusError(
|
||
|
|
"401",
|
||
|
|
request=httpx.Request("POST", "http://x"),
|
||
|
|
response=httpx.Response(401),
|
||
|
|
)
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is not None
|
||
|
|
assert "错误" in result.error
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_returns_empty_on_empty_choices(self, mock_call):
|
||
|
|
mock_call.return_value = {"choices": []}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
result = expand_query(cfg, "锅")
|
||
|
|
assert result.terms == []
|
||
|
|
assert result.error is None
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_extra_hints_appended_to_system_prompt(self, mock_call):
|
||
|
|
"""When extra_hints is non-empty, it should be appended to the system prompt."""
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["扩展词"]'}}]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
expand_query(cfg, "锅", extra_hints="用户物品主要涉及厨房用品")
|
||
|
|
|
||
|
|
# Verify the system prompt includes the extra hints
|
||
|
|
call_args = mock_call.call_args
|
||
|
|
messages = call_args[1]["messages"] if "messages" in call_args[1] else call_args[0][1]
|
||
|
|
system_content = messages[0]["content"]
|
||
|
|
assert "用户物品主要涉及厨房用品" in system_content
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_extra_hints_ignored_when_empty(self, mock_call):
|
||
|
|
"""When extra_hints is empty, system prompt should not change."""
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["扩展词"]'}}]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
expand_query(cfg, "锅", extra_hints="")
|
||
|
|
|
||
|
|
call_args = mock_call.call_args
|
||
|
|
messages = call_args[1]["messages"] if "messages" in call_args[1] else call_args[0][1]
|
||
|
|
system_content = messages[0]["content"]
|
||
|
|
# Should be the base prompt only
|
||
|
|
assert "搬家物品搜索助手" in system_content
|
||
|
|
assert "JSON 字符串数组" in system_content
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_temperature_zero_passed(self, mock_call):
|
||
|
|
"""expand_query should pass temperature=0 for deterministic output."""
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["扩展词"]'}}]
|
||
|
|
}
|
||
|
|
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
|
||
|
|
expand_query(cfg, "锅")
|
||
|
|
|
||
|
|
call_args = mock_call.call_args
|
||
|
|
assert call_args[1]["temperature"] == 0
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# _ai_search: seam function
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestAiSearchSeam:
|
||
|
|
@patch("app.main.expand_query")
|
||
|
|
def test_returns_expanded_terms_and_results(self, mock_expand, client, db_session):
|
||
|
|
"""AI search returns expanded terms and broader results."""
|
||
|
|
box = Box(name="厨房箱", note="装了炒锅和铲子")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_expand.return_value = ExpansionResult(terms=["炒锅", "平底锅", "汤锅"])
|
||
|
|
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
expanded, results, error = _ai_search(db_session, cfg, "平底锅")
|
||
|
|
|
||
|
|
assert "炒锅" in expanded
|
||
|
|
assert error is None
|
||
|
|
assert len(results) >= 1
|
||
|
|
assert any("厨房箱" in r["name"] or "炒锅" in (r.get("note") or "") for r in results)
|
||
|
|
|
||
|
|
@patch("app.main.expand_query")
|
||
|
|
def test_includes_original_query_in_search(self, mock_expand, client, db_session):
|
||
|
|
"""AI search includes the original query term in the search."""
|
||
|
|
box = Box(name="冬季衣物箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_expand.return_value = ExpansionResult(terms=["羽绒服"])
|
||
|
|
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
expanded, results, error = _ai_search(db_session, cfg, "衣物")
|
||
|
|
|
||
|
|
assert error is None
|
||
|
|
assert any("冬季衣物箱" in r["name"] for r in results)
|
||
|
|
|
||
|
|
@patch("app.main.expand_query")
|
||
|
|
def test_empty_expansion_returns_normal_results_no_error(self, mock_expand, client, db_session):
|
||
|
|
"""Legitimate empty expansion (no synonyms found) → normal results, no error."""
|
||
|
|
box = Box(name="书房箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_expand.return_value = ExpansionResult(terms=[])
|
||
|
|
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
expanded, results, error = _ai_search(db_session, cfg, "书房")
|
||
|
|
|
||
|
|
assert expanded == []
|
||
|
|
assert error is None
|
||
|
|
assert any("书房箱" in r["name"] for r in results)
|
||
|
|
|
||
|
|
@patch("app.main.expand_query")
|
||
|
|
def test_llm_failure_returns_normal_results_with_error(self, mock_expand, client, db_session):
|
||
|
|
"""When expand_query signals failure, seam returns normal results + error message."""
|
||
|
|
box = Box(name="厨房箱", note="装了炒锅")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_expand.return_value = ExpansionResult(terms=[], error="AI 搜索请求超时，请稍后再试。")
|
||
|
|
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
expanded, results, error = _ai_search(db_session, cfg, "厨房")
|
||
|
|
|
||
|
|
assert expanded == []
|
||
|
|
assert error is not None
|
||
|
|
assert "超时" in error
|
||
|
|
assert len(results) >= 1
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# _build_search_results: multi-keyword support
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestBuildSearchResultsMultiKeyword:
|
||
|
|
def test_single_keyword_works_as_before(self, db_session):
|
||
|
|
box = Box(name="厨房箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
results = _build_search_results(db_session, "厨房")
|
||
|
|
assert len(results) == 1
|
||
|
|
assert results[0]["name"] == "厨房箱"
|
||
|
|
|
||
|
|
def test_multiple_keywords_match_any(self, db_session):
|
||
|
|
box1 = Box(name="厨房箱")
|
||
|
|
box2 = Box(name="卧室箱")
|
||
|
|
db_session.add_all([box1, box2])
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
results = _build_search_results(db_session, ["厨房", "卧室"])
|
||
|
|
assert len(results) == 2
|
||
|
|
|
||
|
|
def test_multiple_keywords_dedupes_results(self, db_session):
|
||
|
|
"""A box matching multiple keywords appears only once."""
|
||
|
|
box = Box(name="厨房箱", note="装了厨房用品")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
results = _build_search_results(db_session, ["厨房", "用品"])
|
||
|
|
assert len(results) == 1
|
||
|
|
|
||
|
|
def test_empty_keywords_returns_empty(self, db_session):
|
||
|
|
results = _build_search_results(db_session, [])
|
||
|
|
assert results == []
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Routes: GET /search with ai=1
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestSearchRouteAI:
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_ai_search_finds_more_results(self, mock_call, client, db_session):
|
||
|
|
"""Original query misses, but expanded term finds items."""
|
||
|
|
box = Box(name="杂物箱")
|
||
|
|
item = Item(name="炒锅", box=box, is_container=False)
|
||
|
|
db_session.add_all([box, item])
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["炒锅","平底锅","汤锅"]'}}]
|
||
|
|
}
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
# Normal search for "平底锅" — no results
|
||
|
|
response = client.get("/search?q=平底锅")
|
||
|
|
assert "没有找到匹配结果" in response.text
|
||
|
|
|
||
|
|
# AI search for "平底锅" — finds "炒锅" via expansion
|
||
|
|
response = client.get("/search?q=平底锅&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "炒锅" in response.text
|
||
|
|
assert "AI 帮你扩展了" in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_ai_search_includes_original_results(self, mock_call, client, db_session):
|
||
|
|
"""AI search should also include results from original query."""
|
||
|
|
box = Box(name="厨房箱")
|
||
|
|
item1 = Item(name="锅铲", box=box, is_container=False)
|
||
|
|
item2 = Item(name="平底锅", box=box, is_container=False)
|
||
|
|
db_session.add_all([box, item1, item2])
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["炒锅","汤锅"]'}}]
|
||
|
|
}
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=锅&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
# Original result "平底锅" should still be there
|
||
|
|
assert "平底锅" in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_ai_search_shows_expansion_banner(self, mock_call, client, db_session):
|
||
|
|
"""When AI search is activated, a banner shows expanded terms."""
|
||
|
|
box = Box(name="厨房箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["炒锅","平底锅"]'}}]
|
||
|
|
}
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=锅&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "AI 帮你扩展了" in response.text
|
||
|
|
assert "炒锅" in response.text
|
||
|
|
|
||
|
|
def test_ai_search_without_flag_does_normal_search(self, client, db_session):
|
||
|
|
"""Without ai=1, search behaves normally even when AI is configured."""
|
||
|
|
box = Box(name="厨房箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=厨房")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "厨房箱" in response.text
|
||
|
|
assert "AI 帮你扩展了" not in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_ai_search_without_configuration_ignores_flag(self, mock_call, client, db_session):
|
||
|
|
"""ai=1 is ignored when AI is not configured."""
|
||
|
|
box = Box(name="厨房箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
response = client.get("/search?q=厨房&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "厨房箱" in response.text
|
||
|
|
assert "AI 帮你扩展了" not in response.text
|
||
|
|
mock_call.assert_not_called()
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_ai_search_graceful_degradation_on_llm_failure(self, mock_call, client, db_session):
|
||
|
|
"""LLM failure (timeout) → normal results + friendly error banner."""
|
||
|
|
box = Box(name="厨房箱", note="装了炒锅")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
# expand_query catches timeout and returns ExpansionResult with error
|
||
|
|
mock_call.side_effect = httpx.TimeoutException("timeout")
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=厨房&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "厨房箱" in response.text
|
||
|
|
# Should show error banner — timeout is a real failure
|
||
|
|
assert "超时" in response.text or "不可用" in response.text
|
||
|
|
|
||
|
|
def test_ai_search_empty_query_does_nothing(self, client, db_session):
|
||
|
|
"""ai=1 with empty query does not trigger AI."""
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "AI 帮你扩展了" not in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_ai_search_disabled_ignores_flag(self, mock_call, client, db_session):
|
||
|
|
"""ai=1 is ignored when ai_search_enabled is False."""
|
||
|
|
box = Box(name="厨房箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
# Enable LLM but NOT ai_search_enabled
|
||
|
|
client.post(
|
||
|
|
"/settings",
|
||
|
|
data={
|
||
|
|
"enabled": "on",
|
||
|
|
"base_url": "https://api.example.com/v1",
|
||
|
|
"model": "gpt-4o-mini",
|
||
|
|
"api_key": "sk-test-key",
|
||
|
|
},
|
||
|
|
follow_redirects=False,
|
||
|
|
)
|
||
|
|
|
||
|
|
response = client.get("/search?q=厨房&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "厨房箱" in response.text
|
||
|
|
assert "AI 帮你扩展了" not in response.text
|
||
|
|
mock_call.assert_not_called()
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Button visibility on search page
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestAIButtonVisibility:
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_button_visible_when_configured_and_enabled(self, mock_call, client, db_session):
|
||
|
|
"""AI search button is visible when ai_search_enabled and configured."""
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=测试")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "AI 智能搜索" in response.text
|
||
|
|
|
||
|
|
def test_button_hidden_when_not_configured(self, client, db_session):
|
||
|
|
"""AI search button is hidden when LLM is not configured."""
|
||
|
|
response = client.get("/search?q=测试")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "AI 智能搜索" not in response.text
|
||
|
|
|
||
|
|
def test_button_hidden_when_ai_search_disabled(self, client, db_session):
|
||
|
|
"""AI search button is hidden when ai_search_enabled is False."""
|
||
|
|
client.post(
|
||
|
|
"/settings",
|
||
|
|
data={
|
||
|
|
"enabled": "on",
|
||
|
|
"base_url": "https://api.example.com/v1",
|
||
|
|
"model": "gpt-4o-mini",
|
||
|
|
"api_key": "sk-test-key",
|
||
|
|
},
|
||
|
|
follow_redirects=False,
|
||
|
|
)
|
||
|
|
|
||
|
|
response = client.get("/search?q=测试")
|
||
|
|
assert "AI 智能搜索" not in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_button_hidden_on_empty_query(self, mock_call, client, db_session):
|
||
|
|
"""AI search button is not shown when there's no query."""
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search")
|
||
|
|
assert "AI 智能搜索" not in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_button_link_includes_current_query(self, mock_call, client, db_session):
|
||
|
|
"""AI button link includes the current query parameter."""
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=锅")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "ai=1" in response.text
|
||
|
|
from urllib.parse import quote
|
||
|
|
assert f"q={quote('锅')}" in response.text or "q=锅" in response.text
|
||
|
|
|
||
|
|
@patch("app.llm._call_chat_completion")
|
||
|
|
def test_no_button_when_ai_already_activated(self, mock_call, client, db_session):
|
||
|
|
"""When AI is already activated, show status text instead of button."""
|
||
|
|
mock_call.return_value = {
|
||
|
|
"choices": [{"message": {"content": '["炒锅"]'}}]
|
||
|
|
}
|
||
|
|
_enable_ai_search(client, db_session)
|
||
|
|
|
||
|
|
response = client.get("/search?q=锅&ai=1")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "AI 搜索已启用" in response.text
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Settings: ai_search_extra_hints
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestExtraHintsSettings:
|
||
|
|
def test_extra_hints_defaults_to_empty(self, db_session):
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
assert cfg.ai_search_extra_hints == ""
|
||
|
|
|
||
|
|
def test_save_extra_hints(self, db_session):
|
||
|
|
save_app_settings(db_session, ai_search_extra_hints="用户物品主要涉及厨房")
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
assert cfg.ai_search_extra_hints == "用户物品主要涉及厨房"
|
||
|
|
|
||
|
|
def test_save_extra_hints_empty_string(self, db_session):
|
||
|
|
save_app_settings(db_session, ai_search_extra_hints="厨房用品")
|
||
|
|
save_app_settings(db_session, ai_search_extra_hints="")
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
assert cfg.ai_search_extra_hints == ""
|
||
|
|
|
||
|
|
def test_settings_page_has_extra_hints_textarea(self, client):
|
||
|
|
response = client.get("/settings")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'name="ai_search_extra_hints"' in response.text
|
||
|
|
assert "额外领域提示" in response.text
|
||
|
|
|
||
|
|
def test_settings_page_has_ai_search_checkbox(self, client):
|
||
|
|
response = client.get("/settings")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'name="ai_search_enabled"' in response.text
|
||
|
|
assert "启用 AI 智能搜索" in response.text
|
||
|
|
|
||
|
|
def test_save_ai_search_settings_via_route(self, client, db_session):
|
||
|
|
client.post(
|
||
|
|
"/settings",
|
||
|
|
data={
|
||
|
|
"enabled": "on",
|
||
|
|
"base_url": "https://api.example.com/v1",
|
||
|
|
"model": "gpt-4o-mini",
|
||
|
|
"api_key": "sk-key",
|
||
|
|
"ai_search_enabled": "on",
|
||
|
|
"ai_search_extra_hints": "用户物品主要涉及厨房用品",
|
||
|
|
},
|
||
|
|
follow_redirects=False,
|
||
|
|
)
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
assert cfg.ai_search_enabled is True
|
||
|
|
assert cfg.ai_search_extra_hints == "用户物品主要涉及厨房用品"
|
||
|
|
|
||
|
|
def test_save_preserves_extra_hints_on_other_changes(self, client, db_session):
|
||
|
|
"""Changing LLM settings should not clear extra hints."""
|
||
|
|
client.post(
|
||
|
|
"/settings",
|
||
|
|
data={
|
||
|
|
"enabled": "on",
|
||
|
|
"base_url": "https://api.example.com/v1",
|
||
|
|
"model": "gpt-4o-mini",
|
||
|
|
"api_key": "sk-key",
|
||
|
|
"ai_search_enabled": "on",
|
||
|
|
"ai_search_extra_hints": "厨房用品和电子产品",
|
||
|
|
},
|
||
|
|
follow_redirects=False,
|
||
|
|
)
|
||
|
|
client.post(
|
||
|
|
"/settings",
|
||
|
|
data={
|
||
|
|
"enabled": "on",
|
||
|
|
"base_url": "https://api.example.com/v1",
|
||
|
|
"model": "gpt-4o",
|
||
|
|
"api_key": "",
|
||
|
|
"ai_search_enabled": "on",
|
||
|
|
"ai_search_extra_hints": "厨房用品和电子产品",
|
||
|
|
},
|
||
|
|
follow_redirects=False,
|
||
|
|
)
|
||
|
|
cfg = get_app_settings(db_session)
|
||
|
|
assert cfg.ai_search_extra_hints == "厨房用品和电子产品"
|
||
|
|
assert cfg.model == "gpt-4o"
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Regression: existing features still work without AI
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class TestRegressionWithoutAI:
|
||
|
|
def test_normal_search_still_works(self, client, db_session):
|
||
|
|
box = Box(name="测试箱")
|
||
|
|
db_session.add(box)
|
||
|
|
db_session.commit()
|
||
|
|
|
||
|
|
response = client.get("/search?q=测试")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert "测试箱" in response.text
|
||
|
|
|
||
|
|
def test_search_page_no_results(self, client):
|
||
|
|
response = client.get("/search?q=不存在")
|
||
|
|
assert "没有找到匹配结果" in response.text
|
||
|
|
|
||
|
|
def test_search_empty_query(self, client):
|
||
|
|
response = client.get("/search")
|
||
|
|
assert "输入关键词后" in response.text
|