import copy
import pytest
from viano_bug_analysis.prompt_probe import prepare
from viano_bug_analysis.minimal_prompt_test import measure
from viano_bug_analysis.llm import LLMError
from test_minimal_prompt_test import provider


def request():return {'prompt_version':'fixture','messages':[{'role':'system','content':'Test instructions'},
    {'role':'user','content':'Test context'}], 'output_schema':{'type':'object','required':['query'],
    'properties':{'query':{'const':'database.schema'}},'additionalProperties':False}}


def test_native_schema_sent_once_without_wrapper_or_tools():
    r=request();before=copy.deepcopy(r);case=prepare(provider(),r)
    payload=case['payload']
    assert set(payload)=={'systemInstruction','contents','generationConfig'}
    assert payload['contents'][0]['parts'][0]['text']=='Test context'
    assert payload['generationConfig']['responseJsonSchema']['properties']['query']=={'type':'string','enum':['database.schema']}
    assert case['validation_schema']==r['output_schema']
    assert payload['generationConfig']['thinkingConfig']=={'thinkingLevel':'low'}
    assert r==before


def test_validates_the_prepared_schema_not_the_minimal_test_schema():
    case=prepare(provider(),request())
    def post(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"database.schema"}'}]}}],
        'usageMetadata':{'promptTokenCount':200,'candidatesTokenCount':8,'thoughtsTokenCount':30}}
    result=measure(provider(),case,10000,post)
    assert result['status']=='SUCCESS' and result['schema_valid']
    assert result['usage']['input_tokens']==200


@pytest.mark.parametrize('field,value',[('messages',[]),('messages',[{'role':'user','content':'x'}])])
def test_rejects_invalid_envelopes(field,value):
    r=request();r[field]=value
    with pytest.raises(LLMError):prepare(provider(),r)


def test_gemini_projection_types_const_and_keeps_strict_local_validation():
    from viano_bug_analysis.prompt_probe import gemini_schema
    from jsonschema import Draft202012Validator
    r=request()
    r['output_schema']['properties']['query']={'const':'database.schema','minLength':5,'pattern':'^database'}
    original=copy.deepcopy(r)
    c=prepare(provider(),r)
    wire=c['payload']['generationConfig']['responseJsonSchema']
    Draft202012Validator.check_schema(wire)
    assert wire['properties']['query']=={'type':'string','enum':['database.schema']}
    assert r==original and c['validation_schema']==r['output_schema']
    assert gemini_schema({'type':['string','number','boolean','null']})=={
        'anyOf':[{'type':v} for v in ['string','number','boolean','null']]}


def test_local_constraints_not_silently_dropped_from_response_check():
    r=request();r['output_schema']['properties']['query']={'type':'string','pattern':'^database'}
    c=prepare(provider(),r)
    def post(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"invented"}'}]}}],
        'usageMetadata':{'promptTokenCount':200}}
    result=measure(provider(),c,10000,post)
    assert result['status']=='INVALID_JSON' and result['schema_valid'] is False


def test_provider_error_is_bounded_and_redacts_credentials_and_literals(monkeypatch):
    from dataclasses import replace
    from viano_bug_analysis.prompt_probe import provider_diagnostic
    monkeypatch.setenv('PROBE_TEST_SECRET','key-secret-value')
    p=replace(provider(),api_key_env='PROBE_TEST_SECRET')
    detail=provider_diagnostic({'error':{'status':'INVALID_ARGUMENT',
        'message':'generation_config schema failed: key-secret-value "private text" https://example.invalid/private',
        'details':[{'fieldViolations':[{'field':'generation_config.response_json_schema','description':'Too many states'}]}],
        'extra':'do not retain'}},p)
    import json
    text=json.dumps(detail)
    assert 'key-secret-value' not in text and 'private text' not in text
    assert 'https://' not in text and 'do not retain' not in text
    assert detail['status']=='INVALID_ARGUMENT'
    assert detail['field_violations'][0]['description']=='Too many states'


def test_opt_in_http_capture_preserves_controlled_error(monkeypatch):
    import io,json,time,urllib.error
    from viano_bug_analysis.llm.gemini import GeminiProvider
    body=json.dumps({'error':{'status':'INVALID_ARGUMENT','message':'Schema too complex'}}).encode()
    class Opener:
        def open(self,*args,**kwargs):
            raise urllib.error.HTTPError('https://example.invalid',400,'Bad Request',{},io.BytesIO(body))
    monkeypatch.setattr('urllib.request.build_opener',lambda *args:Opener())
    seen=[]
    with pytest.raises(LLMError) as caught:
        GeminiProvider._post('https://example.invalid',{}, {},10000,time.monotonic()+10,error_capture=seen.append)
    assert caught.value.code=='LLM_HTTP_CONFIGURATION' and caught.value.http_status==400
    assert seen[0]['error']['message']=='Schema too complex'


def test_existing_output_is_named_error_without_network(tmp_path,monkeypatch,capsys):
    import json
    from viano_bug_analysis import prompt_probe
    from viano_bug_analysis.llm import GatewayConfig
    path=tmp_path/'request.json';path.write_text(json.dumps(request()))
    output=tmp_path/'result.json';output.write_text('keep original')
    monkeypatch.setattr(prompt_probe.GatewayConfig,'load',lambda _:GatewayConfig(providers=(provider(),),privacy_profile='cloud_allowed'))
    def forbidden(*args):raise AssertionError('Unexpected network call')
    monkeypatch.setattr(prompt_probe,'measure_diagnostic',forbidden)
    result=prompt_probe.main(['--run','--request',str(path),'--output',str(output)])
    assert result==2 and 'PROMPT_PROBE_OUTPUT_EXISTS' in capsys.readouterr().out
    assert output.read_text()=='keep original'


def test_prompt_schema_mode_avoids_native_union_and_keeps_contract_once():
    import json
    r=request();case=prepare(provider(),r,schema_mode='prompt')
    payload=case['payload']
    assert 'responseJsonSchema' not in payload['generationConfig']
    assert payload['generationConfig']['responseMimeType']=='application/json'
    contract=json.dumps(r['output_schema'],ensure_ascii=False,separators=(',',':'))
    assert payload['systemInstruction']['parts'][0]['text'].count(contract)==1
    assert payload['contents'][0]['parts'][0]['text']==r['messages'][1]['content']
    assert case['validation_schema']==r['output_schema']
    def post(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"database.schema"}'}]}}]}
    assert measure(provider(),case,10000,post)['status']=='SUCCESS'
    def bad(*args):return {'candidates':[{'finishReason':'STOP','content':{'parts':[{'text':'{"query":"invented"}'}]}}]}
    assert measure(provider(),case,10000,bad)['status']=='INVALID_JSON'