134 lines
6.7 KiB
Python
134 lines
6.7 KiB
Python
"""Offline JEV transport/contract tests; no live API key or network required."""
|
|
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from agent_framework.decisions import client
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_environment(monkeypatch):
|
|
import os
|
|
for key in list(os.environ):
|
|
if key.startswith(('DECISION_', 'JEV_')) or key == 'AI_GATEWAY_API_KEY':
|
|
monkeypatch.delenv(key)
|
|
|
|
|
|
def answer(questions, *, confidence=.95):
|
|
return {'model': 'typesafe-ai/jev', 'answers': [
|
|
{'name': q['name'], 'type': q['type'], **(
|
|
{'choice': q['choices'][0]['value'], 'confidence': confidence} if q['type'] == 'choice' else
|
|
{'probability': confidence} if q['type'] == 'predicate' else
|
|
{'score': 8.0, 'confidence': confidence})} for q in questions]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_preserves_current_call_and_kwargs(monkeypatch):
|
|
current = AsyncMock(return_value='legacy response')
|
|
monkeypatch.setattr(client, '_post', lambda body: pytest.fail('JEV must not run'))
|
|
result = await client.invoke(type('LLM', (), {'ainvoke': current})(), [{'content': 'x'}],
|
|
decision_point='router', decision_questions=[], profile_name='router', temperature=0)
|
|
assert result == 'legacy response'
|
|
current.assert_awaited_once_with([{'content': 'x'}], profile_name='router', temperature=0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_individual_selection_and_override(monkeypatch):
|
|
monkeypatch.setenv('DECISION_PROVIDER', 'jev')
|
|
monkeypatch.setenv('DECISION_ROUTER_PROVIDER', 'current')
|
|
assert not client.selected('router')
|
|
assert client.selected('route_continuity')
|
|
monkeypatch.setenv('DECISION_PROVIDER', 'current')
|
|
monkeypatch.setenv('DECISION_ROUTE_CONTINUITY_PROVIDER', 'jev')
|
|
assert client.selected('route_continuity')
|
|
assert not client.selected('router')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typed_contract_and_shared_request(monkeypatch):
|
|
monkeypatch.setenv('DECISION_ROUTER_PROVIDER', 'jev')
|
|
requests = []
|
|
def transport(body):
|
|
requests.append(body)
|
|
return answer(body['questions'])
|
|
monkeypatch.setattr(client, '_post', transport)
|
|
llm = type('LLM', (), {'ainvoke': AsyncMock()})()
|
|
result = json.loads(await client.invoke(llm, [{'content': 'context'}], decision_point='router',
|
|
decision_questions=[client.choice('intent', ['billing', 'orders']), client.predicate('allowed'), client.score('score')]))
|
|
assert result['intent'] == 'billing' and result['allowed'] is True and result['score'] == 8
|
|
assert len(requests) == 1 and requests[0]['model'] == 'typesafe-ai/jev'
|
|
assert 'context' in requests[0]['input']
|
|
llm.ainvoke.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('failure', ['network', 'unknown_choice', 'missing', 'duplicate', 'nan', 'low', 'no_confidence', 'wrong_type'])
|
|
async def test_invalid_or_uncertain_jev_falls_back_once(monkeypatch, failure):
|
|
monkeypatch.setenv('DECISION_ROUTER_PROVIDER', 'jev')
|
|
def transport(body):
|
|
if failure == 'network': raise TimeoutError()
|
|
response = answer(body['questions'])
|
|
a = response['answers'][0]
|
|
if failure == 'unknown_choice': a['choice'] = 'undeclared'
|
|
if failure == 'missing': response['answers'] = []
|
|
if failure == 'duplicate': response['answers'] *= 2
|
|
if failure == 'nan': a['confidence'] = float('nan')
|
|
if failure == 'low': a['confidence'] = .4
|
|
if failure == 'no_confidence': a.pop('confidence')
|
|
if failure == 'wrong_type': a['type'] = 'predicate'
|
|
return response
|
|
monkeypatch.setattr(client, '_post', transport)
|
|
llm = type('LLM', (), {'ainvoke': AsyncMock(return_value='legacy')})()
|
|
assert await client.invoke(llm, [], decision_point='router', decision_questions=[client.choice('intent', ['a'])]) == 'legacy'
|
|
llm.ainvoke.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_fallback_surfaces_failure(monkeypatch):
|
|
monkeypatch.setenv('DECISION_ROUTER_PROVIDER', 'jev')
|
|
monkeypatch.setenv('DECISION_FALLBACK_TO_CURRENT', 'false')
|
|
monkeypatch.setattr(client, '_post', lambda body: {'answers': []})
|
|
llm = type('LLM', (), {'ainvoke': AsyncMock()})()
|
|
with pytest.raises(ValueError):
|
|
await client.invoke(llm, [], decision_point='router', decision_questions=[client.predicate('allowed')])
|
|
llm.ainvoke.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generative_bypass_even_with_global_jev(monkeypatch):
|
|
monkeypatch.setenv('DECISION_PROVIDER', 'jev')
|
|
monkeypatch.setattr(client, '_post', lambda body: pytest.fail('rewrite must use current'))
|
|
llm = type('LLM', (), {'ainvoke': AsyncMock(return_value='rewritten')})()
|
|
assert await client.invoke(llm, [], decision_point='rewrite', decision_questions=[], decision_generative=True) == 'rewritten'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_intent_preserves_order_literal_fragments(monkeypatch):
|
|
from types import SimpleNamespace
|
|
monkeypatch.setenv('DECISION_MULTI_INTENT_PROVIDER', 'jev')
|
|
candidates = {'operations': [{'source_text': 'pedido', 'intent': 'x'}, {'source_text': 'fatura', 'intent': 'y'}]}
|
|
llm = type('LLM', (), {'ainvoke': AsyncMock(return_value=json.dumps(candidates))})()
|
|
monkeypatch.setattr(client, '_post', lambda body: answer(body['questions']))
|
|
result = json.loads(await client.multi_intent(llm, [], intents=[SimpleNamespace(name='billing')], text='pedido e fatura'))
|
|
assert [o['source_text'] for o in result['operations']] == ['pedido', 'fatura']
|
|
llm.ainvoke.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_semantic_classifier_and_jev_are_independent(monkeypatch, tmp_path):
|
|
from types import SimpleNamespace
|
|
from agent_framework.routing.enterprise_router import EnterpriseRouter
|
|
config = tmp_path / 'routing.yaml'
|
|
config.write_text('router: {fallback_agent: fallback}\nintents: []\n')
|
|
llm = type('LLM', (), {'ainvoke': AsyncMock(return_value='NO')})()
|
|
router = EnterpriseRouter(SimpleNamespace(ROUTING_CONFIG_PATH=str(config), ENABLE_LLM_ROUTER=False, ENABLE_ROUTE_STICKINESS=False), llm=llm)
|
|
expected = {'allowed_values': ['YES', 'NO'], 'semantic_classifier': {'enabled': True, 'prompt': 'Use meaning: {{ user_input }}'}}
|
|
monkeypatch.setenv('DECISION_WORKFLOW_EXPECTED_INPUT_PROVIDER', 'jev')
|
|
monkeypatch.setattr(client, '_post', lambda body: answer(body['questions']))
|
|
value, _ = await router._classify_expected_input_semantically(text='yes', expected_input=expected, pause_prompt='Confirm?')
|
|
assert value == 'YES'
|
|
value, _ = await router._classify_expected_input_semantically(text='no', expected_input=expected, pause_prompt='Confirm?', component_name='transaction.confirmation')
|
|
assert value == 'NO'
|
|
llm.ainvoke.assert_awaited_once()
|