import asyncio
import os
from threading import Thread
from unittest.mock import Mock

import pytest

from pygls import IS_PYODIDE
from pygls.server import LanguageServer


@pytest.mark.asyncio
@pytest.mark.skipif(IS_PYODIDE, reason='threads are not available in pyodide.')
async def test_tcp_connection_lost():
    loop = asyncio.new_event_loop()

    server = LanguageServer('pygls-test', 'v1', loop=loop)

    server.lsp.connection_made = Mock()
    server.lsp.connection_lost = Mock()

    # Run the server over TCP in a separate thread
    server_thread = Thread(
        target=server.start_tcp,
        args=(
            "127.0.0.1",
            0,
        ),
    )
    server_thread.daemon = True
    server_thread.start()

    # Wait for server to be ready
    while server._server is None:
        await asyncio.sleep(0.5)

    # Simulate client's connection
    port = server._server.sockets[0].getsockname()[1]
    reader, writer = await asyncio.open_connection("127.0.0.1", port)
    await asyncio.sleep(1)

    assert server.lsp.connection_made.called

    # Socket is closed (client's process is terminated)
    writer.close()
    await asyncio.sleep(1)

    assert server.lsp.connection_lost.called


@pytest.mark.asyncio
@pytest.mark.skipif(IS_PYODIDE, reason='threads are not available in pyodide.')
async def test_io_connection_lost():
    # Client to Server pipe.
    csr, csw = os.pipe()
    # Server to client pipe.
    scr, scw = os.pipe()

    server = LanguageServer('pygls-test', 'v1', loop=asyncio.new_event_loop())
    server.lsp.connection_made = Mock()
    server_thread = Thread(
        target=server.start_io,
        args=(
            os.fdopen(csr, "rb"),
            os.fdopen(scw, "wb")
        )
    )
    server_thread.daemon = True
    server_thread.start()

    # Wait for server to be ready
    while not server.lsp.connection_made.called:
        await asyncio.sleep(0.5)

    # Pipe is closed (client's process is terminated)
    os.close(csw)
    server_thread.join()

@@ -8,16 +8,28 @@
 asyncio%0A
+import json%0A
 import o
@@ -30,16 +30,16 @@
 port os%0A
-
 from thr
@@ -176,16 +176,130 @@
 Server%0A%0A
+try:%0A    import websockets%0A%0A    WEBSOCKETS_AVAILABLE = True%0Aexcept ImportError:%0A    WEBSOCKETS_AVAILABLE = False%0A%0A
 %0A@pytest
@@ -2086,24 +2086,24 @@
 terminated)%0A
-
     os.close
@@ -2133,8 +2133,1580 @@
 .join()%0A
+%0A%0A@pytest.mark.asyncio%0A@pytest.mark.skipif(%0A    IS_PYODIDE or not WEBSOCKETS_AVAILABLE,%0A    reason=%22threads are not available in pyodide%22,%0A)%0Aasync def test_ws_server():%0A    %22%22%22Smoke test to ensure we can send/receive messages over websockets%22%22%22%0A%0A    loop = asyncio.new_event_loop()%0A    server = LanguageServer(loop=loop)%0A%0A    # Run the server over Websockets in a separate thread%0A    server_thread = Thread(%0A        target=server.start_ws,%0A        args=(%0A            %22127.0.0.1%22,%0A            0,%0A        ),%0A    )%0A    server_thread.daemon = True%0A    server_thread.start()%0A%0A    # Wait for server to be ready%0A    while server._server is None:%0A        await asyncio.sleep(0.5)%0A%0A    port = server._server.sockets%5B0%5D.getsockname()%5B1%5D%0A    # Simulate client's connection%0A    async with websockets.connect(f%22ws://127.0.0.1:%7Bport%7D%22) as connection:%0A%0A        # Send an 'initialize' request%0A        msg = dict(%0A            jsonrpc=%222.0%22, id=1, method=%22initialize%22, params=dict(capabilities=dict())%0A        )%0A        await connection.send(json.dumps(msg))%0A%0A        response = await connection.recv()%0A        assert %22result%22 in response%0A%0A        # Shut the server down%0A        msg = dict(%0A            jsonrpc=%222.0%22, id=2, method=%22shutdown%22, params=dict(capabilities=dict())%0A        )%0A        await connection.send(json.dumps(msg))%0A%0A        response = await connection.recv()%0A        assert %22result%22 in response%0A%0A        # Finally, tell it to exit%0A        msg = dict(jsonrpc=%222.0%22, id=2, method=%22exit%22, params=None)%0A        await connection.send(json.dumps(msg))%0A%0A    server_thread.join()%0A

Add test for websocket servers