"""Exercise ready checks through real WebSockets against an isolated local server."""
import json
import os
from pathlib import Path
import shutil
import socket
import subprocess
import time

import pytest
from websockets.sync.client import connect


@pytest.fixture
def server_url():
    with socket.socket() as reservation:
        reservation.bind(('127.0.0.1', 0))
        port = reservation.getsockname()[1]
    bun = shutil.which('bun')
    assert bun, 'Bun is required to test the local server'
    process = subprocess.Popen(
        [bun, 'run', 'server.ts'], cwd=Path(__file__).resolve().parent.parent,
        env={**os.environ, 'PORT': str(port)}, stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE, creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
    try:
        deadline = time.monotonic() + 5
        while True:
            if process.poll() is not None:
                pytest.fail(process.stderr.read().decode())
            try:
                with socket.create_connection(('127.0.0.1', port), timeout=.1):
                    break
            except OSError:
                if time.monotonic() >= deadline:
                    pytest.fail('Local server did not start')
                time.sleep(.05)
        yield f'ws://127.0.0.1:{port}/ws'
    finally:
        process.terminate()
        process.wait(timeout=5)
        process.stderr.close()


def send(ws, **message):
    ws.send(json.dumps(message))


def receive(ws, kind, predicate=lambda message: True):
    deadline = time.monotonic() + 3
    while True:
        message = json.loads(ws.recv(timeout=max(.01, deadline - time.monotonic())))
        if message['type'] == kind and predicate(message):
            return message


def test_room_ready_check_lifecycle(server_url):
    file_info = {'name': 'test.mkv', 'size': 1234, 'duration': 60}
    with connect(server_url) as host, connect(server_url) as guest:
        send(host, type='create_room', username='Host', fileInfo=file_info)
        code = receive(host, 'room_created')['code']
        send(guest, type='join_room', username='Guest', code=code)
        receive(guest, 'room_file_check')
        send(guest, type='confirm_join', fileInfo=file_info)
        receive(guest, 'room_joined')
        receive(host, 'user_joined')
        send(host, type='sync', action='play', position=12)
        receive(guest, 'sync')
        send(host, type='ready_check', action='begin')
        pause = receive(host, 'sync')
        assert pause['action'] == 'pause' and pause['position'] >= 12
        check = receive(host, 'ready_check')
        check_id = check['id']
        assert check['requestedBy'] == 'Host'
        assert not any(p['ready'] for p in check['participants'])
        assert receive(guest, 'ready_check')['id'] == check_id

        # Unconfirmed peers cannot create or vote in a check.
        with connect(server_url) as pending:
            send(pending, type='join_room', username='Pending', code=code)
            receive(pending, 'room_file_check')
            send(pending, type='ready_check', action='begin')
            send(pending, type='ready_check', action='vote', id=check_id, ready=True)
        send(host, type='ready_check', action='start', id=check_id)
        send(host, type='ready_check', action='vote', id=check_id, ready=True)
        check = receive(host, 'ready_check')
        assert check['active'] and sum(p['ready'] for p in check['participants']) == 1
        receive(guest, 'ready_check')
        send(host, type='ready_check', action='begin')
        repeated = receive(host, 'ready_check')
        assert repeated['id'] == check_id
        assert sum(p['ready'] for p in repeated['participants']) == 1
        receive(guest, 'ready_check')
        send(guest, type='ready_check', action='vote', id=check_id, ready=True)
        assert all(p['ready'] for p in receive(host, 'ready_check')['participants'])
        receive(guest, 'ready_check')

        # Non-initiators cannot start. Stale IDs cannot change votes.
        send(guest, type='ready_check', action='start', id=check_id)
        send(guest, type='ready_check', action='vote', id='stale', ready=False)
        send(guest, type='ready_check', action='begin')
        assert all(p['ready'] for p in receive(host, 'ready_check')['participants'])
        receive(guest, 'ready_check')

        # Dropping a connection revokes readiness and blocks starting.
        guest.close()
        disconnected = receive(host, 'ready_check')
        guest_state = next(p for p in disconnected['participants'] if p['username'] == 'Guest')
        assert not guest_state['connected'] and not guest_state['ready']
        send(host, type='ready_check', action='start', id=check_id)
        with connect(server_url) as rejoined:
            send(rejoined, type='rejoin_room', username='Guest', code=code)
            restored = receive(rejoined, 'room_rejoined')['readyCheck']
            assert restored['id'] == check_id
            assert not next(p for p in restored['participants'] if p['username'] == 'Guest')['ready']
            receive(host, 'ready_check')
            receive(rejoined, 'ready_check')
            send(rejoined, type='ready_check', action='vote', id=check_id, ready=True)
            receive(host, 'ready_check')
            receive(rejoined, 'ready_check')
            send(host, type='ready_check', action='start', id=check_id)
            assert receive(host, 'sync')['action'] == 'play'
            assert receive(rejoined, 'sync')['action'] == 'play'
            assert not receive(host, 'ready_check')['active']
            assert not receive(rejoined, 'ready_check')['active']

            send(host, type='ready_check', action='begin')
            new_check = receive(host, 'ready_check')
            assert new_check['id'] != check_id
            assert not any(p['ready'] for p in new_check['participants'])
            receive(rejoined, 'ready_check')
            send(host, type='ready_check', action='cancel', id=new_check['id'])
            assert not receive(host, 'ready_check')['active']
            assert not receive(rejoined, 'ready_check')['active']