Add LLM settings integration
test / pytest (push) Successful in 1m13s

Add app_settings migration, settings UI, and OpenAI-compatible httpx LLM client with mocked tests.

Preserve API keys on blank form submissions, require a fresh key when base_url changes, and keep AI search settings untouched for step 3.

Update docs/design LLM integration and step 3 AI search notes, including prompt contract and extra-hints planning.
This commit is contained in:
2026-06-01 20:06:22 +02:00
parent 8b8bd9f38f
commit d36b940981
12 changed files with 1254 additions and 15 deletions
+26 -10
View File
@@ -28,6 +28,7 @@ from app.db import Base, SessionLocal, configure_database
from app.migrate import (
V1_REVISION,
_detect_db_state,
_make_alembic_config,
run_migrations,
verify_schema_is_current,
)
@@ -35,6 +36,18 @@ from app.main import create_app
from fastapi.testclient import TestClient
def _get_head_revision() -> str:
"""Resolve the current Alembic head revision from migration scripts."""
from alembic.script import ScriptDirectory
cfg = _make_alembic_config("sqlite:///") # URL is unused for script lookup
script = ScriptDirectory.from_config(cfg)
return script.get_current_head()
HEAD_REVISION = _get_head_revision()
@pytest.fixture()
def tmp_db_path(tmp_path):
"""Provide a temporary SQLite database path."""
@@ -55,7 +68,7 @@ def tmp_db_url(tmp_db_path):
class TestFreshDBMigration:
"""Empty database gets all tables created by migration."""
def test_creates_all_three_tables(self, tmp_db_url):
def test_creates_all_tables(self, tmp_db_url):
run_migrations(tmp_db_url)
eng = create_engine(tmp_db_url)
tables = set(inspect(eng).get_table_names())
@@ -63,6 +76,7 @@ class TestFreshDBMigration:
assert "boxes" in tables
assert "items" in tables
assert "subitems" in tables
assert "app_settings" in tables
def test_creates_alembic_version_table(self, tmp_db_url):
run_migrations(tmp_db_url)
@@ -77,7 +91,7 @@ class TestFreshDBMigration:
with eng.begin() as conn:
version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
eng.dispose()
assert version == V1_REVISION
assert version == HEAD_REVISION
def test_boxes_table_has_all_columns(self, tmp_db_url):
run_migrations(tmp_db_url)
@@ -147,9 +161,11 @@ class TestUnmanagedDBAdoption2a:
"""Database with existing tables matching baseline gets adopted."""
def _create_old_db(self, db_url: str) -> None:
"""Simulate a pre-Alembic DB: create_all + insert data."""
"""Simulate a pre-Alembic DB: create V1 tables only + insert data."""
eng = create_engine(db_url)
Base.metadata.create_all(bind=eng)
# Only create V1 tables (boxes, items, subitems) — not app_settings
for table_name in ("boxes", "items", "subitems"):
Base.metadata.tables[table_name].create(bind=eng)
with eng.begin() as conn:
conn.execute(text(
"INSERT INTO boxes (name, room, status, created_at, updated_at) "
@@ -170,7 +186,7 @@ class TestUnmanagedDBAdoption2a:
with eng.begin() as conn:
version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
eng.dispose()
assert version == V1_REVISION
assert version == HEAD_REVISION
def test_data_preserved_after_adoption(self, tmp_db_url):
self._create_old_db(tmp_db_url)
@@ -187,14 +203,14 @@ class TestUnmanagedDBAdoption2a:
assert item_count == 1
assert box_name == "Kitchen Box"
def test_no_extra_tables_created(self, tmp_db_url):
def test_no_extra_tables_beyond_migrations(self, tmp_db_url):
self._create_old_db(tmp_db_url)
run_migrations(tmp_db_url)
eng = create_engine(tmp_db_url)
tables = set(inspect(eng).get_table_names())
eng.dispose()
assert tables == {"alembic_version", "boxes", "items", "subitems"}
assert tables == {"alembic_version", "boxes", "items", "subitems", "app_settings"}
def test_adoption_is_idempotent(self, tmp_db_url):
"""Running run_migrations twice does not error or duplicate data."""
@@ -209,7 +225,7 @@ class TestUnmanagedDBAdoption2a:
eng.dispose()
assert box_count == 1
assert version == V1_REVISION
assert version == HEAD_REVISION
# ---------------------------------------------------------------------------
@@ -396,7 +412,7 @@ class TestManagedDBMigration:
with eng.begin() as conn:
version = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
eng.dispose()
assert version == V1_REVISION
assert version == HEAD_REVISION
# ---------------------------------------------------------------------------
@@ -639,7 +655,7 @@ class TestProdDBCopyAdoption:
subitems_after = conn.execute(text("SELECT COUNT(*) FROM subitems")).scalar()
eng.dispose()
assert version == V1_REVISION
assert version == HEAD_REVISION
assert boxes_after == boxes_before
assert items_after == items_before
assert subitems_after == subitems_before
+634
View File
@@ -0,0 +1,634 @@
"""Tests for the settings store, LLM client, and settings routes.
All LLM calls are mocked — CI never touches the network.
"""
from unittest.mock import patch
import pytest
import app.llm as llm_module
from app.llm import LLMResult, expand_query, is_configured
from app.models import AppSetting
from app.settings_store import LLMConfig, get_app_settings, save_app_settings
# Alias to avoid pytest collecting it as a test function
_test_connection = llm_module.test_connection
# ---------------------------------------------------------------------------
# LLMConfig dataclass defaults
# ---------------------------------------------------------------------------
class TestLLMConfigDefaults:
def test_default_values(self):
cfg = LLMConfig()
assert cfg.enabled is False
assert cfg.base_url == "https://api.openai.com/v1"
assert cfg.model == ""
assert cfg.api_key == ""
assert cfg.ai_search_enabled is False
# ---------------------------------------------------------------------------
# settings_store: get_app_settings
# ---------------------------------------------------------------------------
class TestGetAppSettings:
def test_returns_defaults_when_no_rows(self, db_session):
cfg = get_app_settings(db_session)
assert cfg.enabled is False
assert cfg.base_url == "https://api.openai.com/v1"
assert cfg.model == ""
assert cfg.api_key == ""
assert cfg.ai_search_enabled is False
def test_reads_stored_values(self, db_session):
db_session.add(AppSetting(key="llm_enabled", value="true"))
db_session.add(AppSetting(key="llm_base_url", value="https://custom.api/v1"))
db_session.add(AppSetting(key="llm_model", value="gpt-4o"))
db_session.add(AppSetting(key="llm_api_key", value="sk-test-key"))
db_session.add(AppSetting(key="ai_search_enabled", value="true"))
db_session.commit()
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.base_url == "https://custom.api/v1"
assert cfg.model == "gpt-4o"
assert cfg.api_key == "sk-test-key"
assert cfg.ai_search_enabled is True
def test_handles_null_value_as_default(self, db_session):
db_session.add(AppSetting(key="llm_model", value=None))
db_session.commit()
cfg = get_app_settings(db_session)
assert cfg.model == ""
# ---------------------------------------------------------------------------
# settings_store: save_app_settings
# ---------------------------------------------------------------------------
class TestSaveAppSettings:
def test_saves_new_settings(self, db_session):
save_app_settings(
db_session,
enabled=True,
base_url="https://my-api.com/v1",
model="gpt-4o-mini",
api_key="sk-new-key",
)
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.base_url == "https://my-api.com/v1"
assert cfg.model == "gpt-4o-mini"
assert cfg.api_key == "sk-new-key"
def test_updates_existing_settings(self, db_session):
save_app_settings(db_session, enabled=True, model="old-model", api_key="key1")
save_app_settings(db_session, model="new-model")
cfg = get_app_settings(db_session)
assert cfg.model == "new-model"
# enabled was not passed in second save, so it stays unchanged
assert cfg.enabled is True
def test_api_key_none_preserves_old_key(self, db_session):
save_app_settings(db_session, api_key="sk-original")
save_app_settings(db_session, model="gpt-4o", api_key=None)
cfg = get_app_settings(db_session)
assert cfg.api_key == "sk-original"
assert cfg.model == "gpt-4o"
def test_api_key_empty_string_overwrites(self, db_session):
save_app_settings(db_session, api_key="sk-original")
save_app_settings(db_session, api_key="")
cfg = get_app_settings(db_session)
assert cfg.api_key == ""
def test_partial_save_only_updates_specified_fields(self, db_session):
save_app_settings(db_session, enabled=True, model="gpt-4o")
save_app_settings(db_session, base_url="https://new.url/v1")
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.model == "gpt-4o"
assert cfg.base_url == "https://new.url/v1"
# ---------------------------------------------------------------------------
# is_configured
# ---------------------------------------------------------------------------
class TestIsConfigured:
def test_false_when_disabled(self):
cfg = LLMConfig(enabled=False, model="gpt-4o", api_key="sk-key")
assert is_configured(cfg) is False
def test_false_when_no_model(self):
cfg = LLMConfig(enabled=True, model="", api_key="sk-key")
assert is_configured(cfg) is False
def test_false_when_no_api_key(self):
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="")
assert is_configured(cfg) is False
def test_true_when_all_set(self):
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
assert is_configured(cfg) is True
# ---------------------------------------------------------------------------
# test_connection (mocked)
# ---------------------------------------------------------------------------
class TestTestConnection:
def test_returns_failure_when_not_configured(self):
cfg = LLMConfig(enabled=False)
result = _test_connection(cfg)
assert result.success is False
assert "未配置" in result.message
@patch("app.llm._call_chat_completion")
def test_success_when_configured(self, mock_call):
mock_call.return_value = {"choices": [{"message": {"content": "Hi"}}]}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = _test_connection(cfg)
assert result.success is True
assert "连接成功" in result.message
assert "gpt-4o" in result.message
@patch("app.llm._call_chat_completion")
def test_handles_http_error(self, mock_call):
import httpx
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-bad")
result = _test_connection(cfg)
assert result.success is False
assert "401" in result.message
@patch("app.llm._call_chat_completion")
def test_handles_connect_error(self, mock_call):
import httpx
mock_call.side_effect = httpx.ConnectError("refused")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = _test_connection(cfg)
assert result.success is False
assert "无法连接" in result.message
@patch("app.llm._call_chat_completion")
def test_handles_timeout(self, mock_call):
import httpx
mock_call.side_effect = httpx.TimeoutException("timeout")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = _test_connection(cfg)
assert result.success is False
assert "超时" in result.message
# ---------------------------------------------------------------------------
# expand_query (mocked)
# ---------------------------------------------------------------------------
class TestExpandQuery:
def test_returns_original_when_not_configured(self):
cfg = LLMConfig(enabled=False)
result = expand_query(cfg, "")
assert result == [""]
@patch("app.llm._call_chat_completion")
def test_expands_query_successfully(self, mock_call):
mock_call.return_value = {
"choices": [
{"message": {"content": "平底锅\n炒锅\n锅具\n厨房锅"}}
]
}
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert "" in result
assert "平底锅" in result
assert len(result) >= 4
@patch("app.llm._call_chat_completion")
def test_fallback_on_api_failure(self, mock_call):
mock_call.side_effect = Exception("network down")
cfg = LLMConfig(enabled=True, model="gpt-4o", api_key="sk-key")
result = expand_query(cfg, "")
assert result == [""]
@patch("app.llm._call_chat_completion")
def test_fallback_on_empty_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 result == [""]
# ---------------------------------------------------------------------------
# Routes: GET /settings
# ---------------------------------------------------------------------------
class TestSettingsPage:
def test_settings_page_returns_200(self, client):
response = client.get("/settings")
assert response.status_code == 200
def test_settings_page_has_form_elements(self, client):
response = client.get("/settings")
assert "设置" in response.text
assert 'name="enabled"' in response.text
assert 'name="base_url"' in response.text
assert 'name="model"' in response.text
assert 'name="api_key"' in response.text
assert "保存设置" in response.text
assert "测试连接" in response.text
def test_settings_page_shows_nav_link(self, client):
response = client.get("/boxes")
assert "设置" in response.text
assert 'href="/settings"' in response.text
def test_settings_page_no_api_key_echoed(self, client, db_session):
save_app_settings(db_session, api_key="sk-super-secret-key-12345")
response = client.get("/settings")
assert "sk-super-secret-key-12345" not in response.text
assert "已配置" in response.text
def test_settings_page_shows_placeholder_when_no_key(self, client):
response = client.get("/settings")
assert "输入 API Key" in response.text
def test_settings_page_shows_default_base_url(self, client):
response = client.get("/settings")
assert "https://api.openai.com/v1" in response.text
# ---------------------------------------------------------------------------
# Routes: POST /settings
# ---------------------------------------------------------------------------
class TestSaveSettingsRoute:
def test_save_settings_redirects(self, client):
response = client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://my-api.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/settings"
def test_saved_settings_persist(self, client, db_session):
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://my-api.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-test-key",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.enabled is True
assert cfg.base_url == "https://my-api.com/v1"
assert cfg.model == "gpt-4o-mini"
assert cfg.api_key == "sk-test-key"
def test_save_with_blank_api_key_preserves_old(self, client, db_session):
# First save with a key
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-original"},
follow_redirects=False,
)
# Second save without key (blank)
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": ""},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.api_key == "sk-original"
def test_save_disabled_state(self, client, db_session):
# First enable
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-key"},
follow_redirects=False,
)
# Then disable (no 'enabled' checkbox)
client.post(
"/settings",
data={"model": "gpt-4o", "api_key": ""},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.enabled is False
def test_save_settings_no_api_key_in_redirect_page(self, client):
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-secret-key"},
follow_redirects=False,
)
response = client.get("/settings")
assert "sk-secret-key" not in response.text
def test_save_refuses_when_base_url_changes_and_key_blank(self, client, db_session):
"""P1 fix: if base_url changes and api_key is blank, refuse save with error."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://old-api.com/v1",
"model": "gpt-4o",
"api_key": "sk-old-key",
},
follow_redirects=False,
)
# Try saving with different base_url, no key
response = client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://new-api.com/v1",
"model": "gpt-4o",
"api_key": "", # blank + base_url changed → refuse
},
)
assert response.status_code == 200
assert "请重新输入 API Key 后保存" in response.text
# Old config should be unchanged — nothing was saved
cfg = get_app_settings(db_session)
assert cfg.base_url == "https://old-api.com/v1"
assert cfg.api_key == "sk-old-key"
def test_save_preserves_key_when_endpoint_unchanged_and_key_blank(self, client, db_session):
"""P1 fix: if endpoint is unchanged and api_key is blank, keep old key."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-original",
},
follow_redirects=False,
)
# Re-save same endpoint, blank key
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.api_key == "sk-original"
def test_save_preserves_key_when_only_model_changes_and_key_blank(self, client, db_session):
"""Model change alone should not clear the key — same base_url, different model."""
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-original",
},
follow_redirects=False,
)
# Change only model, leave api_key blank
client.post(
"/settings",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"api_key": "",
},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.model == "gpt-4o-mini"
assert cfg.api_key == "sk-original"
def test_save_does_not_touch_ai_search_enabled(self, client, db_session):
"""P2 fix: saving LLM settings must not reset ai_search_enabled."""
db_session.add(AppSetting(key="ai_search_enabled", value="true"))
db_session.commit()
client.post(
"/settings",
data={"enabled": "on", "model": "gpt-4o", "api_key": "sk-key"},
follow_redirects=False,
)
cfg = get_app_settings(db_session)
assert cfg.ai_search_enabled is True
# ---------------------------------------------------------------------------
# Routes: POST /settings/test
# ---------------------------------------------------------------------------
class TestTestConnectionRoute:
@patch("app.llm._call_chat_completion")
def test_test_connection_success(self, mock_call, client):
mock_call.return_value = {"choices": [{"message": {"content": "Hi"}}]}
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-test",
},
)
assert response.status_code == 200
assert "连接成功" in response.text
assert "gpt-4o" in response.text
@patch("app.llm._call_chat_completion")
def test_test_connection_failure(self, mock_call, client):
import httpx
mock_call.side_effect = httpx.HTTPStatusError(
"401",
request=httpx.Request("POST", "http://x"),
response=httpx.Response(401),
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "sk-bad",
},
)
assert response.status_code == 200
assert "连接失败" in response.text
assert "401" in response.text
def test_test_connection_not_configured(self, client):
response = client.post(
"/settings/test",
data={
"enabled": "", # not checked
"base_url": "https://api.openai.com/v1",
"model": "",
"api_key": "",
},
)
assert response.status_code == 200
assert "未配置" in response.text
@patch("app.llm._call_chat_completion")
def test_test_connection_uses_stored_key_when_endpoint_matches(self, mock_call, client, db_session):
"""When api_key is blank but base_url and model match saved config, the stored key should be used."""
mock_call.return_value = {"choices": [{"message": {"content": "Hi"}}]}
# Store a config first
save_app_settings(
db_session,
enabled=True,
base_url="https://api.openai.com/v1",
model="gpt-4o",
api_key="sk-stored-key",
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o",
"api_key": "", # blank → use stored key (endpoint matches)
},
)
assert response.status_code == 200
assert "连接成功" in response.text
@patch("app.llm._call_chat_completion")
def test_test_connection_uses_stored_key_when_only_model_changes(self, mock_call, client, db_session):
"""Model changes under the same base_url can reuse the stored key."""
captured = {}
def fake_call(cfg, **kwargs):
captured["base_url"] = cfg.base_url
captured["model"] = cfg.model
captured["api_key"] = cfg.api_key
return {"choices": [{"message": {"content": "Hi"}}]}
mock_call.side_effect = fake_call
save_app_settings(
db_session,
enabled=True,
base_url="https://api.openai.com/v1",
model="gpt-4o",
api_key="sk-stored-key",
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"api_key": "",
},
)
assert response.status_code == 200
assert "连接成功" in response.text
assert captured == {
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"api_key": "sk-stored-key",
}
def test_test_connection_refuses_stored_key_when_endpoint_changed(self, client, db_session):
"""When base_url changed and api_key is blank, refuse to test."""
save_app_settings(
db_session,
enabled=True,
base_url="https://api.openai.com/v1",
model="gpt-4o",
api_key="sk-stored-key",
)
response = client.post(
"/settings/test",
data={
"enabled": "on",
"base_url": "https://attacker.example/v1", # different endpoint
"model": "gpt-4o",
"api_key": "", # blank → refuse
},
)
assert response.status_code == 200
assert "请重新输入 API Key" in response.text
def test_test_connection_result_shows_on_settings_page(self, client):
"""Test result is rendered on the same settings page."""
response = client.post(
"/settings/test",
data={
"enabled": "",
"base_url": "https://api.openai.com/v1",
"model": "",
"api_key": "",
},
)
assert response.status_code == 200
assert "设置" in response.text
assert "保存设置" in response.text
# ---------------------------------------------------------------------------
# Graceful degradation: unconfigured LLM does not affect existing features
# ---------------------------------------------------------------------------
class TestGracefulDegradation:
def test_boxes_page_works_without_llm_config(self, client):
response = client.get("/boxes")
assert response.status_code == 200
def test_search_page_works_without_llm_config(self, client):
response = client.get("/search?q=test")
assert response.status_code == 200
def test_crud_works_without_llm_config(self, client, db_session):
from app.models import Box
response = client.post(
"/boxes",
data={"name": "No LLM Box"},
follow_redirects=False,
)
assert response.status_code == 303
assert db_session.query(Box).count() == 1