429 lines
24 KiB
Python
429 lines
24 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case, gateway, payload, statement
|
|
from viano_bug_analysis.analysis import CaseInterpreter, InterpretationError
|
|
from viano_bug_analysis.analysis.context import build_context, InterpretationLimits
|
|
from viano_bug_analysis.analysis.entity_resolution import policy, validate_plan
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.lifecycle import CheckpointStore
|
|
from viano_bug_analysis.llm import ProviderResponse, ProviderUnavailable
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.tools import ToolError
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
|
|
|
|
SETTINGS = {"enabled": True, "max_searches": 4, "max_candidates": 8, "max_invalid_responses": 3,
|
|
"catalogs": [{"id": "components", "entity_type": "component", "table": "components",
|
|
"id_column": "id", "name_column": "name", "columns": ["id", "name", "script"]}]}
|
|
PLAN = {"searches": [{"catalog": "components", "term": "switch", "role": "affected_subject", "source_ids": ["O0001"]}]}
|
|
|
|
|
|
class Database:
|
|
def __init__(self, rows=None, truncated=False):
|
|
self.rows = rows if rows is not None else [{"id": 17, "name": "Wall switch", "script": "switch_control"},
|
|
{"id": 42, "name": "Switch operator", "script": "operator_dialog"}]
|
|
self.truncated = truncated
|
|
self.calls = []
|
|
self.unavailable = False
|
|
self.snapshot = "fixture-1"
|
|
|
|
def binding(self):
|
|
return {"kind": "fixture", "snapshot_id": self.snapshot}
|
|
|
|
def validate(self, query):
|
|
assert query["table"] == "components" and query["kind"] == "database.rows"
|
|
if self.unavailable:
|
|
raise ToolError("PROJECT_DATABASE_UNAVAILABLE")
|
|
|
|
def query(self, query):
|
|
self.calls.append(copy.deepcopy(query))
|
|
return {"items": [{"table": "components", "row": r} for r in self.rows], "coverage": "complete_filtered_rows",
|
|
"truncated": self.truncated, "next_offset": 8 if self.truncated else None}
|
|
|
|
|
|
class Model:
|
|
def __init__(self, plan=None, status="selected", candidate="K0001"):
|
|
self.plan = plan or PLAN
|
|
self.status, self.candidate = status, candidate
|
|
self.requests = []
|
|
self.invalid_plan = 0
|
|
self.invalid_assessment = 0
|
|
self.unavailable_stage = None
|
|
self.omit_resolved = 0
|
|
self.linked_capture = False
|
|
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.requests.append(copy.deepcopy(request))
|
|
version = request["prompt_version"]
|
|
if version == self.unavailable_stage:
|
|
raise ProviderUnavailable("LLM_UNAVAILABLE")
|
|
if version.startswith('entity_resolution_') and version.endswith(('_v2','_v3')):
|
|
# This is the channel json_object providers actually receive.
|
|
assert json.loads(request['messages'][1]['content'])['output_schema'] == request['output_schema']
|
|
version = version.replace('_v1', '_v3').replace('_v2', '_v3') if version.startswith('entity_resolution_') else version
|
|
if version == "entity_resolution_plan_v3":
|
|
document = copy.deepcopy(self.plan)
|
|
if self.invalid_plan:
|
|
self.invalid_plan -= 1
|
|
document["searches"][0]["term"] = "invented device"
|
|
elif version == "entity_resolution_assessment_v3":
|
|
document = {"resolutions": [{"search_index": i, "status": self.status,
|
|
"candidate_id": self.candidate if self.status == "selected" else None,
|
|
"reason": "The affected subject is the wall switch; the operator is an interaction participant.",
|
|
"source_ids": ["O0001", f"V{i+1:04d}"]} for i,_ in enumerate(self.plan["searches"])]}
|
|
if self.invalid_assessment:
|
|
self.invalid_assessment -= 1
|
|
document["resolutions"][0]["candidate_id"] = "K9999"
|
|
else:
|
|
document = payload()
|
|
if version in {"case_interpretation_resolved_v3", "case_interpretation_resolved_v4", "case_interpretation_resolved_v5"}:
|
|
document.update(described_behavior=[],provisional_behavior_gap=None)
|
|
data = json.loads(request["messages"][1]["content"])["input"]
|
|
selected = [(k,v) for k,v in data["sources"].items() if v["kind"] == "entity" and v["case_pointer"].startswith("/entity_resolution/")]
|
|
for k,v in selected:
|
|
evidence = k
|
|
if self.linked_capture:
|
|
evidence = next(f for f,s in data['sources'].items() if s['kind']=='fact'
|
|
and s['data'].get('name')=='entity_name_lookup'
|
|
and s['data']['provenance']['source_ref']==v['data']['provenance']['source_ref'])
|
|
document["relevant_entities"].append({"entity_id": k, "reason": "Contextually matched affected switch", "source_ids": [evidence,"O0001"]})
|
|
if self.omit_resolved:
|
|
self.omit_resolved -= 1
|
|
document["relevant_entities"] = payload()["relevant_entities"]
|
|
if version in {'case_interpretation_resolved_v4','case_interpretation_resolved_v5'}:
|
|
from viano_bug_analysis.analysis.quality import unnamed_entity_reasons
|
|
reasons = unnamed_entity_reasons(data)
|
|
for entity in document['relevant_entities']:
|
|
if entity['entity_id'] in reasons:
|
|
entity['reason'] = reasons[entity['entity_id']]
|
|
unresolved = [k for k,v in data["sources"].items() if v["kind"] == "data_quality" and v["case_pointer"].startswith("/entity_resolution/")]
|
|
for k in unresolved:
|
|
document["unknowns"].append(statement("Identity is unresolved, not proof of a missing runtime entity",k))
|
|
return ProviderResponse(document, "fixture-model", 100, 20)
|
|
|
|
|
|
def service_for(store, database=None):
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
project = reader.manifest["project"]
|
|
service = ProjectResearch(project, database=database or Database())
|
|
service.entity_resolution = copy.deepcopy(SETTINGS)
|
|
return service
|
|
|
|
|
|
def run(store, model, service=None, **kwargs):
|
|
return asyncio.run(CaseInterpreter(store, gateway(model), project_research=service, **kwargs).interpret("case-A", "stable-package"))
|
|
|
|
|
|
def context_of(store):
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
ref = reader.manifest["artifacts"]["case"]
|
|
return build_context(json.loads(reader.read_bytes(ref)),ref,InterpretationLimits())
|
|
|
|
|
|
def test_resolves_unrecorded_subject_preserves_case_and_portable_replay(built_case, tmp_path):
|
|
store, _, built, _, _ = built_case
|
|
with PackageReader(built.package) as reader:
|
|
old = {n:reader.read_bytes(n) for n in reader.files if n != "manifest.json"}
|
|
model = Model(); service = service_for(store)
|
|
result = run(store, model, service)
|
|
assert result.status == "COMPLETED" and PackageValidator().validate(result.package).valid
|
|
assert [r["prompt_version"] for r in model.requests] == ["entity_resolution_plan_v3","entity_resolution_assessment_v3","case_interpretation_resolved_v5"]
|
|
assert len(service.database.calls) == 1
|
|
assert service.database.calls[0]["filters"] == [{"column":"name","operator":"contains","value":"switch"}]
|
|
with PackageReader(result.package) as reader:
|
|
for n,data in old.items():
|
|
assert reader.read_bytes(n) == data
|
|
doc = json.loads(reader.read_bytes(result.interpretation_ref))
|
|
context = json.loads(reader.read_bytes(doc["context_ref"]))
|
|
subject = context["sources"]["N0002"]["data"]
|
|
assert subject["id"] == "17" and subject["roles"] == ["affected_subject"]
|
|
assert subject["association"] == "model_contextual_selection_not_causation"
|
|
assert context["sources"]["N0001"]["data"]["id"] == "42"
|
|
assert "captures/" in subject["provenance"]["source_ref"]
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
assert CaseBuilder(store).build("case-A","stable-package").reused
|
|
# No live DB binding required when all records are complete.
|
|
receiving = CheckpointStore(type(store.workspace)(tmp_path / "portable"))
|
|
receiving.intake(result.package)
|
|
again = run(receiving, model)
|
|
assert again.reused and len(model.requests) == 3
|
|
|
|
|
|
@pytest.mark.parametrize("status,rows,truncated", [("ambiguous",None,False),("no_match",[],False),("ambiguous",None,True)])
|
|
def test_unresolved_candidates_remain_explicit_unknowns(built_case,status,rows,truncated):
|
|
store=built_case[0]; model=Model(status=status); service=service_for(store,Database(rows,truncated))
|
|
result=run(store,model,service)
|
|
with PackageReader(result.package) as r:
|
|
doc=json.loads(r.read_bytes(result.interpretation_ref)); context=json.loads(r.read_bytes(doc['context_ref']))
|
|
assert 'N0002' not in context['sources']
|
|
assert any(v['data'].get('code')=='ENTITY_RESOLUTION_UNRESOLVED' for v in context['sources'].values() if isinstance(v['data'],dict))
|
|
assert any('Identity is unresolved' in s['text'] for s in doc['interpretation']['unknowns'])
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["invalid_plan","invalid_assessment"])
|
|
def test_invalid_outputs_corrected_without_repeating_queries(built_case,field):
|
|
store=built_case[0]; model=Model(); setattr(model,field,1); service=service_for(store)
|
|
assert run(store,model,service).status=='REJECTED'
|
|
assert run(store,model,service).status=='COMPLETED'
|
|
assert len(service.database.calls)==1 and len(model.requests)==4
|
|
assert any('validation_feedback' in r['messages'][1]['content'] for r in model.requests)
|
|
assert run(store,model).reused and len(model.requests)==4
|
|
|
|
|
|
def test_invalid_exhaustion_blocks_without_hypotheses_or_budget_mutation(built_case):
|
|
store=built_case[0]; model=Model();model.invalid_plan=3;service=service_for(store)
|
|
for _ in range(3):
|
|
assert run(store,model,service).status=='REJECTED'
|
|
result=run(store,model,service)
|
|
assert result.status=='BLOCKED' and result.code=='ENTITY_RESOLUTION_INVALID_RESPONSES'
|
|
assert len(model.requests)==3 and not service.database.calls
|
|
assert run(store,model,service).status=='BLOCKED'
|
|
|
|
|
|
def test_database_outage_preserves_accepted_plan(built_case,monkeypatch):
|
|
store=built_case[0];model=Model(); service=service_for(store);service.database.unavailable=True
|
|
result=run(store,model,service)
|
|
assert result.status=='BLOCKED' and result.code=='RESEARCH_UNAVAILABLE'
|
|
with PackageReader(result.package) as r:
|
|
assert 'analysis/interpretations/0001/entity-resolution/plan.json' in r.files
|
|
assert r.manifest['artifacts'].get('case_interpretation') is None
|
|
service.database.unavailable=False
|
|
monkeypatch.setattr(store,'retry_due',lambda m: True)
|
|
store.advance('case-A','stable-package','INTERPRETING_CASE',step_id='manual-resume',details={'operation':'manual-resume'})
|
|
assert run(store,model,service).status=='COMPLETED'
|
|
assert len(model.requests)==3
|
|
|
|
|
|
def test_provider_retry_resume_uses_frozen_input(built_case):
|
|
store=built_case[0];model=Model(); model.unavailable_stage='entity_resolution_plan_v3'; service=service_for(store)
|
|
# Tiny positive retry delay permits resumption in this offline fixture.
|
|
g=gateway(model,retry_delay_seconds=0.001)
|
|
owner=CaseInterpreter(store,g,project_research=service)
|
|
first=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert first.status=='RETRY_PENDING'
|
|
model.unavailable_stage=None
|
|
assert asyncio.run(owner.interpret('case-A','stable-package')).status=='COMPLETED'
|
|
|
|
|
|
def test_changed_snapshot_rejected_before_additional_calls(built_case):
|
|
store=built_case[0];model=Model();model.invalid_plan=1;service=service_for(store)
|
|
assert run(store,model,service).status=='REJECTED'
|
|
service.database.snapshot='different'
|
|
with pytest.raises(InterpretationError,match='ENTITY_RESOLUTION_BINDING_CHANGED'):
|
|
run(store,model,service)
|
|
assert len(model.requests)==1
|
|
|
|
|
|
@pytest.mark.parametrize("mutation,code", [
|
|
(lambda p:p['searches'][0].update(term='Unknown NPC'),'ENTITY_RESOLUTION_UNGROUNDED_TERM'),
|
|
(lambda p:p['searches'][0].update(source_ids=['X0001']),'ENTITY_RESOLUTION_SOURCE'),
|
|
(lambda p:p['searches'][0].update(catalog='account'),'ENTITY_RESOLUTION_CATALOG'),
|
|
(lambda p:p['searches'].append(copy.deepcopy(p['searches'][0])),'ENTITY_RESOLUTION_DUPLICATE_SEARCH'),
|
|
])
|
|
def test_plan_validation_rejects_invented_terms_unread_refs_and_unknown_catalogs(built_case,mutation,code):
|
|
context=context_of(built_case[0]); p=copy.deepcopy(PLAN);mutation(p)
|
|
with pytest.raises(InterpretationError,match=code):validate_plan(p,context,policy(SETTINGS))
|
|
|
|
|
|
def test_truncated_search_cannot_be_selected(built_case):
|
|
store=built_case[0]; model=Model();service=service_for(store,Database(truncated=True))
|
|
result=run(store,model,service)
|
|
assert result.status=='REJECTED' and result.code=='ENTITY_RESOLUTION_SELECTION'
|
|
|
|
|
|
def test_search_cannot_be_grounded_in_a_capture_filename(built_case):
|
|
context=context_of(built_case[0]);p=copy.deepcopy(PLAN)
|
|
context['sources']['F0001']['data']['value']={'origin':{'source_ref':'raw/invented device.json'},'row':{'name':'switch'}}
|
|
p['searches'][0].update(term='invented device',source_ids=['F0001'])
|
|
with pytest.raises(InterpretationError,match='ENTITY_RESOLUTION_UNGROUNDED_TERM'):
|
|
validate_plan(p,context,policy(SETTINGS))
|
|
|
|
|
|
def test_empty_plan_still_supports_descriptive_quality_contract(built_case):
|
|
store=built_case[0];model=Model(plan={'searches':[]});service=service_for(store)
|
|
assert run(store,model,service).status=='COMPLETED'
|
|
assert model.requests[-1]['prompt_version']=='case_interpretation_resolved_v5' and not service.database.calls
|
|
|
|
|
|
def test_trial_phase_stop_includes_resolution_and_is_idempotent(built_case):
|
|
from viano_bug_analysis.trial import drive
|
|
store=built_case[0];model=Model();service=service_for(store)
|
|
async def trial():
|
|
return await drive(store,gateway(model),'case-A','stable-package',max_dispatches=6,
|
|
policies={'project_research':service},stop_after='interpretation')
|
|
result=asyncio.run(trial())
|
|
assert result['status']=='PHASE_COMPLETE' and result['state']=='INTERPRETING_CASE'
|
|
assert asyncio.run(trial())['status']=='PHASE_COMPLETE' and len(model.requests)==3
|
|
|
|
|
|
def test_final_interpretation_cannot_discard_resolved_identity(built_case):
|
|
from viano_bug_analysis.analysis.correction import CorrectionConfig
|
|
store=built_case[0];model=Model();model.omit_resolved=1;service=service_for(store)
|
|
result=run(store,model,service,correction_config=CorrectionConfig())
|
|
assert result.status=='REJECTED' and result.code=='INTERPRETATION_RESOLVED_ENTITY_OMITTED'
|
|
result=run(store,model,service,correction_config=CorrectionConfig())
|
|
assert result.status=='COMPLETED' and len(model.requests)==4
|
|
feedback=json.loads(model.requests[-1]['messages'][-1]['content'])['validation_feedback']
|
|
assert any(i.get('constraint')=='retain_resolved_entity' and i['entity_id']=='N0002' for i in feedback['issues'])
|
|
assert len(service.database.calls)==1
|
|
assert run(store,model).reused and len(model.requests)==4
|
|
|
|
|
|
def test_matching_database_capture_is_valid_without_redundant_entity_citation(built_case):
|
|
store=built_case[0];model=Model();model.linked_capture=True;service=service_for(store)
|
|
result=run(store,model,service)
|
|
assert result.status=='COMPLETED' and len(model.requests)==3
|
|
assert run(store,model).reused
|
|
with PackageReader(result.package) as r:
|
|
doc=json.loads(r.read_bytes(result.interpretation_ref))
|
|
assert 'N0002' not in doc['interpretation']['relevant_entities'][-1]['source_ids']
|
|
|
|
|
|
def test_provenance_feedback_identifies_missing_mention_and_wrong_capture(built_case):
|
|
from viano_bug_analysis.analysis.contracts import entity_citation_issues,validate_interpretation
|
|
store=built_case[0];model=Model();service=service_for(store);result=run(store,model,service)
|
|
with PackageReader(result.package) as r:
|
|
doc=json.loads(r.read_bytes(result.interpretation_ref));context=json.loads(r.read_bytes(doc['context_ref']))
|
|
item=doc['interpretation']['relevant_entities'][-1]
|
|
item['source_ids']=['F0001']
|
|
issues=entity_citation_issues(doc['interpretation'],context)
|
|
assert {i['constraint'] for i in issues}=={'cite_original_mention','cite_entity_or_matching_capture'}
|
|
assert next(i for i in issues if i['constraint']=='cite_original_mention')['allowed_source_ids']==['O0001']
|
|
assert 'N0002' in next(i for i in issues if i['constraint']=='cite_entity_or_matching_capture')['allowed_source_ids']
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_ENTITY_PROVENANCE'):
|
|
validate_interpretation(doc['interpretation'],context)
|
|
# Even a lookup-shaped fact cannot substitute an unrelated capture.
|
|
context['sources']['F0001']['data'].update(name='entity_name_lookup',provenance={'source_ref':'different/capture.json'})
|
|
item['source_ids']=['O0001','F0001']
|
|
assert [i['constraint'] for i in entity_citation_issues(doc['interpretation'],context)]==['cite_entity_or_matching_capture']
|
|
|
|
|
|
def test_real_gemini_reply_is_accepted_with_matching_capture_and_legacy_rule_is_preserved():
|
|
from pathlib import Path
|
|
from viano_bug_analysis.analysis.contracts import validate_interpretation
|
|
fixture=json.loads((Path(__file__).parent/'fixtures/interpretation/real-capture-citations.json').read_text(encoding='utf-8'))
|
|
validate_interpretation(fixture['response'],fixture['context'])
|
|
entity=next(e for e in fixture['response']['relevant_entities'] if e['entity_id']=='N0005')
|
|
assert 'N0005' not in entity['source_ids'] and 'F0027' in entity['source_ids']
|
|
assert fixture['context']['sources']['N0005']['data']['id']=='23487'
|
|
old=copy.deepcopy(fixture['context'])
|
|
del old['sources']['N0005']['data']['citation_policy']
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_RESOLVED_ENTITY_OMITTED'):
|
|
validate_interpretation(fixture['response'],old)
|
|
|
|
|
|
@pytest.mark.parametrize('settings', [{'enabled':True,'catalogs':[]},{**SETTINGS,'max_searches':True},{**SETTINGS,'unexpected':5}])
|
|
def test_invalid_operator_configuration(settings):
|
|
with pytest.raises(InterpretationError,match='ENTITY_RESOLUTION_CONFIG'):policy(settings)
|
|
|
|
|
|
@pytest.mark.parametrize("legacy_version", [1, 2])
|
|
def test_legacy_resolution_requests_and_completed_packages_remain_replayable(built_case, legacy_version):
|
|
from viano_bug_analysis.analysis.entity_resolution import EntityResolver, request_for
|
|
store=built_case[0];service=service_for(store);model=Model()
|
|
owner=CaseInterpreter(store,gateway(model),project_research=service)
|
|
context=context_of(store)
|
|
# A v1 artifact has no protocol_version; preserve its original request bytes.
|
|
EntityResolver(owner).save(context,'input',{'context':context,'policy':policy(SETTINGS),'binding':service.database.binding(),
|
|
**({'protocol_version':2} if legacy_version == 2 else {})})
|
|
old=request_for('plan',context,policy(SETTINGS))
|
|
assert old['prompt_version']=='entity_resolution_plan_v1'
|
|
assert 'output_schema' not in json.loads(old['messages'][1]['content'])
|
|
result=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status=='COMPLETED'
|
|
assert model.requests[0]['prompt_version']==f'entity_resolution_plan_v{legacy_version}'
|
|
assert run(store,model).reused and len(model.requests)==3
|
|
|
|
|
|
def test_missing_capture_feedback_resumes_and_replays_without_repeating_research(built_case):
|
|
class MissingCaptureModel(Model):
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
response = await super().generate(config, request, max_response_bytes=max_response_bytes)
|
|
if request['prompt_version'] == 'entity_resolution_assessment_v3':
|
|
data = json.loads(request['messages'][1]['content'])
|
|
if 'validation_feedback' not in data:
|
|
for resolution in response.document['resolutions']:
|
|
resolution['source_ids'] = ['O0001']
|
|
resolution['reason'] = 'The right capture is V0001 but only mentioned here.'
|
|
else:
|
|
fixes = data['validation_feedback']['citation_corrections']
|
|
assert fixes['missing_capture_sources'] == [{
|
|
'search_index':0, 'required_source_id':'V0001',
|
|
'field':'/resolutions/0/source_ids', 'original_mention_source_ids':['O0001']}]
|
|
return response
|
|
store = built_case[0]
|
|
model = MissingCaptureModel()
|
|
service = service_for(store)
|
|
rejected = run(store, model, service)
|
|
assert rejected.status == 'REJECTED'
|
|
assert rejected.code == 'ENTITY_RESOLUTION_CAPTURE_SOURCE'
|
|
assert len(service.database.calls) == 1
|
|
result = run(store, model, service)
|
|
assert result.status == 'COMPLETED'
|
|
assert len(service.database.calls) == 1
|
|
assert len(model.requests) == 4
|
|
with PackageReader(result.package) as reader:
|
|
feedback_ref = 'analysis/interpretations/0001/entity-resolution/feedback-assessment-0002.json'
|
|
assert feedback_ref in reader.files
|
|
assert PackageValidator().validate(result.package).valid
|
|
assert run(store, model).reused
|
|
assert len(model.requests) == 4
|
|
|
|
|
|
def test_third_assessment_can_use_lower_reasoning_without_resetting_history(built_case):
|
|
from dataclasses import replace
|
|
from viano_bug_analysis.llm import LLMError, LLMGateway
|
|
class TruncatedModel(Model):
|
|
failures = 2
|
|
efforts = []
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.efforts.append((request['prompt_version'], config.reasoning_effort))
|
|
if request['prompt_version']=='entity_resolution_assessment_v3' and self.failures:
|
|
self.failures -= 1
|
|
raise LLMError('LLM_INCOMPLETE_OUTPUT')
|
|
return await super().generate(config, request, max_response_bytes=max_response_bytes)
|
|
store = built_case[0]
|
|
model = TruncatedModel()
|
|
service = service_for(store)
|
|
old = gateway(model)
|
|
old.config = replace(old.config, providers=tuple(replace(p, reasoning_effort='medium') for p in old.config.providers))
|
|
owner = CaseInterpreter(store, old, project_research=service)
|
|
for _ in range(2):
|
|
result = asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status == 'REJECTED'
|
|
adjusted = replace(old.config, providers=tuple(replace(p, entity_resolution_reasoning_effort='low') for p in old.config.providers))
|
|
owner = CaseInterpreter(store, LLMGateway(adjusted, {'fixture': model}), project_research=service)
|
|
result = asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status == 'COMPLETED'
|
|
assert len(service.database.calls) == 1
|
|
assert model.efforts == [('entity_resolution_plan_v3','medium'),
|
|
('entity_resolution_assessment_v3','medium'),('entity_resolution_assessment_v3','medium'),
|
|
('entity_resolution_assessment_v3','low'),('case_interpretation_resolved_v5','medium')]
|
|
with PackageReader(result.package) as reader:
|
|
saved = json.loads(reader.read_bytes('analysis/interpretations/0001/entity-resolution/assessment.json'))
|
|
assert saved['payload']['attempt'] == 3
|
|
assert asyncio.run(owner.interpret('case-A','stable-package')).reused
|
|
assert len(model.efforts) == 5
|
|
|
|
|
|
@pytest.mark.parametrize("version", [3,4])
|
|
def test_completed_interpretation_keeps_its_historical_contract(built_case, version):
|
|
from viano_bug_analysis.analysis.entity_resolution import EntityResolver
|
|
store=built_case[0]
|
|
service=service_for(store)
|
|
model=Model()
|
|
owner=CaseInterpreter(store,gateway(model),project_research=service)
|
|
context=context_of(store)
|
|
EntityResolver(owner).save(context,'input',{'context':context,'policy':policy(SETTINGS),
|
|
'binding':service.database.binding(),'protocol_version':3,'interpretation_protocol_version':version})
|
|
result=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status=='COMPLETED'
|
|
assert model.requests[-1]['prompt_version']==f'case_interpretation_resolved_v{version}'
|
|
assert run(store,model).reused
|
|
assert len(model.requests)==3
|