ss-tools/backend/tests/services/test_path_b_batch.py

#region TestPathBBatch [C:3] [TYPE Module] [SEMANTICS test, llm, path-b, batch, isolation, unknown]
# @BRIEF Tests for Path B text-batch isolation: per-dashboard failures don't cascade.
# @RELATION BINDS_TO -> [LLMClient.analyze_dashboard_text_batch]
# @TEST_CONTRACT [{dashboard_id, topology, dataset_health, log_text}] -> {dashboards: [...]}
# @TEST_SCENARIO partial_failure -> 1 of 5 fails → only that record UNKNOWN, rest preserved
# @TEST_SCENARIO all_success -> all 5 PASS with correct per-dashboard data
# @TEST_SCENARIO empty_batch -> empty payload returns 0 dashboards
# @TEST_EDGE: llm_error -> LLM call fails entirely → UNKNOWN for all dashboards
# @TEST_EDGE: malformed_response -> LLM returns invalid JSON → UNKNOWN for all dashboards
# @TEST_EDGE: missing_dashboard_id -> parsed response missing some ids → UNKNOWN defaults
# @INVARIANT Per-dashboard LLM response failures are isolated — surrounding dashboards unaffected (FR-056)

import pytest
from unittest.mock import AsyncMock

from src.plugins.llm_analysis.models import LLMProviderType
from src.plugins.llm_analysis.service import LLMClient


#region _make_client [C:2] [TYPE Function]
# @BRIEF Create LLMClient with mock external dependencies for deterministic testing.
def _make_client():
    """Create an LLMClient with mocked HTTP transport for batch analysis tests."""
    client = LLMClient(
        provider_type=LLMProviderType.LITELLM,
        api_key="sk-test-batch-key",
        base_url="http://localhost:4000/v1",
        default_model="gpt-4o-mini",
    )
    # Replace the real OpenAI client with a mock to prevent network calls
    client.client = AsyncMock()
    return client
#endregion _make_client


#region _make_payloads [C:2] [TYPE Function]
# @BRIEF Create deterministic batch payloads for Path B text batch testing.
def _make_payloads(count: int = 5) -> list[dict]:
    """Generate `count` dashboard payloads, the last one with a KXD dataset error.

    Payload 0..n-2: healthy dashboards
    Payload n-1: has dataset_health indicating KXD connectivity failure (FR-044 level 1 failure)
    """
    payloads = []
    for i in range(count):
        is_broken = i == count - 1
        payloads.append({
            "dashboard_id": f"dash-{i + 1}",
            "topology": f"Chart_A (id: {100 + i})\nChart_B (id: {200 + i})",
            "dataset_health": (
                "ERROR: KXD connection refused — metadata_accessible=false"
                if is_broken
                else "OK: metadata_accessible=true, backend=postgresql"
            ),
            "log_text": f"Session {3000 + i}: loaded successfully",
        })
    return payloads
#endregion _make_payloads


#region test_path_b_batch_partial_failure [C:2] [TYPE Function]
# @BRIEF T047: 1 of 5 dashboards with KXD error → only that dashboard UNKNOWN, rest preserved.
# @TEST_INVARIANT batch_isolation -> VERIFIED_BY: test_path_b_batch_partial_failure
@pytest.mark.anyio
async def test_path_b_batch_partial_failure():
    """Batch of 5 dashboards, last has KXD error → only the last is UNKNOWN."""
    client = _make_client()
    payloads = _make_payloads(count=5)

    # The LLM returns PASS for healthy dashboards and UNKNOWN for the KXD-failed one
    async def _fake_json_completion(_messages):
        return {
            "dashboards": [
                {"dashboard_id": "dash-1", "status": "PASS", "summary": "All metrics OK", "issues": []},
                {"dashboard_id": "dash-2", "status": "PASS", "summary": "All metrics OK", "issues": []},
                {"dashboard_id": "dash-3", "status": "PASS", "summary": "All metrics OK", "issues": []},
                {"dashboard_id": "dash-4", "status": "PASS", "summary": "All metrics OK", "issues": []},
                {
                    "dashboard_id": "dash-5",
                    "status": "UNKNOWN",
                    "summary": "Dataset health check failed — KXD connection refused",
                    "issues": [{"severity": "UNKNOWN", "message": "KXD connection refused for underlying dataset"}],
                },
            ],
        }

    client.get_json_completion = _fake_json_completion

    result = await client.analyze_dashboard_text_batch(
        payloads=payloads,
        prompt_template="Validate the following {total_dashboards} dashboards:\n",
    )

    assert "dashboards" in result
    assert len(result["dashboards"]) == 5

    # First 4 dashboards still PASS
    for i in range(4):
        assert result["dashboards"][i]["dashboard_id"] == f"dash-{i + 1}"
        assert result["dashboards"][i]["status"] == "PASS"

    # Last dashboard (KXD error) is UNKNOWN
    assert result["dashboards"][4]["dashboard_id"] == "dash-5"
    assert result["dashboards"][4]["status"] == "UNKNOWN"
    assert "KXD" in result["dashboards"][4]["summary"]
#endregion test_path_b_batch_partial_failure


#region test_path_b_batch_all_success [C:2] [TYPE Function]
# @BRIEF T047 variant: all 5 healthy dashboards return PASS.
@pytest.mark.anyio
async def test_path_b_batch_all_success():
    """All 5 dashboards healthy → all PASS."""
    client = _make_client()
    payloads = _make_payloads(count=5)
    # Override dataset_health for all to be healthy
    for p in payloads:
        p["dataset_health"] = "OK: metadata_accessible=true, backend=postgresql"

    async def _fake_json_completion(_messages):
        return {
            "dashboards": [
                {"dashboard_id": f"dash-{i + 1}", "status": "PASS",
                 "summary": "Healthy", "issues": []}
                for i in range(5)
            ],
        }

    client.get_json_completion = _fake_json_completion

    result = await client.analyze_dashboard_text_batch(
        payloads=payloads,
        prompt_template="Validate {total_dashboards} dashboards:\n",
    )

    assert all(d["status"] == "PASS" for d in result["dashboards"])
#endregion test_path_b_batch_all_success


#region test_path_b_batch_llm_error_unknown_all [C:2] [TYPE Function]
# @BRIEF T047 edge: complete LLM failure → exception propagates (no silent UNKNOWN).
@pytest.mark.anyio
async def test_path_b_batch_llm_error_unknown_all():
    """LLM call fails entirely → exception propagates from analyze_dashboard_text_batch."""
    client = _make_client()
    payloads = _make_payloads(count=3)

    async def _raise_error(_messages):
        raise RuntimeError("LLM provider returned 503 Service Unavailable")

    client.get_json_completion = _raise_error

    with pytest.raises(RuntimeError, match="503"):
        await client.analyze_dashboard_text_batch(
            payloads=payloads,
            prompt_template="Validate {total_dashboards} dashboards:\n",
        )
#endregion test_path_b_batch_llm_error_unknown_all


#region test_path_b_batch_empty [C:2] [TYPE Function]
# @BRIEF T047 edge: empty payload list returns empty results without LLM call.
@pytest.mark.anyio
async def test_path_b_batch_empty():
    """Empty payload → 0 dashboards, no LLM call."""
    client = _make_client()

    result = await client.analyze_dashboard_text_batch(
        payloads=[],
        prompt_template="Validate {total_dashboards} dashboards:\n",
    )

    assert result == {"dashboards": []}
#endregion test_path_b_batch_empty

#endregion TestPathBBatch