TermiNexus
95a8da86e4
define 宏函数与调用点使用 macro token；常量保持 variable+readonly；多行字符串按行拆分；修正符号/出现点合并顺序导致整行 token 丢失；大文档诊断 scratch 按源大小扩容。
1401 lines
59 KiB
Python
1401 lines
59 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 shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
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}")
|
|
if response.get("method") is not None:
|
|
raise AssertionError(f"response carries a method field: {response}")
|
|
if "result" not in response and "error" not in response:
|
|
raise AssertionError(f"message is not a response: {response}")
|
|
if response.get("jsonrpc") != "2.0":
|
|
raise AssertionError(f"response has no jsonrpc 2.0: {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 _apply_edits(text: str, edits: list[dict[str, Any]]) -> str:
|
|
lines = text.split("\n")
|
|
ordered = sorted(
|
|
edits,
|
|
key=lambda e: (e["range"]["start"]["line"], e["range"]["start"]["character"]),
|
|
reverse=True,
|
|
)
|
|
for edit in ordered:
|
|
start = edit["range"]["start"]
|
|
end = edit["range"]["end"]
|
|
sl = start["line"]
|
|
sc = start["character"]
|
|
el = end["line"]
|
|
ec = end["character"]
|
|
prefix = lines[sl][:sc] if sl < len(lines) else ""
|
|
suffix = lines[el][ec:] if el < len(lines) else ""
|
|
replacement = prefix + edit["newText"] + suffix
|
|
if sl < len(lines):
|
|
lines[sl] = replacement
|
|
else:
|
|
lines.append(replacement)
|
|
if el > sl:
|
|
del lines[sl + 1 : el + 1]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _decode_tokens(data: list[int]) -> list[dict[str, int]]:
|
|
if not isinstance(data, list) or len(data) % 5 != 0:
|
|
raise AssertionError(f"semantic token data is malformed: {data}")
|
|
tokens = []
|
|
line = 0
|
|
char = 0
|
|
for i in range(0, len(data), 5):
|
|
delta_line, delta_char, length, token_type, modifiers = data[i : i + 5]
|
|
line += delta_line
|
|
char = char + delta_char if delta_line == 0 else delta_char
|
|
tokens.append(
|
|
{
|
|
"line": line,
|
|
"character": char,
|
|
"length": length,
|
|
"type": token_type,
|
|
"modifiers": modifiers,
|
|
}
|
|
)
|
|
return tokens
|
|
|
|
|
|
def _find_token(tokens: list[dict[str, int]], line: int, character: int,
|
|
token_type: int) -> bool:
|
|
return any(
|
|
t["line"] == line and t["character"] == character and t["type"] == token_type
|
|
for t in tokens
|
|
)
|
|
|
|
|
|
def _find_token_mods(tokens: list[dict[str, int]], line: int, character: int,
|
|
token_type: int, mod_bit: int) -> bool:
|
|
return any(
|
|
t["line"] == line
|
|
and t["character"] == character
|
|
and t["type"] == token_type
|
|
and (t["modifiers"] & mod_bit) != 0
|
|
for t in tokens
|
|
)
|
|
|
|
|
|
def gate5(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
source = root / "Tools" / "lsp" / "fixtures" / "gate5.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", {})
|
|
rename_provider = capabilities.get("renameProvider")
|
|
if not isinstance(rename_provider, dict) or rename_provider.get("prepareProvider") is not True:
|
|
raise AssertionError(f"missing prepareRename capability: {response}")
|
|
sem = capabilities.get("semanticTokensProvider")
|
|
if not isinstance(sem, dict) or sem.get("full") is not True or sem.get("range") is not True:
|
|
raise AssertionError(f"missing semanticTokens capability: {response}")
|
|
legend = sem.get("legend", {})
|
|
types = legend.get("tokenTypes", [])
|
|
for required in ("class", "function", "parameter", "keyword", "comment", "string", "number"):
|
|
if required not in types:
|
|
raise AssertionError(f"semantic legend misses {required}: {legend}")
|
|
if "readonly" not in legend.get("tokenModifiers", []):
|
|
raise AssertionError(f"semantic legend misses the readonly modifier: {legend}")
|
|
if capabilities.get("codeActionProvider") is not True:
|
|
raise AssertionError(f"missing codeAction capability: {response}")
|
|
|
|
client.notify("initialized", {})
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
field_pos = _position(text, "field: int")
|
|
prepared = client.request(
|
|
"textDocument/prepareRename",
|
|
{"textDocument": {"uri": uri}, "position": field_pos},
|
|
).get("result")
|
|
if not isinstance(prepared, dict) or prepared.get("placeholder") != "field":
|
|
raise AssertionError(f"prepareRename failed: {prepared}")
|
|
if prepared.get("range", {}).get("start") != field_pos:
|
|
raise AssertionError(f"prepareRename range is wrong: {prepared}")
|
|
|
|
renamed = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": uri}, "position": field_pos, "newName": "value"},
|
|
).get("result")
|
|
if not isinstance(renamed, dict):
|
|
raise AssertionError(f"rename failed: {renamed}")
|
|
edits = renamed.get("changes", {}).get(uri)
|
|
if not isinstance(edits, list) or len(edits) < 3:
|
|
raise AssertionError(f"rename produced too few edits: {renamed}")
|
|
new_text = _apply_edits(text, edits)
|
|
if " value: int" not in new_text:
|
|
raise AssertionError(f"rename did not update the declaration: {new_text}")
|
|
if "self.field" in new_text or "self.value" not in new_text:
|
|
raise AssertionError(f"rename did not update field accesses: {new_text}")
|
|
|
|
invalid = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": uri}, "position": field_pos, "newName": "def"},
|
|
).get("result")
|
|
if invalid is not None:
|
|
raise AssertionError(f"reserved word rename should return null: {invalid}")
|
|
|
|
semantic = client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": uri}}
|
|
).get("result")
|
|
if not isinstance(semantic, dict):
|
|
raise AssertionError(f"semanticTokens/full failed: {semantic}")
|
|
tokens = _decode_tokens(semantic.get("data", []))
|
|
keyword_pos = _position(text, "class Thing")
|
|
if _find_token(tokens, keyword_pos["line"], keyword_pos["character"], 7):
|
|
raise AssertionError("keywords must be left to the TextMate grammar")
|
|
if not _find_token(tokens, keyword_pos["line"], keyword_pos["character"] + 6, 1):
|
|
raise AssertionError(f"missing class name token: {tokens[:10]}")
|
|
field_decl = _position(text, "field: int")
|
|
if not _find_token(tokens, field_decl["line"], field_decl["character"], 6):
|
|
raise AssertionError(f"missing property token: {tokens[:20]}")
|
|
arg_decl = _position(text, "arg: int")
|
|
if not _find_token(tokens, arg_decl["line"], arg_decl["character"], 4):
|
|
raise AssertionError(f"missing parameter token: {tokens[:20]}")
|
|
comment_pos = _position(text, "# gate5 comment")
|
|
if not _find_token(tokens, comment_pos["line"], comment_pos["character"], 8):
|
|
raise AssertionError(f"missing comment token: {tokens[:20]}")
|
|
string_pos = _position(text, '"hello"')
|
|
if not _find_token(tokens, string_pos["line"], string_pos["character"], 9):
|
|
raise AssertionError(f"missing string token: {tokens[:20]}")
|
|
number_pos = _position(text, "42")
|
|
if not _find_token(tokens, number_pos["line"], number_pos["character"], 10):
|
|
raise AssertionError(f"missing number token: {tokens[:20]}")
|
|
|
|
define_pos = _position(text, "APP_NAME")
|
|
if not _find_token(tokens, define_pos["line"], define_pos["character"], 5):
|
|
raise AssertionError(f"define name token is misplaced: {define_pos} {tokens}")
|
|
if not _find_token_mods(tokens, define_pos["line"], define_pos["character"], 5, 2):
|
|
raise AssertionError(f"define constant is not readonly: {tokens}")
|
|
for token in tokens:
|
|
if token["line"] == define_pos["line"] and token["character"] == 0:
|
|
if token["length"] == len("APP_NAME"):
|
|
raise AssertionError(f"define name token is at column 0: {token}")
|
|
|
|
value_pos = _position(text, "VALUE: int")
|
|
if not _find_token_mods(tokens, value_pos["line"], value_pos["character"], 5, 2):
|
|
raise AssertionError(f"uppercase constant is not readonly: {tokens}")
|
|
|
|
extern_pos = _position(text, "add(a: int, b: int)")
|
|
if not _find_token(tokens, extern_pos["line"], extern_pos["character"], 2):
|
|
raise AssertionError(f"extern name token is misplaced: {extern_pos} {tokens}")
|
|
for token in tokens:
|
|
if token["line"] == extern_pos["line"] and token["character"] == 0:
|
|
if token["length"] == len("add"):
|
|
raise AssertionError(f"extern name token is at column 0: {token}")
|
|
|
|
ranged = client.request(
|
|
"textDocument/semanticTokens/range",
|
|
{
|
|
"textDocument": {"uri": uri},
|
|
"range": {
|
|
"start": {"line": keyword_pos["line"], "character": 0},
|
|
"end": {"line": keyword_pos["line"], "character": 80},
|
|
},
|
|
},
|
|
).get("result")
|
|
if not isinstance(ranged, dict):
|
|
raise AssertionError(f"semanticTokens/range failed: {ranged}")
|
|
ranged_tokens = _decode_tokens(ranged.get("data", []))
|
|
if not ranged_tokens:
|
|
raise AssertionError(f"semanticTokens/range returned nothing: {ranged}")
|
|
for token in ranged_tokens:
|
|
if token["line"] != keyword_pos["line"]:
|
|
raise AssertionError(f"range tokens escaped the request: {token}")
|
|
|
|
actions = client.request(
|
|
"textDocument/codeAction",
|
|
{
|
|
"textDocument": {"uri": uri},
|
|
"range": {"start": {"line": 0, "character": 0}, "end": {"line": 0, "character": 0}},
|
|
"context": {"diagnostics": []},
|
|
},
|
|
).get("result")
|
|
if not isinstance(actions, list):
|
|
raise AssertionError(f"codeAction failed: {actions}")
|
|
organize = next(
|
|
(a for a in actions if a.get("kind") == "source.organizeImports"), None
|
|
)
|
|
if organize is None:
|
|
raise AssertionError(f"organize imports action missing: {actions}")
|
|
organize_edits = organize.get("edit", {}).get("changes", {}).get(uri)
|
|
if not isinstance(organize_edits, list) or not organize_edits:
|
|
raise AssertionError(f"organize imports edit is empty: {organize}")
|
|
organized = _apply_edits(text, organize_edits)
|
|
expected = (
|
|
"from .gate3_helper import helper_func\n"
|
|
"from .gate3_stub import StubThing\n"
|
|
"import stdlib\n"
|
|
)
|
|
if not organized.startswith(expected):
|
|
raise AssertionError(f"organize imports produced the wrong block: {organized[:200]}")
|
|
if organized.count("from .gate3_helper import helper_func") != 1:
|
|
raise AssertionError(f"organize imports did not deduplicate: {organized[:200]}")
|
|
|
|
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("rename: OK")
|
|
print("semantic-tokens: OK")
|
|
print("code-action: OK")
|
|
return 0
|
|
|
|
|
|
def _tree_state(path: Path) -> tuple | None:
|
|
if not path.exists():
|
|
return None
|
|
items = []
|
|
for candidate in sorted(path.rglob("*")):
|
|
if candidate.is_file():
|
|
stat = candidate.stat()
|
|
items.append(
|
|
(
|
|
str(candidate.relative_to(path)).replace("\\", "/"),
|
|
stat.st_size,
|
|
stat.st_mtime_ns,
|
|
)
|
|
)
|
|
return tuple(items)
|
|
|
|
|
|
def gate7(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
workspace = root.parent
|
|
source = root / "Tools" / "lsp" / "fixtures" / "gate3.vp"
|
|
uri = source.as_uri()
|
|
text = source.read_text(encoding="utf-8")
|
|
output_before = _tree_state(root / "output")
|
|
temp_before = _tree_state(root / "temp")
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": workspace.as_uri(), "capabilities": {}},
|
|
timeout=15.0,
|
|
)
|
|
if not isinstance(response.get("result"), dict):
|
|
raise AssertionError(f"initialize failed from the repository root: {response}")
|
|
client.notify("initialized", {})
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
stats = client.request("viper/lspStats").get("result")
|
|
if not isinstance(stats, dict):
|
|
raise AssertionError(f"lspStats failed after project adoption: {stats}")
|
|
project = str(stats.get("project") or "").replace("\\", "/")
|
|
source_dir = str(stats.get("sourceDir") or "").replace("\\", "/")
|
|
includes = str(stats.get("includes") or "").replace("\\", "/")
|
|
if not project.endswith("ViperC/project.vpj"):
|
|
raise AssertionError(f"the subproject was not adopted: {stats}")
|
|
if not source_dir.endswith("ViperC/App"):
|
|
raise AssertionError(f"source_dir is wrong after adoption: {stats}")
|
|
if "includes" not in includes:
|
|
raise AssertionError(f"includes is wrong after adoption: {stats}")
|
|
|
|
symbols = client.request(
|
|
"workspace/symbol", {"query": "LspDocuments"}, timeout=60.0
|
|
).get("result")
|
|
if not isinstance(symbols, list) or len(symbols) == 0:
|
|
raise AssertionError(f"workspace/symbol failed after adoption: {symbols}")
|
|
if not any(
|
|
item["location"]["uri"].endswith("LspDocuments.vp") and item["kind"] == 5
|
|
for item in symbols
|
|
):
|
|
raise AssertionError(f"workspace/symbol missed the LspDocuments class: {symbols}")
|
|
|
|
pos = _position(text, "helper_func", 1)
|
|
definition = client.request(
|
|
"textDocument/definition", {"textDocument": {"uri": uri}, "position": pos}
|
|
).get("result")
|
|
if not isinstance(definition, list) or not definition:
|
|
raise AssertionError(f"definition failed after adoption: {definition}")
|
|
if not definition[0]["uri"].endswith("gate3_helper.vp"):
|
|
raise AssertionError(f"definition target is wrong after adoption: {definition}")
|
|
|
|
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")
|
|
|
|
output_after = _tree_state(root / "output")
|
|
temp_after = _tree_state(root / "temp")
|
|
if output_before != output_after:
|
|
raise AssertionError("--lsp modified ViperC/output")
|
|
if temp_before != temp_after:
|
|
raise AssertionError("--lsp modified ViperC/temp")
|
|
|
|
mixed = subprocess.run(
|
|
[exe, "--lsp", "--project", str(root / "project.vpj")],
|
|
capture_output=True,
|
|
timeout=30,
|
|
)
|
|
if mixed.returncode == 0:
|
|
raise AssertionError("--lsp accepted --project")
|
|
stderr = mixed.stderr.decode("utf-8", "replace")
|
|
if "standalone" not in stderr and "--lsp" not in stderr:
|
|
raise AssertionError(f"--lsp rejection message is unclear: {stderr}")
|
|
|
|
print("workspace-root-adoption: OK")
|
|
print("product-side-effects: OK")
|
|
print("standalone-mode: OK")
|
|
return 0
|
|
|
|
|
|
def _apply_workspace(changes: dict[str, list[dict[str, Any]]],
|
|
texts: dict[str, str]) -> dict[str, str]:
|
|
out = dict(texts)
|
|
for uri, edits in changes.items():
|
|
for name in list(out.keys()):
|
|
if uri.endswith(name):
|
|
out[name] = _apply_edits(out[name], edits)
|
|
return out
|
|
|
|
|
|
def gate8(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
fixtures = root / "Tools" / "lsp" / "fixtures"
|
|
# The fixtures live outside the project's source roots; copy them into an isolated workspace
|
|
# so the cross-file index contains both files.
|
|
workspace = Path(tempfile.mkdtemp(prefix="transpyc_lsp_gate8_"))
|
|
path_a = workspace / "gate8_a.vp"
|
|
path_b = workspace / "gate8_b.vp"
|
|
shutil.copy(fixtures / "gate8_a.vp", path_a)
|
|
shutil.copy(fixtures / "gate8_b.vp", path_b)
|
|
uri_a = path_a.as_uri()
|
|
text_a = path_a.read_text(encoding="utf-8")
|
|
text_b = path_b.read_text(encoding="utf-8")
|
|
texts = {"gate8_a.vp": text_a, "gate8_b.vp": text_b}
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": workspace.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_a, "languageId": "viper", "version": 1, "text": text_a}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
shared_pos = _position(text_a, "Shared")
|
|
prepared = client.request(
|
|
"textDocument/prepareRename",
|
|
{"textDocument": {"uri": uri_a}, "position": shared_pos},
|
|
).get("result")
|
|
if not isinstance(prepared, dict) or prepared.get("placeholder") != "Shared":
|
|
raise AssertionError(f"prepareRename on a class failed: {prepared}")
|
|
|
|
references = client.request(
|
|
"textDocument/references",
|
|
{
|
|
"textDocument": {"uri": uri_a},
|
|
"position": shared_pos,
|
|
"context": {"includeDeclaration": True},
|
|
},
|
|
timeout=90.0,
|
|
).get("result")
|
|
if not isinstance(references, list) or len(references) < 3:
|
|
raise AssertionError(f"cross-file references are incomplete: {references}")
|
|
ref_files = {Path(item["uri"]).name for item in references}
|
|
if "gate8_a.vp" not in ref_files or "gate8_b.vp" not in ref_files:
|
|
raise AssertionError(f"references did not cross files: {references}")
|
|
|
|
helper_pos = _position(text_a, "helper(x: int)")
|
|
helper_refs = client.request(
|
|
"textDocument/references",
|
|
{
|
|
"textDocument": {"uri": uri_a},
|
|
"position": helper_pos,
|
|
"context": {"includeDeclaration": False},
|
|
},
|
|
).get("result")
|
|
if not isinstance(helper_refs, list) or len(helper_refs) < 2:
|
|
raise AssertionError(f"helper references are incomplete: {helper_refs}")
|
|
if not any(Path(item["uri"]).name == "gate8_b.vp" for item in helper_refs):
|
|
raise AssertionError(f"helper references missed the other file: {helper_refs}")
|
|
|
|
renamed = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": uri_a}, "position": helper_pos, "newName": "helper2"},
|
|
timeout=90.0,
|
|
).get("result")
|
|
if not isinstance(renamed, dict):
|
|
raise AssertionError(f"cross-file rename failed: {renamed}")
|
|
changes = renamed.get("changes", {})
|
|
if len(changes) != 2:
|
|
raise AssertionError(f"rename did not touch both files: {changes}")
|
|
updated = _apply_workspace(changes, texts)
|
|
if "def helper2(" not in updated["gate8_a.vp"]:
|
|
raise AssertionError(f"rename missed the declaration: {updated['gate8_a.vp']}")
|
|
if "import Shared, helper2" not in updated["gate8_b.vp"]:
|
|
raise AssertionError(f"rename missed the import: {updated['gate8_b.vp']}")
|
|
if "helper2(v)" not in updated["gate8_b.vp"]:
|
|
raise AssertionError(f"rename missed the call: {updated['gate8_b.vp']}")
|
|
|
|
renamed = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": uri_a}, "position": shared_pos, "newName": "Thing"},
|
|
timeout=90.0,
|
|
).get("result")
|
|
if not isinstance(renamed, dict):
|
|
raise AssertionError(f"class rename failed: {renamed}")
|
|
updated = _apply_workspace(renamed.get("changes", {}), texts)
|
|
if "class Thing:" not in updated["gate8_a.vp"]:
|
|
raise AssertionError(f"class rename missed the declaration: {updated['gate8_a.vp']}")
|
|
if "s: Thing" not in updated["gate8_a.vp"]:
|
|
raise AssertionError(f"class rename missed the annotation: {updated['gate8_a.vp']}")
|
|
if "Shared" in updated["gate8_b.vp"]:
|
|
raise AssertionError(f"class rename left the old name: {updated['gate8_b.vp']}")
|
|
|
|
param_pos = _position(text_a, "x: int")
|
|
renamed = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": uri_a}, "position": param_pos, "newName": "y"},
|
|
).get("result")
|
|
if not isinstance(renamed, dict):
|
|
raise AssertionError(f"local rename failed: {renamed}")
|
|
if len(renamed.get("changes", {})) != 1:
|
|
raise AssertionError(f"local rename crossed files: {renamed}")
|
|
local = _apply_workspace(renamed.get("changes", {}), texts)
|
|
if "def helper(y: int)" not in local["gate8_a.vp"]:
|
|
raise AssertionError(f"local rename missed the parameter: {local['gate8_a.vp']}")
|
|
if "s.value + y" not in local["gate8_a.vp"]:
|
|
raise AssertionError(f"local rename missed the use: {local['gate8_a.vp']}")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri_a}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
client.request("shutdown")
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
shutil.rmtree(workspace, ignore_errors=True)
|
|
print("cross-file-references: OK")
|
|
print("cross-file-rename: OK")
|
|
print("local-rename: OK")
|
|
return 0
|
|
|
|
|
|
def gate9(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
source = root / "Tools" / "lsp" / "fixtures" / "gate9.vp"
|
|
uri = source.as_uri()
|
|
text = source.read_text(encoding="utf-8")
|
|
lines = text.split("\n")
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root.as_uri(), "capabilities": {}},
|
|
)
|
|
capabilities = response.get("result", {}).get("capabilities", {})
|
|
sem = capabilities.get("semanticTokensProvider")
|
|
if not isinstance(sem, dict):
|
|
raise AssertionError(f"missing semanticTokensProvider: {response}")
|
|
legend = sem.get("legend", {})
|
|
types = legend.get("tokenTypes", [])
|
|
mods = legend.get("tokenModifiers", [])
|
|
for required in ("macro", "variable", "keyword", "string", "number", "comment"):
|
|
if required not in types:
|
|
raise AssertionError(f"semantic legend misses {required}: {legend}")
|
|
if "readonly" not in mods:
|
|
raise AssertionError(f"semantic legend misses readonly: {legend}")
|
|
macro_idx = types.index("macro")
|
|
variable_idx = types.index("variable")
|
|
keyword_idx = types.index("keyword")
|
|
comment_idx = types.index("comment")
|
|
string_idx = types.index("string")
|
|
number_idx = types.index("number")
|
|
readonly_bit = 1 << mods.index("readonly")
|
|
mod_mask = (1 << len(mods)) - 1
|
|
|
|
client.notify("initialized", {})
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
semantic = client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": uri}}
|
|
).get("result")
|
|
if not isinstance(semantic, dict):
|
|
raise AssertionError(f"semanticTokens/full failed: {semantic}")
|
|
tokens = _decode_tokens(semantic.get("data", []))
|
|
if not tokens:
|
|
raise AssertionError("semantic tokens are empty")
|
|
|
|
last_line = -1
|
|
last_end = -1
|
|
for token in tokens:
|
|
if token["length"] <= 0:
|
|
raise AssertionError(f"zero-length token: {token}")
|
|
if token["line"] < 0 or token["line"] >= len(lines):
|
|
raise AssertionError(f"token line is out of range: {token}")
|
|
line_len = len(lines[token["line"]])
|
|
if token["character"] < 0 or token["character"] + token["length"] > line_len:
|
|
raise AssertionError(f"token range is out of the line: {token}")
|
|
if token["type"] < 0 or token["type"] >= len(types):
|
|
raise AssertionError(f"token type is outside the legend: {token}")
|
|
if (token["modifiers"] & ~mod_mask) != 0:
|
|
raise AssertionError(f"token modifiers are outside the legend: {token}")
|
|
if token["type"] == keyword_idx:
|
|
raise AssertionError(f"keywords must be left to TextMate: {token}")
|
|
if token["line"] == last_line:
|
|
if token["character"] < last_end:
|
|
raise AssertionError(f"overlapping tokens on line {last_line}: {token}")
|
|
last_end = token["character"] + token["length"]
|
|
else:
|
|
last_line = token["line"]
|
|
last_end = token["character"] + token["length"]
|
|
|
|
add_decl = _position(text, "define ADD")
|
|
add_decl["character"] += len("define ")
|
|
if not _find_token(tokens, add_decl["line"], add_decl["character"], macro_idx):
|
|
raise AssertionError(f"define macro declaration is not a macro token: {add_decl}")
|
|
add_call = _position(text, "ADD(x")
|
|
if not _find_token(tokens, add_call["line"], add_call["character"], macro_idx):
|
|
raise AssertionError(f"define macro call is not a macro token: {add_call}")
|
|
|
|
max_decl = _position(text, "MAX_VALUE")
|
|
if not _find_token_mods(tokens, max_decl["line"], max_decl["character"],
|
|
variable_idx, readonly_bit):
|
|
raise AssertionError(f"define constant is not readonly: {max_decl}")
|
|
max_call = _position(text, "MAX_VALUE", 1)
|
|
if not _find_token_mods(tokens, max_call["line"], max_call["character"],
|
|
variable_idx, readonly_bit):
|
|
raise AssertionError(f"define constant use is not readonly: {max_call}")
|
|
|
|
for keyword in ("define", "def", "class", "return"):
|
|
pos = _position(text, keyword)
|
|
if _find_token(tokens, pos["line"], pos["character"], keyword_idx):
|
|
raise AssertionError(f"keyword {keyword} must not be a semantic token")
|
|
|
|
comment_pos = _position(text, "# gate9 comment")
|
|
if not _find_token(tokens, comment_pos["line"], comment_pos["character"], comment_idx):
|
|
raise AssertionError(f"missing comment token: {comment_pos}")
|
|
first_string = _position(text, '"""alpha')
|
|
if not _find_token(tokens, first_string["line"], first_string["character"], string_idx):
|
|
raise AssertionError(f"missing first line of the multi-line string: {first_string}")
|
|
second_string = _position(text, "beta")
|
|
if not _find_token(tokens, second_string["line"], second_string["character"], string_idx):
|
|
raise AssertionError(f"missing second line of the multi-line string: {second_string}")
|
|
number_pos = _position(text, "42")
|
|
if not _find_token(tokens, number_pos["line"], number_pos["character"], number_idx):
|
|
raise AssertionError(f"missing number token: {number_pos}")
|
|
|
|
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("semantic-legend: OK")
|
|
print("semantic-audit: OK")
|
|
print("semantic-macro: 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("--gate5", action="store_true", help="run rename/semantic/code-action checks")
|
|
parser.add_argument("--gate7", action="store_true", help="run the product gate checks")
|
|
parser.add_argument("--gate8", action="store_true", help="run cross-file references/rename checks")
|
|
parser.add_argument("--gate9", action="store_true", help="run semantic highlighting audits")
|
|
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.gate9:
|
|
return gate9(exe)
|
|
if args.gate8:
|
|
return gate8(exe)
|
|
if args.gate7:
|
|
return gate7(exe)
|
|
if args.gate5:
|
|
return gate5(exe)
|
|
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())
|