breakpilot-compliance/backend-compliance/tests/test_rag_client.py

"""Tests for ComplianceRAGClient."""

import pytest
from unittest.mock import AsyncMock, patch, MagicMock

from compliance.services.rag_client import ComplianceRAGClient, RAGSearchResult


class TestComplianceRAGClient:
    """Tests for the RAG client proxy."""

    def setup_method(self):
        self.client = ComplianceRAGClient(base_url="http://test-sdk:8090")

    @pytest.mark.asyncio
    async def test_search_success(self):
        """Successful search returns parsed results."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "query": "DSGVO Art. 35",
            "results": [
                {
                    "text": "Art. 35 DSGVO regelt die Datenschutz-Folgenabschaetzung...",
                    "regulation_code": "eu_2016_679",
                    "regulation_name": "DSGVO",
                    "regulation_short": "DSGVO",
                    "category": "regulation",
                    "article": "Art. 35",
                    "paragraph": "",
                    "source_url": "https://example.com",
                    "score": 0.92,
                }
            ],
            "count": 1,
        }

        with patch("httpx.AsyncClient") as MockClient:
            mock_instance = AsyncMock()
            mock_instance.post.return_value = mock_response
            mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
            mock_instance.__aexit__ = AsyncMock(return_value=False)
            MockClient.return_value = mock_instance

            results = await self.client.search("DSGVO Art. 35", collection="bp_compliance_ce")

        assert len(results) == 1
        assert results[0].regulation_code == "eu_2016_679"
        assert results[0].score == 0.92
        assert "Art. 35" in results[0].text

    @pytest.mark.asyncio
    async def test_search_failure_returns_empty(self):
        """Network errors return empty list, never raise."""
        with patch("httpx.AsyncClient") as MockClient:
            mock_instance = AsyncMock()
            mock_instance.post.side_effect = Exception("Connection refused")
            mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
            mock_instance.__aexit__ = AsyncMock(return_value=False)
            MockClient.return_value = mock_instance

            results = await self.client.search("test query")

        assert results == []

    @pytest.mark.asyncio
    async def test_search_http_error_returns_empty(self):
        """HTTP errors return empty list."""
        mock_response = MagicMock()
        mock_response.status_code = 500
        mock_response.text = "Internal Server Error"

        with patch("httpx.AsyncClient") as MockClient:
            mock_instance = AsyncMock()
            mock_instance.post.return_value = mock_response
            mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
            mock_instance.__aexit__ = AsyncMock(return_value=False)
            MockClient.return_value = mock_instance

            results = await self.client.search("test query")

        assert results == []

    def test_format_for_prompt(self):
        """format_for_prompt produces Markdown output."""
        results = [
            RAGSearchResult(
                text="Die Verarbeitung personenbezogener Daten...",
                regulation_code="eu_2016_679",
                regulation_name="DSGVO",
                regulation_short="DSGVO",
                category="regulation",
                article="Art. 35",
                paragraph="",
                source_url="https://example.com",
                score=0.9,
            ),
            RAGSearchResult(
                text="Risikobewertung fuer KI-Systeme...",
                regulation_code="eu_2024_1689",
                regulation_name="AI Act",
                regulation_short="AI Act",
                category="regulation",
                article="",
                paragraph="",
                source_url="https://example.com",
                score=0.85,
            ),
        ]

        output = self.client.format_for_prompt(results)

        assert "## Relevanter Rechtskontext" in output
        assert "**DSGVO**" in output
        assert "Art. 35" in output
        assert "**AI Act**" in output

    def test_format_for_prompt_empty(self):
        """Empty results return empty string."""
        assert self.client.format_for_prompt([]) == ""

    def test_format_for_prompt_truncation(self):
        """Long text is truncated to 400 chars."""
        results = [
            RAGSearchResult(
                text="A" * 500,
                regulation_code="test",
                regulation_name="Test",
                regulation_short="Test",
                category="test",
                article="",
                paragraph="",
                source_url="",
                score=0.5,
            ),
        ]

        output = self.client.format_for_prompt(results)
        assert "..." in output


class TestCollectionMapping:
    """Tests for regulation → collection mapping in AIComplianceAssistant."""

    def test_eu_regulations_map_to_ce(self):
        from compliance.services.ai_compliance_assistant import AIComplianceAssistant
        assistant = AIComplianceAssistant.__new__(AIComplianceAssistant)

        assert assistant._collection_for_regulation("DSGVO") == "bp_compliance_ce"
        assert assistant._collection_for_regulation("GDPR") == "bp_compliance_ce"
        assert assistant._collection_for_regulation("AI_ACT") == "bp_compliance_ce"
        assert assistant._collection_for_regulation("NIS2") == "bp_compliance_ce"
        assert assistant._collection_for_regulation("CRA") == "bp_compliance_ce"

    def test_de_regulations_map_to_recht(self):
        from compliance.services.ai_compliance_assistant import AIComplianceAssistant
        assistant = AIComplianceAssistant.__new__(AIComplianceAssistant)

        assert assistant._collection_for_regulation("BDSG") == "bp_compliance_recht"
        assert assistant._collection_for_regulation("TDDDG") == "bp_compliance_recht"
        assert assistant._collection_for_regulation("TKG") == "bp_compliance_recht"

    def test_unknown_regulation_defaults_to_ce(self):
        from compliance.services.ai_compliance_assistant import AIComplianceAssistant
        assistant = AIComplianceAssistant.__new__(AIComplianceAssistant)

        assert assistant._collection_for_regulation("UNKNOWN") == "bp_compliance_ce"
        assert assistant._collection_for_regulation("") == "bp_compliance_ce"

    def test_case_insensitive(self):
        from compliance.services.ai_compliance_assistant import AIComplianceAssistant
        assistant = AIComplianceAssistant.__new__(AIComplianceAssistant)

        assert assistant._collection_for_regulation("dsgvo") == "bp_compliance_ce"
        assert assistant._collection_for_regulation("bdsg") == "bp_compliance_recht"