Merge origin/main into feat/export-service
同步 main（054f704），解决 contracts.py / main.py / README.md / 技术栈说明 的合并冲突。 - contracts.py：保留 pydantic 多行导入并新增 RequestOverride - main.py：合并 lifespan（导出孤儿清理 + 转写/本地模型生命周期） - README.md / 技术栈说明：文档取 main 最新版本 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Code
@@ -19,5 +19,19 @@ def _isolate_data_dir(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("APP_VAULT_PATH", str(tmp_path / "vault"))
|
||||
# 清除 lru 缓存，让本次测试内的 get_settings() 读到临时目录
|
||||
get_settings.cache_clear()
|
||||
# Unit tests explicitly inject deterministic embeddings. Production uses real models.
|
||||
from app import container as container_module
|
||||
from app.services import note_service
|
||||
from app.retrieval.engine import engine
|
||||
from app.retrieval.embedding import HashEmbeddingProvider
|
||||
from app.providers.routing import ModelRoutingService
|
||||
def test_routing(providers, credentials):
|
||||
return ModelRoutingService(providers, credentials, local_embedding=HashEmbeddingProvider())
|
||||
monkeypatch.setattr(container_module, "_local_model_routing", test_routing)
|
||||
monkeypatch.setattr(container_module.container.model_routing, "local_embedding", HashEmbeddingProvider())
|
||||
monkeypatch.setattr(note_service, "embedding", HashEmbeddingProvider())
|
||||
test_embedding = HashEmbeddingProvider()
|
||||
monkeypatch.setattr(engine, "embedding", test_embedding)
|
||||
monkeypatch.setattr(engine, "_routed_defaults", (test_embedding, engine.vector_store))
|
||||
yield
|
||||
get_settings.cache_clear()
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import asyncio
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from app.contracts import ChatRequest, Message, ModelEvent, ModelEventType, SearchRequest
|
||||
from app.routes import chat, utc_now
|
||||
from app.services import note_service
|
||||
from app.services.chat_context import prepare
|
||||
|
||||
|
||||
@pytest.mark.parametrize('enabled', [True, False])
|
||||
def test_chat_stream_retrieves_real_notes_and_emits_sources(monkeypatch, enabled):
|
||||
received = []
|
||||
|
||||
class Adapter:
|
||||
async def stream(self, request):
|
||||
received.append(request)
|
||||
yield ModelEvent(event=ModelEventType.text_delta, sequence=0, data={'text': 'answer [1]'}, timestamp=utc_now())
|
||||
yield ModelEvent(event=ModelEventType.done, sequence=1, data={}, timestamp=utc_now())
|
||||
|
||||
monkeypatch.setattr('app.routes.provider_or_404', lambda _: SimpleNamespace(adapter=Adapter()))
|
||||
|
||||
async def scenario():
|
||||
note = await note_service.create_note(title='Orchard', markdown='apple orchard knowledge', folder=None, tags=[])
|
||||
request = ChatRequest(provider_id='test', model='test', use_rag=enabled,
|
||||
system='Keep original instructions',
|
||||
messages=[Message(role='user', content='apple')],
|
||||
retrieval=SearchRequest(query='apple', mode='fts'))
|
||||
response = await chat(request)
|
||||
chunks = [chunk async for chunk in response.body_iterator]
|
||||
events = [json.loads(chunk.split('data: ', 1)[1]) for chunk in chunks]
|
||||
assert [e['sequence'] for e in events] == list(range(len(events)))
|
||||
assert events[-1]['event'] == 'Done'
|
||||
assert received[0].messages == request.messages
|
||||
if enabled:
|
||||
assert events[0]['event'] == 'Citation'
|
||||
assert events[0]['data']['note_id'] == note.note_id
|
||||
assert 'apple orchard knowledge' in received[0].system
|
||||
assert 'Keep original instructions' in received[0].system
|
||||
else:
|
||||
assert all(e['event'] != 'Citation' for e in events)
|
||||
assert received[0].system == request.system
|
||||
assert request.system == 'Keep original instructions'
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_empty_knowledge_base_has_no_invented_citations():
|
||||
async def scenario():
|
||||
request = ChatRequest(provider_id='test', model='test', messages=[Message(role='user', content='missing')])
|
||||
grounded, sources = await prepare(request)
|
||||
assert sources == []
|
||||
assert '不要编造' in grounded.system
|
||||
asyncio.run(scenario())
|
||||
@@ -0,0 +1,121 @@
|
||||
import asyncio
|
||||
from datetime import datetime, timezone
|
||||
from types import SimpleNamespace
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
|
||||
from app.contracts import ChatRequest, ModelEvent, ModelEventType
|
||||
from app.main import app
|
||||
from app.services import chat_history
|
||||
|
||||
|
||||
def test_chat_history_survives_new_connections_and_deletes_messages() -> None:
|
||||
conversation = chat_history.create("Persistent chat", "conversation-1")
|
||||
chat_history.append_message(
|
||||
conversation.conversation_id,
|
||||
message_id="user-1",
|
||||
role="user",
|
||||
content="question",
|
||||
)
|
||||
chat_history.append_message(
|
||||
conversation.conversation_id,
|
||||
message_id="assistant-1",
|
||||
role="assistant",
|
||||
content="answer",
|
||||
citations=[{"note_id": "note-1", "heading_path": ["Heading"]}],
|
||||
usage={"input_tokens": 2, "output_tokens": 1, "total_tokens": 3},
|
||||
)
|
||||
|
||||
listed, total = chat_history.list_conversations(50, 0)
|
||||
messages, message_total = chat_history.list_messages("conversation-1", 50, 0)
|
||||
assert total == 1
|
||||
assert listed[0].message_count == 2
|
||||
assert message_total == 2
|
||||
assert messages[1].citations[0]["note_id"] == "note-1"
|
||||
assert messages[1].usage["total_tokens"] == 3
|
||||
|
||||
assert chat_history.delete("conversation-1") is True
|
||||
assert chat_history.list_conversations(50, 0)[1] == 0
|
||||
|
||||
|
||||
def test_chat_stream_persists_user_and_assistant_messages(monkeypatch) -> None:
|
||||
from app import routes
|
||||
|
||||
class Adapter:
|
||||
async def stream(self, _request):
|
||||
now = datetime.now(timezone.utc)
|
||||
yield ModelEvent(event=ModelEventType.text_delta, data={"text": "persisted answer"}, timestamp=now)
|
||||
yield ModelEvent(event=ModelEventType.usage, data={"input_tokens": 4, "output_tokens": 2}, timestamp=now)
|
||||
yield ModelEvent(event=ModelEventType.done, timestamp=now)
|
||||
|
||||
monkeypatch.setattr(routes, "provider_or_404", lambda _provider_id: SimpleNamespace(adapter=Adapter()))
|
||||
payload = {
|
||||
"provider_id": "configured",
|
||||
"model": "model",
|
||||
"conversation_id": "conversation-stream",
|
||||
"user_message_id": "user-stream",
|
||||
"assistant_message_id": "assistant-stream",
|
||||
"conversation_title": "Persist this",
|
||||
"use_rag": False,
|
||||
"messages": [{"role": "user", "content": "question"}],
|
||||
}
|
||||
with TestClient(app) as client:
|
||||
with client.stream("POST", "/api/chat", json=payload) as response:
|
||||
assert response.status_code == 200
|
||||
assert "persisted answer" in "".join(response.iter_text())
|
||||
messages = client.get("/api/chat/conversations/conversation-stream/messages").json()["items"]
|
||||
conversations = client.get("/api/chat/conversations").json()["items"]
|
||||
assert [message["content"] for message in messages] == ["question", "persisted answer"]
|
||||
assert messages[1]["usage"]["total_tokens"] == 6
|
||||
assert conversations[0]["title"] == "Persist this"
|
||||
assert conversations[0]["message_count"] == 2
|
||||
|
||||
|
||||
def test_chat_conversation_crud_api() -> None:
|
||||
with TestClient(app) as client:
|
||||
created = client.post("/api/chat/conversations", json={"conversation_id": "crud", "title": "CRUD"})
|
||||
assert created.status_code == 201
|
||||
assert client.get("/api/chat/conversations").json()["page"]["total"] == 1
|
||||
assert client.get("/api/chat/conversations/crud/messages").json()["items"] == []
|
||||
assert client.delete("/api/chat/conversations/crud").status_code == 200
|
||||
missing = client.get("/api/chat/conversations/crud/messages")
|
||||
assert missing.status_code == 404
|
||||
assert missing.json()["error"]["code"] == "CONVERSATION_NOT_FOUND"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("close_early", [True, False])
|
||||
@pytest.mark.parametrize("deleted", [True, False])
|
||||
def test_stream_finalization_respects_conversation_deletion(monkeypatch, close_early, deleted) -> None:
|
||||
from app import routes
|
||||
|
||||
class Adapter:
|
||||
async def stream(self, _request):
|
||||
now = datetime.now(timezone.utc)
|
||||
yield ModelEvent(event=ModelEventType.text_delta, data={"text": "partial answer"}, timestamp=now)
|
||||
yield ModelEvent(event=ModelEventType.done, timestamp=now)
|
||||
|
||||
monkeypatch.setattr(routes, "provider_or_404", lambda _: SimpleNamespace(adapter=Adapter()))
|
||||
|
||||
async def scenario():
|
||||
response = await routes.chat(ChatRequest(
|
||||
provider_id="configured", model="model", conversation_id="stream",
|
||||
use_rag=False, messages=[{"role": "user", "content": "question"}],
|
||||
))
|
||||
await anext(response.body_iterator)
|
||||
if deleted:
|
||||
assert chat_history.delete("stream")
|
||||
if close_early:
|
||||
await response.body_iterator.aclose()
|
||||
else:
|
||||
async for _ in response.body_iterator:
|
||||
pass
|
||||
if deleted:
|
||||
assert chat_history.get("stream") is None
|
||||
assert chat_history.list_conversations(50, 0)[1] == 0
|
||||
else:
|
||||
messages, total = chat_history.list_messages("stream", 50, 0)
|
||||
assert total == 2
|
||||
assert [message.content for message in messages] == ["question", "partial answer"]
|
||||
|
||||
asyncio.run(scenario())
|
||||
@@ -0,0 +1,139 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from app.local_models import manager
|
||||
from app.local_models.runtime import Runtime
|
||||
from app.providers.base import ProviderError
|
||||
|
||||
|
||||
def test_download_resumes_partial_and_checks_digest(monkeypatch):
|
||||
payload = b'verified-model-weights'
|
||||
entry = {'path':'model.safetensors','size':len(payload),'hash':hashlib.sha256(payload).hexdigest(),
|
||||
'algorithm':'sha256','url':'https://fixture.invalid/weights'}
|
||||
async def manifest(client, spec):
|
||||
return [entry]
|
||||
monkeypatch.setattr(manager, '_manifest', manifest)
|
||||
path = manager.model_path('bekko')
|
||||
path.mkdir(parents=True)
|
||||
(path/'model.safetensors.partial').write_bytes(payload[:5])
|
||||
requests = []
|
||||
def respond(request):
|
||||
requests.append(request)
|
||||
assert request.headers['range'] == 'bytes=5-'
|
||||
return httpx.Response(206, headers={'content-range':f'bytes 5-{len(payload)-1}/{len(payload)}'},content=payload[5:])
|
||||
original = httpx.AsyncClient
|
||||
monkeypatch.setattr(manager.httpx,'AsyncClient',lambda **kwargs:original(**kwargs,transport=httpx.MockTransport(respond)))
|
||||
asyncio.run(manager._download('bekko'))
|
||||
assert manager.read_state('bekko')['status'] == 'installed'
|
||||
assert (path/'model.safetensors').read_bytes() == payload
|
||||
assert manager.valid_file(path/'model.safetensors',entry)
|
||||
(path/'model.safetensors').write_bytes(b'x'*len(payload))
|
||||
assert not manager.valid_file(path/'model.safetensors',entry)
|
||||
assert len(requests) == 1
|
||||
|
||||
|
||||
def test_local_model_missing_is_explicit():
|
||||
with pytest.raises(ProviderError) as error:
|
||||
asyncio.run(Runtime().infer('qwen3-asr','transcription',{'source':'missing.wav'}))
|
||||
assert error.value.code == 'LOCAL_MODEL_NOT_INSTALLED'
|
||||
|
||||
|
||||
def test_cancel_reaps_active_model_process(monkeypatch):
|
||||
import app.local_models.runtime as module
|
||||
monkeypatch.setattr(module,'read_state',lambda key:{'status':'installed'})
|
||||
monkeypatch.setattr(module,'interpreter',lambda *_:Path(sys.executable))
|
||||
class Input:
|
||||
def write(self, value):
|
||||
request = json.loads(value)
|
||||
assert request['config']['device'] == 'cpu'
|
||||
async def drain(self):
|
||||
pass
|
||||
def close(self):
|
||||
pass
|
||||
class Process:
|
||||
returncode = None
|
||||
stdin = Input()
|
||||
def __init__(self):
|
||||
self.stdout = asyncio.StreamReader()
|
||||
self.killed = False
|
||||
def kill(self):
|
||||
self.killed = True
|
||||
self.returncode = -9
|
||||
self.stdout.feed_eof()
|
||||
async def wait(self):
|
||||
return self.returncode
|
||||
async def scenario():
|
||||
started = asyncio.Event()
|
||||
process = Process()
|
||||
async def spawn(*args, **kwargs):
|
||||
assert kwargs['env']['HF_HUB_OFFLINE'] == '1'
|
||||
started.set()
|
||||
return process
|
||||
monkeypatch.setattr(module.asyncio,'create_subprocess_exec',spawn)
|
||||
runtime = Runtime()
|
||||
task = asyncio.create_task(runtime.infer('qwen3-asr','transcription',{'source':'fixture.wav'}))
|
||||
await started.wait()
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
assert process.killed and not runtime.active
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("cancel", [False, True])
|
||||
def test_subprocess_fallback_runs_and_reaps_real_worker(monkeypatch, tmp_path, cancel):
|
||||
import app.local_models.runtime as module
|
||||
import app.local_models.process as process_module
|
||||
|
||||
monkeypatch.setattr(module, 'read_state', lambda key: {'status': 'installed'})
|
||||
monkeypatch.setattr(module, 'interpreter', lambda *_: Path(sys.executable))
|
||||
worker = tmp_path / 'worker.py'
|
||||
worker.write_text(
|
||||
'import json,sys,time\n'
|
||||
'request=json.load(sys.stdin)\n'
|
||||
'print(json.dumps({"progress": 1}),flush=True)\n'
|
||||
+ ('time.sleep(60)\n' if cancel else '')
|
||||
+ 'print(json.dumps({"result": [[1.0,0.0]], "usage": {"input_tokens": 2}}),flush=True)\n',
|
||||
encoding='utf-8',
|
||||
)
|
||||
processes = []
|
||||
original = process_module.ThreadedProcess
|
||||
|
||||
def spawn(args, **kwargs):
|
||||
process = original((sys.executable, str(worker)), **kwargs)
|
||||
processes.append(process)
|
||||
return process
|
||||
|
||||
async def unsupported(*args, **kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
monkeypatch.setattr(module.asyncio, 'create_subprocess_exec', unsupported)
|
||||
monkeypatch.setattr(process_module, 'ThreadedProcess', spawn)
|
||||
|
||||
async def scenario():
|
||||
runtime = Runtime()
|
||||
started = asyncio.Event()
|
||||
token = module.runtime_progress.set(lambda message: started.set())
|
||||
try:
|
||||
task = asyncio.create_task(runtime.infer('bekko', 'embedding', {'texts': ['test']}))
|
||||
await asyncio.wait_for(started.wait(), 10)
|
||||
if cancel:
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
else:
|
||||
assert await task == [[1.0, 0.0]]
|
||||
assert not runtime.active and not runtime.active_files and not runtime.waiters
|
||||
assert processes[0].returncode is not None
|
||||
assert processes[0].process.stdin.closed
|
||||
assert processes[0].process.stdout.closed
|
||||
finally:
|
||||
module.runtime_progress.reset(token)
|
||||
|
||||
asyncio.run(scenario())
|
||||
@@ -0,0 +1,132 @@
|
||||
"""Durability, cancellation and optimistic editing without model downloads."""
|
||||
import asyncio
|
||||
from contextlib import closing
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.contracts import TranscriptEditRequest
|
||||
from app.database.db import connect
|
||||
from app.errors import ApiError
|
||||
from app.services import transcription_service as jobs
|
||||
from app.services.attachment_service import attachment_path
|
||||
|
||||
|
||||
def text_attachment():
|
||||
path = attachment_path("lecture.txt")
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("原始识别内容", encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def test_idempotency_edit_history_and_event_replay():
|
||||
text_attachment()
|
||||
|
||||
async def scenario():
|
||||
first = await jobs.create_transcription("lecture.txt", idempotency_key="submit-1")
|
||||
repeated = await jobs.create_transcription("lecture.txt", idempotency_key="submit-1")
|
||||
assert first.job_id == repeated.job_id
|
||||
assert first.status == "completed"
|
||||
with pytest.raises(ApiError) as conflict:
|
||||
await jobs.create_transcription("lecture.txt", language="en", idempotency_key="submit-1")
|
||||
assert conflict.value.code == "IDEMPOTENCY_CONFLICT"
|
||||
revised = jobs.edit(first.job_id, TranscriptEditRequest(revision=1, text="校对内容"))
|
||||
assert revised.original_text == "原始识别内容"
|
||||
assert revised.revision == 2
|
||||
with pytest.raises(ApiError) as stale:
|
||||
jobs.edit(first.job_id, TranscriptEditRequest(revision=1, text="覆盖"))
|
||||
assert stale.value.code == "VERSION_CONFLICT"
|
||||
with closing(connect()) as conn:
|
||||
assert conn.execute("SELECT COUNT(*) FROM media_revisions").fetchone()[0] == 1
|
||||
events = jobs.events(first.job_id)
|
||||
assert [e["event"] for e in events] == ["Queued", "TranscriptionStarted", "Completed", "Revised"]
|
||||
assert jobs.events(first.job_id, events[-2]["sequence"]) == events[-1:]
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_cancel_before_start_retry_and_restart_recovery():
|
||||
text_attachment()
|
||||
|
||||
async def scenario():
|
||||
job = await jobs.create_transcription("lecture.txt", wait=False)
|
||||
cancelled = await jobs.cancel(job.job_id)
|
||||
assert cancelled.status == "cancelled"
|
||||
next_job = await jobs.retry(job.job_id)
|
||||
assert next_job.previous_job_id == job.job_id
|
||||
assert next_job.job_id != job.job_id
|
||||
await jobs._tasks[jobs.task_key(next_job.job_id)]
|
||||
assert jobs.require_job(next_job.job_id).status == "completed"
|
||||
# Simulate a persisted job left behind by a stopped process.
|
||||
cancelled.status = "running"
|
||||
jobs.save(cancelled, "TranscriptionStarted")
|
||||
jobs.recover_interrupted()
|
||||
assert jobs.require_job(job.job_id).error_code == "TRANSCRIPTION_INTERRUPTED"
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_controlled_upload_and_async_http_flow():
|
||||
from app.main import app
|
||||
with TestClient(app) as client:
|
||||
assert client.post("/api/media/attachments?filename=a.wav", content=b"").status_code == 422
|
||||
uploaded = client.post("/api/media/attachments?filename=lecture.txt", content="真实转写文本".encode())
|
||||
assert uploaded.status_code == 201
|
||||
attachment_id = uploaded.json()["attachment_id"]
|
||||
assert client.get(f"/api/media/attachments/{attachment_id}").content == "真实转写文本".encode()
|
||||
response = client.post("/api/media/transcriptions", json={"attachment_id": attachment_id})
|
||||
assert response.status_code == 202 and response.json()["status"] == "queued"
|
||||
job_id = response.json()["job_id"]
|
||||
events = client.get(f"/api/media/transcriptions/{job_id}/events")
|
||||
assert "event: Completed" in events.text
|
||||
assert client.get("/api/media/transcriptions").json()["page"]["total"] == 1
|
||||
assert client.get(f"/api/media/transcriptions/{job_id}").json()["text"] == "真实转写文本"
|
||||
assert client.get(f"/api/media/transcriptions/{job_id}/events", headers={"Last-Event-ID": "bad"}).status_code == 422
|
||||
|
||||
|
||||
def test_terminology_export_and_privacy_cleanup():
|
||||
from app.main import app
|
||||
text_attachment()
|
||||
with TestClient(app) as client:
|
||||
created = client.post('/api/media/transcriptions', json={'attachment_id':'lecture.txt','terminology':{'识别':'校对'}}).json()
|
||||
job_id = created['job_id']
|
||||
client.get(f'/api/media/transcriptions/{job_id}/events')
|
||||
job = client.get(f'/api/media/transcriptions/{job_id}').json()
|
||||
assert job['text'] == '原始校对内容' and job['original_text'] == '原始识别内容'
|
||||
first = client.post(f'/api/media/transcriptions/{job_id}/notes', json={'title':'课程'}).json()
|
||||
again = client.post(f'/api/media/transcriptions/{job_id}/notes', json={'title':'课程'}).json()
|
||||
assert first['note_id'] == again['note_id']
|
||||
response = client.delete('/api/media/attachments/lecture.txt')
|
||||
assert first['note_id'] in response.json()['retained_note_ids']
|
||||
cleaned = client.get(f'/api/media/transcriptions/{job_id}').json()
|
||||
assert cleaned['text'] is None and cleaned['original_text'] is None and cleaned['corrections'] == []
|
||||
assert client.post(f'/api/media/transcriptions/{job_id}/retry').status_code == 409
|
||||
assert client.get('/api/media/attachments/lecture.txt').status_code == 404
|
||||
|
||||
|
||||
def test_local_only_export_and_rebuild_keep_local_embedding_policy(monkeypatch):
|
||||
from types import SimpleNamespace
|
||||
from app.contracts import TranscriptNoteRequest, IndexRebuildRequest
|
||||
from app.local_models.runtime import LocalEmbedding
|
||||
from app.retrieval import routed_vectors
|
||||
from app.services import note_service, index_service
|
||||
from app.services.media_notes import create_transcript_note
|
||||
calls = []
|
||||
class Routing:
|
||||
async def embed(self, texts, *, local_only=False):
|
||||
calls.append(local_only)
|
||||
assert local_only
|
||||
return SimpleNamespace(source='local', model_id='local-test', dimensions=2,
|
||||
vectors=[[1.0, 0.0] for _ in texts], fallback_reason=None)
|
||||
monkeypatch.setattr(routed_vectors, 'get_model_routing', lambda: Routing())
|
||||
monkeypatch.setattr(note_service, 'embedding', LocalEmbedding())
|
||||
text_attachment()
|
||||
async def scenario():
|
||||
job = await jobs.create_transcription('lecture.txt', local_only=True)
|
||||
note = await create_transcript_note(job.job_id, TranscriptNoteRequest(title='Private'))
|
||||
assert note.markdown.startswith('---\nembedding_local_only: true\n---')
|
||||
await note_service.update_note(note.note_id, markdown=note.markdown.replace(
|
||||
'embedding_local_only: true', 'embedding_local_only: true # keep local'))
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
assert len(calls) >= 3 and all(calls)
|
||||
asyncio.run(scenario())
|
||||
@@ -641,9 +641,14 @@ def test_api_speech_failure_reports_reason_in_503_and_transcription_job(api):
|
||||
assert match.status_code == 503
|
||||
assert match.json()["error"]["code"] == "LOCAL_MODEL_NOT_INSTALLED"
|
||||
assert match.json()["error"]["details"] == {"fallback_reason": "PROVIDER_UNAVAILABLE"}
|
||||
transcript = api.client.post("/api/media/transcriptions", json={"attachment_id": source.name, "language": "zh"})
|
||||
assert transcript.status_code == 202
|
||||
job = transcript.json()
|
||||
with api.client:
|
||||
transcript = api.client.post("/api/media/transcriptions", json={"attachment_id": source.name, "language": "zh"})
|
||||
assert transcript.status_code == 202
|
||||
job = transcript.json()
|
||||
assert job["status"] == "queued"
|
||||
stream = api.client.get(f"/api/media/transcriptions/{job['job_id']}/events")
|
||||
assert "event: Failed" in stream.text
|
||||
job = api.client.get(f"/api/media/transcriptions/{job['job_id']}").json()
|
||||
assert job["status"] == "failed" and job["error_code"] == "LOCAL_MODEL_NOT_INSTALLED"
|
||||
assert job["fallback_reason"] == "PROVIDER_UNAVAILABLE"
|
||||
assert api.client.get(f"/api/media/transcriptions/{job['job_id']}").json() == job
|
||||
@@ -661,3 +666,55 @@ def test_out_of_float_range_json_number_is_invalid_remote_and_falls_back(rig, au
|
||||
result = run(media_call(rig, capability, audio))
|
||||
assert result.source == "local" and result.score == rig.speech.score
|
||||
assert result.fallback_reason == "PROVIDER_INVALID_RESPONSE"
|
||||
|
||||
|
||||
def test_remote_segments_are_validated_and_local_only_skips_api(rig, audio):
|
||||
bind(rig, "transcription")
|
||||
rig.http.handler = lambda request: response({"text":"内容", "segments":[{"start":0,"end":1.5,"text":"内容"}]})
|
||||
result = run(rig.service.transcribe(audio[0], "zh"))
|
||||
assert result.source == "api" and result.segments[0].end_time == 1.5
|
||||
rig.http.handler = lambda request: response({"text":"内容", "segments":[{"start":2,"end":1,"text":"内容"}]})
|
||||
assert run(rig.service.transcribe(audio[0], "zh")).fallback_reason == "PROVIDER_INVALID_RESPONSE"
|
||||
count = len(rig.requests)
|
||||
result = run(rig.service.transcribe(audio[0], "zh", local_only=True))
|
||||
assert result.source == "local" and len(rig.requests) == count
|
||||
|
||||
|
||||
def test_embedding_local_only_does_not_change_normal_api_fallback(rig):
|
||||
bind(rig)
|
||||
result = run(rig.service.embed(['private'], local_only=True))
|
||||
assert result.source == 'local' and result.fallback_reason is None
|
||||
assert rig.requests == [] and rig.credentials.calls == []
|
||||
rig.http.handler = lambda request: response({'data': [{'index': 0, 'embedding': [1, 0, 0]}]})
|
||||
assert run(rig.service.embed(['normal'])).source == 'api'
|
||||
rig.http.handler = lambda request: response({}, status=503)
|
||||
result = run(rig.service.embed(['fallback']))
|
||||
assert result.source == 'local' and result.fallback_reason
|
||||
|
||||
|
||||
@pytest.mark.parametrize('api_failure', [False, True])
|
||||
def test_local_embedding_identity_and_device_are_frozen_during_inference(rig, monkeypatch, api_failure):
|
||||
import app.local_models.runtime as module
|
||||
config = module.RuntimeConfig(embedding_model='bekko')
|
||||
monkeypatch.setattr(module, 'configuration', lambda: module.runtime_context.get() or config)
|
||||
calls = []
|
||||
async def infer(key, *args, **kwargs):
|
||||
calls.append(key)
|
||||
config.embedding_model = 'granite'
|
||||
config.device = 'cuda'
|
||||
await asyncio.sleep(0)
|
||||
assert module.configuration().embedding_model == key
|
||||
assert module.configuration().device == ('cpu' if len(calls) == 1 else 'cuda')
|
||||
return [[1.0] + [0.0] * 383]
|
||||
monkeypatch.setattr(module.runtime, 'infer', infer)
|
||||
rig.service.local_embedding = module.LocalEmbedding()
|
||||
if api_failure:
|
||||
bind(rig)
|
||||
rig.http.handler = lambda request: response({}, status=503)
|
||||
first = run(rig.service.embed(['first']))
|
||||
assert 'bekko' in first.model_id
|
||||
assert module.runtime_context.get() is None
|
||||
second = run(rig.service.embed(['second']))
|
||||
assert 'granite' in second.model_id
|
||||
assert calls == ['bekko', 'granite']
|
||||
assert bool(first.fallback_reason) == api_failure
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
"""Finalization regressions: device recovery, durable facts and guarded writes."""
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
from contextlib import closing
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.errors import ApiError
|
||||
from app.providers.base import ProviderError
|
||||
|
||||
|
||||
@pytest.mark.parametrize('code,retries', [('LOCAL_CUDA_OOM', True), ('LOCAL_CUDA_INIT_FAILED', True),
|
||||
('LOCAL_INFERENCE_FAILED', False), ('LOCAL_RUNTIME_DEPENDENCY_MISSING', False)])
|
||||
def test_cuda_retries_only_device_failures_in_reaped_process(monkeypatch, code, retries):
|
||||
import app.local_models.runtime as module
|
||||
from app.services import model_diagnostics
|
||||
from app.services.usage_service import connection
|
||||
monkeypatch.setattr(module, 'configuration', lambda: module.RuntimeConfig(device='cuda'))
|
||||
monkeypatch.setattr(module, 'read_state', lambda key: {'status': 'installed'})
|
||||
monkeypatch.setattr(module, 'interpreter', lambda *_: Path(sys.executable))
|
||||
events = []
|
||||
|
||||
class Process:
|
||||
def __init__(self):
|
||||
from types import SimpleNamespace
|
||||
self.stdin = SimpleNamespace(write=self.write, drain=self.drain, close=lambda: None)
|
||||
self.stdout = asyncio.StreamReader()
|
||||
self.returncode = None
|
||||
self.device = None
|
||||
def write(self, raw):
|
||||
self.device = json.loads(raw)['config']['device']
|
||||
events.append('start-' + self.device)
|
||||
result = {'error_code': code} if self.device == 'cuda' else {'result': [[1, 0]], 'usage': {'input_tokens': 2}, 'diagnostics': {'actual_device': 'cpu'}}
|
||||
self.stdout.feed_data((json.dumps(result) + '\n').encode())
|
||||
self.stdout.feed_eof()
|
||||
async def drain(self):
|
||||
pass
|
||||
async def close(self):
|
||||
pass
|
||||
async def wait(self):
|
||||
self.returncode = 0
|
||||
events.append('reaped-' + self.device)
|
||||
def kill(self):
|
||||
self.returncode = -9
|
||||
|
||||
async def spawn(*args, **kwargs):
|
||||
if events:
|
||||
assert events[-1] == 'reaped-cuda'
|
||||
return Process()
|
||||
monkeypatch.setattr(module.asyncio, 'create_subprocess_exec', spawn)
|
||||
|
||||
async def scenario():
|
||||
runtime = module.Runtime()
|
||||
if retries:
|
||||
assert await runtime.infer('bekko', 'embedding', {'texts': ['private text']}) == [[1, 0]]
|
||||
else:
|
||||
with pytest.raises(ProviderError) as error:
|
||||
await runtime.infer('bekko', 'embedding', {'texts': ['private text']})
|
||||
assert error.value.code == code
|
||||
assert not runtime.active and not runtime.waiters
|
||||
asyncio.run(scenario())
|
||||
assert events == (['start-cuda', 'reaped-cuda', 'start-cpu', 'reaped-cpu'] if retries else ['start-cuda', 'reaped-cuda'])
|
||||
records = model_diagnostics.recent()
|
||||
assert records[0]['error_code'] == code
|
||||
assert 'private text' not in json.dumps(records)
|
||||
if retries:
|
||||
assert records[-1]['requested_device'] == 'cuda' and records[-1]['actual_device'] == 'cpu'
|
||||
assert records[-1]['fallback_reason'] == code
|
||||
assert records[0]['request_id'] == records[1]['request_id']
|
||||
assert records[0]['attempt_id'] != records[1]['attempt_id']
|
||||
with closing(connection()) as conn:
|
||||
assert conn.execute('SELECT COUNT(*) FROM model_usage').fetchone()[0] == (2 if retries else 1)
|
||||
|
||||
|
||||
def test_cpu_failure_does_not_loop_and_interactive_precedes_index(monkeypatch):
|
||||
import app.local_models.runtime as module
|
||||
async def scenario():
|
||||
runtime = module.Runtime()
|
||||
entered, release = asyncio.Event(), asyncio.Event()
|
||||
order = []
|
||||
async def execute(key, operation, payload, config, diagnostics):
|
||||
order.append(payload['name'])
|
||||
if payload['name'] == 'running':
|
||||
entered.set()
|
||||
await release.wait()
|
||||
return {'result': []}
|
||||
monkeypatch.setattr(runtime, '_execute', execute)
|
||||
first = asyncio.create_task(runtime.infer('bekko', 'embedding', {'name': 'running'}))
|
||||
await entered.wait()
|
||||
background = asyncio.create_task(runtime.infer('bekko', 'embedding', {'name': 'index'}, priority=20))
|
||||
query = asyncio.create_task(runtime.infer('bekko', 'embedding', {'name': 'query'}, priority=0))
|
||||
await asyncio.sleep(0)
|
||||
release.set()
|
||||
await asyncio.gather(first, background, query)
|
||||
assert order == ['running', 'query', 'index']
|
||||
calls = []
|
||||
async def failed(key, operation, payload, config, diagnostics):
|
||||
calls.append(config.device)
|
||||
raise ProviderError('LOCAL_CUDA_OOM', 'simulated')
|
||||
monkeypatch.setattr(runtime, '_execute', failed)
|
||||
monkeypatch.setattr(module, 'configuration', lambda: module.RuntimeConfig(device='cuda'))
|
||||
with pytest.raises(ProviderError):
|
||||
await runtime.infer('bekko', 'embedding', {})
|
||||
assert calls == ['cuda', 'cpu'] and not runtime.active
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_durable_diagnostics_are_bounded_and_disk_size_is_real():
|
||||
from app.services import model_diagnostics
|
||||
from app.local_models import manager
|
||||
for index in range(205):
|
||||
model_diagnostics.record(model='bekko', status='failed', error_code='TEST', payload='secret', elapsed_seconds=index)
|
||||
records = model_diagnostics.recent()
|
||||
assert len(records) == 200 and records[0]['elapsed_seconds'] == 5
|
||||
assert 'secret' not in json.dumps(records)
|
||||
path = manager.model_path('bekko')
|
||||
path.mkdir(parents=True)
|
||||
(path / 'weights.partial').write_bytes(b'1234567')
|
||||
assert manager.disk_bytes('bekko') == 7
|
||||
|
||||
|
||||
def test_upload_key_replay_and_content_conflict():
|
||||
from app.main import app
|
||||
with TestClient(app) as client:
|
||||
headers = {'Idempotency-Key': 'stable-upload-123456'}
|
||||
first = client.post('/api/media/attachments?filename=lecture.txt', content=b'original', headers=headers)
|
||||
again = client.post('/api/media/attachments?filename=lecture.txt', content=b'original', headers=headers)
|
||||
assert first.status_code == again.status_code == 201
|
||||
assert first.json()['attachment_id'] == again.json()['attachment_id']
|
||||
assert client.post('/api/media/attachments?filename=lecture.txt', content=b'changed', headers=headers).status_code == 409
|
||||
changed_name = client.post('/api/media/attachments?filename=lecture.md', content=b'original', headers=headers)
|
||||
assert changed_name.status_code == 409 and changed_name.json()['error']['code'] == 'IDEMPOTENCY_CONFLICT'
|
||||
assert client.get('/api/media/attachments/' + first.json()['attachment_id']).content == b'original'
|
||||
|
||||
|
||||
def test_updated_transcript_note_keeps_identity_and_rejects_user_edits():
|
||||
from app.contracts import TranscriptNoteRequest, TranscriptEditRequest, IndexRebuildRequest
|
||||
from app.services import transcription_service as jobs, note_service, index_service
|
||||
from app.services.media_notes import create_transcript_note
|
||||
from app.services.attachment_service import attachment_path
|
||||
path = attachment_path('lecture.txt')
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text('original', encoding='utf-8')
|
||||
async def scenario():
|
||||
job = await jobs.create_transcription('lecture.txt', local_only=True)
|
||||
options = TranscriptNoteRequest(title='Lecture')
|
||||
first = await create_transcript_note(job.job_id, options)
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
jobs.edit(job.job_id, TranscriptEditRequest(revision=1, text='revised'))
|
||||
update = options.model_copy(update={'update_existing': True})
|
||||
second = await create_transcript_note(job.job_id, update)
|
||||
assert first.note_id == second.note_id and 'revised' in second.markdown
|
||||
assert 'embedding_local_only: true' in second.markdown
|
||||
again = await create_transcript_note(job.job_id, update)
|
||||
assert again.note_id == first.note_id
|
||||
await note_service.update_note(first.note_id, markdown='User edits')
|
||||
jobs.edit(job.job_id, TranscriptEditRequest(revision=2, text='third revision'))
|
||||
with pytest.raises(ApiError) as error:
|
||||
await create_transcript_note(job.job_id, update)
|
||||
assert error.value.code == 'NOTE_CONTENT_CONFLICT'
|
||||
assert (await note_service.get_note(first.note_id)).markdown == 'User edits'
|
||||
copy = await create_transcript_note(job.job_id, options)
|
||||
assert copy.note_id != first.note_id
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_audio_usage_is_separate_and_unknown_durations_stay_null():
|
||||
from app.services.usage_service import UsageAttempt, aggregate
|
||||
now = datetime.now(timezone.utc)
|
||||
first = UsageAttempt('local', 'asr', 'local', 'transcription', source='local')
|
||||
first.observe({'audio_seconds': 2.25, 'usage': {}})
|
||||
first.persist(); first.persist()
|
||||
unknown = UsageAttempt('remote', 'asr', 'openai_compatible', 'transcription')
|
||||
unknown.persist()
|
||||
result = aggregate(now - timedelta(days=1), now + timedelta(days=1))
|
||||
assert result['audio_request_count'] == 2 and result['audio_covered_requests'] == 1
|
||||
assert result['audio_seconds'] == 2.25 and result['totals']['input_tokens'] is None
|
||||
remote = aggregate(now - timedelta(days=1), now + timedelta(days=1), source='api')
|
||||
assert remote['audio_seconds'] is None
|
||||
|
||||
|
||||
def test_request_rule_import_rejects_credentials_and_host_fields():
|
||||
from app.main import app
|
||||
with TestClient(app) as client:
|
||||
path = '/api/providers/request-rules/validate'
|
||||
body = {'version': 1, 'request_overrides': [{'body': {'enable_thinking': False}}]}
|
||||
assert client.post(path, json=body).status_code == 200
|
||||
for bad in ({'api_key': 'secret'}, {'nested': {'authorization': 'secret'}}, {'stream': False}):
|
||||
body['request_overrides'][0]['body'] = bad
|
||||
assert client.post(path, json=body).status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.parametrize('stream', [False, True])
|
||||
def test_inference_probe_uses_adapter_body_and_no_vault_context(monkeypatch, stream):
|
||||
import httpx
|
||||
from app.container import container
|
||||
from app.main import app
|
||||
original = container.provider_factory.build
|
||||
requests = []
|
||||
def respond(request):
|
||||
data = json.loads(request.content)
|
||||
requests.append(data)
|
||||
assert data['enable_thinking'] is False and data['stream'] == stream
|
||||
assert data['messages'] == [{'role': 'user', 'content': 'Reply with OK.'}]
|
||||
assert not data.get('tools')
|
||||
if stream:
|
||||
return httpx.Response(200, text='data: {"choices":[{"delta":{"content":"OK"},"finish_reason":null}]}\n\ndata: [DONE]\n\n')
|
||||
return httpx.Response(200, json={'choices': [{'message': {'role': 'assistant', 'content': 'OK'}, 'finish_reason': 'stop'}]})
|
||||
def build(config):
|
||||
adapter = original(config)
|
||||
adapter.transport = httpx.MockTransport(respond)
|
||||
return adapter
|
||||
monkeypatch.setattr(container.provider_factory, 'build', build)
|
||||
with TestClient(app) as client:
|
||||
response = client.post('/api/providers/request-probe', json={'stream': stream, 'provider': {
|
||||
'name': 'Probe', 'provider_type': 'openai_compatible', 'base_url': 'https://fixture.invalid/v1',
|
||||
'default_model': 'test', 'request_overrides': [{'body': {'enable_thinking': False}}]}})
|
||||
assert response.status_code == 200, response.text
|
||||
assert len(requests) == 1
|
||||
@@ -0,0 +1,46 @@
|
||||
import asyncio
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from app.contracts import IndexRebuildRequest
|
||||
from app.knowledge.parser import parse_note
|
||||
from app.services import index_service, note_service
|
||||
|
||||
|
||||
@pytest.mark.parametrize(('header', 'expected'), [
|
||||
('tags:\n- python\n- rust', ['python', 'rust']),
|
||||
('tags:\n - python\n - rust', ['python', 'rust']),
|
||||
('"tags": ["a,b", "quote\\\"tag", "path\\\\tag"] # comment', ['a,b', 'quote"tag', 'path\\tag']),
|
||||
('tags: [on, yes, "true", "001"]', ['on', 'yes', 'true', '001']),
|
||||
('tags: python, rust', ['python', 'rust']),
|
||||
('tags: []', []),
|
||||
('tags: null', []),
|
||||
])
|
||||
def test_yaml_tags_are_parsed_as_complete_values(header, expected):
|
||||
now = datetime.now(timezone.utc)
|
||||
note = parse_note(
|
||||
markdown=f'---\ntitle: "Demo: YAML"\n{header}\n---\n# Body',
|
||||
file_path='demo.md', folder='', created_at=now, updated_at=now,
|
||||
)
|
||||
assert note.tags == expected
|
||||
assert note.title == 'Demo: YAML'
|
||||
|
||||
|
||||
def test_saved_metadata_survives_full_index_rebuild():
|
||||
async def scenario():
|
||||
note = await note_service.create_note(title='Demo', markdown='# Body', folder=None, tags=['old'])
|
||||
for tags, yaml_tags in [
|
||||
(['python', 'a,b', 'on'], '\n - python\n - a,b\n - on'),
|
||||
([], ' []'),
|
||||
]:
|
||||
markdown = f'---\ntitle: "Demo: updated"\ntags:{yaml_tags}\n---\n# Body\n'
|
||||
saved = await note_service.update_note(note.note_id, markdown=markdown, tags=tags)
|
||||
assert saved.tags == tags
|
||||
job = await index_service.rebuild(IndexRebuildRequest())
|
||||
assert job.status == 'completed'
|
||||
restored = await note_service.get_note(note.note_id)
|
||||
assert restored.tags == tags
|
||||
assert restored.title == 'Demo: updated'
|
||||
assert restored.markdown == markdown
|
||||
asyncio.run(scenario())
|
||||
@@ -0,0 +1,205 @@
|
||||
import sqlite3
|
||||
from datetime import datetime, timezone
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
import pytest
|
||||
|
||||
from app.database import migrations
|
||||
from app.database.db import _load_extension
|
||||
from app.errors import ApiError
|
||||
from app.knowledge.parser import parse_note
|
||||
|
||||
|
||||
def parsed(value):
|
||||
return parse_note(markdown='---\nembedding_local_only: '+value+'\n---\nbody', file_path='note.md', folder='',
|
||||
created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc))
|
||||
|
||||
|
||||
@pytest.mark.parametrize('value,expected', [('true', True), ('true # keep local', True), ('TRUE # comment', True), ('false # explicit', False)])
|
||||
def test_policy_parses_yaml_boolean_with_comments(value, expected):
|
||||
assert parsed(value).embedding_local_only is expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize('value', ['truth', '1', '', 'null', '"true"', '[true]', '{broken', 'true\nembedding_local_only: false'])
|
||||
def test_invalid_policy_never_silently_enables_remote(value):
|
||||
with pytest.raises(ApiError) as error:
|
||||
parsed(value)
|
||||
assert error.value.code == 'INVALID_EMBEDDING_POLICY'
|
||||
|
||||
|
||||
def connection(path, factory=sqlite3.Connection):
|
||||
conn = sqlite3.connect(path, isolation_level=None, factory=factory)
|
||||
conn.row_factory = sqlite3.Row
|
||||
_load_extension(conn)
|
||||
return conn
|
||||
|
||||
|
||||
def seed_v5(path, monkeypatch):
|
||||
conn = connection(path)
|
||||
with monkeypatch.context() as patch:
|
||||
patch.setattr(migrations, 'MIGRATIONS', migrations.MIGRATIONS[:5])
|
||||
migrations.migrate(conn)
|
||||
conn.execute("INSERT INTO search_history(query) VALUES ('retained')")
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('failure', [sqlite3.OperationalError, KeyboardInterrupt])
|
||||
def test_migration_and_version_write_rollback_together(tmp_path, monkeypatch, failure):
|
||||
path = tmp_path / 'migration.db'
|
||||
seed_v5(path, monkeypatch)
|
||||
class Interrupted(sqlite3.Connection):
|
||||
def execute(self, sql, parameters=()):
|
||||
if sql.startswith('INSERT INTO schema_migrations') and parameters[0] == 6:
|
||||
raise failure('interrupted')
|
||||
return super().execute(sql, parameters)
|
||||
conn = connection(path, Interrupted)
|
||||
try:
|
||||
with pytest.raises(failure):
|
||||
migrations.migrate(conn)
|
||||
assert not conn.in_transaction
|
||||
assert not any(r['name'] == 'embedding_local_only' for r in conn.execute('pragma table_info(blocks)'))
|
||||
finally:
|
||||
conn.close()
|
||||
conn = connection(path)
|
||||
try:
|
||||
migrations.migrate(conn)
|
||||
assert conn.execute('select count(*) from schema_migrations where version=6').fetchone()[0] == 1
|
||||
assert conn.execute('select query from search_history').fetchone()[0] == 'retained'
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_old_partial_v6_recovers_without_duplicate_column(tmp_path, monkeypatch):
|
||||
path = tmp_path / 'partial.db'
|
||||
seed_v5(path, monkeypatch)
|
||||
conn = connection(path)
|
||||
try:
|
||||
conn.executescript(migrations.MIGRATIONS[5])
|
||||
migrations.migrate(conn)
|
||||
migrations.migrate(conn)
|
||||
assert conn.execute('select count(*) from schema_migrations where version=6').fetchone()[0] == 1
|
||||
assert conn.execute('select query from search_history').fetchone()[0] == 'retained'
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_concurrent_connections_can_upgrade(tmp_path, monkeypatch):
|
||||
path = tmp_path / 'concurrent.db'
|
||||
seed_v5(path, monkeypatch)
|
||||
def upgrade(_):
|
||||
conn = connection(path)
|
||||
try:
|
||||
migrations.migrate(conn)
|
||||
return conn.execute('select count(*) from schema_migrations where version=6').fetchone()[0]
|
||||
finally:
|
||||
conn.close()
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
assert list(pool.map(upgrade, range(2))) == [1, 1]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('header', ['"embedding_local_only": true # comment', ' embedding_local_only: true', 'embedding_local_only:\n true', 'local: &local true\nembedding_local_only: *local'])
|
||||
def test_policy_supports_yaml_key_and_scalar_forms(header):
|
||||
note = parse_note(markdown='---\n'+header+'\n---\nbody',file_path='note.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
|
||||
assert note.embedding_local_only
|
||||
|
||||
|
||||
def test_merge_policy_is_rejected_instead_of_ignored():
|
||||
with pytest.raises(ApiError):
|
||||
parsed('true\n<<: {embedding_local_only: false}')
|
||||
with pytest.raises(ApiError):
|
||||
parsed('!!bool invalid')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('bom', ['', '\ufeff'])
|
||||
@pytest.mark.parametrize('newline', ['\n', '\r\n', '\r'])
|
||||
@pytest.mark.parametrize('closing', ['---', '...'])
|
||||
def test_frontmatter_boundaries_preserve_policy_and_utf16_offsets(bom, newline, closing):
|
||||
markdown = bom + newline.join(['--- ', 'title: Sample', 'embedding_local_only: true # local', closing+' ', '# Heading', '', 'private \U0001f600'])
|
||||
note = parse_note(markdown=markdown, file_path='note.md', folder='', created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc))
|
||||
assert note.embedding_local_only and note.title == 'Sample'
|
||||
assert all('embedding_local_only' not in block.content for block in note.blocks)
|
||||
block = next(block for block in note.blocks if block.content == 'private \U0001f600')
|
||||
original = markdown.encode('utf-16-le')[block.start_offset*2:block.end_offset*2].decode('utf-16-le')
|
||||
assert original == block.content
|
||||
|
||||
|
||||
@pytest.mark.parametrize('ending', ['', '\n---not-a-delimiter', '\n----'])
|
||||
def test_unclosed_frontmatter_is_rejected_even_with_bom(ending):
|
||||
for bom in ['', '\ufeff']:
|
||||
markdown = bom+'---\nembedding_local_only: true'+ending
|
||||
with pytest.raises(ApiError) as error:
|
||||
parse_note(markdown=markdown, file_path='note.md', folder='', created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc))
|
||||
assert error.value.code == 'INVALID_EMBEDDING_POLICY'
|
||||
|
||||
|
||||
def test_boundary_matching_does_not_truncate_yaml_keys():
|
||||
markdown = '---\n---metadata: value\nembedding_local_only: true\n---\nbody'
|
||||
note = parse_note(markdown=markdown,file_path='note.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
|
||||
assert note.embedding_local_only
|
||||
|
||||
|
||||
def test_bom_save_and_invalid_update_never_use_remote(monkeypatch):
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
from app.local_models.runtime import LocalEmbedding
|
||||
from app.retrieval import routed_vectors
|
||||
from app.services import note_service, index_service
|
||||
from app.contracts import IndexRebuildRequest
|
||||
from app.config import get_settings
|
||||
calls=[]
|
||||
class Routing:
|
||||
async def embed(self, texts, *, local_only=False):
|
||||
calls.append(local_only)
|
||||
assert local_only
|
||||
return SimpleNamespace(source='local', model_id='local-test', dimensions=2, vectors=[[1.0,0.0] for _ in texts], fallback_reason=None)
|
||||
monkeypatch.setattr(routed_vectors, 'get_model_routing', lambda: Routing())
|
||||
monkeypatch.setattr(note_service, 'embedding', LocalEmbedding())
|
||||
async def scenario():
|
||||
markdown='\ufeff---\nembedding_local_only: true\n---\nprivate text'
|
||||
note=await note_service.create_note(title='Private',markdown=markdown,folder=None,tags=[])
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
count=len(calls)
|
||||
with pytest.raises(ApiError):
|
||||
await note_service.update_note(note.note_id,markdown='\ufeff---\nembedding_local_only: true\nprivate text')
|
||||
assert len(calls)==count
|
||||
assert (get_settings().vault_path/note.file_path).read_text(encoding='utf-8')==markdown
|
||||
assert (await note_service.get_note(note.note_id)).markdown==markdown
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.parametrize('markdown', ['---', '---\n\n# Title\n\nNormal body', '---\n\nNormal body\n\n---\n\nLast paragraph', '---\n\n```python\nprint(1)\n```\n---'])
|
||||
def test_thematic_breaks_are_not_frontmatter(markdown):
|
||||
note = parse_note(markdown=markdown,file_path='ordinary.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
|
||||
assert not note.embedding_local_only
|
||||
assert note.blocks[0].content == '---'
|
||||
assert any(block.content == markdown.split('\n\n')[-1] for block in note.blocks) or '```' in markdown
|
||||
|
||||
|
||||
@pytest.mark.parametrize('header', ['title: Sample\nembedding_local_only: true', '"embedding_local_only": true', 'title: [broken\nembedding_local_only: true', '{embedding_local_only: true'])
|
||||
def test_unclosed_metadata_still_fails_closed(header):
|
||||
with pytest.raises(ApiError) as error:
|
||||
parse_note(markdown='---\n'+header,file_path='private.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
|
||||
assert error.value.code == 'INVALID_EMBEDDING_POLICY'
|
||||
|
||||
|
||||
def test_thematic_break_note_can_save_and_rebuild():
|
||||
import asyncio
|
||||
from app.services import note_service, index_service
|
||||
from app.contracts import IndexRebuildRequest
|
||||
async def scenario():
|
||||
markdown='---\n\n# Title\n\nNormal body'
|
||||
note=await note_service.create_note(title='Divider',markdown=markdown,folder=None,tags=[])
|
||||
assert note.blocks[0].content == '---'
|
||||
assert (await index_service.rebuild(IndexRebuildRequest())).status == 'completed'
|
||||
loaded=await note_service.get_note(note.note_id)
|
||||
assert loaded.markdown == markdown
|
||||
assert [b.content for b in loaded.blocks] == [b.content for b in note.blocks]
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_thematic_break_with_policy_example_is_ordinary_markdown():
|
||||
markdown='---\n\n```yaml\nembedding_local_only: true\n```\n\n---\n\nExplanation'
|
||||
note=parse_note(markdown=markdown,file_path='example.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
|
||||
assert not note.embedding_local_only
|
||||
assert any('embedding_local_only: true' in block.content for block in note.blocks)
|
||||
assert note.blocks[0].content=='---'
|
||||
@@ -464,3 +464,158 @@ def test_missing_runtime_uses_unchanged_local_retrieval(runtime, monkeypatch):
|
||||
assert runtime.calls == []
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def production_engine(monkeypatch):
|
||||
from app.local_models.runtime import LocalEmbedding
|
||||
embedding = LocalEmbedding()
|
||||
monkeypatch.setattr(note_service, "embedding", embedding)
|
||||
return RetrievalEngine(embedding, LexicalReranker(), SqliteVecStore(), route_embeddings=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", ["api", "local"])
|
||||
def test_real_embedding_route_rebuilds_missing_space(runtime, production_engine, source):
|
||||
from app.errors import ApiError
|
||||
runtime.source = source
|
||||
|
||||
async def scenario():
|
||||
await seed()
|
||||
runtime.model_id = "new-configured-space"
|
||||
with pytest.raises(ApiError) as error:
|
||||
await production_engine.search(request())
|
||||
assert error.value.code == "SEMANTIC_INDEX_UNAVAILABLE"
|
||||
assert "Embedding 已可用" in error.value.message
|
||||
assert error.value.details["source"] == source
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
assert (await production_engine.search(request())).items
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_real_embedding_failure_is_not_reported_as_missing_configuration(runtime, production_engine):
|
||||
from app.errors import ApiError
|
||||
|
||||
async def scenario():
|
||||
await seed()
|
||||
runtime.error = ApiError(503, "LOCAL_MODEL_TIMEOUT", "本地模型推理超时。", {"fallback_reason": "PROVIDER_TIMEOUT"})
|
||||
with pytest.raises(ApiError) as error:
|
||||
await production_engine.search(request())
|
||||
assert error.value.code == "LOCAL_MODEL_TIMEOUT"
|
||||
assert error.value.details["fallback_reason"] == "PROVIDER_TIMEOUT"
|
||||
assert (await production_engine.search(SearchRequest(query="apple", mode=SearchMode.hybrid))).items
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", ["inference", "storage", "space_change"])
|
||||
def test_real_embedding_rebuild_failure_preserves_index(runtime, production_engine, monkeypatch, failure):
|
||||
from app.errors import ApiError
|
||||
|
||||
async def scenario():
|
||||
await seed()
|
||||
tables = ("notes", "blocks", "blocks_fts", "index_meta", "routed_block_vectors")
|
||||
before = {table: [tuple(r) for r in rows(f"SELECT * FROM {table}")] for table in tables}
|
||||
if failure == "inference":
|
||||
runtime.error = ApiError(503, "LOCAL_MODEL_TIMEOUT", "本地模型推理超时。")
|
||||
elif failure == "storage":
|
||||
monkeypatch.setattr(routed_vectors, "store_remote", lambda *args: None)
|
||||
else:
|
||||
original = runtime.embed
|
||||
async def changing(texts):
|
||||
runtime.model_id += "x"
|
||||
return await original(texts)
|
||||
monkeypatch.setattr(runtime, "embed", changing)
|
||||
with pytest.raises(ApiError):
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
assert index_service.get_status().status == "failed"
|
||||
after = {table: [tuple(r) for r in rows(f"SELECT * FROM {table}")] for table in tables}
|
||||
assert before == after
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_empty_vault_vector_search_returns_empty(runtime, production_engine):
|
||||
assert asyncio.run(production_engine.search(request())).items == []
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def policy_runtime(monkeypatch):
|
||||
class PolicyRuntime:
|
||||
fallback = False
|
||||
calls = []
|
||||
async def embed(self, texts, *, local_only=False):
|
||||
self.calls.append((list(texts), local_only))
|
||||
local = local_only or self.fallback
|
||||
dim = 3 if local else 2
|
||||
return SimpleNamespace(source='local' if local else 'api', model_id='local-space' if local else 'api-space',
|
||||
dimensions=dim, vectors=[[1.0] + [0.0] * (dim - 1) for _ in texts],
|
||||
fallback_reason='PROVIDER_TIMEOUT' if self.fallback and not local_only else None)
|
||||
runtime = PolicyRuntime()
|
||||
monkeypatch.setattr(routed_vectors, 'get_model_routing', lambda: runtime)
|
||||
return runtime
|
||||
|
||||
|
||||
async def seed_policies():
|
||||
normal = await note_service.create_note(title='Normal', markdown='apple public', folder=None, tags=[])
|
||||
private = await note_service.create_note(title='Private', markdown='---\nembedding_local_only: true\n---\napple private', folder=None, tags=[])
|
||||
return normal, private
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fallback', [False, True])
|
||||
def test_mixed_policy_rebuild_and_retrieval(policy_runtime, production_engine, fallback):
|
||||
policy_runtime.fallback = fallback
|
||||
async def scenario():
|
||||
notes = await seed_policies()
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
for mode in (SearchMode.vector, SearchMode.hybrid):
|
||||
result = await production_engine.search(SearchRequest(query='apple', mode=mode))
|
||||
assert {item.note_id for item in result.items} == {note.note_id for note in notes}
|
||||
for texts, local_only in policy_runtime.calls:
|
||||
if any('private' in text for text in texts):
|
||||
assert local_only
|
||||
if not fallback:
|
||||
assert {r[0] for r in rows('SELECT DISTINCT space_id FROM routed_block_vectors')} == {'api-space', 'local-space'}
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_local_only_vault_never_requests_api_for_search(policy_runtime, production_engine):
|
||||
async def scenario():
|
||||
await note_service.create_note(title='Private', markdown='---\nembedding_local_only: true\n---\napple private', folder=None, tags=[])
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
assert (await production_engine.search(request())).items
|
||||
assert all(local_only for _, local_only in policy_runtime.calls)
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_partition_storage_failure_rolls_back_all_partitions(policy_runtime, production_engine, monkeypatch):
|
||||
from app.errors import ApiError
|
||||
async def scenario():
|
||||
await seed_policies()
|
||||
before = [tuple(row) for row in rows('SELECT * FROM routed_block_vectors ORDER BY block_id')]
|
||||
original = routed_vectors.store_remote
|
||||
def fail_local(conn, ids, batch):
|
||||
if batch.source != 'local':
|
||||
original(conn, ids, batch)
|
||||
monkeypatch.setattr(routed_vectors, 'store_remote', fail_local)
|
||||
with pytest.raises(ApiError) as error:
|
||||
await index_service.rebuild(IndexRebuildRequest())
|
||||
assert error.value.code == 'SEMANTIC_INDEX_WRITE_FAILED'
|
||||
assert [tuple(row) for row in rows('SELECT * FROM routed_block_vectors ORDER BY block_id')] == before
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_missing_partition_does_not_silently_return_partial_hits(policy_runtime, production_engine):
|
||||
from app.errors import ApiError
|
||||
async def scenario():
|
||||
await seed_policies()
|
||||
conn = connect()
|
||||
try:
|
||||
conn.execute("DELETE FROM routed_block_vectors WHERE space_id='local-space'")
|
||||
finally:
|
||||
conn.close()
|
||||
with pytest.raises(ApiError) as error:
|
||||
await production_engine.search(request())
|
||||
assert error.value.code == 'SEMANTIC_INDEX_UNAVAILABLE'
|
||||
assert (await production_engine.search(request(SearchMode.hybrid))).items
|
||||
asyncio.run(scenario())
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from app.errors import ApiError
|
||||
from app.local_models import components, runtime
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolate(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(components, 'ROOT', tmp_path / 'cuda')
|
||||
monkeypatch.setattr(components, 'state', {'status': 'unchecked', 'stage': '', 'cuda_available': None})
|
||||
monkeypatch.setattr(components, 'task', None)
|
||||
|
||||
|
||||
def test_status_checks_without_installing_and_detects_existing_cuda(monkeypatch):
|
||||
python = components.ROOT / 'Scripts/python.exe'
|
||||
python.parent.mkdir(parents=True)
|
||||
python.touch()
|
||||
calls = []
|
||||
async def execute(args, timeout):
|
||||
calls.append(args)
|
||||
return [json.dumps({'torch': '2.9.1+cu128', 'cuda_available': True})]
|
||||
monkeypatch.setattr(components, 'execute', execute)
|
||||
async def scenario():
|
||||
assert (await components.status())['status'] == 'checking'
|
||||
await components.task
|
||||
assert (await components.status())['status'] == 'installed'
|
||||
assert len(calls) == 1 and calls[0][0] == str(python)
|
||||
assert components.ready()
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name != 'nt', reason='Windows installer')
|
||||
def test_install_deduplicates_and_failure_can_retry(monkeypatch):
|
||||
monkeypatch.setattr(components.shutil, 'which', lambda name: 'uv.exe')
|
||||
async def scenario():
|
||||
entered, release = asyncio.Event(), asyncio.Event()
|
||||
calls = []
|
||||
async def execute(args, timeout):
|
||||
calls.append(args)
|
||||
entered.set()
|
||||
await release.wait()
|
||||
raise RuntimeError('private exception')
|
||||
monkeypatch.setattr(components, 'execute', execute)
|
||||
await components.install()
|
||||
await entered.wait()
|
||||
first = components.task
|
||||
await components.install()
|
||||
assert first is components.task
|
||||
release.set()
|
||||
await first
|
||||
assert components.state['status'] == 'failed'
|
||||
assert 'private exception' not in str(components.state)
|
||||
await components.install()
|
||||
await components.task
|
||||
assert len(calls) == 2 and '-RuntimeDirectory' in calls[0]
|
||||
assert not components.ready()
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name != 'nt', reason='Windows installer')
|
||||
def test_install_refuses_active_inference(monkeypatch):
|
||||
monkeypatch.setattr(runtime.runtime, 'active', {1: 'bekko'})
|
||||
async def scenario():
|
||||
with pytest.raises(ApiError) as exc:
|
||||
await components.install()
|
||||
assert exc.value.code == 'MODEL_IN_USE'
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_interpreter_keeps_cpu_default_and_respects_explicit_override(monkeypatch):
|
||||
monkeypatch.delenv('APP_MODEL_PYTHON', raising=False)
|
||||
python = components.ROOT / 'Scripts/python.exe'
|
||||
python.parent.mkdir(parents=True)
|
||||
python.touch()
|
||||
(components.ROOT / 'ready.json').write_text('{}')
|
||||
monkeypatch.setattr(runtime, 'configuration', lambda: runtime.RuntimeConfig(device='cpu'))
|
||||
assert runtime.interpreter() != python
|
||||
# A queued attempt keeps its frozen device even after the saved setting changes.
|
||||
assert runtime.interpreter(runtime.RuntimeConfig(device='cuda')) == python
|
||||
assert runtime.interpreter(runtime.RuntimeConfig(device='cpu')) != python
|
||||
monkeypatch.setenv('APP_MODEL_PYTHON', 'explicit-python.exe')
|
||||
assert str(runtime.interpreter()) == 'explicit-python.exe'
|
||||
@@ -0,0 +1,22 @@
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
from app.services import search_history
|
||||
|
||||
|
||||
def test_history_survives_new_clients_and_clear():
|
||||
with TestClient(app) as client:
|
||||
for query in ['first', 'second', ' first ']:
|
||||
assert client.post('/api/search', json={'query': query, 'mode': 'fts'}).status_code == 200
|
||||
assert client.get('/api/search/history').json() == {'queries': ['first', 'second']}
|
||||
with TestClient(app) as client:
|
||||
assert client.get('/api/search/history').json() == {'queries': ['first', 'second']}
|
||||
assert client.delete('/api/search/history').json() == {'queries': []}
|
||||
assert search_history.list_queries() == []
|
||||
|
||||
|
||||
def test_history_is_bounded_and_blank_queries_are_ignored():
|
||||
for number in range(12):
|
||||
search_history.record(str(number))
|
||||
search_history.record(' ')
|
||||
assert search_history.list_queries() == [str(number) for number in range(11, 1, -1)]
|
||||
@@ -0,0 +1,94 @@
|
||||
import asyncio
|
||||
import json
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from contextlib import closing
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.contracts import ModelRequest, ProviderConfig, ProviderType
|
||||
from app.providers.factory import ProviderFactory
|
||||
from app.request_overrides import RequestOverride, apply_overrides
|
||||
from app.services.usage_service import UsageAttempt, aggregate, connection
|
||||
|
||||
|
||||
def summary():
|
||||
now = datetime.now(timezone.utc)
|
||||
return aggregate(now - timedelta(days=1), now + timedelta(days=1))
|
||||
|
||||
|
||||
def test_cumulative_usage_deduplicates_and_missing_is_not_zero():
|
||||
attempt = UsageAttempt("test", "chat", "openai_compatible")
|
||||
attempt.observe({"usage": {"prompt_tokens": 100, "completion_tokens": 2, "prompt_tokens_details": {"cached_tokens": 75}}})
|
||||
attempt.persist()
|
||||
attempt.observe({"usage": {"completion_tokens": 5}})
|
||||
attempt.observe({"usage": {"completion_tokens": 3}})
|
||||
attempt.persist()
|
||||
incomplete = UsageAttempt("test", "chat", "openai_compatible")
|
||||
incomplete.persist()
|
||||
result = summary()
|
||||
assert result["request_count"] == 2
|
||||
assert result["totals"]["input_tokens"] == 100
|
||||
assert result["totals"]["output_tokens"] == 5
|
||||
assert result["totals"]["cache_write_tokens"] is None
|
||||
assert result["cache_hit_rate"] == .75
|
||||
assert result["coverage"]["input_tokens"] == 1
|
||||
|
||||
|
||||
def test_anthropic_cache_is_added_once_and_raw_text_is_not_saved():
|
||||
attempt = UsageAttempt("test", "claude", "anthropic_messages")
|
||||
attempt.observe({"message": {"usage": {"input_tokens": 10, "cache_read_input_tokens": 80,
|
||||
"cache_creation_input_tokens": 20, "output_tokens": 0, "secret": "private text"}}})
|
||||
attempt.observe({"usage": {"output_tokens": 12}})
|
||||
attempt.persist()
|
||||
counts = summary()["totals"]
|
||||
assert counts["input_tokens"] == 110 and counts["total_tokens"] == 122
|
||||
assert counts["cache_miss_tokens"] == 10
|
||||
with closing(connection()) as conn:
|
||||
assert "private text" not in conn.execute("SELECT raw_json FROM model_usage").fetchone()[0]
|
||||
|
||||
|
||||
def test_override_rules_merge_and_respect_capability_and_stream():
|
||||
rules = [RequestOverride(body={"stream_options": {"include_usage": True, "extra": 1}, "stop": ["one"]}),
|
||||
RequestOverride(model="special", stream=True, body={"stream_options": {"extra": 2}, "stop": ["two"], "temperature": None}),
|
||||
RequestOverride(capability="embedding", body={"dimensions": 384})]
|
||||
base = {"model": "special", "messages": [], "stream": True}
|
||||
result = apply_overrides(base, rules, "chat", stream=True)
|
||||
assert result["stream_options"] == {"include_usage": True, "extra": 2}
|
||||
assert result["stop"] == ["two"] and result["temperature"] is None
|
||||
assert "dimensions" not in result and "stop" not in base
|
||||
assert apply_overrides(base, rules, "chat")["stop"] == ["one"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body", [{"model":"other"}, {"messages":[]}, {"tools":[]}, {"stream":False},
|
||||
{"metadata":{"api_key":"hidden"}}, {"stream_options":{"include_usage": "false"}}])
|
||||
def test_unsafe_or_invalid_overrides_are_rejected(body):
|
||||
with pytest.raises(ValidationError):
|
||||
RequestOverride(body=body)
|
||||
|
||||
|
||||
def test_real_adapter_body_and_usage_persistence():
|
||||
class Credentials:
|
||||
def resolve(self, key):
|
||||
return None
|
||||
config = ProviderConfig(provider_id="wire", provider_type=ProviderType.openai_compatible, name="Wire", base_url="https://model.invalid/v1",
|
||||
request_overrides=[RequestOverride(stream=True, body={"stream_options":{"include_usage":False},"enable_thinking":False})])
|
||||
adapter = ProviderFactory(Credentials()).build(config)
|
||||
captured = []
|
||||
def respond(request):
|
||||
captured.append(json.loads(request.content))
|
||||
return httpx.Response(200, headers={"content-type":"text/event-stream"}, content=(
|
||||
'data: {"choices":[{"delta":{"content":"ok"},"finish_reason":null}]}\n\n'
|
||||
'data: {"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":1}}\n\n'
|
||||
'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'
|
||||
'data: [DONE]\n\n'))
|
||||
adapter.transport = httpx.MockTransport(respond)
|
||||
async def consume():
|
||||
return [event async for event in adapter.stream(ModelRequest(provider_id="wire", model="special", messages=[]))]
|
||||
asyncio.run(consume())
|
||||
assert captured[0]["enable_thinking"] is False
|
||||
assert captured[0]["stream_options"]["include_usage"] is False
|
||||
result = summary()
|
||||
assert result["request_count"] == 1 and result["totals"]["input_tokens"] == 10
|
||||
assert result["complete_requests"] == 1
|
||||
Reference in New Issue
Block a user