484 lines
17 KiB
Python
484 lines
17 KiB
Python
"""Unit tests for Bicep proxy message injection."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from ilsp.bicep_lsp.modules import BicepModuleCatalog
|
|
from ilsp.bicep_lsp.proxy import _ProxySession, _frame, _inject_completions
|
|
|
|
|
|
def test_frame_produces_correct_header():
|
|
body = b'{"jsonrpc":"2.0"}'
|
|
framed = _frame(body)
|
|
assert framed.startswith(b"Content-Length: 17\r\n\r\n")
|
|
assert framed.endswith(body)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_modules():
|
|
BicepModuleCatalog._modules = []
|
|
yield
|
|
|
|
|
|
def _completion_response(items: list) -> dict:
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"result": {"isIncomplete": False, "items": items},
|
|
}
|
|
|
|
|
|
def _make_module(name, versions=None, schema=None):
|
|
versions = versions or ["1.0.0", "latest"]
|
|
path = f"bicep/modules/{name}"
|
|
return {
|
|
"name": name,
|
|
"path": path,
|
|
"ref_path": f"modules/{name}", # what appears after 'br/modules:' in bicep files
|
|
"versions": versions,
|
|
"latest": versions[-1],
|
|
"registry": "iactemplatereg.azurecr.io",
|
|
"schema": schema or {},
|
|
}
|
|
|
|
|
|
# ── Existing injection tests (unchanged behaviour) ─────────────────────────────
|
|
|
|
def test_standard_items_not_downgraded_without_lru():
|
|
"""Without LRU modules, standard items keep their original sortText."""
|
|
msg = _completion_response([{"label": "Microsoft.Storage", "sortText": "az"}])
|
|
out = json.loads(_inject_completions(msg))
|
|
assert out["result"]["items"][0]["sortText"] == "az"
|
|
|
|
|
|
def test_lru_modules_injected_at_top():
|
|
BicepModuleCatalog._modules = [_make_module("appservice", ["2.3.0", "latest"])]
|
|
|
|
msg = _completion_response([{"label": "Microsoft.Web/sites", "sortText": "az"}])
|
|
out = json.loads(_inject_completions(msg))
|
|
items = out["result"]["items"]
|
|
|
|
assert items[0]["label"] == "appservice"
|
|
assert items[0]["sortText"].startswith("0_lru_")
|
|
assert items[1]["label"] == "Microsoft.Web/sites"
|
|
assert items[1]["sortText"].startswith("1_az_")
|
|
|
|
|
|
def test_list_result_also_handled():
|
|
BicepModuleCatalog._modules = [_make_module("roleassignments", ["2.0.0"])]
|
|
|
|
msg = {"jsonrpc": "2.0", "id": 2, "result": [{"label": "az-item", "sortText": "az"}]}
|
|
out = json.loads(_inject_completions(msg))
|
|
assert isinstance(out["result"], list)
|
|
assert out["result"][0]["label"] == "roleassignments"
|
|
|
|
|
|
def test_non_completion_message_passthrough():
|
|
msg = {"jsonrpc": "2.0", "method": "initialized", "params": {}}
|
|
out = json.loads(_inject_completions(msg))
|
|
assert out["method"] == "initialized"
|
|
|
|
|
|
# ── Version completion tests ───────────────────────────────────────────────────
|
|
|
|
def test_version_completions_injected_on_version_context():
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x", "2.0.x", "latest"],
|
|
schema={
|
|
"1.1.x": {"parameters": {"principalId": {"type": "string", "description": ""}}},
|
|
"2.0.x": {"parameters": {"assignments": {"type": "array", "description": ""}}},
|
|
"latest": {"parameters": {"assignments": {"type": "array", "description": ""}}},
|
|
},
|
|
)]
|
|
|
|
msg = _completion_response([{"label": "az-builtin", "sortText": "az"}])
|
|
context = {"type": "version", "module": "roleassignments", "prefix": ""}
|
|
out = json.loads(_inject_completions(msg, context))
|
|
items = out["result"]["items"]
|
|
|
|
labels = [i["label"] for i in items]
|
|
assert "1.1.x" in labels
|
|
assert "2.0.x" in labels
|
|
assert "latest" in labels
|
|
# LS items are replaced entirely — private registry versions only
|
|
assert all(i["sortText"].startswith("0_lru_ver_") for i in items)
|
|
assert not any(i["label"] == "az-builtin" for i in items)
|
|
|
|
|
|
def test_version_items_include_param_detail():
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"appservice",
|
|
versions=["2.3.2"],
|
|
schema={"2.3.2": {"parameters": {
|
|
"projectName": {"type": "string", "description": ""},
|
|
"environmentType": {"type": "string", "description": ""},
|
|
}}},
|
|
)]
|
|
|
|
items = BicepModuleCatalog.version_completion_items("appservice")
|
|
assert len(items) == 1
|
|
assert items[0]["label"] == "2.3.2"
|
|
assert "projectName" in items[0]["detail"]
|
|
assert items[0]["kind"] == 12 # Value
|
|
|
|
|
|
def test_version_completions_unknown_module_returns_empty():
|
|
items = BicepModuleCatalog.version_completion_items("nonexistent")
|
|
assert items == []
|
|
|
|
|
|
# ── Param completion tests ─────────────────────────────────────────────────────
|
|
|
|
_ROLEASSIGNMENTS_SCHEMA = {
|
|
"1.1.x": {
|
|
"parameters": {
|
|
"environmentType": {
|
|
"type": "string",
|
|
"description": "",
|
|
"allowed": ["DEV", "TEST", "PROD"],
|
|
},
|
|
"roleDefinitionIds": {
|
|
"type": "array",
|
|
"description": "The role definition ID.",
|
|
},
|
|
"principalId": {
|
|
"type": "string",
|
|
"description": "The principal ID.",
|
|
},
|
|
"principalType": {
|
|
"type": "string",
|
|
"description": "The principal type.",
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
def test_param_completions_injected_on_param_context():
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema=_ROLEASSIGNMENTS_SCHEMA,
|
|
)]
|
|
|
|
msg = _completion_response([])
|
|
context = {"type": "param", "module": "roleassignments", "version": "1.1.x"}
|
|
out = json.loads(_inject_completions(msg, context))
|
|
items = out["result"]["items"]
|
|
|
|
labels = [i["label"] for i in items]
|
|
assert "environmentType" in labels
|
|
assert "roleDefinitionIds" in labels
|
|
assert "principalId" in labels
|
|
assert "principalType" in labels
|
|
# First item is now the snippet completion
|
|
assert items[0]["label"] == "⚡ Fill params block"
|
|
assert items[0]["kind"] == 15 # Snippet
|
|
# Second item should be a regular param
|
|
assert items[1]["sortText"].startswith("0_lru_param_")
|
|
assert items[1]["kind"] == 5 # Field
|
|
|
|
|
|
def test_param_completion_items_have_insert_text():
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments", versions=["1.1.x"], schema=_ROLEASSIGNMENTS_SCHEMA
|
|
)]
|
|
items = BicepModuleCatalog.param_completion_items("roleassignments", "1.1.x")
|
|
# First item should be the snippet completion
|
|
assert items[0]["label"] == "⚡ Fill params block"
|
|
assert items[0]["insertTextFormat"] == 2
|
|
assert "principalId" in items[0]["insertText"]
|
|
# Individual param items should have ": " suffix
|
|
for item in items[1:]:
|
|
assert item["insertText"].endswith(": ")
|
|
|
|
|
|
def test_param_completions_fallback_to_closest_version():
|
|
"""When the exact version isn't in schema, fall back to any available version."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x", "1.1.4"],
|
|
schema=_ROLEASSIGNMENTS_SCHEMA, # only has "1.1.x"
|
|
)]
|
|
items = BicepModuleCatalog.param_completion_items("roleassignments", "1.1.4")
|
|
assert any(i["label"] == "principalId" for i in items)
|
|
|
|
|
|
def test_param_completions_unknown_module_returns_empty():
|
|
items = BicepModuleCatalog.param_completion_items("nonexistent", "1.0.0")
|
|
assert items == []
|
|
|
|
|
|
# ── _ProxySession context detection tests ─────────────────────────────────────
|
|
|
|
def _make_session_with_doc(uri: str, lines: list[str]) -> _ProxySession:
|
|
session = _ProxySession()
|
|
session.update_doc(uri, "\n".join(lines))
|
|
return session
|
|
|
|
|
|
URI = "file:///test/main.bicep"
|
|
|
|
|
|
def test_detect_version_context():
|
|
lines = ["module x 'br/modules:roleassignments:' = {}"]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 0, "character": 37})
|
|
assert ctx["type"] == "version"
|
|
assert ctx["module"] == "roleassignments"
|
|
|
|
|
|
def test_detect_version_context_partial_prefix():
|
|
lines = ["module x 'br/modules:roleassignments:1.1' = {}"]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 0, "character": 40})
|
|
assert ctx["type"] == "version"
|
|
assert ctx["module"] == "roleassignments"
|
|
assert ctx["prefix"] == "1.1"
|
|
|
|
|
|
def test_detect_module_path_context():
|
|
lines = ["module x 'br/modules:role' = {}"]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 0, "character": 25})
|
|
assert ctx["type"] == "module_path"
|
|
|
|
|
|
def test_detect_param_context():
|
|
lines = [
|
|
"module myMod 'br/modules:roleassignments:1.1.x' = {",
|
|
" name: 'test'",
|
|
" params: {",
|
|
" ", # ← cursor here (blank line, no param name yet)
|
|
" }",
|
|
"}",
|
|
]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 3, "character": 4})
|
|
assert ctx["type"] == "param"
|
|
assert ctx["module"] == "roleassignments"
|
|
assert ctx["version"] == "1.1.x"
|
|
|
|
|
|
def test_detect_param_context_empty_version():
|
|
"""Params context should work even when module ref has no version yet."""
|
|
lines = [
|
|
"module keyVault 'br/modules:modules/keyvault:' = {",
|
|
" params: {",
|
|
" ", # ← cursor here
|
|
" }",
|
|
"}",
|
|
]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 2, "character": 4})
|
|
assert ctx["type"] == "param"
|
|
assert ctx["module"] == "modules/keyvault"
|
|
assert ctx["version"] == ""
|
|
|
|
|
|
def test_detect_param_value_context_no_quote():
|
|
"""Cursor after 'principalType: ' (no opening quote) → param_value context."""
|
|
lines = [
|
|
"module myMod 'br/modules:roleassignments:1.1.x' = {",
|
|
" params: {",
|
|
" principalType: ", # ← cursor at end
|
|
" }",
|
|
"}",
|
|
]
|
|
session = _make_session_with_doc(URI, lines)
|
|
# character = len(" principalType: ") = 19
|
|
ctx = session._detect_context(URI, {"line": 2, "character": 19})
|
|
assert ctx["type"] == "param_value"
|
|
assert ctx["module"] == "roleassignments"
|
|
assert ctx["param"] == "principalType"
|
|
assert ctx["has_open_quote"] is False
|
|
|
|
|
|
def test_detect_param_value_context_open_quote():
|
|
"""Cursor after 'principalType: \\'' → param_value with has_open_quote=True."""
|
|
lines = [
|
|
"module myMod 'br/modules:roleassignments:1.1.x' = {",
|
|
" params: {",
|
|
" principalType: '", # ← cursor after opening quote
|
|
" }",
|
|
"}",
|
|
]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 2, "character": 20})
|
|
assert ctx["type"] == "param_value"
|
|
assert ctx["has_open_quote"] is True
|
|
|
|
|
|
def test_detect_param_value_context_in_array():
|
|
"""Cursor inside array value → param_value context."""
|
|
lines = [
|
|
"module myMod 'br/modules:roleassignments:1.1.x' = {",
|
|
" params: {",
|
|
" roles: ['KEY_VAULT_", # ← cursor inside array element
|
|
" }",
|
|
"}",
|
|
]
|
|
session = _make_session_with_doc(URI, lines)
|
|
# character = len(" roles: ['KEY_VAULT_") = 23
|
|
ctx = session._detect_context(URI, {"line": 2, "character": 23})
|
|
assert ctx["type"] == "param_value"
|
|
assert ctx["module"] == "roleassignments"
|
|
assert ctx["param"] == "roles"
|
|
assert ctx["has_open_quote"] is True
|
|
|
|
|
|
def test_param_value_items_from_catalog_allowed():
|
|
"""environmentType completions come from catalog 'allowed' field."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema={"1.1.x": {"parameters": {
|
|
"environmentType": {"type": "string", "allowed": ["DEV", "TEST", "PROD"]},
|
|
}}},
|
|
)]
|
|
items = BicepModuleCatalog.param_value_completion_items(
|
|
"roleassignments", "1.1.x", "environmentType"
|
|
)
|
|
labels = [i["label"] for i in items]
|
|
assert labels == ["DEV", "TEST", "PROD"]
|
|
# Without open quote, insertText should wrap in quotes
|
|
assert items[0]["insertText"] == "'DEV'"
|
|
|
|
|
|
def test_param_value_items_open_quote():
|
|
"""When has_open_quote=True, insertText closes the quote but doesn't open one."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema={"1.1.x": {"parameters": {
|
|
"environmentType": {"type": "string", "allowed": ["DEV", "TEST", "PROD"]},
|
|
}}},
|
|
)]
|
|
items = BicepModuleCatalog.param_value_completion_items(
|
|
"roleassignments", "1.1.x", "environmentType", has_open_quote=True
|
|
)
|
|
assert items[0]["insertText"] == "DEV'"
|
|
|
|
|
|
def test_param_value_items_known_enum_fallback():
|
|
"""principalType uses _KNOWN_ENUMS fallback when not in catalog."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema={"1.1.x": {"parameters": {
|
|
"principalType": {"type": "string"}, # no 'allowed' in catalog
|
|
}}},
|
|
)]
|
|
items = BicepModuleCatalog.param_value_completion_items(
|
|
"roleassignments", "1.1.x", "principalType"
|
|
)
|
|
labels = [i["label"] for i in items]
|
|
assert "Group" in labels
|
|
assert "ServicePrincipal" in labels
|
|
assert "User" in labels
|
|
|
|
|
|
def test_param_value_items_roles_enum():
|
|
"""roles parameter uses _KNOWN_ENUMS for Azure role completions."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema={"1.1.x": {"parameters": {
|
|
"roles": {"type": "array"}, # no 'allowed' in catalog
|
|
}}},
|
|
)]
|
|
items = BicepModuleCatalog.param_value_completion_items(
|
|
"roleassignments", "1.1.x", "roles"
|
|
)
|
|
labels = [i["label"] for i in items]
|
|
assert "KEY_VAULT_SECRETS_USER" in labels
|
|
assert "STORAGE_BLOB_DATA_READER" in labels
|
|
assert "CONTRIBUTOR" in labels
|
|
|
|
|
|
def test_param_value_items_empty_for_free_string():
|
|
"""A plain string param with no allowed values returns no completions."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema={"1.1.x": {"parameters": {
|
|
"principalId": {"type": "string"},
|
|
}}},
|
|
)]
|
|
items = BicepModuleCatalog.param_value_completion_items(
|
|
"roleassignments", "1.1.x", "principalId"
|
|
)
|
|
assert items == []
|
|
|
|
|
|
def test_param_value_injected_in_completion_response():
|
|
"""Full pipeline: param_value context injects enum completions at top."""
|
|
BicepModuleCatalog._modules = [_make_module(
|
|
"roleassignments",
|
|
versions=["1.1.x"],
|
|
schema={"1.1.x": {"parameters": {
|
|
"environmentType": {"type": "string", "allowed": ["DEV", "TEST", "PROD"]},
|
|
}}},
|
|
)]
|
|
msg = _completion_response([{"label": "existing", "sortText": "z"}])
|
|
ctx = {"type": "param_value", "module": "roleassignments", "version": "1.1.x",
|
|
"param": "environmentType", "has_open_quote": False}
|
|
out = json.loads(_inject_completions(msg, ctx))
|
|
labels = [i["label"] for i in out["result"]["items"]]
|
|
# LRU enum values only — LS completions are replaced entirely
|
|
assert labels == ["DEV", "TEST", "PROD"]
|
|
|
|
|
|
def test_detect_unknown_context_outside_module():
|
|
lines = ["var x = 'hello'"]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 0, "character": 10})
|
|
assert ctx["type"] == "unknown"
|
|
|
|
|
|
def test_detect_no_param_context_after_params_closed():
|
|
"""Cursor after the closing brace of params should NOT be param context."""
|
|
lines = [
|
|
"module myMod 'br/modules:roleassignments:1.1.x' = {",
|
|
" params: {",
|
|
" }",
|
|
" ", # ← cursor here (outside params block)
|
|
"}",
|
|
]
|
|
session = _make_session_with_doc(URI, lines)
|
|
ctx = session._detect_context(URI, {"line": 3, "character": 2})
|
|
# params block is closed, so should NOT be param context
|
|
assert ctx["type"] != "param"
|
|
|
|
|
|
# ── Session request tracking ───────────────────────────────────────────────────
|
|
|
|
def test_session_records_and_pops_context():
|
|
session = _ProxySession()
|
|
session.update_doc(URI, "\n".join([
|
|
"module m 'br/modules:appservice:2.3.0' = {",
|
|
" params: {",
|
|
" ",
|
|
" }",
|
|
"}",
|
|
]))
|
|
msg = {
|
|
"jsonrpc": "2.0",
|
|
"id": 42,
|
|
"method": "textDocument/completion",
|
|
"params": {
|
|
"textDocument": {"uri": URI},
|
|
"position": {"line": 2, "character": 4},
|
|
},
|
|
}
|
|
session.record_completion_request(msg)
|
|
ctx = session.pop_context(42)
|
|
assert ctx["type"] == "param"
|
|
assert ctx["module"] == "appservice"
|
|
|
|
# Second pop returns unknown
|
|
assert session.pop_context(42)["type"] == "unknown"
|
|
|