test_fuzzy_grounding.py

import sys
import unittest
from unittest.mock import patch, AsyncMock
from pathlib import Path

# Add backend directory to sys.path
sys.path.insert(0, str(Path(__file__).parent.resolve()))

from app.services import engine, rag

class TestFuzzyGroundingGates(unittest.IsolatedAsyncioTestCase):

    def setUp(self):
        # Clear engine config cache to reload with fresh synonyms
        engine._load_domain_config.cache_clear()

    @patch("app.services.rag.retrieve")
    async def test_synonym_overlap_fever_jvara(self, mock_retrieve):
        # Query containing English synonym "fever" should match context containing Sanskrit "jvara"
        mock_chunk = rag.RetrievedChunk(
            chunk_id="test_1",
            content="Jvara is a condition caused by vitiated doshas, affecting the body and mind.",
            source="Charaka Samhita",
            chapter="Chapter 1",
            section="Sutrasthana",
            relevance_score=0.9
        )
        mock_retrieve.return_value = [mock_chunk]

        # Mock connection and streaming
        with patch("app.services.llm.check_connection", new_callable=AsyncMock) as mock_conn, \
             patch("app.services.llm.stream_completion") as mock_stream:
            mock_conn.return_value = True
            
            async def dummy_generator(*args, **kwargs):
                yield "Test response"
            mock_stream.return_value = dummy_generator()

            is_grounded = None
            async for event in engine.stream_answer("ayurveda", "Explain how fever manifests.", []):
                if event.get("type") == "citations":
                    is_grounded = event.get("is_grounded")

            self.assertTrue(is_grounded, "English synonym 'fever' failed to match Sanskrit context 'jvara'!")

    @patch("app.services.rag.retrieve")
    async def test_phonetic_overlap_vasti_basti(self, mock_retrieve):
        # Query containing "vasti" should match context containing "basti" via phonetic hashing
        mock_chunk = rag.RetrievedChunk(
            chunk_id="test_2",
            content="Basti therapy is the chief treatment for Vata disorders.",
            source="Charaka Samhita",
            chapter="Chapter 2",
            section="Sutrasthana",
            relevance_score=0.9
        )
        mock_retrieve.return_value = [mock_chunk]

        with patch("app.services.llm.check_connection", new_callable=AsyncMock) as mock_conn, \
             patch("app.services.llm.stream_completion") as mock_stream:
            mock_conn.return_value = True
            
            async def dummy_generator(*args, **kwargs):
                yield "Test response"
            mock_stream.return_value = dummy_generator()

            is_grounded = None
            async for event in engine.stream_answer("ayurveda", "What is the function of vasti?", []):
                if event.get("type") == "citations":
                    is_grounded = event.get("is_grounded")

            self.assertTrue(is_grounded, "Phonetic variation 'vasti' failed to match context 'basti'!")

    @patch("app.services.rag.retrieve")
    async def test_adversarial_rejection_uranium(self, mock_retrieve):
        # Query containing "uranium" should fail lexical gate on Vata/Basti context
        mock_chunk = rag.RetrievedChunk(
            chunk_id="test_3",
            content="Basti therapy is the chief treatment for Vata disorders.",
            source="Charaka Samhita",
            chapter="Chapter 2",
            section="Sutrasthana",
            relevance_score=0.9
        )
        mock_retrieve.return_value = [mock_chunk]

        with patch("app.services.llm.check_connection", new_callable=AsyncMock) as mock_conn:
            mock_conn.return_value = True

            is_grounded = None
            async for event in engine.stream_answer("ayurveda", "Can I treat Vata using uranium?", []):
                if event.get("type") == "citations":
                    is_grounded = event.get("is_grounded")

            self.assertFalse(is_grounded, "Adversarial query containing 'uranium' incorrectly cleared the grounding gate!")

    @patch("app.services.rag.retrieve")
    async def test_phonetic_sammurchana_sammurshana(self, mock_retrieve):
        # Query spelling "Sammurchana" should match context spelling "sammurshana"
        mock_chunk = rag.RetrievedChunk(
            chunk_id="test_4",
            content="The pathogenesis involves dosha-dushya sammurshana in the srotas.",
            source="Charaka Samhita",
            chapter="Chapter 3",
            section="Sutrasthana",
            relevance_score=0.9
        )
        mock_retrieve.return_value = [mock_chunk]

        with patch("app.services.llm.check_connection", new_callable=AsyncMock) as mock_conn, \
             patch("app.services.llm.stream_completion") as mock_stream:
            mock_conn.return_value = True
            
            async def dummy_generator(*args, **kwargs):
                yield "Test response"
            mock_stream.return_value = dummy_generator()

            is_grounded = None
            async for event in engine.stream_answer("ayurveda", "How does Sammurchana occur?", []):
                if event.get("type") == "citations":
                    is_grounded = event.get("is_grounded")

            self.assertTrue(is_grounded, "Sanskrit phonetic variation 'Sammurchana' failed to match context 'sammurshana'!")

if __name__ == "__main__":
    unittest.main()