iLSP/tests/test_yaml_proxy.py

"""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_")