87 lines
4.2 KiB
Python
87 lines
4.2 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from test_entity_resolution import SETTINGS
|
|
from test_llm_gateway import http_endpoint, request
|
|
from viano_bug_analysis.analysis.entity_resolution import request_for, policy
|
|
from viano_bug_analysis.llm import GatewayConfig, LLMError, LLMGateway, ProviderConfig
|
|
|
|
|
|
@pytest.mark.parametrize("effort", [None, "low", "medium", "high"])
|
|
def test_reasoning_and_output_limits_reach_transport(http_endpoint, effort):
|
|
url, seen, _ = http_endpoint
|
|
provider = ProviderConfig(id="test", model="fixture", base_url=url,
|
|
reasoning_effort=effort, max_output_tokens=16384)
|
|
asyncio.run(LLMGateway(GatewayConfig((provider,))).complete_round(request()))
|
|
body = seen[0]["body"]
|
|
assert body[provider.token_parameter] == 16384
|
|
assert body.get("reasoning_effort") == effort
|
|
assert ("reasoning_effort" in body) == (effort is not None)
|
|
snapshot = GatewayConfig((provider,)).snapshot()["providers"][0]
|
|
assert ("reasoning_effort" in snapshot) == (effort is not None)
|
|
|
|
|
|
@pytest.mark.parametrize("effort", [True, 1024, [], "unlimited"])
|
|
def test_invalid_reasoning_configuration(effort):
|
|
with pytest.raises(LLMError, match="LLM_CONFIG"):
|
|
ProviderConfig(id="test", model="fixture", base_url="http://localhost", reasoning_effort=effort)
|
|
|
|
|
|
@pytest.mark.parametrize("stage", ["plan", "assessment"])
|
|
def test_identity_projection_keeps_readable_evidence_and_legacy_requests(stage):
|
|
context = {"sources": {
|
|
"O0001": {"kind":"observation", "data":{"text":"The switch is missing"}},
|
|
"F0001": {"kind":"fact", "data":{"name":"description", "value":"Switch operator"}},
|
|
"N0001": {"kind":"entity", "data":{"id":"42"}},
|
|
"X0001": {"kind":"unread_reference", "data":{"path":"raw/" + "x" * 10000}},
|
|
"F0002": {"kind":"fact_reference", "data":{"path":"raw/fact"}}}}
|
|
before = copy.deepcopy(context)
|
|
args = dict(plan={"searches":[]}, captures=[], candidates=[]) if stage == "assessment" else {}
|
|
old = request_for(stage, context, policy(SETTINGS), protocol_version=2, **args)
|
|
new = request_for(stage, context, policy(SETTINGS), protocol_version=3, **args)
|
|
old_data = json.loads(old["messages"][1]["content"])
|
|
data = json.loads(new["messages"][1]["content"])
|
|
assert old_data["context"] == context == before
|
|
assert set(data["context"]["sources"]) == {"O0001", "F0001", "N0001"}
|
|
for key, value in data["context"]["sources"].items():
|
|
assert value == context["sources"][key]
|
|
assert len(new["messages"][1]["content"]) < len(old["messages"][1]["content"]) / 2
|
|
assert data["output_schema"] == new["output_schema"]
|
|
|
|
|
|
@pytest.mark.parametrize("version,expected", [
|
|
("entity_resolution_plan_v1", "low"),
|
|
("entity_resolution_assessment_v3", "low"),
|
|
("case_interpretation_resolved_v3", "medium"),
|
|
("hypothesis_generation_v1", "medium"),
|
|
("investigation_v1", "medium"),
|
|
("entity_resolution_assessment_v3_suffix", "medium"),
|
|
])
|
|
def test_identity_override_is_scoped_at_http_transport(http_endpoint, version, expected):
|
|
url, seen, _ = http_endpoint
|
|
provider = ProviderConfig(id="test", model="fixture", base_url=url,
|
|
reasoning_effort="medium", entity_resolution_reasoning_effort="low", max_output_tokens=16384)
|
|
config = GatewayConfig((provider,))
|
|
query = request()
|
|
query['prompt_version'] = version
|
|
asyncio.run(LLMGateway(config).complete_round(query))
|
|
assert seen[0]['body']['reasoning_effort'] == expected
|
|
assert seen[0]['body'][provider.token_parameter] == 16384
|
|
assert provider.reasoning_effort == "medium"
|
|
assert config.snapshot()['providers'][0]['entity_resolution_reasoning_effort'] == "low"
|
|
|
|
|
|
@pytest.mark.parametrize("effort", [True, 8192, [], "unlimited"])
|
|
def test_invalid_identity_reasoning_override(effort):
|
|
with pytest.raises(LLMError, match="LLM_CONFIG"):
|
|
ProviderConfig(id="test", model="fixture", base_url="http://localhost",
|
|
entity_resolution_reasoning_effort=effort)
|
|
|
|
|
|
def test_omitted_identity_override_preserves_previous_snapshot():
|
|
provider = ProviderConfig(id="test", model="fixture", base_url="http://localhost", reasoning_effort="medium")
|
|
assert 'entity_resolution_reasoning_effort' not in GatewayConfig((provider,)).snapshot()['providers'][0]
|