776 lines
32 KiB
Python
776 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""Gate 0 LSP probe: framing selfcheck and lifecycle checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import queue
|
|
import random
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def encode_message(message: dict[str, Any]) -> bytes:
|
|
body = json.dumps(message, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
return f"Content-Length: {len(body)}\r\n\r\n".encode("ascii") + body
|
|
|
|
|
|
def selfcheck() -> int:
|
|
cases = [
|
|
{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
|
|
{"jsonrpc": "2.0", "id": 2, "method": "x", "params": {"text": "中文🙂"}},
|
|
]
|
|
for message in cases:
|
|
framed = encode_message(message)
|
|
header, _, body = framed.partition(b"\r\n\r\n")
|
|
assert header.startswith(b"Content-Length: "), header
|
|
declared = int(header.split(b":", 1)[1].strip())
|
|
assert declared == len(body), (declared, len(body))
|
|
decoded = json.loads(body.decode("utf-8"))
|
|
assert decoded == message, (decoded, message)
|
|
print("selfcheck: OK")
|
|
return 0
|
|
|
|
|
|
class Reader(threading.Thread):
|
|
def __init__(self, stream) -> None:
|
|
super().__init__(daemon=True)
|
|
self.stream = stream
|
|
self.messages: queue.Queue[dict[str, Any]] = queue.Queue()
|
|
self.error: str | None = None
|
|
|
|
def run(self) -> None:
|
|
try:
|
|
while True:
|
|
headers: dict[str, str] = {}
|
|
while True:
|
|
line = self.stream.readline()
|
|
if not line:
|
|
return
|
|
if line in (b"\r\n", b"\n"):
|
|
break
|
|
key, _, value = line.decode("ascii").partition(":")
|
|
headers[key.strip().lower()] = value.strip()
|
|
length = int(headers["content-length"])
|
|
body = self.stream.read(length)
|
|
if len(body) != length:
|
|
self.error = "short body"
|
|
return
|
|
self.messages.put(json.loads(body.decode("utf-8")))
|
|
except Exception as exc: # pragma: no cover - diagnostics path
|
|
self.error = str(exc)
|
|
|
|
def recv(self, timeout: float = 5.0) -> dict[str, Any]:
|
|
try:
|
|
return self.messages.get(timeout=timeout)
|
|
except queue.Empty as exc:
|
|
raise TimeoutError(f"no response within {timeout}s") from exc
|
|
|
|
|
|
class Client:
|
|
def __init__(self, exe: str) -> None:
|
|
self.proc = subprocess.Popen(
|
|
[exe, "--lsp"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=None,
|
|
)
|
|
assert self.proc.stdin is not None and self.proc.stdout is not None
|
|
self.reader = Reader(self.proc.stdout)
|
|
self.reader.start()
|
|
self.next_id = 0
|
|
|
|
def notify(self, method: str, params: dict[str, Any] | None = None) -> None:
|
|
message: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
|
|
if params is not None:
|
|
message["params"] = params
|
|
self.proc.stdin.write(encode_message(message))
|
|
self.proc.stdin.flush()
|
|
|
|
def request(self, method: str, params: dict[str, Any] | None = None,
|
|
timeout: float = 5.0) -> dict[str, Any]:
|
|
self.next_id += 1
|
|
message: dict[str, Any] = {"jsonrpc": "2.0", "id": self.next_id, "method": method}
|
|
if params is not None:
|
|
message["params"] = params
|
|
self.proc.stdin.write(encode_message(message))
|
|
self.proc.stdin.flush()
|
|
while True:
|
|
response = self.reader.recv(timeout=timeout)
|
|
if "id" not in response:
|
|
continue
|
|
if response.get("id") != self.next_id:
|
|
raise AssertionError(f"id mismatch: {response}")
|
|
return response
|
|
|
|
def wait_notification(self, method: str) -> dict[str, Any]:
|
|
while True:
|
|
message = self.reader.recv()
|
|
if message.get("method") == method:
|
|
return message
|
|
|
|
def close(self) -> int:
|
|
self.notify("exit")
|
|
return self.proc.wait(timeout=5)
|
|
|
|
|
|
def lifecycle(exe: str) -> int:
|
|
client = Client(exe)
|
|
# A health probe must answer before initialize; plugins use it to confirm the server is alive.
|
|
for method in ("$/ping", "viper/ping", "ping"):
|
|
ping = client.request(method, {})
|
|
ping_result = ping.get("result")
|
|
if not isinstance(ping_result, dict) or ping_result.get("status") != "ok":
|
|
raise AssertionError(f"{method} did not return an ok result: {ping}")
|
|
if ping_result.get("server") != "viperc-lsp":
|
|
raise AssertionError(f"{method} returned the wrong server name: {ping}")
|
|
|
|
response = client.request("initialize", {"processId": os.getpid(), "capabilities": {}})
|
|
result = response.get("result")
|
|
if not isinstance(result, dict):
|
|
raise AssertionError(f"initialize result is not an object: {response}")
|
|
capabilities = result.get("capabilities")
|
|
if not isinstance(capabilities, dict) or capabilities.get("positionEncoding") != "utf-16":
|
|
raise AssertionError(f"missing utf-16 positionEncoding: {response}")
|
|
server_info = result.get("serverInfo")
|
|
if not isinstance(server_info, dict) or server_info.get("name") != "viperc-lsp":
|
|
raise AssertionError(f"unexpected serverInfo: {response}")
|
|
|
|
client.notify("initialized", {})
|
|
|
|
# Unsupported features must not surface as protocol errors to the client.
|
|
unknown = client.request("unknown/method", {})
|
|
if "error" in unknown:
|
|
raise AssertionError(f"unknown method returned an error: {unknown}")
|
|
if "result" not in unknown or unknown.get("result") is not None:
|
|
raise AssertionError(f"unknown method did not return a null result: {unknown}")
|
|
|
|
unsupported = client.request(
|
|
"textDocument/formatting",
|
|
{"textDocument": {"uri": "file:///unsupported.vp"}, "options": {"tabSize": 4, "insertSpaces": True}},
|
|
)
|
|
if "error" in unsupported or unsupported.get("result") is not None:
|
|
raise AssertionError(f"unsupported feature should answer null: {unsupported}")
|
|
|
|
client.notify("unknown/notification", {})
|
|
ping = client.request("$/ping", {})
|
|
if ping.get("result", {}).get("status") != "ok":
|
|
raise AssertionError(f"server stopped answering after an unknown notification: {ping}")
|
|
|
|
shutdown = client.request("shutdown")
|
|
if shutdown.get("result") is not None:
|
|
raise AssertionError(f"shutdown result is not null: {shutdown}")
|
|
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
print("lifecycle: OK")
|
|
print("ping: OK")
|
|
return 0
|
|
|
|
|
|
def documents(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
root_uri = root.as_uri()
|
|
uri = (root / "Tools" / "lsp" / "fixtures" / "gate1.vp").as_uri()
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root_uri, "capabilities": {}},
|
|
)
|
|
result = response.get("result")
|
|
if not isinstance(result, dict):
|
|
raise AssertionError(f"initialize result is not an object: {response}")
|
|
capabilities = result.get("capabilities")
|
|
if not isinstance(capabilities, dict):
|
|
raise AssertionError(f"missing capabilities: {response}")
|
|
if capabilities.get("positionEncoding") != "utf-16":
|
|
raise AssertionError(f"positionEncoding is not utf-16: {response}")
|
|
if capabilities.get("textDocumentSync") != 2:
|
|
raise AssertionError(f"textDocumentSync is not incremental: {response}")
|
|
client.notify("initialized", {})
|
|
|
|
text = "def f(a: int) -> int:\r\n return a\r\n\r\n# 中文🙂\r\nx: int = 1\r\n"
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
|
|
stats = client.request("viper/lspStats").get("result")
|
|
if not isinstance(stats, dict) or stats.get("documentCount") != 1:
|
|
raise AssertionError(f"didOpen was not stored: {stats}")
|
|
source_dir = str(stats.get("sourceDir", "")).replace("\\", "/")
|
|
if not source_dir.endswith("ViperC/App"):
|
|
raise AssertionError(f"workspace sourceDir is wrong: {stats}")
|
|
if "includes" not in str(stats.get("includes", "")):
|
|
raise AssertionError(f"workspace includes is wrong: {stats}")
|
|
if not str(stats.get("project", "")).endswith("project.vpj"):
|
|
raise AssertionError(f"workspace project is wrong: {stats}")
|
|
|
|
doc = client.request("viper/lspDocument", {"uri": uri}).get("result")
|
|
if not isinstance(doc, dict) or doc.get("text") != text or doc.get("version") != 1:
|
|
raise AssertionError(f"initial document text/version mismatch: {doc}")
|
|
|
|
lines = text.splitlines()
|
|
choices: list[tuple[int, int]] = []
|
|
for line_no, line_text in enumerate(lines):
|
|
offsets = [0]
|
|
units = 0
|
|
for ch in line_text:
|
|
units += 2 if ord(ch) > 0xFFFF else 1
|
|
offsets.append(units)
|
|
for character in offsets:
|
|
choices.append((line_no, character))
|
|
random.seed(1234)
|
|
for _ in range(10000):
|
|
line_no, character = random.choice(choices)
|
|
result = client.request(
|
|
"viper/lspPosition",
|
|
{"uri": uri, "line": line_no, "character": character},
|
|
).get("result")
|
|
if not isinstance(result, dict) or result.get("line") != line_no or result.get("character") != character:
|
|
raise AssertionError(
|
|
f"UTF-16 round-trip failed for ({line_no},{character}): {result}"
|
|
)
|
|
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": 2},
|
|
"contentChanges": [
|
|
{
|
|
"range": {
|
|
"start": {"line": 1, "character": 11},
|
|
"end": {"line": 1, "character": 12},
|
|
},
|
|
"text": "a + 1",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
doc = client.request("viper/lspDocument", {"uri": uri}).get("result")
|
|
if not isinstance(doc, dict) or "return a + 1" not in doc.get("text", ""):
|
|
raise AssertionError(f"incremental ASCII edit failed: {doc}")
|
|
if doc.get("version") != 2:
|
|
raise AssertionError(f"version was not updated: {doc}")
|
|
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": 3},
|
|
"contentChanges": [
|
|
{
|
|
"range": {
|
|
"start": {"line": 3, "character": 4},
|
|
"end": {"line": 3, "character": 6},
|
|
},
|
|
"text": "X",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
doc = client.request("viper/lspDocument", {"uri": uri}).get("result")
|
|
if not isinstance(doc, dict) or "# 中文X" not in doc.get("text", ""):
|
|
raise AssertionError(f"UTF-16 surrogate edit failed: {doc}")
|
|
|
|
replacement = "def f() -> int:\r\n return 42\r\n"
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": 4},
|
|
"contentChanges": [{"text": replacement}],
|
|
},
|
|
)
|
|
doc = client.request("viper/lspDocument", {"uri": uri}).get("result")
|
|
if not isinstance(doc, dict) or doc.get("text") != replacement or doc.get("version") != 4:
|
|
raise AssertionError(f"full replacement failed: {doc}")
|
|
|
|
client.notify("textDocument/didSave", {"textDocument": {"uri": uri}})
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
stats = client.request("viper/lspStats").get("result")
|
|
if not isinstance(stats, dict) or stats.get("documentCount") != 0:
|
|
raise AssertionError(f"didClose was not applied: {stats}")
|
|
|
|
edit_text = "x = 0\n"
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": edit_text}},
|
|
)
|
|
for i in range(10000):
|
|
digit = "1" if (i % 2) == 0 else "0"
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": 2 + i},
|
|
"contentChanges": [
|
|
{
|
|
"range": {
|
|
"start": {"line": 0, "character": 4},
|
|
"end": {"line": 0, "character": 5},
|
|
},
|
|
"text": digit,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
stats = client.request("viper/lspStats").get("result")
|
|
if not isinstance(stats, dict) or stats.get("documentCount") != 1:
|
|
raise AssertionError(f"10k edits lost the document: {stats}")
|
|
doc = client.request("viper/lspDocument", {"uri": uri}).get("result")
|
|
if not isinstance(doc, dict) or doc.get("text") != edit_text:
|
|
raise AssertionError(f"10k edits produced unexpected text: {doc}")
|
|
if doc.get("version") != 10001:
|
|
raise AssertionError(f"10k edits produced unexpected version: {doc}")
|
|
if int(stats.get("usedBytes", 0)) > 8 * 1024 * 1024:
|
|
raise AssertionError(f"document pool grew unexpectedly: {stats}")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
client.request("shutdown")
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
print("documents: OK")
|
|
print("positions: OK")
|
|
print("10k-edits: OK")
|
|
return 0
|
|
|
|
|
|
def gate2(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
uri = (root / "Tools" / "lsp" / "fixtures" / "gate2.vp").as_uri()
|
|
broken = "def a(:\n pass\n\ndef b(:\n pass\n\ndef c(:\n pass\n"
|
|
fixed = "def a() -> int:\n return 1\n"
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root.as_uri(), "capabilities": {}},
|
|
)
|
|
if not isinstance(response.get("result"), dict):
|
|
raise AssertionError(f"initialize failed: {response}")
|
|
client.notify("initialized", {})
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": broken}},
|
|
)
|
|
publish = client.wait_notification("textDocument/publishDiagnostics")
|
|
diagnostics = publish.get("params", {}).get("diagnostics")
|
|
if not isinstance(diagnostics, list) or len(diagnostics) < 3:
|
|
raise AssertionError(f"expected at least 3 syntax diagnostics: {publish}")
|
|
for diagnostic in diagnostics:
|
|
if diagnostic.get("severity") != 1 or not diagnostic.get("message"):
|
|
raise AssertionError(f"invalid diagnostic shape: {diagnostic}")
|
|
if diagnostic.get("range", {}).get("start") is None:
|
|
raise AssertionError(f"diagnostic has no range: {diagnostic}")
|
|
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": 2},
|
|
"contentChanges": [{"text": fixed}],
|
|
},
|
|
)
|
|
publish = client.wait_notification("textDocument/publishDiagnostics")
|
|
diagnostics = publish.get("params", {}).get("diagnostics")
|
|
if not isinstance(diagnostics, list) or len(diagnostics) != 0:
|
|
raise AssertionError(f"fixed document still has diagnostics: {publish}")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
publish = client.wait_notification("textDocument/publishDiagnostics")
|
|
diagnostics = publish.get("params", {}).get("diagnostics")
|
|
if not isinstance(diagnostics, list) or len(diagnostics) != 0:
|
|
raise AssertionError(f"didClose did not clear diagnostics: {publish}")
|
|
|
|
client.request("shutdown")
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
print("diagnostics: OK")
|
|
print("recovery: OK")
|
|
return 0
|
|
|
|
|
|
def _position(text: str, needle: str, occurrence: int = 0) -> dict[str, int]:
|
|
start = -1
|
|
for _ in range(occurrence + 1):
|
|
start = text.index(needle, start + 1)
|
|
line = text.count("\n", 0, start)
|
|
column = start - (text.rfind("\n", 0, start) + 1)
|
|
return {"line": line, "character": column}
|
|
|
|
|
|
def _symbols_by_name(items: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
|
return {item["name"]: item for item in items}
|
|
|
|
|
|
def gate3(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
fixtures = root / "Tools" / "lsp" / "fixtures"
|
|
source = fixtures / "gate3.vp"
|
|
helper = fixtures / "gate3_helper.vp"
|
|
stub = fixtures / "gate3_stub.vpi"
|
|
uri = source.as_uri()
|
|
text = source.read_text(encoding="utf-8")
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root.as_uri(), "capabilities": {}},
|
|
)
|
|
capabilities = response.get("result", {}).get("capabilities", {})
|
|
for capability in (
|
|
"documentSymbolProvider",
|
|
"workspaceSymbolProvider",
|
|
"definitionProvider",
|
|
"referencesProvider",
|
|
"documentHighlightProvider",
|
|
"hoverProvider",
|
|
):
|
|
if capabilities.get(capability) is not True:
|
|
raise AssertionError(f"missing capability {capability}: {response}")
|
|
client.notify("initialized", {})
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
outline = client.request(
|
|
"textDocument/documentSymbol", {"textDocument": {"uri": uri}}
|
|
).get("result")
|
|
if not isinstance(outline, list):
|
|
raise AssertionError(f"documentSymbol did not return a list: {outline}")
|
|
top = _symbols_by_name(outline)
|
|
for required in ("Widget", "make_widget", "COUNT"):
|
|
if required not in top:
|
|
raise AssertionError(f"documentSymbol missing {required}: {outline}")
|
|
widget_children = _symbols_by_name(top["Widget"].get("children", []))
|
|
for required in ("name", "__init__", "greet"):
|
|
if required not in widget_children:
|
|
raise AssertionError(f"Widget has no child {required}: {top.get('Widget')}")
|
|
if widget_children["name"]["kind"] != 8:
|
|
raise AssertionError(f"Widget.name should be a field: {widget_children['name']}")
|
|
if top["Widget"]["kind"] != 5 or top["make_widget"]["kind"] != 12:
|
|
raise AssertionError(f"unexpected symbol kinds: {top}")
|
|
if top["COUNT"]["kind"] != 14:
|
|
raise AssertionError(f"COUNT should be a constant: {top['COUNT']}")
|
|
if top["Widget"]["selectionRange"]["start"] != _position(text, "Widget", 0):
|
|
raise AssertionError(f"Widget selectionRange is misplaced: {top['Widget']}")
|
|
if top["Widget"]["range"]["start"]["line"] != _position(text, "class Widget", 0)["line"]:
|
|
raise AssertionError(f"Widget range start is misplaced: {top['Widget']}")
|
|
if top["make_widget"]["range"]["start"]["line"] <= top["Widget"]["range"]["start"]["line"]:
|
|
raise AssertionError(f"Widget range does not stop before make_widget: {top['Widget']}")
|
|
|
|
def definition(needle: str, occurrence: int, offset: int = 0) -> list[dict[str, Any]]:
|
|
pos = _position(text, needle, occurrence)
|
|
pos["character"] += offset
|
|
result = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": pos},
|
|
).get("result")
|
|
if not isinstance(result, list):
|
|
raise AssertionError(f"definition did not return a list: {result}")
|
|
return result
|
|
|
|
widget_def = definition("Widget", 2)
|
|
if not widget_def or not widget_def[0]["uri"].endswith("gate3.vp"):
|
|
raise AssertionError(f"same-file definition failed: {widget_def}")
|
|
if widget_def[0]["range"]["start"] != _position(text, "Widget", 0):
|
|
raise AssertionError(f"definition range is misplaced: {widget_def}")
|
|
|
|
helper_def = definition("helper_func", 1)
|
|
if not helper_def or not helper_def[0]["uri"].endswith("gate3_helper.vp"):
|
|
raise AssertionError(f"cross-file definition failed: {helper_def}")
|
|
if helper_def[0]["range"]["start"]["line"] != 3:
|
|
raise AssertionError(f"cross-file definition line is wrong: {helper_def}")
|
|
|
|
stub_def = definition("StubThing", 1)
|
|
if not stub_def or not stub_def[0]["uri"].endswith("gate3_stub.vpi"):
|
|
raise AssertionError(f".vpi definition failed: {stub_def}")
|
|
if stub_def[0]["range"]["start"] != {"line": 2, "character": 6}:
|
|
raise AssertionError(f".vpi definition range is wrong: {stub_def}")
|
|
|
|
field_def = definition("self.name", 0, 5)
|
|
if not field_def or field_def[0]["range"]["start"] != {"line": 11, "character": 4}:
|
|
raise AssertionError(f"field definition failed: {field_def}")
|
|
|
|
hover_pos = _position(text, "make_widget", 0)
|
|
hover = client.request(
|
|
"textDocument/hover", {"textDocument": {"uri": uri}, "position": hover_pos}
|
|
).get("result")
|
|
contents = hover.get("contents", {}) if isinstance(hover, dict) else {}
|
|
value = contents.get("value", "") if isinstance(contents, dict) else ""
|
|
if "def make_widget(name: str) -> Widget" not in value:
|
|
raise AssertionError(f"hover signature missing: {hover}")
|
|
if "Build one widget." not in value:
|
|
raise AssertionError(f"hover docstring missing: {hover}")
|
|
|
|
stub_pos = _position(text, "StubThing", 1)
|
|
hover = client.request(
|
|
"textDocument/hover", {"textDocument": {"uri": uri}, "position": stub_pos}
|
|
).get("result")
|
|
value = hover.get("contents", {}).get("value", "") if isinstance(hover, dict) else ""
|
|
if "class StubThing" not in value:
|
|
raise AssertionError(f".vpi hover failed: {hover}")
|
|
|
|
ref_pos = _position(text, "self.name", 0)
|
|
ref_pos["character"] += 5
|
|
references = client.request(
|
|
"textDocument/references",
|
|
{
|
|
"textDocument": {"uri": uri},
|
|
"position": ref_pos,
|
|
"context": {"includeDeclaration": True},
|
|
},
|
|
).get("result")
|
|
if not isinstance(references, list) or len(references) < 3:
|
|
raise AssertionError(f"references are incomplete: {references}")
|
|
for reference in references:
|
|
if reference.get("uri") != uri:
|
|
raise AssertionError(f"references crossed files unexpectedly: {references}")
|
|
|
|
highlights = client.request(
|
|
"textDocument/documentHighlight",
|
|
{"textDocument": {"uri": uri}, "position": _position(text, "Widget", 2)},
|
|
).get("result")
|
|
if not isinstance(highlights, list) or len(highlights) < 2:
|
|
raise AssertionError(f"documentHighlight is incomplete: {highlights}")
|
|
|
|
workspace = client.request(
|
|
"workspace/symbol", {"query": "LspDocuments"}, timeout=60.0
|
|
).get("result")
|
|
if not isinstance(workspace, list) or len(workspace) == 0:
|
|
raise AssertionError(f"workspace/symbol returned nothing: {workspace}")
|
|
if not any(item["location"]["uri"].endswith("LspDocuments.vp") for item in workspace):
|
|
raise AssertionError(f"workspace/symbol missed LspDocuments.vp: {workspace}")
|
|
if not any(item["kind"] == 5 for item in workspace):
|
|
raise AssertionError(f"workspace/symbol lost class kinds: {workspace}")
|
|
|
|
again = client.request(
|
|
"workspace/symbol", {"query": "LspServer"}, timeout=60.0
|
|
).get("result")
|
|
if not isinstance(again, list) or len(again) == 0:
|
|
raise AssertionError(f"cached workspace/symbol returned nothing: {again}")
|
|
stats = client.request("viper/lspStats").get("result")
|
|
if not isinstance(stats, dict):
|
|
raise AssertionError(f"lspStats failed after reindex: {stats}")
|
|
if int(stats.get("requestUsedBytes", 0)) > 8 * 1024 * 1024:
|
|
raise AssertionError(f"message pool was not reclaimed: {stats}")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
client.request("shutdown")
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
print("outline: OK")
|
|
print("navigate: OK")
|
|
print("workspace-symbol: OK")
|
|
return 0
|
|
|
|
|
|
def _position_after(text: str, needle: str, delta: int = 0) -> dict[str, int]:
|
|
pos = _position(text, needle)
|
|
pos["character"] += len(needle) + delta
|
|
return pos
|
|
|
|
|
|
def _end_position(text: str) -> dict[str, int]:
|
|
line = text.count("\n")
|
|
column = len(text) - (text.rfind("\n") + 1)
|
|
return {"line": line, "character": column}
|
|
|
|
|
|
def _replace(client: "Client", uri: str, version: int, text: str) -> None:
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{
|
|
"textDocument": {"uri": uri, "version": version},
|
|
"contentChanges": [{"text": text}],
|
|
},
|
|
)
|
|
|
|
|
|
def _completion_labels(client: "Client", uri: str,
|
|
pos: dict[str, int]) -> list[str]:
|
|
result = client.request(
|
|
"textDocument/completion",
|
|
{
|
|
"textDocument": {"uri": uri},
|
|
"position": pos,
|
|
"context": {"triggerKind": 1},
|
|
},
|
|
).get("result")
|
|
if not isinstance(result, dict) or not isinstance(result.get("items"), list):
|
|
raise AssertionError(f"completion result is malformed: {result}")
|
|
return [item["label"] for item in result["items"] if isinstance(item, dict)]
|
|
|
|
|
|
def _signature(client: "Client", uri: str,
|
|
pos: dict[str, int]) -> dict[str, Any]:
|
|
result = client.request(
|
|
"textDocument/signatureHelp",
|
|
{"textDocument": {"uri": uri}, "position": pos},
|
|
).get("result")
|
|
if result is not None and not isinstance(result, dict):
|
|
raise AssertionError(f"signatureHelp result is malformed: {result}")
|
|
return result or {}
|
|
|
|
|
|
def gate4(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
source = root / "Tools" / "lsp" / "fixtures" / "gate4.vp"
|
|
uri = source.as_uri()
|
|
text = source.read_text(encoding="utf-8")
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root.as_uri(), "capabilities": {}},
|
|
)
|
|
capabilities = response.get("result", {}).get("capabilities", {})
|
|
completion_provider = capabilities.get("completionProvider")
|
|
if not isinstance(completion_provider, dict):
|
|
raise AssertionError(f"missing completionProvider: {response}")
|
|
if "." not in completion_provider.get("triggerCharacters", []):
|
|
raise AssertionError(f"completion trigger characters are wrong: {completion_provider}")
|
|
signature_provider = capabilities.get("signatureHelpProvider")
|
|
if not isinstance(signature_provider, dict):
|
|
raise AssertionError(f"missing signatureHelpProvider: {response}")
|
|
if "(" not in signature_provider.get("triggerCharacters", []):
|
|
raise AssertionError(f"signature trigger characters are wrong: {signature_provider}")
|
|
|
|
client.notify("initialized", {})
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
return_pos = _position(text, " return x")
|
|
return_pos["character"] = 4
|
|
labels = _completion_labels(client, uri, return_pos)
|
|
for required in ("g", "x", "Gadget", "GLOBAL_COUNT", "helper_func", "overload", "pair"):
|
|
if required not in labels:
|
|
raise AssertionError(f"scope completion missed {required}: {labels}")
|
|
for builtin in ("len", "print"):
|
|
if builtin not in labels:
|
|
raise AssertionError(f"scope completion missed builtin {builtin}: {labels}")
|
|
|
|
labels = _completion_labels(client, uri, _position_after(text, "self."))
|
|
for required in ("label", "run", "__init__"):
|
|
if required not in labels:
|
|
raise AssertionError(f"self member completion missed {required}: {labels}")
|
|
|
|
labels = _completion_labels(client, uri, _position_after(text, "Gadget."))
|
|
for required in ("label", "run", "__init__"):
|
|
if required not in labels:
|
|
raise AssertionError(f"class member completion missed {required}: {labels}")
|
|
|
|
labels = _completion_labels(client, uri, _position_after(text, "StubThing."))
|
|
for required in ("value", "run"):
|
|
if required not in labels:
|
|
raise AssertionError(f".vpi member completion missed {required}: {labels}")
|
|
|
|
import_text = "from .gate3_"
|
|
_replace(client, uri, 2, import_text)
|
|
labels = _completion_labels(client, uri, _end_position(import_text))
|
|
for required in ("gate3_helper", "gate3_stub"):
|
|
if required not in labels:
|
|
raise AssertionError(f"import completion missed {required}: {labels}")
|
|
|
|
overload_text = (
|
|
"def overload(a: int) -> int:\n"
|
|
" return a\n"
|
|
"\n"
|
|
"def overload(a: str) -> int:\n"
|
|
" return 0\n"
|
|
"\n"
|
|
"overload("
|
|
)
|
|
_replace(client, uri, 3, overload_text)
|
|
help = _signature(client, uri, _end_position(overload_text))
|
|
signatures = help.get("signatures")
|
|
if not isinstance(signatures, list) or len(signatures) != 2:
|
|
raise AssertionError(f"overload signatureHelp is wrong: {help}")
|
|
for signature in signatures:
|
|
if "def overload" not in signature.get("label", ""):
|
|
raise AssertionError(f"overload signature label is wrong: {signature}")
|
|
if len(signature.get("parameters", [])) != 1:
|
|
raise AssertionError(f"overload parameter list is wrong: {signature}")
|
|
if help.get("activeParameter") != 0:
|
|
raise AssertionError(f"activeParameter should be 0: {help}")
|
|
|
|
pair_text = (
|
|
"def pair(a: int, b: int) -> int:\n"
|
|
" return a\n"
|
|
"\n"
|
|
"pair(1, "
|
|
)
|
|
_replace(client, uri, 4, pair_text)
|
|
help = _signature(client, uri, _end_position(pair_text))
|
|
signatures = help.get("signatures")
|
|
if not isinstance(signatures, list) or len(signatures) != 1:
|
|
raise AssertionError(f"pair signatureHelp is wrong: {help}")
|
|
if len(signatures[0].get("parameters", [])) != 2:
|
|
raise AssertionError(f"pair parameter list is wrong: {signatures[0]}")
|
|
if help.get("activeParameter") != 1:
|
|
raise AssertionError(f"activeParameter should be 1: {help}")
|
|
|
|
cross_text = "from .gate3_helper import helper_func\n\nhelper_func("
|
|
_replace(client, uri, 5, cross_text)
|
|
help = _signature(client, uri, _end_position(cross_text))
|
|
signatures = help.get("signatures")
|
|
if not isinstance(signatures, list) or len(signatures) != 1:
|
|
raise AssertionError(f"cross-file signatureHelp is wrong: {help}")
|
|
if "def helper_func" not in signatures[0].get("label", ""):
|
|
raise AssertionError(f"cross-file signature label is wrong: {signatures[0]}")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
client.request("shutdown")
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
print("completion: OK")
|
|
print("signature: OK")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="ViperC LSP probe")
|
|
parser.add_argument("--selfcheck", action="store_true", help="run pure-Python framing checks")
|
|
parser.add_argument("--gate1", action="store_true", help="run document/position checks")
|
|
parser.add_argument("--gate2", action="store_true", help="run syntax diagnostic checks")
|
|
parser.add_argument("--gate3", action="store_true", help="run symbol/navigation checks")
|
|
parser.add_argument("--gate4", action="store_true", help="run completion/signature checks")
|
|
parser.add_argument("--exe", help="path to viperc.exe")
|
|
args = parser.parse_args()
|
|
|
|
if args.selfcheck:
|
|
return selfcheck()
|
|
if not args.exe:
|
|
parser.error("--exe is required unless --selfcheck is used")
|
|
exe = os.path.abspath(args.exe)
|
|
if not os.path.isfile(exe):
|
|
print(f"probe: executable not found: {exe}", file=sys.stderr)
|
|
return 2
|
|
if args.gate4:
|
|
return gate4(exe)
|
|
if args.gate3:
|
|
return gate3(exe)
|
|
if args.gate2:
|
|
return gate2(exe)
|
|
if args.gate1:
|
|
return documents(exe)
|
|
return lifecycle(exe)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|