import json
from pathlib import Path
import socket
import struct
import tempfile
import time
import unittest
from unittest.mock import patch

from witcher_devkit import native, wire


def response(request=1, **extra):
    return dict(session='test',request=request,build='build',ok=True,generation=10,data={},**extra)


def output(value, size=25):
    data=json.dumps(value,ensure_ascii=False)
    parts=[data[i:i+size] for i in range(0,len(data),size)]
    return ''.join(f'PDPART:test:{value["request"]}:{i}:{p}\n' for i,p in enumerate(parts))+f'PDEND:test:{value["request"]}:{len(parts)}\n'


class ProtocolTests(unittest.TestCase):
    def test_utf8_and_utf16(self):
        packet=wire.encode_packet(['Geralt — Łódź',-4,27])
        self.assertEqual(wire.decode_payload(packet[4:-2]),['Geralt — Łódź',-4,27])
        s='Geralt 🐺'; data=s.encode('utf-16-be')
        self.assertEqual(wire.decode_payload(b'\x9c\x16\x81\x16'+struct.pack('>H',len(data)//2)+data),[s])

    def test_fragmented_receive(self):
        packet=wire.encode_packet(['Remote',1,'value'])
        class Fragmented:
            def __init__(self):self.data=packet
            def settimeout(self,t):pass
            def recv(self,n):
                result=self.data[:min(n,1)];self.data=self.data[len(result):];return result
        c=wire.Connection.__new__(wire.Connection);c.socket=Fragmented();c.deadline=time.monotonic()+1
        self.assertEqual(c.receive(),['Remote',1,'value'])

    def test_bad_frames_and_bounds(self):
        for payload in (b'\x81\x32\x00',b'\xac\x08\x81\x16\x00\x05x',b'\x00\x00'):
            with self.assertRaises(ValueError):wire.decode_payload(payload)
        with self.assertRaises(ValueError):wire.encode_packet(['x'*32768])
        with self.assertRaises(ValueError):wire.encode_packet(['x'*30000]*3)

    def test_committed_chunk_identity(self):
        r=response();self.assertEqual(native.parse_reply(output(r),'test',1),r)
        for bad in (output(r).replace('PDEND','BROKEN'),output(r).replace('PDPART:test:1:0:','PDPART:other:1:0:'),output(r)+'PDEND:test:1:1\n',output(r)+output(r)):
            with self.assertRaises((RuntimeError,ValueError)):native.parse_reply(bad,'test',1)

    def test_no_injection_or_unbounded_input(self):
        for args in ('0|0|0|0|AttackLight|2\');exit()','0|0|0|0|AttackLight|121','nan|0|0|0|None|60','2|0|0|0|None|60','0|0|0|0|Bogus|60'):
            with self.assertRaises(ValueError):native.validate('input',args)
        native.validate('input','0|1|-.25|0|AttackLight|60')
        with self.assertRaises(ValueError):native.validate('ui_click',"resume');exit()")
        for args in ('0|0|0|0|1024|0|0|50','0|0|0|0|0|0|256|50','0|0|0|0|0|0|0|10001'):
            with self.assertRaises(ValueError):native.validate('pad',args)

    def test_late_broadcast_reply_is_skipped(self):
        c=wire.Connection.__new__(wire.Connection);c.timeout=1
        c.send=lambda _:None
        packets=iter([[0x12345678,-2129264632,'PDPART:test:1:0:old'],
                      [0x12345678,-2129264632,'PDPART:test:2:0:current']])
        c.receive=lambda:next(packets)
        result,all_packets=c.execute('snapshot',markers=['PDPART:test:2:0:'])
        self.assertEqual(result[2],'PDPART:test:2:0:current')
        self.assertEqual(len(all_packets),2)


class SessionTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
        self.path=Path(self.tmp.name)/'session.json'
        self.s={'session':'test','request':0,'build':'build','process':{'pid':1},'uncertain':None}
        self.path.write_text(json.dumps(self.s))
        self.patches=[patch.object(native,'ROOT',self.path.parent),patch.object(native,'STATE_ROOT',self.path.parent),patch.object(native,'STATE',self.path),patch.object(native.legacy,'process_identity',return_value={'pid':1}),patch.object(native,'record')]
        for p in self.patches:p.start();self.addCleanup(p.stop)

    def test_uncertain_mutation_is_not_replayed(self):
        class Failed:
            def execute(self,expression,markers=None):raise TimeoutError('lost ack')
        with self.assertRaises(TimeoutError):native.call('pause',connection=Failed())
        self.assertEqual(json.loads(self.path.read_text())['uncertain']['command'],'pause')
        with self.assertRaisesRegex(RuntimeError,'uncertain'):native.call('pause',connection=Failed())
        self.assertEqual(json.loads(self.path.read_text())['request'],1)

    def test_read_does_not_clear_uncertain_mutation(self):
        self.s['uncertain']={'command':'pause','request':0};self.path.write_text(json.dumps(self.s))
        class Reply:
            def execute(self,expression,markers=None):return [0,0,output(response(),size=9999)],[]
        native.call('snapshot',connection=Reply())
        self.assertIsNotNone(json.loads(self.path.read_text())['uncertain'])

    def test_cached_paging_executes_original_only_once(self):
        calls=[];r=response();r['data']={'long':'x'*3000}
        lines=output(r,size=400).splitlines(keepends=True)
        class Reply:
            timeout=5
            def execute(self,expression,markers=None):
                calls.append(expression)
                if expression.startswith('pathdevkit('):return [0,0,lines[0]+lines[-1]],[]
                index=int(expression.rsplit(',',1)[1][:-1]);return [0,0,lines[index]],[]
        self.assertEqual(native.call('snapshot',connection=Reply()),r)
        self.assertEqual(sum(x.startswith('pathdevkit(') for x in calls),1)
        self.assertGreater(len(calls),1)

    def test_build_drift_keeps_mutation_uncertain(self):
        r=response();r['build']='unexpected'
        class Reply:
            def execute(self,expression,markers=None):return [0,0,output(r,size=9999)],[]
        with self.assertRaisesRegex(RuntimeError,'build'):native.call('pause',connection=Reply())
        self.assertIsNotNone(json.loads(self.path.read_text())['uncertain'])

    def test_reused_process_is_rejected(self):
        with patch.object(native.legacy,'process_identity',return_value={'pid':1,'start':'different'}):
            with self.assertRaisesRegex(RuntimeError,'ended'):native.state()


if __name__=='__main__':unittest.main()