Spaces:
Runtime error
Runtime error
| """ | |
| Tests du contexte envoye au modele et des citations. | |
| Le contexte ne contenait que highlights['content'], soit les 400 premiers | |
| caracteres du texte de la LOI : le modele devait citer des articles dont il | |
| n'avait jamais lu le texte. Et format_matched_articles, seule fonction censee | |
| ajouter ce texte, lisait un attribut inexistant (`snippet` au lieu de | |
| `content_snippet`) : sa branche ne s'est jamais declenchee. | |
| """ | |
| from unittest.mock import AsyncMock, MagicMock | |
| import pytest | |
| from app.schemas.search import ChunkResult | |
| from app.services.prompts import ( | |
| CONTEXT_MAX_CHARS, | |
| CONTEXT_TRUNCATION_MARK, | |
| build_context_string, | |
| ) | |
| from app.services.rag_service import RAGService, _normalize_article_number | |
| def _chunk(article_id: int, number: str, content: str, **kwargs) -> ChunkResult: | |
| defaults = dict( | |
| law_id=kwargs.pop("law_id", 1), | |
| reference=kwargs.pop("reference", "LOI-2024-001"), | |
| law_title=kwargs.pop("law_title", "Code OHADA"), | |
| relevance_score=kwargs.pop("relevance_score", 0.9), | |
| ) | |
| return ChunkResult( | |
| article_id=article_id, | |
| number=number, | |
| content=content, | |
| excerpt=content[:120], | |
| **defaults, | |
| **kwargs, | |
| ) | |
| def rag_service(): | |
| service = RAGService.__new__(RAGService) | |
| service.db = MagicMock() | |
| service.llm = AsyncMock() | |
| service.search_service = MagicMock() | |
| return service | |
| class TestContextBuilding: | |
| def test_context_contains_article_number_and_full_content(self): | |
| chunk = _chunk( | |
| 161, "161", | |
| "Les dirigeants sociaux sont responsables des fautes commises " | |
| "dans l'exercice de leurs fonctions envers la société et les tiers.", | |
| article_title="Responsabilité des dirigeants", | |
| section="TITRE III", | |
| page_number=47, | |
| ) | |
| context = build_context_string([chunk]) | |
| assert "Article 161" in context | |
| assert "Responsabilité des dirigeants" in context | |
| assert "TITRE III" in context | |
| assert "Page 47" in context | |
| # Le CONTENU, pas seulement un extrait de 400 caracteres de la loi. | |
| assert "envers la société et les tiers" in context | |
| def test_context_respects_global_budget(self): | |
| chunks = [ | |
| _chunk(i, str(i), "A" * 5000) for i in range(1, 21) | |
| ] | |
| context = build_context_string(chunks) | |
| assert len(context) <= CONTEXT_MAX_CHARS * 1.1 | |
| assert "Article 1" in context | |
| def test_first_chunk_survives_even_if_oversized(self): | |
| """Un contexte tronque vaut mieux qu'un contexte vide.""" | |
| chunks = [_chunk(1, "1", "B" * 100_000)] | |
| context = build_context_string(chunks) | |
| assert context | |
| assert "Article 1" in context | |
| assert CONTEXT_TRUNCATION_MARK in context | |
| def test_truncation_falls_on_a_paragraph_boundary(self): | |
| content = ("Paragraphe un." + "\n\n" + "x" * 200) * 60 | |
| chunks = [_chunk(1, "1", content)] | |
| context = build_context_string(chunks) | |
| assert CONTEXT_TRUNCATION_MARK in context | |
| def test_multilingual_context_gets_a_translation_note(self): | |
| chunks = [ | |
| _chunk(1, "1", "Texte en français", language="fr"), | |
| _chunk(2, "1", "English text", language="en", law_id=2, | |
| reference="LAW-2024-002", law_title="Commercial Code"), | |
| ] | |
| context = build_context_string(chunks) | |
| assert "traduis" in context.lower() | |
| def test_empty_chunks_give_empty_context(self): | |
| assert build_context_string([]) == "" | |
| class TestArticlePinning: | |
| def test_requested_article_moves_to_the_front(self, rag_service): | |
| chunks = [ | |
| _chunk(1, "1", "Premier"), | |
| _chunk(2, "42", "Quarante-deux"), | |
| ] | |
| pinned = rag_service._pin_requested_article("Que dit l'article 42 ?", chunks) | |
| assert pinned[0].number == "42" | |
| assert len(pinned) == 2 | |
| def test_pinning_does_not_touch_the_database(self, rag_service): | |
| """ | |
| Le texte est deja dans le contexte : aller le rechercher ouvrait un | |
| moteur synchrone de plus a chaque question. | |
| """ | |
| chunks = [_chunk(1, "1", "Premier"), _chunk(2, "42", "Quarante-deux")] | |
| rag_service._pin_requested_article("article 42", chunks) | |
| rag_service.db.execute.assert_not_called() | |
| def test_unknown_article_leaves_order_unchanged(self, rag_service): | |
| chunks = [_chunk(1, "1", "Premier"), _chunk(2, "2", "Deux")] | |
| assert rag_service._pin_requested_article("article 99", chunks) == chunks | |
| class TestArticleNumberNormalisation: | |
| def test_normalisation(self, raw, expected): | |
| assert _normalize_article_number(raw) == expected | |
| def test_does_not_confuse_1_with_10(self): | |
| """ | |
| Les numeros etaient parfois compares par inclusion : "1" correspondait | |
| alors a "10", "11" et "100", et la mauvaise citation etait rattachee. | |
| """ | |
| assert _normalize_article_number("1") != _normalize_article_number("10") | |
| class TestCitations: | |
| def test_citation_binds_article_id(self, rag_service): | |
| chunks = [_chunk(161, "161", "Les dirigeants sont responsables.")] | |
| answer = "Selon l'article 161 du Code OHADA, les dirigeants sont responsables." | |
| citations = rag_service._extract_citations(answer, chunks) | |
| assert len(citations) == 1 | |
| assert citations[0].article_id == 161 | |
| assert citations[0].law_id == 1 | |
| def test_citation_excerpt_comes_from_the_chunk(self, rag_service): | |
| chunks = [_chunk(161, "161", "Contenu unique et reconnaissable.")] | |
| answer = "Selon l'article 161 du Code OHADA." | |
| citations = rag_service._extract_citations(answer, chunks) | |
| assert "reconnaissable" in citations[0].excerpt | |
| rag_service.db.execute.assert_not_called() | |
| def test_citation_rejects_an_article_absent_from_the_context(self, rag_service): | |
| """ | |
| La validation ne portait que sur le titre de la loi, jamais sur le | |
| numero : un article invente par le modele passait. | |
| """ | |
| chunks = [_chunk(161, "161", "Les dirigeants sont responsables.")] | |
| answer = "Selon l'article 999 du Code OHADA, tout est permis." | |
| assert rag_service._extract_citations(answer, chunks) == [] | |
| def test_sources_fallback_uses_chunk_content(self, rag_service): | |
| chunks = [_chunk(161, "161", "Contenu de repli.")] | |
| sources = rag_service._create_sources_from_results(chunks, "Une question") | |
| assert len(sources) == 1 | |
| assert sources[0].article_id == 161 | |
| assert "repli" in sources[0].excerpt | |
| class TestEventLoop: | |
| """Le chemin RAG ne doit rien appeler de bloquant sur la boucle.""" | |
| async def test_gemini_generate_runs_off_the_event_loop(self, monkeypatch): | |
| """ | |
| GeminiService.generate est `async def` mais appelle un client | |
| SYNCHRONE : sans deport, la boucle d'evenements etait gelee pendant | |
| tout l'aller-retour avec le modele, a chaque question. | |
| """ | |
| import threading | |
| from app.services.gemini_service import GeminiService | |
| service = GeminiService.__new__(GeminiService) | |
| service.model_name = "test-model" | |
| service.SYSTEM_INSTRUCTION = "sys" | |
| loop_thread = threading.get_ident() | |
| seen = {} | |
| class _Response: | |
| text = "réponse" | |
| class _Models: | |
| def generate_content(self, **kwargs): | |
| seen["thread"] = threading.get_ident() | |
| return _Response() | |
| service.client = type("C", (), {"models": _Models()})() | |
| result = await service.generate(prompt="Question ?") | |
| assert result["response"] == "réponse" | |
| assert seen["thread"] != loop_thread | |
| class TestFallbackSearch: | |
| """ | |
| Repli multilingue : recherche d'un article precis quand la recherche | |
| generale ne rend rien. | |
| Cette fonction utilise `text(...)` sans que le module l'importe : chaque | |
| appel levait un NameError, avale par son propre except, et le repli n'a | |
| jamais rien renvoye. Un test d'integration est le seul moyen de s'en | |
| apercevoir, l'exception etant masquee. | |
| """ | |
| async def constitution(self, db_session): | |
| from app.models.law import Article, Law | |
| law = Law( | |
| reference="CONST-1996", | |
| title="Constitution de la République du Cameroun", | |
| content="Constitution.", | |
| type="constitution", | |
| language="fr", | |
| status="published", | |
| ) | |
| db_session.add(law) | |
| await db_session.flush() | |
| db_session.add(Article( | |
| law_id=law.id, | |
| number="8", | |
| content="Le Président de la République est le Chef de l'État.", | |
| order=8, | |
| )) | |
| await db_session.commit() | |
| return law | |
| async def test_finds_the_article_in_the_named_document(self, db_session, constitution): | |
| service = RAGService.__new__(RAGService) | |
| service.db = db_session | |
| service.llm = AsyncMock() | |
| service.search_service = MagicMock() | |
| chunks = await service._fallback_search("Que dit l'article 8 de la constitution ?") | |
| assert chunks, "le repli doit trouver l'article 8 de la constitution" | |
| assert chunks[0].number == "8" | |
| assert chunks[0].article_id is not None | |
| assert "Chef de l'État" in chunks[0].content | |
| assert chunks[0].source == "fallback" | |
| async def test_returns_empty_without_an_article_number(self, db_session, constitution): | |
| service = RAGService.__new__(RAGService) | |
| service.db = db_session | |
| service.llm = AsyncMock() | |
| service.search_service = MagicMock() | |
| assert await service._fallback_search("Parle-moi du droit du travail") == [] | |
| class TestPseudoArticleNumbers: | |
| """ | |
| PREAMBULE et LEGAL_BASIS ne sont pas des numeros d'article. | |
| Ce sont les pseudo-numeros que le decoupeur attribue au preambule et aux | |
| visas pour ne perdre aucun caractere du document. Presentes tels quels, ils | |
| ressortaient en "Article LEGAL_BASIS" dans le contexte du modele et dans les | |
| citations affichees a l'utilisateur — invitant a citer un article qui | |
| n'existe pas. | |
| """ | |
| def test_context_labels_the_legal_basis(self): | |
| from app.services.prompts import format_chunk_block | |
| chunk = _chunk(1, "LEGAL_BASIS", "Vu la Constitution ;") | |
| block = format_chunk_block(chunk, 1) | |
| assert "Visas et base légale" in block | |
| assert "Article LEGAL_BASIS" not in block | |
| def test_context_keeps_real_article_numbers(self): | |
| from app.services.prompts import format_chunk_block | |
| block = format_chunk_block(_chunk(1, "39", "Contenu."), 1) | |
| assert "Article 39" in block | |
| def test_citation_drops_the_pseudo_number(self, rag_service): | |
| chunks = [_chunk(61, "LEGAL_BASIS", "Vu la Constitution et la loi.")] | |
| sources = rag_service._create_sources_from_results(chunks, "une question") | |
| assert sources[0].article_number is None | |
| # L'identifiant, lui, reste : le chunk existe bien en base. | |
| assert sources[0].article_id == 61 | |
| def test_search_result_omits_pseudo_articles(self): | |
| from app.services.search_service import SearchService | |
| chunks = [ | |
| _chunk(61, "LEGAL_BASIS", "Vu la Constitution."), | |
| _chunk(62, "39", "Article trente-neuf."), | |
| ] | |
| results = SearchService._chunks_to_search_results(chunks) | |
| numbers = [a.number for a in results[0].matched_articles] | |
| assert "39" in numbers | |
| assert "LEGAL_BASIS" not in numbers | |
| class TestStreaming: | |
| """ | |
| Le flux passe par la surface asynchrone native du client. | |
| La version synchrone bloquait la boucle d'evenements a chaque morceau recu : | |
| sur une reponse de plusieurs secondes, le serveur ne traitait plus aucune | |
| autre requete pendant tout le flux — ce qui annule l'interet meme du | |
| streaming. | |
| """ | |
| async def test_uses_the_async_client(self, monkeypatch): | |
| import asyncio | |
| from app.services.gemini_service import GeminiService | |
| service = GeminiService.__new__(GeminiService) | |
| service.model_name = "test-model" | |
| service.SYSTEM_INSTRUCTION = "sys" | |
| class _Chunk: | |
| def __init__(self, text): | |
| self.text = text | |
| async def _stream(**kwargs): | |
| async def _gen(): | |
| for piece in ("Selon ", "l'article 161, ", "les dirigeants..."): | |
| await asyncio.sleep(0) | |
| yield _Chunk(piece) | |
| return _gen() | |
| sync_called = {"n": 0} | |
| def _sync_stream(**kwargs): | |
| sync_called["n"] += 1 | |
| raise AssertionError("la surface synchrone ne doit plus etre utilisee") | |
| service.client = type("C", (), { | |
| "aio": type("A", (), {"models": type("M", (), { | |
| "generate_content_stream": staticmethod(_stream)})()})(), | |
| "models": type("M2", (), { | |
| "generate_content_stream": staticmethod(_sync_stream)})(), | |
| })() | |
| pieces = [chunk async for chunk in service.generate_stream(prompt="q")] | |
| assert "".join(pieces) == "Selon l'article 161, les dirigeants..." | |
| assert sync_called["n"] == 0 | |
| async def test_does_not_block_the_event_loop(self): | |
| """ | |
| Une autre tache doit continuer a s'executer PENDANT le flux. | |
| C'est la propriete que la version synchrone violait, et qu'aucun test | |
| d'egalite de texte ne peut detecter. | |
| """ | |
| import asyncio | |
| from app.services.gemini_service import GeminiService | |
| service = GeminiService.__new__(GeminiService) | |
| service.model_name = "test-model" | |
| service.SYSTEM_INSTRUCTION = "sys" | |
| class _Chunk: | |
| def __init__(self, text): | |
| self.text = text | |
| async def _stream(**kwargs): | |
| async def _gen(): | |
| for i in range(5): | |
| await asyncio.sleep(0.01) | |
| yield _Chunk(f"morceau{i} ") | |
| return _gen() | |
| service.client = type("C", (), { | |
| "aio": type("A", (), {"models": type("M", (), { | |
| "generate_content_stream": staticmethod(_stream)})()})() | |
| })() | |
| ticks = 0 | |
| async def _other_work(): | |
| nonlocal ticks | |
| while True: | |
| await asyncio.sleep(0.005) | |
| ticks += 1 | |
| task = asyncio.create_task(_other_work()) | |
| pieces = [chunk async for chunk in service.generate_stream(prompt="q")] | |
| task.cancel() | |
| assert len(pieces) == 5 | |
| assert ticks > 3, "la boucle etait bloquee pendant le flux" | |
| async def test_flux_rapporte_sa_raison_d_arret(): | |
| """Un flux coupe par le budget ne levait rien et ne se signalait pas.""" | |
| from types import SimpleNamespace | |
| from app.services.gemini_service import GeminiService | |
| service = GeminiService.__new__(GeminiService) | |
| service.model_name = "gemini-3-flash-preview" | |
| service.SYSTEM_INSTRUCTION = "sys" | |
| def _morceau(texte, raison=None): | |
| candidat = SimpleNamespace(finish_reason=SimpleNamespace(name=raison) if raison else None) | |
| return SimpleNamespace(text=texte, candidates=[candidat]) | |
| async def _stream(**kwargs): | |
| async def _gen(): | |
| yield _morceau("Un militaire qui abuse ") | |
| yield _morceau("de sa position pour", "MAX_TOKENS") | |
| return _gen() | |
| service.client = SimpleNamespace(aio=SimpleNamespace(models=SimpleNamespace( | |
| generate_content_stream=_stream))) | |
| raisons = [] | |
| pieces = [p async for p in service.generate_stream(prompt="q", fin=raisons.append)] | |
| assert "".join(pieces) == "Un militaire qui abuse de sa position pour" | |
| assert raisons == ["MAX_TOKENS"] | |