Language Client¶
The pytest-lsp LanguageClient supports the following LSP requests and notifications.
textDocument/publishDiagnostics¶
The client maintains a record of any diagnostics published by the server in a dictionary indexed by a text document’s uri.
test_server.py
@pytest.mark.asyncio
async def test_diagnostics(client: LanguageClient):
"""Ensure that the server implements diagnostics correctly."""
test_uri = "file:///path/to/file.txt"
client.text_document_did_open(
DidOpenTextDocumentParams(
text_document=TextDocumentItem(
uri=test_uri,
language_id="plaintext",
version=1,
text="The file's contents",
)
)
)
# Wait for the server to publish its diagnostics
await client.wait_for_notification(TEXT_DOCUMENT_PUBLISH_DIAGNOSTICS)
assert test_uri in client.diagnostics
assert client.diagnostics[test_uri][0].message == "There is an error here."
Note
While the client has the (rather useful!) ability to wait_for_notification() messages from the server, this is not something covered by the LSP Specification.
server.py
@server.feature(TEXT_DOCUMENT_DID_OPEN)
def did_open(ls: LanguageServer, params: DidOpenTextDocumentParams):
ls.publish_diagnostics(
params.text_document.uri,
[
Diagnostic(
message="There is an error here.",
range=Range(
start=Position(line=1, character=1),
end=Position(line=1, character=10),
),
)
],
)
window/logMessage¶
Any window/logMessage notifications sent from the server will be accessible via the client’s log_messages attribute.
test_server.py
@pytest.mark.asyncio
async def test_completions(client: LanguageClient):
results = await client.text_document_completion_async(
params=CompletionParams(
position=Position(line=1, character=0),
text_document=TextDocumentIdentifier(uri="file:///path/to/file.txt"),
)
)
assert results is not None
if isinstance(results, CompletionList):
items = results.items
else:
items = results
labels = [item.label for item in items]
assert labels == [f"item-{i}" for i in range(10)]
for idx, log_message in enumerate(client.log_messages):
assert log_message.message == f"Suggesting item {idx}"
server.py
@server.feature(TEXT_DOCUMENT_COMPLETION)
def completion(ls: LanguageServer, params: CompletionParams):
items = []
for i in range(10):
ls.show_message_log(f"Suggesting item {i}")
items.append(CompletionItem(label=f"item-{i}"))
return items
If a test case fails pytest-lsp will also include any captured log messages in the error report
================================== test session starts ====================================
platform linux -- Python 3.11.2, pytest-7.2.0, pluggy-1.0.0
rootdir: /..., configfile: tox.ini
plugins: typeguard-2.13.3, asyncio-0.20.2, lsp-0.2.1
asyncio: mode=Mode.AUTO
collected 1 item
test_server.py F [100%]
======================================== FAILURES =========================================
____________________________________ test_completions _____________________________________
client = <pytest_lsp.client.LanguageClient object at 0x7f38f144a690>
...
E assert False
test_server.py:35: AssertionError
---------------------------- Captured window/logMessages call -----------------------------
LOG: Suggesting item 0
LOG: Suggesting item 1
LOG: Suggesting item 2
LOG: Suggesting item 3
LOG: Suggesting item 4
LOG: Suggesting item 5
LOG: Suggesting item 6
LOG: Suggesting item 7
LOG: Suggesting item 8
LOG: Suggesting item 9
================================ short test summary info ==================================
FAILED test_server.py::test_completions - assert False
=================================== 1 failed in 1.02s =====================================
window/showDocument¶
Similar to window/logMessage above, the client records any window/showDocument notifications and are accessible via its shown_documents attribute.
test_server.py
@pytest.mark.asyncio
async def test_completions(client: LanguageClient):
test_uri = "file:///path/to/file.txt"
results = await client.text_document_completion_async(
params=CompletionParams(
position=Position(line=1, character=0),
text_document=TextDocumentIdentifier(uri=test_uri),
)
)
assert results is not None
if isinstance(results, CompletionList):
items = results.items
else:
items = results
labels = [item.label for item in items]
assert labels == [f"item-{i}" for i in range(10)]
assert client.shown_documents[0].uri == test_uri
server.py
@server.feature(TEXT_DOCUMENT_COMPLETION)
async def completion(ls: LanguageServer, params: CompletionParams):
items = []
await ls.show_document_async(ShowDocumentParams(uri=params.text_document.uri))
for i in range(10):
items.append(CompletionItem(label=f"item-{i}"))
return items
window/showMessage¶
Similar to window/logMessage above, the client records any window/showMessage notifications and are accessible via its messages attribute.
test_server.py
@pytest.mark.asyncio
async def test_completions(client: LanguageClient):
results = await client.text_document_completion_async(
params=CompletionParams(
position=Position(line=1, character=0),
text_document=TextDocumentIdentifier(uri="file:///path/to/file.txt"),
)
)
assert results is not None
if isinstance(results, CompletionList):
items = results.items
else:
items = results
labels = [item.label for item in items]
assert labels == [f"item-{i}" for i in range(10)]
for idx, shown_message in enumerate(client.messages):
assert shown_message.message == f"Suggesting item {idx}"
server.py
@server.feature(TEXT_DOCUMENT_COMPLETION)
def completion(ls: LanguageServer, params: CompletionParams):
items = []
for i in range(10):
ls.show_message(f"Suggesting item {i}")
items.append(CompletionItem(label=f"item-{i}"))
return items
window/workDoneProgress/create¶
The client can respond to window/workDoneProgress/create requests and handle associated $/progress notifications
test_server.py
@pytest.mark.asyncio
async def test_progress(client: LanguageClient):
result = await client.workspace_execute_command_async(
params=types.ExecuteCommandParams(command="do.progress")
)
assert result == "a result"
progress = client.progress_reports["a-token"]
assert progress == [
types.WorkDoneProgressBegin(title="Indexing", percentage=0),
types.WorkDoneProgressReport(message="25%", percentage=25),
types.WorkDoneProgressReport(message="50%", percentage=50),
types.WorkDoneProgressReport(message="75%", percentage=75),
types.WorkDoneProgressEnd(message="Finished"),
]
server.py
@server.command("do.progress")
async def do_progress(ls: LanguageServer, *args):
token = "a-token"
await ls.progress.create_async(token)
# Begin
ls.progress.begin(
token,
types.WorkDoneProgressBegin(title="Indexing", percentage=0),
)
# Report
for i in range(1, 4):
ls.progress.report(
token,
types.WorkDoneProgressReport(message=f"{i * 25}%", percentage=i * 25),
)
# End
ls.progress.end(token, types.WorkDoneProgressEnd(message="Finished"))
return "a result"
workspace/configuration¶
The client can respond to workspace/configuration requests.
The client supports settings different configuration values for different scope_uris as well as getting/setting specific configuration sections.
However, to keep the implementation simple the client will not fallback to more general configuration scopes if it cannot find a value in the requested scope.
See the documentation on set_configuration() and get_configuration() for details
test_server.py
@pytest.mark.asyncio
async def test_configuration(client: LanguageClient):
global_config = {"values": {"a": 42, "c": 4}}
workspace_uri = "file://workspace/file.txt"
workspace_config = {"a": 1, "c": 1}
client.set_configuration(global_config)
client.set_configuration(
workspace_config, section="values", scope_uri=workspace_uri
)
result = await client.workspace_execute_command_async(
params=types.ExecuteCommandParams(command="server.configuration")
)
assert result == 5
server.py
@server.command("server.configuration")
async def configuration(ls: LanguageServer, *args):
results = await ls.get_configuration_async(
types.WorkspaceConfigurationParams(
items=[
types.ConfigurationItem(scope_uri="file://workspace/file.txt"),
types.ConfigurationItem(section="not.found"),
types.ConfigurationItem(section="values.c"),
]
)
)
a = results[0]["values"]["a"]
assert results[1] is None
c = results[2]
return a + c