"""Tests for YAML LSP proxy — document format detection and context detection.""" import json import pytest from ilsp.yaml_lsp.proxy import ( _detect_doc_format, _detect_azdo_context, _detect_gha_context, _inject_completions, _LspFrameBuffer, ) from ilsp.yaml_lsp.catalog import PipelineTemplateCatalog # ── LspFrameBuffer ───────────────────────────────────────────────────────────── class TestLspFrameBuffer: def test_single_frame(self): body = b'{"jsonrpc":"2.0"}' framed = f"Content-Length: {len(body)}\r\n\r\n".encode() + body buf = _LspFrameBuffer() frames = buf.feed(framed) assert frames == [body] def test_two_frames_in_one_feed(self): body1 = b'{"a":1}' body2 = b'{"b":2}' def frame(b): return f"Content-Length: {len(b)}\r\n\r\n".encode() + b buf = _LspFrameBuffer() frames = buf.feed(frame(body1) + frame(body2)) assert frames == [body1, body2] def test_partial_frame_buffered(self): body = b'{"partial":true}' header = f"Content-Length: {len(body)}\r\n\r\n".encode() buf = _LspFrameBuffer() assert buf.feed(header + body[:3]) == [] frames = buf.feed(body[3:]) assert frames == [body] # ── Document format detection ────────────────────────────────────────────────── class TestDocFormatDetection: def test_azdo_template_line(self): lines = [ "trigger: [main]", "stages:", " - stage: Build", " jobs:", " - template: tasks/k8s/deploy.yaml@pipeline-templates", ] assert _detect_doc_format(lines) == "azdo" def test_azdo_pipeline_keywords(self): lines = ["trigger: [main]", "stages:", " - stage: Build"] assert _detect_doc_format(lines) == "azdo" def test_gha_workflow_call(self): lines = [ "on:", " workflow_call:", " inputs:", " environment:", " type: string", ] assert _detect_doc_format(lines) == "gha" def test_gha_runs_on(self): lines = [ "on: [push]", "jobs:", " build:", " runs-on: ubuntu-latest", ] assert _detect_doc_format(lines) == "gha" def test_unknown_empty(self): assert _detect_doc_format([]) == "unknown" # ── AzDO context detection ──────────────────────────────────────────────────── class TestAzdoContextDetection: def test_template_path_context(self): lines = [ " - template: tasks/k8s/", ] ctx = _detect_azdo_context(lines, 0, len(lines[0])) assert ctx["type"] == "template_path" assert ctx["format"] == "azdo" def test_param_name_context(self): lines = [ " - template: tasks/k8s/deploy.yaml@pipeline-templates", " parameters:", " projectName: ", " ", ] # Cursor on line 3 (blank), after parameters block ctx = _detect_azdo_context(lines, 3, 10) # Should detect we're in a parameters block under a template assert ctx["format"] == "azdo" # type is param_name (we're inside parameters block) assert ctx["type"] in ("param_name", "unknown") def test_template_path_prefix_extracted(self): lines = [" - template: tasks/k8s/dep"] ctx = _detect_azdo_context(lines, 0, len(lines[0])) assert ctx["type"] == "template_path" assert "dep" in ctx["prefix"] # ── GHA context detection ────────────────────────────────────────────────────── class TestGhaContextDetection: def test_workflow_ref_context(self): lines = [ " - name: Deploy", " uses: LRU-Digital/CommonLoginTestApplication/.github/workflows/k8s.yml@", ] ctx = _detect_gha_context(lines, 1, len(lines[1])) assert ctx["type"] == "workflow_ref" assert ctx["format"] == "gha" def test_input_name_context(self): lines = [ " uses: LRU-Digital/CommonLoginTestApplication/.github/workflows/k8s.yml@main", " with:", " ", ] ctx = _detect_gha_context(lines, 2, 8) assert ctx["format"] == "gha" assert ctx["type"] in ("input_name", "unknown") def test_workflow_ref_prefix_extracted(self): lines = [" uses: LRU-Digital/Repo/.github/workflows/dep"] ctx = _detect_gha_context(lines, 0, len(lines[0])) assert ctx["type"] == "workflow_ref" assert "dep" in ctx["prefix"] # ── Completion injection ─────────────────────────────────────────────────────── _SAMPLE_CATALOG_DATA = { "synced_at": "2025-01-01T00:00:00+00:00", "template_count": 1, "templates": { "tasks/k8s/deploy.yaml@pipeline-templates": { "format": "azdo", "title": "deploy", "path": "tasks/k8s/deploy.yaml", "alias": "pipeline-templates", "parameters": [ {"name": "projectName", "type": "string", "required": True}, {"name": "environment", "type": "string", "required": True, "allowed": ["dev", "test", "prod"]}, ], }, }, } @pytest.fixture(autouse=True) def load_catalog(tmp_path, monkeypatch): import json as _json f = tmp_path / "pipeline_templates_catalog.json" f.write_text(_json.dumps(_SAMPLE_CATALOG_DATA)) import ilsp.yaml_lsp.catalog as mod monkeypatch.setattr(mod, "_CATALOG_PATHS", [f]) PipelineTemplateCatalog.load() class TestInjection: def test_azdo_template_path_injects_at_top(self): msg = {"jsonrpc": "2.0", "id": 1, "result": {"items": [{"label": "existing"}]}} ctx = {"type": "template_path", "format": "azdo", "prefix": ""} modified = json.loads(_inject_completions(msg, ctx)) labels = [i["label"] for i in modified["result"]["items"]] assert "tasks/k8s/deploy.yaml" in labels # LRU items appear before standard items lru_idx = next(i for i, l in enumerate(labels) if "deploy.yaml" in l) existing_idx = next(i for i, l in enumerate(labels) if l == "existing") assert lru_idx < existing_idx def test_azdo_param_names_injected(self): msg = {"jsonrpc": "2.0", "id": 2, "result": {"items": []}} ctx = { "type": "param_name", "format": "azdo", "template_key": "tasks/k8s/deploy.yaml@pipeline-templates", } modified = json.loads(_inject_completions(msg, ctx)) names = [i.get("filterText", i["label"]) for i in modified["result"]["items"]] assert "projectName" in names assert "environment" in names def test_azdo_allowed_values_injected(self): msg = {"jsonrpc": "2.0", "id": 3, "result": {"items": []}} ctx = { "type": "param_value", "format": "azdo", "template_key": "tasks/k8s/deploy.yaml@pipeline-templates", "param": "environment", } modified = json.loads(_inject_completions(msg, ctx)) vals = [i["label"] for i in modified["result"]["items"]] assert "dev" in vals assert "test" in vals assert "prod" in vals def test_unknown_context_no_injection(self): msg = {"jsonrpc": "2.0", "id": 4, "result": {"items": [{"label": "x"}]}} ctx = {"type": "unknown", "format": "unknown"} modified = json.loads(_inject_completions(msg, ctx)) assert modified["result"]["items"] == [{"label": "x"}] def test_no_result_passthrough(self): msg = {"jsonrpc": "2.0", "id": 5} ctx = {"type": "template_path", "format": "azdo"} modified = json.loads(_inject_completions(msg, ctx)) assert modified == msg def test_existing_items_downgraded(self): msg = { "jsonrpc": "2.0", "id": 6, "result": {"items": [{"label": "standard", "sortText": "aaa"}]}, } ctx = {"type": "template_path", "format": "azdo", "prefix": ""} modified = json.loads(_inject_completions(msg, ctx)) items = modified["result"]["items"] std = next(i for i in items if i["label"] == "standard") assert std["sortText"].startswith("9_")