23 lines
626 B
Python
23 lines
626 B
Python
import unittest
|
|
from unittest.mock import AsyncMock
|
|
|
|
from rag.events import publish_knowledge_update
|
|
|
|
|
|
class KnowledgeEventTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_publishes_update_event_after_successful_ingestion(self):
|
|
publisher = AsyncMock()
|
|
|
|
await publish_knowledge_update(
|
|
publisher, doc_id="doc-1", collection_name="fin_faq", chunk_count=4
|
|
)
|
|
|
|
publisher.assert_awaited_once_with(
|
|
"event:knowledge_update",
|
|
{"doc_id": "doc-1", "collection_name": "fin_faq", "chunk_count": 4},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|