# Path: pygls/server.py
# class LanguageServer(Server):
#     """A class that represents Language server using Language Server Protocol.
# 
#     This class can be extended and it can be passed as a first argument to
#     registered commands/features.
# 
#     Args:
#         protocol_cls(LanguageServerProtocol): LSP or any subclass of it
#         max_workers(int, optional): Number of workers for `ThreadPool` and
#                                     `ThreadPoolExecutor`
#     """
# 
#     def __init__(self, loop=None, protocol_cls=LanguageServerProtocol, max_workers: int = 2):
#         if not issubclass(protocol_cls, LanguageServerProtocol):
#             raise TypeError('Protocol class should be subclass of LanguageServerProtocol')
#         super().__init__(protocol_cls, loop, max_workers)
# 
#     def apply_edit(self, edit: WorkspaceEdit, label: str = None) -> ApplyWorkspaceEditResponse:
#         """Sends apply edit request to the client."""
#         return self.lsp.apply_edit(edit, label)
# 
#     def command(self, command_name: str) -> Callable[[F], F]:
#         """Decorator used to register custom commands.
# 
#         Example:
#             @ls.command('myCustomCommand')
#             def my_cmd(ls, a, b, c):
#                 pass
#         """
#         return self.lsp.fm.command(command_name)
# 
#     @property
#     def client_capabilities(self) -> ClientCapabilities:
#         """Return client capabilities."""
#         return self.lsp.client_capabilities
# 
#     def feature(
#         self, feature_name: str, options: Optional[Any] = None,
#     ) -> Callable[[F], F]:
#         """Decorator used to register LSP features.
# 
#         Example:
#             @ls.feature('textDocument/completion', triggerCharacters=['.'])
#             def completions(ls, params: CompletionRequest):
#                 return CompletionList(False, [CompletionItem("Completion 1")])
#         """
#         return self.lsp.fm.feature(feature_name, options)
# 
#     def get_configuration(self, params: ConfigurationParams,
#                           callback: Optional[ConfigCallbackType] = None) -> Future:
#         """Gets the configuration settings from the client."""
#         return self.lsp.get_configuration(params, callback)
# 
#     def get_configuration_async(self, params: ConfigurationParams) -> asyncio.Future:
#         """Gets the configuration settings from the client. Should be called with `await`"""
#         return self.lsp.get_configuration_async(params)
# 
#     def log_trace(self, message: str, verbose: Optional[str] = None) -> None:
#         """Sends trace notification to the client."""
#         self.lsp.log_trace(message, verbose)
# 
#     @property
#     def progress(self) -> Progress:
#         """Gets the object to manage client's progress bar."""
#         return self.lsp.progress
# 
#     def publish_diagnostics(self, doc_uri: str, diagnostics: List[Diagnostic]):
#         """Sends diagnostic notification to the client."""
#         self.lsp.publish_diagnostics(doc_uri, diagnostics)
# 
#     def register_capability(self, params: RegistrationParams,
#                             callback: Optional[Callable[[], None]] = None) -> Future:
#         """Register a new capability on the client."""
#         return self.lsp.register_capability(params, callback)
# 
#     def register_capability_async(self, params: RegistrationParams) -> asyncio.Future:
#         """Register a new capability on the client. Should be called with `await`"""
#         return self.lsp.register_capability_async(params)
# 
#     def semantic_tokens_refresh(self, callback: Optional[Callable[[], None]] = None) -> Future:
#         """Request a refresh of all semantic tokens."""
#         return self.lsp.semantic_tokens_refresh(callback)
# 
#     def semantic_tokens_refresh_async(self) -> asyncio.Future:
#         """Request a refresh of all semantic tokens. Should be called with `await`"""
#         return self.lsp.semantic_tokens_refresh_async()
# 
#     def send_notification(self, method: str, params: object = None) -> None:
#         """Sends notification to the client."""
#         self.lsp.notify(method, params)
# 
#     @property
#     def server_capabilities(self) -> ServerCapabilities:
#         """Return server capabilities."""
#         return self.lsp.server_capabilities
# 
#     def show_document(self, params: ShowDocumentParams,
#                       callback: Optional[ShowDocumentCallbackType] = None) -> Future:
#         """Display a particular document in the user interface."""
#         return self.lsp.show_document(params, callback)
# 
#     def show_document_async(self, params: ShowDocumentParams) -> asyncio.Future:
#         """Display a particular document in the user interface. Should be called with `await`"""
#         return self.lsp.show_document_async(params)
# 
#     def show_message(self, message, msg_type=MessageType.Info) -> None:
#         """Sends message to the client to display message."""
#         self.lsp.show_message(message, msg_type)
# 
#     def show_message_log(self, message, msg_type=MessageType.Log) -> None:
#         """Sends message to the client's output channel."""
#         self.lsp.show_message_log(message, msg_type)
# 
#     def thread(self) -> Callable[[F], F]:
#         """Decorator that mark function to execute it in a thread."""
#         return self.lsp.thread()
# 
#     def unregister_capability(self, params: UnregistrationParams,
#                               callback: Optional[Callable[[], None]] = None) -> Future:
#         """Unregister a new capability on the client."""
#         return self.lsp.unregister_capability(params, callback)
# 
#     def unregister_capability_async(self, params: UnregistrationParams) -> asyncio.Future:
#         """Unregister a new capability on the client. Should be called with `await`"""
#         return self.lsp.unregister_capability_async(params)
# 
#     @property
#     def workspace(self) -> Workspace:
#         """Returns in-memory workspace."""
#         return self.lsp.workspace


import asyncio
import os
import pytest
from threading import Thread
from unittest.mock import Mock
from pygls.server import LanguageServer




@pytest.mark.asyncio
async def test_tcp_connection_lost():
    loop = asyncio.new_event_loop()


# Path: pygls/server.py
# class LanguageServer(Server):
#     """A class that represents Language server using Language Server Protocol.
# 
#     This class can be extended and it can be passed as a first argument to
#     registered commands/features.
# 
#     Args:
#         protocol_cls(LanguageServerProtocol): LSP or any subclass of it
#         max_workers(int, optional): Number of workers for `ThreadPool` and
#                                     `ThreadPoolExecutor`
#     """
# 
#     def __init__(self, loop=None, protocol_cls=LanguageServerProtocol, max_workers: int = 2):
#         if not issubclass(protocol_cls, LanguageServerProtocol):
#             raise TypeError('Protocol class should be subclass of LanguageServerProtocol')
#         super().__init__(protocol_cls, loop, max_workers)
# 
#     def apply_edit(self, edit: WorkspaceEdit, label: str = None) -> ApplyWorkspaceEditResponse:
#         """Sends apply edit request to the client."""
#         return self.lsp.apply_edit(edit, label)
# 
#     def command(self, command_name: str) -> Callable[[F], F]:
#         """Decorator used to register custom commands.
# 
#         Example:
#             @ls.command('myCustomCommand')
#             def my_cmd(ls, a, b, c):
#                 pass
#         """
#         return self.lsp.fm.command(command_name)
# 
#     @property
#     def client_capabilities(self) -> ClientCapabilities:
#         """Return client capabilities."""
#         return self.lsp.client_capabilities
# 
#     def feature(
#         self, feature_name: str, options: Optional[Any] = None,
#     ) -> Callable[[F], F]:
#         """Decorator used to register LSP features.
# 
#         Example:
#             @ls.feature('textDocument/completion', triggerCharacters=['.'])
#             def completions(ls, params: CompletionRequest):
#                 return CompletionList(False, [CompletionItem("Completion 1")])
#         """
#         return self.lsp.fm.feature(feature_name, options)
# 
#     def get_configuration(self, params: ConfigurationParams,
#                           callback: Optional[ConfigCallbackType] = None) -> Future:
#         """Gets the configuration settings from the client."""
#         return self.lsp.get_configuration(params, callback)
# 
#     def get_configuration_async(self, params: ConfigurationParams) -> asyncio.Future:
#         """Gets the configuration settings from the client. Should be called with `await`"""
#         return self.lsp.get_configuration_async(params)
# 
#     def log_trace(self, message: str, verbose: Optional[str] = None) -> None:
#         """Sends trace notification to the client."""
#         self.lsp.log_trace(message, verbose)
# 
#     @property
#     def progress(self) -> Progress:
#         """Gets the object to manage client's progress bar."""
#         return self.lsp.progress
# 
#     def publish_diagnostics(self, doc_uri: str, diagnostics: List[Diagnostic]):
#         """Sends diagnostic notification to the client."""
#         self.lsp.publish_diagnostics(doc_uri, diagnostics)
# 
#     def register_capability(self, params: RegistrationParams,
#                             callback: Optional[Callable[[], None]] = None) -> Future:
#         """Register a new capability on the client."""
#         return self.lsp.register_capability(params, callback)
# 
#     def register_capability_async(self, params: RegistrationParams) -> asyncio.Future:
#         """Register a new capability on the client. Should be called with `await`"""
#         return self.lsp.register_capability_async(params)
# 
#     def semantic_tokens_refresh(self, callback: Optional[Callable[[], None]] = None) -> Future:
#         """Request a refresh of all semantic tokens."""
#         return self.lsp.semantic_tokens_refresh(callback)
# 
#     def semantic_tokens_refresh_async(self) -> asyncio.Future:
#         """Request a refresh of all semantic tokens. Should be called with `await`"""
#         return self.lsp.semantic_tokens_refresh_async()
# 
#     def send_notification(self, method: str, params: object = None) -> None:
#         """Sends notification to the client."""
#         self.lsp.notify(method, params)
# 
#     @property
#     def server_capabilities(self) -> ServerCapabilities:
#         """Return server capabilities."""
#         return self.lsp.server_capabilities
# 
#     def show_document(self, params: ShowDocumentParams,
#                       callback: Optional[ShowDocumentCallbackType] = None) -> Future:
#         """Display a particular document in the user interface."""
#         return self.lsp.show_document(params, callback)
# 
#     def show_document_async(self, params: ShowDocumentParams) -> asyncio.Future:
#         """Display a particular document in the user interface. Should be called with `await`"""
#         return self.lsp.show_document_async(params)
# 
#     def show_message(self, message, msg_type=MessageType.Info) -> None:
#         """Sends message to the client to display message."""
#         self.lsp.show_message(message, msg_type)
# 
#     def show_message_log(self, message, msg_type=MessageType.Log) -> None:
#         """Sends message to the client's output channel."""
#         self.lsp.show_message_log(message, msg_type)
# 
#     def thread(self) -> Callable[[F], F]:
#         """Decorator that mark function to execute it in a thread."""
#         return self.lsp.thread()
# 
#     def unregister_capability(self, params: UnregistrationParams,
#                               callback: Optional[Callable[[], None]] = None) -> Future:
#         """Unregister a new capability on the client."""
#         return self.lsp.unregister_capability(params, callback)
# 
#     def unregister_capability_async(self, params: UnregistrationParams) -> asyncio.Future:
#         """Unregister a new capability on the client. Should be called with `await`"""
#         return self.lsp.unregister_capability_async(params)
# 
#     @property
#     def workspace(self) -> Workspace:
#         """Returns in-memory workspace."""
#         return self.lsp.workspace


# Path: tests/test_server_connection.py
import asyncio
import os
import pytest
from threading import Thread
from unittest.mock import Mock
from pygls.server import LanguageServer





@pytest.mark.asyncio
async def test_tcp_connection_lost():
    loop = asyncio.new_event_loop()


    server = LanguageServer(loop=loop)