82 lines
3.8 KiB
Python
82 lines
3.8 KiB
Python
"""Question/engine binding across actual HTTP/MCP boundaries, without an LLM."""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from test_claude_packages import backend, ROOT
|
|
from packages.claude import client
|
|
|
|
|
|
QUESTIONS=[
|
|
'Čomu sa venuje diplomová práca Jána Holpa? Stručne ju predstav a uveď zdroj.',
|
|
'Porovnaj zadanie a výsledky Ondreja Megelu a Jána Holpa.\nPlán nie je výsledok; uveď zdroje.',
|
|
'Ktorí študenti majú tag "RAG" a zároveň diplomový projekt z roku 2026?']
|
|
|
|
|
|
@pytest.mark.parametrize('engine',['rag','graphrag'])
|
|
def test_latest_question_and_all_conditions_cross_http_unchanged(backend,engine):
|
|
for question in QUESTIONS:
|
|
r=client.query_backend(question,engine,limit=7)
|
|
assert r['accepted_for_delivery'] and r['query']==question and r['engine']==engine
|
|
path,body,_=backend['requests'][-1]
|
|
assert path=='/answer' and body['query']==question and body['engine']==engine and body['limit']==7
|
|
assert len(backend['requests'])==len(QUESTIONS)
|
|
|
|
|
|
@pytest.mark.parametrize('engine',['rag','graphrag'])
|
|
def test_only_a_matching_leading_command_is_removed(backend,engine):
|
|
question='Aké práce má Ján Holp v roku 2020? Zachovaj výraz "RAG".'
|
|
r=client.query_backend(' /'+engine.upper()+' '+question+' ',engine)
|
|
assert r['accepted_for_delivery'] and r['query']==question
|
|
assert backend['requests'][-1][1]['query']==question
|
|
other='graphrag' if engine=='rag' else 'rag'
|
|
r=client.query_backend('/'+other+' '+question,engine)
|
|
assert r['reason']=='engine_mismatch' and not r['accepted_for_delivery']
|
|
assert len(backend['requests'])==1
|
|
|
|
|
|
@pytest.mark.parametrize('question',['/rag','/rag ','/graphrag'])
|
|
def test_command_without_a_question_never_reaches_backend(backend,question):
|
|
engine='graphrag' if question.startswith('/graphrag') else 'rag'
|
|
assert client.query_backend(question,engine)['reason']=='invalid_query'
|
|
assert not backend['requests']
|
|
|
|
|
|
@pytest.mark.parametrize('change',[
|
|
{'query':'A previous unrelated question','engine':'rag'},
|
|
{'query':QUESTIONS[0],'engine':'graphrag'},
|
|
])
|
|
def test_stale_question_or_wrong_engine_cannot_deliver_another_answer(backend,change):
|
|
backend['raw']=json.dumps({**backend['checked'],'answer':'OTHER_ANSWER_SHOULD_NOT_ESCAPE',**change}).encode()
|
|
r=client.query_backend(QUESTIONS[0],'rag')
|
|
assert r['reason']=='question_mismatch' and not r['accepted_for_delivery']
|
|
assert 'OTHER_ANSWER' not in json.dumps(r)
|
|
|
|
|
|
def test_missing_question_echo_is_not_accepted(backend):
|
|
backend['raw']=json.dumps({**backend['checked'],'engine':'rag'}).encode()
|
|
r=client.query_backend(QUESTIONS[0],'rag')
|
|
assert r['reason']=='invalid_backend_response' and not r['accepted_for_delivery']
|
|
|
|
|
|
def test_consecutive_mcp_calls_do_not_reuse_previous_queries(backend):
|
|
from mcp import ClientSession,StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
from packages.claude.mcp_server import QUESTION_POLICY
|
|
async def run():
|
|
params=StdioServerParameters(command=sys.executable,args=[str(ROOT/'packages/claude/mcp_server.py')],
|
|
env={**os.environ,'PYTHONDONTWRITEBYTECODE':'1'})
|
|
async with stdio_client(params) as (read,write):
|
|
async with ClientSession(read,write) as session:
|
|
initialized=await session.initialize()
|
|
assert QUESTION_POLICY in initialized.instructions
|
|
for engine,question in [('rag',QUESTIONS[0]),('graphrag',QUESTIONS[2]),('rag',QUESTIONS[1])]:
|
|
result=await session.call_tool('zpwiki_'+engine,{'query':question})
|
|
assert not result.isError and result.structuredContent['query']==question
|
|
assert backend['requests'][-1][1]['query']==question
|
|
assert backend['requests'][-1][1]['engine']==engine
|
|
asyncio.run(run())
|