""" Real-file LSP test — opens actual Bicep and azure-pipelines.yml files from Henrik's IaC repos and verifies completions are returned. """ import asyncio import json import struct import sys import pathlib import aiohttp BASE_URL = sys.argv[1] if len(sys.argv) > 1 else "https://ilsp.i80.dk" BICEP_WS = BASE_URL.replace("http", "ws") + "/bicep" YAML_WS = BASE_URL.replace("http", "ws") + "/yaml" PASS = 0 FAIL = 0 def ok(label, detail=""): global PASS; PASS += 1 print(f" ✓ {label}" + (f" ({detail})" if detail else "")) def fail(label, detail=""): global FAIL; FAIL += 1 print(f" ✗ {label}" + (f" — {detail}" if detail else "")) def _frame(obj: dict) -> bytes: body = json.dumps(obj).encode() header = f"Content-Length: {len(body)}\r\n\r\n".encode() return header + body async def _recv_until_id(ws, req_id, timeout=10.0): deadline = asyncio.get_event_loop().time() + timeout while asyncio.get_event_loop().time() < deadline: remaining = deadline - asyncio.get_event_loop().time() try: msg = await asyncio.wait_for(ws.receive(), timeout=remaining) except asyncio.TimeoutError: break if msg.type not in (aiohttp.WSMsgType.BINARY, aiohttp.WSMsgType.TEXT): continue raw = msg.data if isinstance(msg.data, (bytes, bytearray)) else msg.data.encode() text = raw.decode(errors="replace") for chunk in text.split("Content-Length:")[1:]: try: body = chunk.split("\r\n\r\n", 1)[1] parsed = json.loads(body) if parsed.get("id") == req_id: return parsed except Exception: pass return None async def _initialize_bicep(ws): await ws.send_bytes(_frame({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"processId": None, "rootUri": None, "capabilities": {}} })) resp = await _recv_until_id(ws, 1, timeout=10) if not resp: return False await ws.send_bytes(_frame({"jsonrpc": "2.0", "method": "initialized", "params": {}})) return True async def _initialize_yaml(ws): await ws.send_bytes(_frame({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "processId": None, "rootUri": None, "capabilities": {"workspace": {"configuration": True}} } })) resp = await _recv_until_id(ws, 1, timeout=10) if not resp: return False await ws.send_bytes(_frame({"jsonrpc": "2.0", "method": "initialized", "params": {}})) await asyncio.sleep(0.5) return True async def test_bicep_real_file(ws, label, file_path, cursor_line_contains, req_id): """Open a real Bicep file and request completion at a module ref line.""" p = pathlib.Path(file_path) if not p.exists(): print(f"\n ⚠ Skipped: {p.name} not found") return content = p.read_text() lines = content.splitlines() target_line = next( (i for i, l in enumerate(lines) if cursor_line_contains in l), None ) if target_line is None: print(f"\n ⚠ Skipped: pattern '{cursor_line_contains}' not in {p.name}") return uri = p.as_uri() char = len(lines[target_line]) await ws.send_bytes(_frame({ "jsonrpc": "2.0", "method": "textDocument/didOpen", "params": {"textDocument": {"uri": uri, "languageId": "bicep", "version": 1, "text": content}} })) await asyncio.sleep(0.5) await ws.send_bytes(_frame({ "jsonrpc": "2.0", "id": req_id, "method": "textDocument/completion", "params": {"textDocument": {"uri": uri}, "position": {"line": target_line, "character": char}, "context": {"triggerKind": 1}} })) resp = await _recv_until_id(ws, req_id, timeout=10) if resp is None: fail(f"Bicep {p.name}: no response") return result = resp.get("result", {}) items = result.get("items", result) if isinstance(result, dict) else result if not isinstance(items, list): items = [] labels = [i.get("label", "") for i in items] if labels: ok(f"Bicep {p.name}", f"{len(labels)} completions: {', '.join(labels[:4])}") else: fail(f"Bicep {p.name}: 0 completions", f"line {target_line}: {lines[target_line].strip()}") async def test_yaml_real_file(ws, label, file_path, req_id): """Open a real azure-pipelines.yml and request task completion.""" p = pathlib.Path(file_path) if not p.exists(): print(f"\n ⚠ Skipped: {p.name} not found") return content = p.read_text() lines = content.splitlines() # Find first '- task: ' line task_line = next((i for i, l in enumerate(lines) if "- task:" in l), None) if task_line is None: print(f"\n ⚠ Skipped: no '- task:' in {p.name}") return # Position cursor at end of '- task: ' (before the task name) raw = lines[task_line] char = raw.index("task:") + len("task: ") uri = p.as_uri() await ws.send_bytes(_frame({ "jsonrpc": "2.0", "method": "textDocument/didOpen", "params": {"textDocument": {"uri": uri, "languageId": "yaml", "version": 1, "text": content}} })) await asyncio.sleep(1.5) await ws.send_bytes(_frame({ "jsonrpc": "2.0", "id": req_id, "method": "textDocument/completion", "params": {"textDocument": {"uri": uri}, "position": {"line": task_line, "character": char}, "context": {"triggerKind": 1}} })) resp = await _recv_until_id(ws, req_id, timeout=12) if resp is None: fail(f"YAML {p.name}: no response") return result = resp.get("result", {}) items = result.get("items", []) if isinstance(result, dict) else [] labels = [i.get("label", "") for i in items] task_labels = [l for l in labels if "@" in l] if task_labels: ok(f"YAML {p.name}", f"{len(task_labels)} task@version: {', '.join(task_labels[:3])}") elif labels: ok(f"YAML {p.name} (generic)", f"{len(labels)} items") else: fail(f"YAML {p.name}: 0 completions at '- task:' line {task_line}") async def main(): print(f"\niLSP — real file test") print(f"Target: {BASE_URL}") print("═" * 60) # ── Bicep files ────────────────────────────────────────────── BICEP_FILES = [ # Module definitions ("/Users/lrihni/IdeaProjects/Bitbucket/IaC_Modules/iac-module-roleassignments/modules/roleAssignments.bicep", "principalObjectType", 10), ("/Users/lrihni/IdeaProjects/Bitbucket/IaC_Modules/iac-module-keyvault/modules/keyVault.bicep", "import", 11), # Consumer files with br/modules: references ("/Users/lrihni/IdeaProjects/Bitbucket/IaC/IaC-Matematikfessor/IaC/Matematikfessor.bicep", "br/modules:appservice", 20), ("/Users/lrihni/IdeaProjects/Bitbucket/IaC/IaC-Lounge/IaC/Lounge.bicep", "br/modules:", 21), ("/Users/lrihni/IdeaProjects/Bitbucket/IaC/IaC-Achievement/IaC/Achievement.bicep", "br/modules:", 22), ("/Users/lrihni/IdeaProjects/Bitbucket/IaC/IaC-Next/IaC/Next.bicep", "br/modules:", 23), ("/Users/lrihni/IdeaProjects/Bitbucket/IaC/iac-lrurag-ai/IaC/LRI-GenAIRAG.bicep", "br/modules:", 24), ] # ── YAML pipeline files ────────────────────────────────────── YAML_FILES = [ "/Users/lrihni/IdeaProjects/Bitbucket/UmbPlatform/UmbPlatform/UmbPlatform.Web/azure-pipelines.yml", "/Users/lrihni/IdeaProjects/Bitbucket/UmbPlatform/UmbWebbog/UmbWebbog.Web/azure-pipelines.yml", "/Users/lrihni/IdeaProjects/Bitbucket/MitAlinea/MitAlinea.UI/MitAlinea.UI/azure-pipelines.yml", "/Users/lrihni/IdeaProjects/Bitbucket/Next_Production/Next/Alinea.Products.Next.UI/azure-pipelines.yml", "/Users/lrihni/IdeaProjects/Bitbucket/IaC/IaC-Content/azure-pipelines.yml", "/Users/lrihni/IdeaProjects/Bitbucket/IaC/IaC-Homework/azure-pipelines.yml", "/Users/lrihni/IdeaProjects/Bitbucket/Next_Production/Next Api/Alinea.Api.Next/azure-pipelines.yml", ] async with aiohttp.ClientSession() as session: # ── Bicep ── print("\n── Bicep (real module files) ───────────────────────────────") async with session.ws_connect(BICEP_WS, timeout=aiohttp.ClientTimeout(total=30)) as ws: ok_init = await _initialize_bicep(ws) if not ok_init: fail("Bicep initialize failed"); return for file_path, pattern, req_id in BICEP_FILES: await test_bicep_real_file(ws, "", file_path, pattern, req_id) # ── YAML ── print("\n── YAML azure-pipelines.yml (task completions) ─────────────") async with session.ws_connect(YAML_WS, timeout=aiohttp.ClientTimeout(total=60)) as ws: ok_init = await _initialize_yaml(ws) if not ok_init: fail("YAML initialize failed"); return for i, file_path in enumerate(YAML_FILES, start=100): await test_yaml_real_file(ws, "", file_path, i) print(f"\n{'═' * 60}") print(f"Results: {PASS} passed, {FAIL} failed") return 0 if FAIL == 0 else 1 if __name__ == "__main__": sys.exit(asyncio.run(main()))