TermiNexus
1968748ea5
Adds LspSigCache: per-file signature cache under <project>/.viperc-lsp-cache, keyed by content hash+size, invalidated on didChange/didSave; resolves from-import, module-attribute and wildcard calls without building the workspace index. Extends LspTypes with list/array/pointer structural checks, array/string initializer lengths and list-literal element checks; adds return-type consistency, augmented-assignment narrowing, constant folding for unary/binop and string/list subscripts. Diagnostics use a per-line occurrence bucket, a dedicated cache work pool and statement/expression budgets so large include files stay responsive. gate13 covers the new rules.
2483 lines
115 KiB
Python
2483 lines
115 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
|
|
import time
|
|
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}")
|
|
|
|
# Lexer overflow: exactly one diagnostic, the range covers the whole literal, and the rest of
|
|
# the file keeps its symbols/semantic tokens (recovery parsing).
|
|
overflow = "def a(l: int):\n pass\n\ndef main():\n a(1024000000000000000000)\n"
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 3, "text": overflow}},
|
|
)
|
|
publish = client.wait_notification("textDocument/publishDiagnostics")
|
|
diagnostics = publish.get("params", {}).get("diagnostics")
|
|
if not isinstance(diagnostics, list) or len(diagnostics) != 1:
|
|
raise AssertionError(f"expected exactly one overflow diagnostic: {publish}")
|
|
overflow_diag = diagnostics[0]
|
|
if overflow_diag["range"]["start"] != {"line": 4, "character": 6} or \
|
|
overflow_diag["range"]["end"] != {"line": 4, "character": 28}:
|
|
raise AssertionError(f"overflow range does not cover the literal: {overflow_diag}")
|
|
if "1024000000000000000000" not in overflow_diag["message"]:
|
|
raise AssertionError(f"overflow diagnostic lost the literal text: {overflow_diag}")
|
|
sem = client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": uri}}, timeout=120.0
|
|
).get("result")
|
|
sem_tokens = _decode_tokens(sem.get("data", [])) if isinstance(sem, dict) else []
|
|
legend = response.get("result", {}).get("capabilities", {}) \
|
|
.get("semanticTokensProvider", {}).get("legend", {}).get("tokenTypes", [])
|
|
function_type = legend.index("function")
|
|
functions = {(t["line"], t["character"]) for t in sem_tokens if t["type"] == function_type}
|
|
if (0, 4) not in functions or (3, 4) not in functions:
|
|
raise AssertionError(f"semantic tokens lost after a lexer error: {sem_tokens}")
|
|
# Fixing the literal must clear the overflow diagnostic (lexer fail-fast state is global and
|
|
# must be reset for the next parse, otherwise the stale error sticks to every later edit).
|
|
fixed_overflow = overflow.replace("1024000000000000000000", "1")
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{"textDocument": {"uri": uri, "version": 4},
|
|
"contentChanges": [{"text": fixed_overflow}]},
|
|
)
|
|
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 overflow literal still reports diagnostics: {publish}")
|
|
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")
|
|
|
|
# A pull-diagnostics client must not also receive publishDiagnostics (that shows every
|
|
# problem twice in the Problems panel).
|
|
pull_client = Client(exe)
|
|
pull_client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root.as_uri(),
|
|
"capabilities": {"textDocument": {"diagnostic": {}}}},
|
|
)
|
|
pull_client.notify("initialized", {})
|
|
pull_client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 4, "text": overflow}},
|
|
)
|
|
time.sleep(0.4)
|
|
pushed: list[dict[str, Any]] = []
|
|
while not pull_client.reader.messages.empty():
|
|
pushed.append(pull_client.reader.messages.get())
|
|
if any(m.get("method") == "textDocument/publishDiagnostics" for m in pushed):
|
|
raise AssertionError(f"pull client received duplicate push diagnostics: {pushed}")
|
|
pulled = pull_client.request(
|
|
"textDocument/diagnostic", {"textDocument": {"uri": uri}}, timeout=60.0
|
|
).get("result")
|
|
items = pulled.get("items", []) if isinstance(pulled, dict) else []
|
|
if len(items) != 1:
|
|
raise AssertionError(f"pull diagnostics returned {len(items)} items: {pulled}")
|
|
pull_client.notify(
|
|
"textDocument/didChange",
|
|
{"textDocument": {"uri": uri, "version": 5},
|
|
"contentChanges": [{"text": fixed_overflow}]},
|
|
)
|
|
pulled = pull_client.request(
|
|
"textDocument/diagnostic", {"textDocument": {"uri": uri}}, timeout=60.0
|
|
).get("result")
|
|
items = pulled.get("items", []) if isinstance(pulled, dict) else []
|
|
if len(items) != 0:
|
|
raise AssertionError(f"pull diagnostics kept the stale overflow: {pulled}")
|
|
pull_client.request("shutdown")
|
|
pull_client.close()
|
|
|
|
# `vipeye.disablePullDiagnostics` must hide diagnosticProvider and fall back to push.
|
|
push_client = Client(exe)
|
|
init = push_client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": root.as_uri(),
|
|
"capabilities": {"textDocument": {"diagnostic": {}}},
|
|
"initializationOptions": {"disablePullDiagnostics": True}},
|
|
)
|
|
caps = init.get("result", {}).get("capabilities", {})
|
|
if "diagnosticProvider" in caps:
|
|
raise AssertionError(f"disablePullDiagnostics still advertises pull: {init}")
|
|
push_client.notify("initialized", {})
|
|
push_client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 6, "text": overflow}},
|
|
)
|
|
publish = push_client.wait_notification("textDocument/publishDiagnostics")
|
|
if len(publish.get("params", {}).get("diagnostics", [])) != 1:
|
|
raise AssertionError(f"disablePullDiagnostics did not push: {publish}")
|
|
push_client.request("shutdown")
|
|
push_client.close()
|
|
|
|
print("diagnostics: OK")
|
|
print("recovery: OK")
|
|
print("overflow-diagnostics: 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},
|
|
},
|
|
timeout=30.0,
|
|
).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"},
|
|
timeout=30.0,
|
|
).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}")
|
|
|
|
# Project-relative imports must resolve before any index rebuild: the adopted project's
|
|
# source_dir/includes have to replace the repository-root fallback roots.
|
|
app_doc = root / "App" / "lib" / "lsp" / "LspDocuments.vp"
|
|
app_text = app_doc.read_text(encoding="utf-8")
|
|
app_uri = app_doc.as_uri()
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": app_uri, "languageId": "viper", "version": 1, "text": app_text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
pool_pos = _position(app_text, "LspPools.LspPool")
|
|
pool_pos["character"] += len("LspPools.")
|
|
imported_def = client.request(
|
|
"textDocument/definition", {"textDocument": {"uri": app_uri}, "position": pool_pos},
|
|
timeout=60.0,
|
|
).get("result")
|
|
if not isinstance(imported_def, list) or not imported_def:
|
|
raise AssertionError(f"project import did not resolve after adoption: {imported_def}")
|
|
if not imported_def[0]["uri"].endswith("LspPools.vp"):
|
|
raise AssertionError(f"project import resolved to the wrong file: {imported_def}")
|
|
legend = response.get("result", {}).get("capabilities", {}).get(
|
|
"semanticTokensProvider", {}).get("legend", {})
|
|
sem_types = legend.get("tokenTypes", [])
|
|
semantic = client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": app_uri}}, timeout=60.0
|
|
).get("result")
|
|
sem_tokens = _decode_tokens(semantic.get("data", [])) if isinstance(semantic, dict) else []
|
|
if not _find_token(sem_tokens, pool_pos["line"], pool_pos["character"],
|
|
sem_types.index("class")):
|
|
raise AssertionError(f"imported class was not highlighted: {pool_pos}")
|
|
|
|
mem_pos = _position(app_text, "memhub.MemManager")
|
|
mem_pos["character"] += len("memhub.")
|
|
mem_def = client.request(
|
|
"textDocument/definition", {"textDocument": {"uri": app_uri}, "position": mem_pos},
|
|
timeout=60.0,
|
|
).get("result")
|
|
if not isinstance(mem_def, list) or not mem_def:
|
|
raise AssertionError(f"includes definition failed after adoption: {mem_def}")
|
|
if not mem_def[0]["uri"].endswith("memhub.vp"):
|
|
raise AssertionError(f"includes definition target is wrong: {mem_def}")
|
|
if not _find_token(sem_tokens, mem_pos["line"], mem_pos["character"],
|
|
sem_types.index("class")):
|
|
raise AssertionError(f"includes class was not highlighted: {mem_pos}")
|
|
|
|
# A large include-root document must tokenize through the same project roots without
|
|
# exhausting the request pool via on-demand re-parsing.
|
|
parser_doc = root.parent / "includes" / "ast" / "parser.vp"
|
|
parser_text = parser_doc.read_text(encoding="utf-8")
|
|
parser_uri = parser_doc.as_uri()
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": parser_uri, "languageId": "viper", "version": 1,
|
|
"text": parser_text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
parser_semantic = client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": parser_uri}},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(parser_semantic, dict) or not parser_semantic.get("data"):
|
|
raise AssertionError(f"include-root semantic tokens failed: {parser_semantic}")
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": parser_uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": app_uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
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}")
|
|
|
|
# Attribute bases stay in source order and keep their token (`self.value`).
|
|
self_use = _position(text, "return self.value")
|
|
self_use["character"] += len("return ")
|
|
if not _find_token(tokens, self_use["line"], self_use["character"], 4):
|
|
raise AssertionError(f"missing attribute base token: {self_use}")
|
|
|
|
# Builtin value types and array element/container types use the class token.
|
|
class_idx = types.index("class")
|
|
builtin_cases = (
|
|
("x: int", 3, "int"),
|
|
("s: str", 3, "str"),
|
|
("xs: []int", 6, "array element int"),
|
|
("ys: [4]str", 7, "fixed array element str"),
|
|
("zs: list[int]", 4, "list container"),
|
|
("zs: list[int]", 9, "list element int"),
|
|
("[2]int", 3, "array return element int"),
|
|
)
|
|
for needle, delta, label in builtin_cases:
|
|
pos = _position(text, needle)
|
|
pos["character"] += delta
|
|
if not _find_token(tokens, pos["line"], pos["character"], class_idx):
|
|
raise AssertionError(f"missing builtin type token for {label}: {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 gate10(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 the module, the stub and the consumer.
|
|
workspace = Path(tempfile.mkdtemp(prefix="transpyc_lsp_gate10_"))
|
|
names = ("gate10_types.vp", "gate10_stub.vpi", "gate10_extra.vp", "gate10_use.vp",
|
|
"gate10_wild_leaf.vp", "gate10_wild_reexport.vp",
|
|
"gate10_wild_b.vp", "gate10_wild_c.vp")
|
|
for name in names:
|
|
shutil.copy(fixtures / name, workspace / name)
|
|
shutil.copytree(fixtures / "gate10_pkg", workspace / "gate10_pkg")
|
|
type_path = workspace / "gate10_types.vp"
|
|
stub_path = workspace / "gate10_stub.vpi"
|
|
use_path = workspace / "gate10_use.vp"
|
|
uri = use_path.as_uri()
|
|
text = use_path.read_text(encoding="utf-8")
|
|
texts = {name: (workspace / name).read_text(encoding="utf-8") for name in names}
|
|
|
|
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, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
diagnostics = client.wait_notification("textDocument/publishDiagnostics")
|
|
reported = diagnostics.get("params", {}).get("diagnostics", [])
|
|
if reported:
|
|
raise AssertionError(f"gate10 fixture has diagnostics: {reported}")
|
|
|
|
legend = response.get("result", {}).get("capabilities", {}).get("semanticTokensProvider", {}).get("legend", {})
|
|
sem_types = legend.get("tokenTypes", [])
|
|
sem_mods = legend.get("tokenModifiers", [])
|
|
semantic = client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": uri}}, timeout=120.0
|
|
).get("result")
|
|
if not isinstance(semantic, dict):
|
|
raise AssertionError(f"semanticTokens/full failed: {semantic}")
|
|
sem_tokens = _decode_tokens(semantic.get("data", []))
|
|
for label, needle, delta, token_name in (
|
|
("imported class", "import Widget", 7, "class"),
|
|
("imported macro", "import Widget", 15, "macro"),
|
|
("macro call", "return DOUBLE", 7, "macro"),
|
|
("member field", "s.value", 2, "property"),
|
|
("member method", "s.resize", 2, "method"),
|
|
("star class", "e: Extra", 3, "class"),
|
|
("star function", "return extra_fn", 7, "function"),
|
|
("star constant", "EXTRA_CONST", 0, "variable"),
|
|
("star member", "e.amount", 2, "property"),
|
|
):
|
|
pos = _position(text, needle)
|
|
pos["character"] += delta
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"], sem_types.index(token_name)):
|
|
raise AssertionError(f"missing cross-module semantic token for {label}: {pos}")
|
|
for needle, label in (("return extra_fn(EXTRA_CONST + 1)", "else block"),
|
|
("return extra_fn(EXTRA_CONST + 2)", "inline else")):
|
|
pos = _position(text, needle)
|
|
pos["character"] += 7
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"], sem_types.index("function")):
|
|
raise AssertionError(f"missing cross-module token in {label}: {pos}")
|
|
else_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": pos},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(else_def, list) or not else_def:
|
|
raise AssertionError(f"missing definition in {label}: {else_def}")
|
|
if not else_def[0]["uri"].replace("\\", "/").endswith("gate10_extra.vp"):
|
|
raise AssertionError(f"wrong definition in {label}: {else_def}")
|
|
re_pos = _position(text, "gate10_pkg.reexport_fn")
|
|
re_pos["character"] += len("gate10_pkg.")
|
|
if not _find_token(sem_tokens, re_pos["line"], re_pos["character"], sem_types.index("function")):
|
|
raise AssertionError(f"package re-export member is not highlighted: {re_pos}")
|
|
re_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": re_pos},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(re_def, list) or not re_def:
|
|
raise AssertionError(f"package re-export definition failed: {re_def}")
|
|
if not re_def[0]["uri"].replace("\\", "/").endswith("gate10_pkg/leaf.vp"):
|
|
raise AssertionError(f"package re-export definition is wrong: {re_def}")
|
|
for needle, delta, label in (
|
|
("return extra_fn(EXTRA_CONST) if", 7, "ternary body"),
|
|
("else extra_fn(EXTRA_CONST + 1)", 5, "ternary else"),
|
|
):
|
|
pos = _position(text, needle)
|
|
pos["character"] += delta
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"], sem_types.index("function")):
|
|
raise AssertionError(f"missing function token in {label}: {pos}")
|
|
t_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": pos},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(t_def, list) or not t_def:
|
|
raise AssertionError(f"missing definition in {label}: {t_def}")
|
|
if not t_def[0]["uri"].replace("\\", "/").endswith("gate10_extra.vp"):
|
|
raise AssertionError(f"wrong definition in {label}: {t_def}")
|
|
wild_pos = _position(text, "gate10_wild_c.wild_b")
|
|
wild_pos["character"] += len("gate10_wild_c.")
|
|
if not _find_token(sem_tokens, wild_pos["line"], wild_pos["character"], sem_types.index("function")):
|
|
raise AssertionError(f"wildcard re-export chain member is not highlighted: {wild_pos}")
|
|
wild_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": wild_pos},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(wild_def, list) or not wild_def:
|
|
raise AssertionError(f"wildcard re-export chain definition failed: {wild_def}")
|
|
if not wild_def[0]["uri"].replace("\\", "/").endswith("gate10_wild_leaf.vp"):
|
|
raise AssertionError(f"wildcard re-export chain definition is wrong: {wild_def}")
|
|
for needle, delta, label in (
|
|
("import extra_fn as aliased_fn", len("import "), "aliased import original name"),
|
|
("import extra_fn as aliased_fn", len("import extra_fn as "), "aliased import alias"),
|
|
("return aliased_fn", len("return "), "aliased import use"),
|
|
):
|
|
pos = _position(text, needle)
|
|
pos["character"] += delta
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"],
|
|
sem_types.index("function")):
|
|
raise AssertionError(f"{label} is not highlighted: {pos}")
|
|
a_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": pos},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(a_def, list) or not a_def:
|
|
raise AssertionError(f"{label} definition failed: {a_def}")
|
|
if not a_def[0]["uri"].replace("\\", "/").endswith("gate10_extra.vp"):
|
|
raise AssertionError(f"{label} definition is wrong: {a_def}")
|
|
const_pos = _position(text, "EXTRA_CONST")
|
|
if not _find_token_mods(sem_tokens, const_pos["line"], const_pos["character"],
|
|
sem_types.index("variable"),
|
|
1 << sem_mods.index("readonly")):
|
|
raise AssertionError(f"wildcard constant is not readonly: {const_pos}")
|
|
star_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": {"line": _position(text, "e: Extra")["line"],
|
|
"character": _position(text, "e: Extra")["character"] + 3}},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(star_def, list) or not star_def:
|
|
raise AssertionError(f"wildcard definition failed: {star_def}")
|
|
if not star_def[0]["uri"].endswith("gate10_extra.vp"):
|
|
raise AssertionError(f"wildcard definition target is wrong: {star_def}")
|
|
|
|
# Dotted module paths: every segment is a namespace and ctrl-click opens that module file.
|
|
dotted = _position(text, "gate10_pkg.sub.mod")
|
|
for delta, segment in ((0, "pkg"), (len("gate10_pkg."), "sub"),
|
|
(len("gate10_pkg.sub."), "mod")):
|
|
pos = {"line": dotted["line"], "character": dotted["character"] + delta}
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"],
|
|
sem_types.index("namespace")):
|
|
raise AssertionError(f"dotted import segment is not highlighted: {segment} {pos}")
|
|
for delta, suffix in ((len("gate10_pkg."), "gate10_pkg/sub/__init__.vp"),
|
|
(len("gate10_pkg.sub."), "gate10_pkg/sub/mod.vp")):
|
|
pos = {"line": dotted["line"], "character": dotted["character"] + delta}
|
|
target = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": pos},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(target, list) or not target:
|
|
raise AssertionError(f"dotted import definition failed: {pos} -> {target}")
|
|
got = target[0]["uri"].replace("\\", "/")
|
|
if not got.endswith(suffix):
|
|
raise AssertionError(f"dotted import definition target is wrong: {pos} -> {got}")
|
|
|
|
alias_line = _position(text, "import gate10_pkg.sub as pkgsub")
|
|
for delta, segment in ((7, "gate10_pkg"), (18, "sub"), (25, "pkgsub")):
|
|
pos = {"line": alias_line["line"], "character": alias_line["character"] + delta}
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"],
|
|
sem_types.index("namespace")):
|
|
raise AssertionError(f"aliased dotted import is not highlighted: {segment} {pos}")
|
|
alias_sub = {"line": alias_line["line"], "character": alias_line["character"] + 18}
|
|
alias_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": alias_sub},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(alias_def, list) or not alias_def:
|
|
raise AssertionError(f"aliased dotted import definition failed: {alias_def}")
|
|
if not alias_def[0]["uri"].replace("\\", "/").endswith("gate10_pkg/sub/__init__.vp"):
|
|
raise AssertionError(f"aliased dotted import target is wrong: {alias_def}")
|
|
|
|
# Alias repeating the last path segment must still be a separate occurrence at the `as` end.
|
|
same_line = _position(text, "import gate10_pkg.sub as sub")
|
|
for delta, segment in ((7, "gate10_pkg"), (18, "sub"), (25, "sub")):
|
|
pos = {"line": same_line["line"], "character": same_line["character"] + delta}
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"],
|
|
sem_types.index("namespace")):
|
|
raise AssertionError(f"repeated alias import is not highlighted: {segment} {pos}")
|
|
same_alias = {"line": same_line["line"], "character": same_line["character"] + 25}
|
|
same_def = client.request(
|
|
"textDocument/definition",
|
|
{"textDocument": {"uri": uri}, "position": same_alias},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(same_def, list) or not same_def:
|
|
raise AssertionError(f"repeated alias definition failed: {same_def}")
|
|
if not same_def[0]["uri"].replace("\\", "/").endswith("gate10_pkg/sub/__init__.vp"):
|
|
raise AssertionError(f"repeated alias target is wrong: {same_def}")
|
|
|
|
ghost_line = _position(text, "import ghost_lib.path")
|
|
for delta, segment in ((7, "ghost_lib"), (17, "path")):
|
|
pos = {"line": ghost_line["line"], "character": ghost_line["character"] + delta}
|
|
if not _find_token(sem_tokens, pos["line"], pos["character"],
|
|
sem_types.index("namespace")):
|
|
raise AssertionError(f"unresolved module path is not highlighted: {segment} {pos}")
|
|
|
|
def at(needle: str, delta: int = 0, occurrence: int = 0) -> dict[str, int]:
|
|
pos = _position(text, needle, occurrence)
|
|
pos["character"] += delta
|
|
return pos
|
|
|
|
def request(method: str, pos: dict[str, int], extra: dict[str, Any] | None = None,
|
|
timeout: float = 90.0) -> Any:
|
|
params: dict[str, Any] = {"textDocument": {"uri": uri}, "position": pos}
|
|
if extra:
|
|
params.update(extra)
|
|
return client.request(method, params, timeout=timeout).get("result")
|
|
|
|
def hover_text(pos: dict[str, int]) -> str:
|
|
result = request("textDocument/hover", pos)
|
|
if not isinstance(result, dict):
|
|
return ""
|
|
return result.get("contents", {}).get("value", "")
|
|
|
|
def definition_of(pos: dict[str, int]) -> tuple[str, int, int]:
|
|
result = request("textDocument/definition", pos)
|
|
if not isinstance(result, list) or len(result) != 1:
|
|
raise AssertionError(f"definition failed: {result}")
|
|
item = result[0]
|
|
start = item["range"]["start"]
|
|
return Path(item["uri"]).name, start["line"], start["character"]
|
|
|
|
expected_field = _position(texts["gate10_types.vp"], "value: int")
|
|
expected_resize = _position(texts["gate10_types.vp"], "def resize")
|
|
expected_resize["character"] += len("def ")
|
|
expected_run = _position(texts["gate10_stub.vpi"], "def run")
|
|
expected_run["character"] += len("def ")
|
|
expected_count = _position(texts["gate10_stub.vpi"], "count: int")
|
|
|
|
# Annotated parameter of a from-import class resolves to the field/method declaration.
|
|
value_pos = at("s.value", 2)
|
|
if "int" not in hover_text(value_pos):
|
|
raise AssertionError(f"hover on s.value failed: {hover_text(value_pos)}")
|
|
if definition_of(value_pos) != ("gate10_types.vp", expected_field["line"], expected_field["character"]):
|
|
raise AssertionError(f"definition on s.value failed: {definition_of(value_pos)}")
|
|
prepared = request("textDocument/prepareRename", value_pos)
|
|
if not isinstance(prepared, dict) or prepared.get("placeholder") != "value":
|
|
raise AssertionError(f"prepareRename on s.value failed: {prepared}")
|
|
|
|
resize_pos = at("s.resize", 2)
|
|
if "def resize(self, width: int) -> int" not in hover_text(resize_pos):
|
|
raise AssertionError(f"hover on s.resize failed: {hover_text(resize_pos)}")
|
|
if definition_of(resize_pos) != ("gate10_types.vp", expected_resize["line"], expected_resize["character"]):
|
|
raise AssertionError(f"definition on s.resize failed: {definition_of(resize_pos)}")
|
|
|
|
# Annotation through an `import module` path: module.Class.value / module.Class.method.
|
|
if "int" not in hover_text(at("w.value", 2)):
|
|
raise AssertionError("hover on w.value failed")
|
|
if definition_of(at("w.value", 2))[0] != "gate10_types.vp":
|
|
raise AssertionError("definition on w.value missed the field")
|
|
if "def resize" not in hover_text(at("w.resize", 2)):
|
|
raise AssertionError("hover on w.resize failed")
|
|
if "int" not in hover_text(at("Widget.value", 7)):
|
|
raise AssertionError("hover on module.Class.field failed")
|
|
if definition_of(at("Widget.value", 7))[0] != "gate10_types.vp":
|
|
raise AssertionError("definition on module.Class.field missed the field")
|
|
|
|
# .vpi members resolve through the same chain.
|
|
if "def run" not in hover_text(at("t.run", 2)):
|
|
raise AssertionError("hover on .vpi method failed")
|
|
if definition_of(at("t.run", 2)) != ("gate10_stub.vpi", expected_run["line"], expected_run["character"]):
|
|
raise AssertionError(f"definition on .vpi method failed: {definition_of(at('t.run', 2))}")
|
|
if definition_of(at("t.count", 2)) != ("gate10_stub.vpi", expected_count["line"], expected_count["character"]):
|
|
raise AssertionError(f"definition on .vpi field failed: {definition_of(at('t.count', 2))}")
|
|
if definition_of(at("StubThing.count", 10))[0] != "gate10_stub.vpi":
|
|
raise AssertionError("definition on .vpi class field chain failed")
|
|
|
|
# Literal hovers carry the numeric spelling, storage widths and float representability.
|
|
module_hover = hover_text(at("import gate10_types", 7))
|
|
if "module gate10_types" not in module_hover or "gate10_types.vp" not in module_hover:
|
|
raise AssertionError(f"module hover lacks name/path: {module_hover}")
|
|
int_hover = hover_text(at("return 128", 7))
|
|
for piece in ("dec 128", "hex 0x80", "binary 0b1000_0000",
|
|
"bits 8 unsigned, 9 signed", "uint u8", "int i16",
|
|
"float f16 exact, f32 exact, f64 exact"):
|
|
if piece not in int_hover:
|
|
raise AssertionError(f"integer hover lacks {piece!r}: {int_hover}")
|
|
hex_hover = hover_text(at("return 0x10", 7))
|
|
for piece in ("dec 16", "hex 0x10", "binary 0b1_0000"):
|
|
if piece not in hex_hover:
|
|
raise AssertionError(f"hex hover lacks {piece!r}: {hex_hover}")
|
|
bin_hover = hover_text(at("return 0b1010", 7))
|
|
for piece in ("dec 10", "hex 0xA", "binary 0b1010"):
|
|
if piece not in bin_hover:
|
|
raise AssertionError(f"binary hover lacks {piece!r}: {bin_hover}")
|
|
sep_hover = hover_text(at("return 1_000", 7))
|
|
if "dec 1000" not in sep_hover:
|
|
raise AssertionError(f"digit separator changed the value: {sep_hover}")
|
|
for needle, expected in (
|
|
("return 128_u8", "declared u8 (fits)"),
|
|
("return 1_u100", "declared u100 (fits)"),
|
|
("return 128_u7", "declared u7 (overflow)"),
|
|
):
|
|
suffix_hover = hover_text(at(needle, 7))
|
|
if expected not in suffix_hover:
|
|
raise AssertionError(f"suffix hover lacks {expected!r}: {suffix_hover}")
|
|
float_hover = hover_text(at("return 3.5", 7))
|
|
if "hex 0x400C000000000000" not in float_hover:
|
|
raise AssertionError(f"float hover has the wrong bit pattern: {float_hover}")
|
|
if "f16 exact, f32 exact, f64 exact" not in float_hover:
|
|
raise AssertionError(f"float hover misreports exactness: {float_hover}")
|
|
round_hover = hover_text(at("return 3.14", 7))
|
|
if "hex 0x40091EB851EB851F" not in round_hover:
|
|
raise AssertionError(f"float hover has the wrong bit pattern: {round_hover}")
|
|
if "f16 rounded, f32 rounded, f64 exact" not in round_hover:
|
|
raise AssertionError(f"float hover misreports exactness: {round_hover}")
|
|
string_hover = hover_text(at('return len("hello")', len("return len(")))
|
|
for piece in ("type str", "length 5 bytes"):
|
|
if piece not in string_hover:
|
|
raise AssertionError(f"string hover lacks {piece!r}: {string_hover}")
|
|
|
|
def reference_ranges(pos: dict[str, int]) -> list[tuple[str, int, int]]:
|
|
result = request("textDocument/references", pos,
|
|
{"context": {"includeDeclaration": True}})
|
|
if not isinstance(result, list):
|
|
raise AssertionError(f"references failed: {result}")
|
|
return [
|
|
(Path(item["uri"]).name,
|
|
item["range"]["start"]["line"],
|
|
item["range"]["start"]["character"])
|
|
for item in result
|
|
]
|
|
|
|
field_refs = reference_ranges(value_pos)
|
|
files = {entry[0] for entry in field_refs}
|
|
if "gate10_types.vp" not in files or "gate10_use.vp" not in files:
|
|
raise AssertionError(f"field references did not cross files: {field_refs}")
|
|
declaration = ("gate10_types.vp", expected_field["line"], expected_field["character"])
|
|
if field_refs.count(declaration) != 1:
|
|
raise AssertionError(f"field declaration was not reported exactly once: {field_refs}")
|
|
if len(field_refs) != len(set(field_refs)):
|
|
raise AssertionError(f"field references contain duplicates: {field_refs}")
|
|
|
|
method_refs = reference_ranges(resize_pos)
|
|
if {"gate10_types.vp", "gate10_use.vp"} - {entry[0] for entry in method_refs}:
|
|
raise AssertionError(f"method references did not cross files: {method_refs}")
|
|
stub_refs = reference_ranges(at("t.run", 2))
|
|
if {"gate10_stub.vpi", "gate10_use.vp"} - {entry[0] for entry in stub_refs}:
|
|
raise AssertionError(f".vpi method references did not cross files: {stub_refs}")
|
|
|
|
def rename(pos: dict[str, int], new_name: str) -> dict[str, list[dict[str, Any]]]:
|
|
result = request("textDocument/rename", pos, {"newName": new_name})
|
|
if not isinstance(result, dict):
|
|
raise AssertionError(f"rename to {new_name} failed: {result}")
|
|
changes = result.get("changes", {})
|
|
for furi, edits in changes.items():
|
|
seen: set[tuple[int, int, int, int]] = set()
|
|
for edit in edits:
|
|
start = edit["range"]["start"]
|
|
end = edit["range"]["end"]
|
|
key = (start["line"], start["character"], end["line"], end["character"])
|
|
if key in seen:
|
|
raise AssertionError(f"duplicate edit in {furi}: {key}")
|
|
seen.add(key)
|
|
return changes
|
|
|
|
field_changes = rename(value_pos, "data")
|
|
updated = _apply_workspace(field_changes, texts)
|
|
if "data: int" not in updated["gate10_types.vp"]:
|
|
raise AssertionError(f"field rename missed the declaration: {updated['gate10_types.vp']}")
|
|
if "self.data = value" not in updated["gate10_types.vp"]:
|
|
raise AssertionError(f"field rename missed a write: {updated['gate10_types.vp']}")
|
|
if "def __init__(self, value: int)" not in updated["gate10_types.vp"]:
|
|
raise AssertionError("field rename touched the same-named parameter")
|
|
if "s.data" not in updated["gate10_use.vp"] or "w.data" not in updated["gate10_use.vp"]:
|
|
raise AssertionError(f"field rename did not reach the consumer: {updated['gate10_use.vp']}")
|
|
|
|
method_changes = rename(resize_pos, "draw")
|
|
updated = _apply_workspace(method_changes, texts)
|
|
if "def draw(self, width: int)" not in updated["gate10_types.vp"]:
|
|
raise AssertionError(f"method rename missed the declaration: {updated['gate10_types.vp']}")
|
|
if "s.draw(v)" not in updated["gate10_use.vp"] or "w.draw(w.value)" not in updated["gate10_use.vp"]:
|
|
raise AssertionError(f"method rename did not reach the consumer: {updated['gate10_use.vp']}")
|
|
|
|
stub_changes = rename(at("t.run", 2), "walk")
|
|
updated = _apply_workspace(stub_changes, texts)
|
|
if "def walk(self, steps: int)" not in updated["gate10_stub.vpi"]:
|
|
raise AssertionError(f".vpi method rename missed the declaration: {updated['gate10_stub.vpi']}")
|
|
if "t.walk(t.count)" not in updated["gate10_use.vp"]:
|
|
raise AssertionError(f".vpi method rename missed the call: {updated['gate10_use.vp']}")
|
|
|
|
count_changes = rename(at("t.count", 2), "total")
|
|
updated = _apply_workspace(count_changes, texts)
|
|
if "total: int" not in updated["gate10_stub.vpi"]:
|
|
raise AssertionError(f".vpi field rename missed the declaration: {updated['gate10_stub.vpi']}")
|
|
if "t.run(t.total)" not in updated["gate10_use.vp"] or "StubThing.total" not in updated["gate10_use.vp"]:
|
|
raise AssertionError(f".vpi field rename missed a use: {updated['gate10_use.vp']}")
|
|
|
|
# Renaming from the declaration uses the same cross-file identity.
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
type_uri = type_path.as_uri()
|
|
type_text = texts["gate10_types.vp"]
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": type_uri, "languageId": "viper", "version": 1, "text": type_text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
decl_pos = _position(type_text, "value: int")
|
|
result = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": type_uri}, "position": decl_pos, "newName": "data"},
|
|
timeout=90.0,
|
|
).get("result")
|
|
if not isinstance(result, dict):
|
|
raise AssertionError(f"rename from the declaration failed: {result}")
|
|
updated = _apply_workspace(result.get("changes", {}), texts)
|
|
if "class Widget:\n data: int" not in updated["gate10_types.vp"]:
|
|
raise AssertionError(f"declaration rename missed the field: {updated['gate10_types.vp']}")
|
|
if "s.data" not in updated["gate10_use.vp"]:
|
|
raise AssertionError(f"declaration rename missed the consumer: {updated['gate10_use.vp']}")
|
|
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": type_uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
client.request("shutdown")
|
|
code = client.close()
|
|
if code != 0:
|
|
raise AssertionError(f"exit code is {code}, expected 0")
|
|
|
|
# VS Code's URI serializer percent-encodes the drive colon (file:///d%3A/...). A decoded
|
|
# path starting with "/" prevents project adoption and leaves the index roots empty.
|
|
def vscode_uri(path: Path) -> str:
|
|
p = path.resolve().as_posix()
|
|
if len(p) >= 2 and p[1] == ":":
|
|
p = p[0].lower() + "%3A" + p[2:]
|
|
return "file:///" + p
|
|
|
|
enc_client = Client(exe)
|
|
enc_init = enc_client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": vscode_uri(workspace), "capabilities": {}},
|
|
)
|
|
if not isinstance(enc_init.get("result"), dict):
|
|
raise AssertionError(f"encoded-URI initialize failed: {enc_init}")
|
|
enc_client.notify("initialized", {})
|
|
enc_client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": vscode_uri(use_path), "languageId": "viper",
|
|
"version": 1, "text": text}},
|
|
)
|
|
enc_client.wait_notification("textDocument/publishDiagnostics")
|
|
enc_stats = enc_client.request("viper/lspStats").get("result") or {}
|
|
enc_source = str(enc_stats.get("sourceDir") or "")
|
|
if enc_source.startswith("/"):
|
|
raise AssertionError(f"encoded URI root was not decoded: {enc_stats}")
|
|
if Path(enc_source).resolve() != workspace.resolve():
|
|
raise AssertionError(f"encoded URI sourceDir mismatch: {enc_stats}")
|
|
enc_sem = enc_client.request(
|
|
"textDocument/semanticTokens/full", {"textDocument": {"uri": vscode_uri(use_path)}},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(enc_sem, dict):
|
|
raise AssertionError(f"encoded URI semanticTokens failed: {enc_sem}")
|
|
enc_tokens = _decode_tokens(enc_sem.get("data", []))
|
|
enc_pos = _position(text, "s.value")
|
|
enc_pos["character"] += 2
|
|
if not _find_token(enc_tokens, enc_pos["line"], enc_pos["character"], sem_types.index("property")):
|
|
raise AssertionError(f"encoded URI lost cross-file member token: {enc_pos}")
|
|
enc_client.request("shutdown")
|
|
if enc_client.close() != 0:
|
|
raise AssertionError("encoded URI client did not exit cleanly")
|
|
|
|
shutil.rmtree(workspace, ignore_errors=True)
|
|
print("member-annotations: OK")
|
|
print("member-module-chain: OK")
|
|
print("member-vpi: OK")
|
|
print("member-rename: OK")
|
|
return 0
|
|
|
|
|
|
def gate11(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
fixtures = root / "Tools" / "lsp" / "fixtures"
|
|
workspace = Path(tempfile.mkdtemp(prefix="transpyc_lsp_gate11_"))
|
|
names = ("gate11_types.vp", "gate11_use.vp", "gate11_other.vp")
|
|
for name in names:
|
|
shutil.copy(fixtures / name, workspace / name)
|
|
type_path = workspace / names[0]
|
|
use_path = workspace / names[1]
|
|
other_path = workspace / names[2]
|
|
texts = {name: (workspace / name).read_text(encoding="utf-8") for name in names}
|
|
|
|
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", {})
|
|
for path, name in ((type_path, names[0]), (use_path, names[1])):
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": path.as_uri(), "languageId": "viper", "version": 1,
|
|
"text": texts[name]}},
|
|
)
|
|
diagnostics = client.wait_notification("textDocument/publishDiagnostics")
|
|
reported = diagnostics.get("params", {}).get("diagnostics", [])
|
|
if reported:
|
|
raise AssertionError(f"{name} has diagnostics: {reported}")
|
|
|
|
use_uri = use_path.as_uri()
|
|
|
|
def references(text: str, needle: str, delta: int = 0,
|
|
occurrence: int = 0) -> tuple[list[dict[str, Any]], float]:
|
|
pos = _position(text, needle, occurrence)
|
|
pos["character"] += delta
|
|
started = time.time()
|
|
result = client.request(
|
|
"textDocument/references",
|
|
{"textDocument": {"uri": use_uri}, "position": pos,
|
|
"context": {"includeDeclaration": True}},
|
|
timeout=120.0,
|
|
).get("result")
|
|
elapsed = time.time() - started
|
|
if not isinstance(result, list):
|
|
raise AssertionError(f"references failed: {result}")
|
|
return result, elapsed
|
|
|
|
initial, initial_sec = references(texts[names[1]], "item.value", 5)
|
|
files = {Path(item["uri"]).name for item in initial}
|
|
if files != set(names):
|
|
raise AssertionError(f"initial references missed files: {files}")
|
|
if len(initial) != len({(item["uri"], item["range"]["start"]["line"],
|
|
item["range"]["start"]["character"]) for item in initial}):
|
|
raise AssertionError(f"initial references contain duplicates: {initial}")
|
|
|
|
# Unsaved edits in both open documents: prepend lines so every indexed position shifts.
|
|
type_text = "\n" + texts[names[0]]
|
|
use_text = ("# shifted\n" + texts[names[1]] +
|
|
"\n\ndef extra(item: Item) -> int:\n return item.value\n")
|
|
for path, text in ((type_path, type_text), (use_path, use_text)):
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{"textDocument": {"uri": path.as_uri(), "version": 2},
|
|
"contentChanges": [{"text": text}]},
|
|
)
|
|
diagnostics = client.wait_notification("textDocument/publishDiagnostics")
|
|
reported = diagnostics.get("params", {}).get("diagnostics", [])
|
|
if reported:
|
|
raise AssertionError(f"unsaved edit produced diagnostics: {reported}")
|
|
|
|
current, current_sec = references(use_text, "item.value", 5)
|
|
expected_decl = _position(type_text, "value: int")
|
|
expected_extra = _position(use_text, "item.value", 2)
|
|
expected_extra["character"] += 5
|
|
decl_hits = 0
|
|
has_extra = False
|
|
for item in current:
|
|
name = Path(item["uri"]).name
|
|
start = item["range"]["start"]
|
|
if name == names[0]:
|
|
source = type_text
|
|
elif name == names[1]:
|
|
source = use_text
|
|
else:
|
|
source = texts[names[2]]
|
|
line = source.split("\n")[start["line"]]
|
|
if line[start["character"]:start["character"] + 5] != "value":
|
|
raise AssertionError(f"stale reference position: {name} {start}")
|
|
if name == names[0] and start == expected_decl:
|
|
decl_hits += 1
|
|
if name == names[1] and start == expected_extra:
|
|
has_extra = True
|
|
if decl_hits != 1:
|
|
raise AssertionError(f"declaration was not reported at the shifted position: {current}")
|
|
if not has_extra:
|
|
raise AssertionError(f"new unsaved occurrence was missed: {current}")
|
|
|
|
# Rename with two documents open must use live contents for both.
|
|
rename_pos = _position(use_text, "item.value")
|
|
rename_pos["character"] += 5
|
|
renamed = client.request(
|
|
"textDocument/rename",
|
|
{"textDocument": {"uri": use_uri}, "position": rename_pos, "newName": "data"},
|
|
timeout=120.0,
|
|
).get("result")
|
|
if not isinstance(renamed, dict):
|
|
raise AssertionError(f"multi-document rename failed: {renamed}")
|
|
changes = renamed.get("changes", {})
|
|
if {Path(uri).name for uri in changes} != set(names):
|
|
raise AssertionError(f"rename missed open or closed files: {changes}")
|
|
for uri, edits in changes.items():
|
|
seen: set[tuple[int, int, int, int]] = set()
|
|
for edit in edits:
|
|
start = edit["range"]["start"]
|
|
end = edit["range"]["end"]
|
|
key = (start["line"], start["character"], end["line"], end["character"])
|
|
if key in seen:
|
|
raise AssertionError(f"duplicate rename edit in {uri}: {key}")
|
|
seen.add(key)
|
|
updated = _apply_workspace(changes, {**texts, names[0]: type_text, names[1]: use_text})
|
|
if " data: int" not in updated[names[0]]:
|
|
raise AssertionError(f"rename missed the shifted declaration: {updated[names[0]]}")
|
|
if "self.data = value" not in updated[names[0]]:
|
|
raise AssertionError(f"rename missed a write: {updated[names[0]]}")
|
|
if "def __init__(self, value: int)" not in updated[names[0]]:
|
|
raise AssertionError("rename touched the same-named parameter")
|
|
if "return item.data" not in updated[names[1]] or "item.bump(item.data)" not in updated[names[1]]:
|
|
raise AssertionError(f"rename missed open-document uses: {updated[names[1]]}")
|
|
if "item.data" not in updated[names[2]]:
|
|
raise AssertionError(f"rename missed the closed file: {updated[names[2]]}")
|
|
|
|
# A watched-file change invalidates the index; the new on-disk occurrence must appear.
|
|
other_text = texts[names[2]] + "\n\ndef other2(item: Item) -> int:\n return item.value\n"
|
|
other_path.write_text(other_text, encoding="utf-8")
|
|
client.notify(
|
|
"workspace/didChangeWatchedFiles",
|
|
{"changes": [{"uri": other_path.as_uri(), "type": 2}]},
|
|
)
|
|
refreshed, refreshed_sec = references(use_text, "item.value", 5)
|
|
other_hits = [item for item in refreshed if Path(item["uri"]).name == names[2]]
|
|
if len(other_hits) != 2:
|
|
raise AssertionError(f"indexed file was not refreshed: {other_hits}")
|
|
|
|
if current_sec > 0.3:
|
|
raise AssertionError(f"incremental references took {current_sec:.2f}s, expected < 0.3s")
|
|
|
|
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(f"index-freshness: OK (first {initial_sec:.2f}s, incremental {current_sec:.2f}s, "
|
|
f"refresh {refreshed_sec:.2f}s)")
|
|
print("multi-document-rename: OK")
|
|
print("watched-files: OK")
|
|
return 0
|
|
|
|
|
|
def gate12(exe: str) -> int:
|
|
root = Path(__file__).resolve().parents[2]
|
|
fixtures = root / "Tools" / "lsp" / "fixtures"
|
|
workspace = Path(tempfile.mkdtemp(prefix="transpyc_lsp_gate12_"))
|
|
source = workspace / "gate12.vp"
|
|
shutil.copy(fixtures / "gate12.vp", source)
|
|
text = source.read_text(encoding="utf-8")
|
|
uri = source.as_uri()
|
|
|
|
client = Client(exe)
|
|
response = client.request(
|
|
"initialize",
|
|
{"processId": os.getpid(), "rootUri": workspace.as_uri(), "capabilities": {}},
|
|
)
|
|
capabilities = response.get("result", {}).get("capabilities", {})
|
|
provider = capabilities.get("diagnosticProvider")
|
|
if not isinstance(provider, dict):
|
|
raise AssertionError(f"missing diagnosticProvider: {response}")
|
|
if provider.get("interFileDependencies") is not False:
|
|
raise AssertionError(f"diagnosticProvider advertises inter-file dependencies: {provider}")
|
|
if provider.get("workspaceDiagnostics") is not False:
|
|
raise AssertionError(f"diagnosticProvider advertises workspace diagnostics: {provider}")
|
|
client.notify("initialized", {})
|
|
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": 1, "text": text}},
|
|
)
|
|
pushed = client.wait_notification("textDocument/publishDiagnostics")
|
|
if pushed.get("params", {}).get("diagnostics"):
|
|
raise AssertionError(f"clean fixture has push diagnostics: {pushed}")
|
|
|
|
pulled = client.request(
|
|
"textDocument/diagnostic", {"textDocument": {"uri": uri}}
|
|
).get("result")
|
|
if not isinstance(pulled, dict) or pulled.get("kind") != "full":
|
|
raise AssertionError(f"pull diagnostics result is malformed: {pulled}")
|
|
if pulled.get("items") != []:
|
|
raise AssertionError(f"clean fixture has pull diagnostics: {pulled}")
|
|
|
|
broken = 'def broken(:\n pass\n'
|
|
client.notify(
|
|
"textDocument/didChange",
|
|
{"textDocument": {"uri": uri, "version": 2}, "contentChanges": [{"text": broken}]},
|
|
)
|
|
pushed = client.wait_notification("textDocument/publishDiagnostics")
|
|
if not pushed.get("params", {}).get("diagnostics"):
|
|
raise AssertionError(f"syntax error did not produce push diagnostics: {pushed}")
|
|
pulled = client.request(
|
|
"textDocument/diagnostic", {"textDocument": {"uri": uri}}
|
|
).get("result")
|
|
items = pulled.get("items", []) if isinstance(pulled, dict) else []
|
|
if pulled.get("kind") != "full" or not items:
|
|
raise AssertionError(f"syntax error did not produce pull diagnostics: {pulled}")
|
|
first = items[0]
|
|
if not first.get("message") or not isinstance(first.get("range"), dict):
|
|
raise AssertionError(f"pull diagnostic item is incomplete: {first}")
|
|
|
|
# A response that cannot fit the 1 MiB cap must be an error, not a closed connection.
|
|
big_lines = ["class Big:"]
|
|
for i in range(5000):
|
|
big_lines.append(f" field_{i:05d}_long_name: int")
|
|
big_text = "\n".join(big_lines) + "\n"
|
|
big_path = workspace / "gate12_big.vp"
|
|
big_path.write_text(big_text, encoding="utf-8")
|
|
big_uri = big_path.as_uri()
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": big_uri, "languageId": "viper", "version": 1, "text": big_text}},
|
|
)
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
oversized = client.request(
|
|
"textDocument/documentSymbol", {"textDocument": {"uri": big_uri}}, timeout=120.0
|
|
)
|
|
error = oversized.get("error", {})
|
|
if error.get("code") != -32603 or "limit" not in error.get("message", ""):
|
|
raise AssertionError(f"oversized response was not reported: {oversized}")
|
|
pong = client.request("$/ping", {}).get("result")
|
|
if not isinstance(pong, dict):
|
|
raise AssertionError(f"server closed after an oversized response: {pong}")
|
|
|
|
# Cancelled ids are rejected with RequestCancelled at the dispatch boundary.
|
|
def raw_request(message: dict[str, Any]) -> dict[str, Any]:
|
|
client.proc.stdin.write(encode_message(message))
|
|
client.proc.stdin.flush()
|
|
while True:
|
|
reply = client.reader.recv(timeout=120.0)
|
|
if reply.get("id") == message["id"]:
|
|
return reply
|
|
|
|
hover_params = {"textDocument": {"uri": uri}, "position": {"line": 0, "character": 0}}
|
|
client.notify("$/cancelRequest", {"id": 4242})
|
|
reply = raw_request({"jsonrpc": "2.0", "id": 4242,
|
|
"method": "textDocument/hover", "params": hover_params})
|
|
if reply.get("error", {}).get("code") != -32800:
|
|
raise AssertionError(f"integer cancel was not honoured: {reply}")
|
|
|
|
client.notify("$/cancelRequest", {"id": "cancel-me"})
|
|
reply = raw_request({"jsonrpc": "2.0", "id": "cancel-me",
|
|
"method": "textDocument/hover", "params": hover_params})
|
|
if reply.get("error", {}).get("code") != -32800:
|
|
raise AssertionError(f"string cancel was not honoured: {reply}")
|
|
|
|
after = client.request("$/ping", {}).get("result")
|
|
if not isinstance(after, dict):
|
|
raise AssertionError(f"server stopped after cancellations: {after}")
|
|
|
|
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("pull-diagnostics: OK")
|
|
print("response-limit: OK")
|
|
print("cancel-request: OK")
|
|
return 0
|
|
|
|
|
|
def gate13(exe: str) -> int:
|
|
"""Semantic diagnostics: non-void functions that can fall off the end."""
|
|
root = Path(__file__).resolve().parents[2]
|
|
workspace = Path(tempfile.mkdtemp(prefix="transpyc_lsp_gate13_"))
|
|
uri = (workspace / "gate13.vp").as_uri()
|
|
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", {})
|
|
|
|
cases = [
|
|
("pointer return", "def f() -> int*:\n pass\n", 1, "int*", 0, "fall off"),
|
|
("array return", "def f() -> []int:\n pass\n", 1, "[]int", 0, "fall off"),
|
|
("has return", "def f() -> int:\n return 1\n", 0, None, 0, None),
|
|
("if without else", "def f() -> int:\n if 1:\n return 1\n", 1, "int", 0, "fall off"),
|
|
("if with else", "def f() -> int:\n if 1:\n return 1\n else:\n return 2\n", 0, None, 0, None),
|
|
("void function", "def f():\n pass\n", 0, None, 0, None),
|
|
("None return type", "def f() -> None:\n pass\n", 0, None, 0, None),
|
|
("bare return", "def f() -> int:\n return\n", 0, None, 0, None),
|
|
("raise", "def f() -> int:\n raise 1\n", 0, None, 0, None),
|
|
("ellipsis stub", "def f() -> int:\n ...\n", 0, None, 0, None),
|
|
("nested function", "def outer() -> int:\n def inner() -> int*:\n pass\n return 1\n",
|
|
1, "int*", 0, "fall off"),
|
|
("extern declaration", "extern def f() -> int\n", 0, None, 0, None),
|
|
("narrow literal", "def f() -> int:\n x: u8 = 300\n return 0\n", 1, "300", 0, "narrowing"),
|
|
("negative fits bit pattern", "def f() -> int:\n x: u8 = -1\n return 0\n", 0, None, 0, None),
|
|
("str to int", "def f() -> int:\n x: int = \"s\"\n return 0\n", 1, "\"s\"", 0, "pointer-to-integer"),
|
|
("scalar to aggregate", "def f() -> int:\n x: [4]char = 0\n return 0\n", 1, "0", 0, "aggregate"),
|
|
("call arg type", "def g(p: u8) -> int:\n return 0\n\ndef f() -> int:\n return g(300)\n", 1, "g(300)", 2, "narrowing"),
|
|
("call missing args", "def h(a: int, b: int) -> int:\n return 0\n\ndef f() -> int:\n return h(1)\n", 1, "h(1)", 0, "at least"),
|
|
("call extra args", "def h(a: int) -> int:\n return 0\n\ndef f() -> int:\n return h(1, 2)\n", 1, "h(1, 2)", 0, "at most"),
|
|
("double to float arg", "def g(p: float) -> int:\n return 0\n\ndef f() -> int:\n d: double = 1.0\n return g(d)\n", 1, "g(d)", 2, "double to float"),
|
|
("return type mismatch", "def f() -> int:\n return \"s\"\n", 1, "\"s\"", 0, "pointer-to-integer"),
|
|
("aug assign", "def f() -> int:\n x: u8 = 1\n x += 300\n return 0\n", 1, "300", 0, "augmented"),
|
|
("pointer pointee", "def g(p: int*) -> int:\n return 0\n\ndef f(q: u8*) -> int:\n return g(q)\n", 1, "g(q)", 2, "incompatible pointer"),
|
|
("array length", "def f() -> int:\n x: [4]int = [1, 2, 3, 4, 5]\n return 0\n", 1, None, 0, "array initializer"),
|
|
("string length", "def f() -> int:\n x: [4]char = \"hello\"\n return 0\n", 1, "\"hello\"", 0, "string literal length"),
|
|
("list element", "def f() -> int:\n x: [4]int = [1, \"s\"]\n return 0\n", 1, "\"s\"", 0, "pointer-to-integer"),
|
|
]
|
|
version = 1
|
|
for label, text, expected, needle, delta, message_sub in cases:
|
|
client.notify(
|
|
"textDocument/didOpen" if version == 1 else "textDocument/didChange",
|
|
{"textDocument": {"uri": uri, "languageId": "viper", "version": version, "text": text}}
|
|
if version == 1 else
|
|
{"textDocument": {"uri": uri, "version": version}, "contentChanges": [{"text": text}]},
|
|
)
|
|
version += 1
|
|
publish = client.wait_notification("textDocument/publishDiagnostics")
|
|
diagnostics = publish.get("params", {}).get("diagnostics", [])
|
|
if len(diagnostics) != expected:
|
|
raise AssertionError(f"gate13 {label}: expected {expected} diagnostics, got {publish}")
|
|
if expected == 1:
|
|
diagnostic = diagnostics[0]
|
|
if needle is not None:
|
|
start = _position(text, needle)
|
|
start["character"] += delta
|
|
line_text = text.split("\n")[start["line"]]
|
|
end_len = 0
|
|
i = start["character"]
|
|
stop_chars = " \t,():;=+-/%<>&|^]}"
|
|
if start["character"] < len(line_text) and line_text[start["character"]] == "[":
|
|
stop_chars = " \t,():;=+-/%<>&|^"
|
|
while i + end_len < len(line_text) and line_text[i + end_len] not in stop_chars:
|
|
end_len += 1
|
|
if end_len <= 0:
|
|
end_len = 1
|
|
end = {"line": start["line"], "character": start["character"] + end_len}
|
|
if diagnostic["range"]["start"] != start or diagnostic["range"]["end"] != end:
|
|
raise AssertionError(f"gate13 {label}: range misses {needle!r}: {diagnostic}")
|
|
if message_sub not in diagnostic["message"]:
|
|
raise AssertionError(f"gate13 {label}: message is wrong: {diagnostic}")
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
|
|
client.wait_notification("textDocument/publishDiagnostics")
|
|
|
|
# Stub declarations (.vpi) intentionally have no body and must never be flagged.
|
|
stub_uri = (workspace / "gate13_stub.vpi").as_uri()
|
|
stub_text = "def stub_fn() -> int:\n pass\n"
|
|
client.notify(
|
|
"textDocument/didOpen",
|
|
{"textDocument": {"uri": stub_uri, "languageId": "viper", "version": version, "text": stub_text}},
|
|
)
|
|
publish = client.wait_notification("textDocument/publishDiagnostics")
|
|
if publish.get("params", {}).get("diagnostics", []):
|
|
raise AssertionError(f"gate13 flagged a .vpi stub: {publish}")
|
|
client.notify("textDocument/didClose", {"textDocument": {"uri": stub_uri}})
|
|
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("semantic-diagnostics: OK")
|
|
print("missing-return: 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("--gate10", action="store_true", help="run member resolution checks")
|
|
parser.add_argument("--gate11", action="store_true", help="run index freshness/multi-document checks")
|
|
parser.add_argument("--gate12", action="store_true", help="run protocol/diagnostic checks")
|
|
parser.add_argument("--gate13", action="store_true", help="run semantic diagnostic 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.gate13:
|
|
return gate13(exe)
|
|
if args.gate12:
|
|
return gate12(exe)
|
|
if args.gate11:
|
|
return gate11(exe)
|
|
if args.gate10:
|
|
return gate10(exe)
|
|
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())
|