93 lines
4.9 KiB
Python
93 lines
4.9 KiB
Python
import asyncio
|
|
import zipfile
|
|
from types import SimpleNamespace
|
|
import pytest
|
|
from app.services import chat_attachments as service
|
|
from app.contracts import ChatRequest, ModelCapability
|
|
|
|
@pytest.mark.parametrize('suffix,name,xml,expected', [
|
|
('.docx','word/document.xml','<document><p><t>Hello</t></p><p><t>World</t></p></document>','Hello\nWorld'),
|
|
('.pptx','ppt/slides/slide1.xml','<slide><p><t>Title</t></p></slide>','第 1 页\nTitle'),
|
|
])
|
|
def test_office_text_extraction(tmp_path,suffix,name,xml,expected):
|
|
path=tmp_path/('file'+suffix)
|
|
with zipfile.ZipFile(path,'w') as z: z.writestr(name,xml)
|
|
assert service.extract_document(path)==(expected,False)
|
|
|
|
def test_markdown_truncation_and_invalid_document(tmp_path):
|
|
path=tmp_path/'file.md';path.write_text('a'*200001,encoding='utf-8')
|
|
text,truncated=service.extract_document(path)
|
|
assert len(text)==200000 and truncated
|
|
path=tmp_path/'file.docx';path.write_bytes(b'invalid')
|
|
with pytest.raises(zipfile.BadZipFile): service.extract_document(path)
|
|
|
|
def test_native_vision_precedes_registered_fallback(tmp_path):
|
|
path=tmp_path/'image.png';path.write_bytes(b'\x89PNG\r\n\x1a\nimage')
|
|
seen=[]
|
|
class Adapter:
|
|
async def list_models(self): return []
|
|
async def complete(self,request):
|
|
seen.append(request)
|
|
return SimpleNamespace(text='image description')
|
|
provider=SimpleNamespace(config=SimpleNamespace(capabilities=[ModelCapability.vision]),adapter=Adapter())
|
|
request=ChatRequest(provider_id='mock',model='mock',messages=[])
|
|
result=asyncio.run(service.describe_image(path,request,provider))
|
|
assert result[1]=='native' and seen[0].messages[0].images[0].startswith('data:image/png;base64,')
|
|
|
|
def test_fallback_order_is_mcp_then_plugin(tmp_path,monkeypatch):
|
|
from app.container import container
|
|
from app.contracts import ToolDefinition
|
|
path=tmp_path/'image.png';path.write_bytes(b'\x89PNG\r\n\x1a\nimage')
|
|
definitions=[ToolDefinition(name='plugin.image',description='',source='plugin'),ToolDefinition(name='mcp.image',description='',source='mcp_server')]
|
|
monkeypatch.setattr(container.tools,'definitions',lambda:definitions)
|
|
seen=[]
|
|
async def execute(call,context):
|
|
seen.append(call.name)
|
|
if call.name == 'mcp.image': raise TimeoutError('MCP timeout')
|
|
return SimpleNamespace(success=True,output={'text':'fallback'})
|
|
monkeypatch.setattr(container.tools,'execute',execute)
|
|
class Adapter:
|
|
async def list_models(self): return []
|
|
provider=SimpleNamespace(config=SimpleNamespace(capabilities=[]),adapter=Adapter())
|
|
request=ChatRequest(provider_id='mock',model='mock',messages=[],image_fallback_tools=['plugin.image','mcp.image'])
|
|
result=asyncio.run(service.describe_image(path,request,provider))
|
|
assert seen==['mcp.image','plugin.image'] and result[1]=='plugin.image'
|
|
|
|
|
|
def test_audio_uses_persistent_transcription_and_returns_text_context(tmp_path,monkeypatch):
|
|
from app.services import transcription_service as jobs
|
|
from app.services.attachment_service import attachment_path
|
|
path=attachment_path('audio.wav');path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(b'audio')
|
|
seen=[]
|
|
async def transcribe(attachment_id,**kwargs):
|
|
seen.append((attachment_id,kwargs))
|
|
return SimpleNamespace(status='completed',text='transcript',job_id='job_test',warnings=[])
|
|
monkeypatch.setattr(jobs,'create_transcription',transcribe)
|
|
request=ChatRequest(provider_id='mock',model='mock',messages=[],attachments=['audio.wav'])
|
|
result=asyncio.run(service.prepare(request,None))
|
|
assert seen==[('audio.wav',{'wait':True})]
|
|
assert result.attachments==[] and 'transcript' in result.system
|
|
assert result.metadata['chat_attachment_context'][0]['route']=='transcription:job_test'
|
|
|
|
|
|
def test_legacy_ppt_reads_unicode_text_records(tmp_path,monkeypatch):
|
|
import io,struct,olefile
|
|
path=tmp_path/'legacy.ppt';path.write_bytes(b'compound-file-fixture')
|
|
text='旧版演示文稿'.encode('utf-16-le');data=struct.pack('<HHI',0,4000,len(text))+text
|
|
class Ole:
|
|
def __enter__(self): return self
|
|
def __exit__(self,*args): pass
|
|
def openstream(self,name):
|
|
assert name=='PowerPoint Document'
|
|
return io.BytesIO(data)
|
|
monkeypatch.setattr(olefile,'OleFileIO',lambda path:Ole())
|
|
assert service.extract_document(path)==('旧版演示文稿',False)
|
|
|
|
|
|
def test_compatible_provider_serializes_native_image_parts():
|
|
from app.providers.openai_compatible import OpenAICompatibleProvider
|
|
from app.contracts import ModelRequest, Message
|
|
request=ModelRequest(provider_id='p',model='m',messages=[Message(role='user',content='describe',images=['data:image/png;base64,aW1hZ2U='])])
|
|
wire=OpenAICompatibleProvider._messages(None,request)
|
|
assert wire[0]['content']==[{'type':'text','text':'describe'},{'type':'image_url','image_url':{'url':'data:image/png;base64,aW1hZ2U='}}]
|