"""Public-result framing with synthetic JSONL commands; no Ansible or remote calls."""
from dataclasses import replace
import json,os,pwd,socket,struct,sys,threading,time
from pathlib import Path
import pytest
from aim_webgui.config import Settings
from aim_webgui.core.client import CoreClient
from aim_webgui.core.executor import Executor
from aim_webgui.core.jsonio import loads
from aim_webgui.core.reports import encoded
from aim_webgui.credentials import wire
from aim_webgui.errors import WebError
from evidence_fixtures import result,plan


def large_fixture(tmp_path, *, full_budget=False):
    decl={'protocol':'aim_output_v1','schema':'synthetic_rows_v1','scope':'per_host','required':True,'sensitivity':'safe',
          'max_bytes_per_host':1048576,'data_schema':{'type':'object','properties':{'rows':{'type':'array','maxItems':20000,
          'items':{'type':'string','maxLength':8192}}},'required':['rows'],'additionalProperties':False}}
    hosts=['test'+str(i)+'.example' for i in range(16 if full_budget else 4)]
    rows=['\u2603'*2600]*128 if full_budget else ['synthetic-\u2603'*90]*500
    final=result(decl,hosts,report_data={'rows':rows})
    assert 1024*1024<len(encoded(final))<16*1024*1024
    script=tmp_path/'wire_fixture.py';payload=tmp_path/'fixture-result.json';payload.write_bytes(encoded(final))
    script.write_text('''import os,sys,json
request=json.load(sys.stdin)
fd=int(sys.argv[sys.argv.index('--credentials-fd')+1])
with os.fdopen(fd) as stream: credentials=json.load(stream)
assert credentials['vault_password']=='SYNTHETIC-PRIVATE-ONLY'
assert 'credentials' not in request
r=json.load(open('''+repr(str(payload))+'''))
e={'event_version':'1.0','run_id':'fixture-run','sequence':1,'timestamp':'2026-09-20T10:00:00Z','kind':'result','status':'succeeded','result':r}
for value in [{'type':'event','event':e},{'type':'response','api_version':'1.0','ok':True,'result':r}]:
 raw=(json.dumps(value,ensure_ascii=False)+'\\n').encode()
 for i in range(0,len(raw),16381):os.write(1,raw[i:i+16381])
''')
    if full_budget:script.write_text(script.read_text().replace('ensure_ascii=False','ensure_ascii=True'))
    settings=Settings(core_transport='stdio',core_command=(sys.executable,str(script)),core_config=tmp_path/'synthetic.yml')
    return settings,decl,hosts,final


def test_large_result_stdio_and_duplicate_event_is_not_report_sample(tmp_path):
    settings,decl,hosts,expected=large_fixture(tmp_path);events=[]
    r=CoreClient(settings).request('execute',request=plan(decl,hosts)['core_request'],expected_revision='a'*64,
        credentials={'vault_password':'SYNTHETIC-PRIVATE-ONLY'},result_contract=decl,event_sink=events.append)
    assert r['operation_result']==expected['operation_result']
    assert len(events)==1 and 'operation_result'not in events[0] and 'result'not in events[0]


def test_large_result_through_executor_socket_and_unchanged_secret_limits(tmp_path):
    settings,decl,hosts,expected=large_fixture(tmp_path)
    path=tmp_path/'core.sock'
    who=pwd.getpwuid(os.geteuid()).pw_name
    settings=replace(settings,core_transport='unix',core_socket=path,core_executor_user=who,core_client_user=who)
    server=socket.socket(socket.AF_UNIX);server.bind(str(path));path.chmod(0o660);server.listen(1)
    executor=Executor(settings)
    def run():
        conn,_=server.accept();executor.handle(conn)
    thread=threading.Thread(target=run);thread.start()
    try:
        r=CoreClient(settings).request('execute',request=plan(decl,hosts)['core_request'],expected_revision='a'*64,
            result_contract=decl,credentials={'vault_password':'SYNTHETIC-PRIVATE-ONLY'})
        assert r['operation_result']==expected['operation_result']
    finally:thread.join(timeout=10);server.close()
    assert not thread.is_alive()
    assert wire.SECRET_LIMIT==8192 and wire.MAX_MESSAGE==1048576


@pytest.mark.parametrize('raw',[b'{"a":1,"a":2}',b'{"a":NaN}',b'['*49+b'0'+b']'*49,b'{"a":"\xff"}',b'{"a":'])
def test_strict_decoder_rejects_ambiguous_or_torn_json(raw):
    with pytest.raises((ValueError,UnicodeError)):loads(raw)


def test_frame_bounds_before_allocation_and_torn_frames():
    for body in (struct.pack('!I',33*1024*1024),struct.pack('!I',30)+b'{}'):
        a,b=socket.socketpair()
        try:
            a.sendall(body);a.shutdown(socket.SHUT_WR)
            with pytest.raises(ValueError):wire.receive(b,32*1024*1024,decoder=loads)
        finally:a.close();b.close()


def test_near_run_limit_with_escaped_unicode_and_repeated_final(tmp_path):
    settings,decl,hosts,expected=large_fixture(tmp_path,full_budget=True)
    assert 15*1024*1024<len(encoded(expected))<16*1024*1024
    r=CoreClient(settings).request('execute',request=plan(decl,hosts)['core_request'],expected_revision='a'*64,
        result_contract=decl,credentials={'vault_password':'SYNTHETIC-PRIVATE-ONLY'})
    assert r['operation_result']==expected['operation_result']