scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-18 04:44:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 123

test(guardrails): cover session-id fallbacks and mark_texts_scanned guards

Browse source 
Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-07-14 23:20:18 +00:00
parent a0f765638e
commit 4fd2a5f482
1 changed files with 69 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
tests/test_litellm/integrations/test_custom_guardrail.py
View file

@ -1845,3 +1845,72 @@ class TestOnlyScanNewMessages:
cache.async_set_cache.assert_awaited_once()
assert cache.async_set_cache.await_args.kwargs["ttl"] == GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS
@pytest.mark.asyncio
async def test_session_id_from_metadata_is_used_for_dedupe(self):
guardrail = self._guardrail()
cache = self._cache()
request = {"metadata": {"session_id": "sess-meta"}}
texts = ["shared message"]
await guardrail.filter_new_texts_for_session(texts=texts, request_data=request, cache=cache)
await guardrail.mark_texts_scanned(texts=texts, request_data=request, cache=cache)
again = await guardrail.filter_new_texts_for_session(texts=texts, request_data=request, cache=cache)
assert again == []
@pytest.mark.asyncio
async def test_session_id_from_litellm_metadata_is_used_for_dedupe(self):
guardrail = self._guardrail()
cache = self._cache()
request = {"litellm_metadata": {"session_id": "sess-lmeta"}}
texts = ["shared message"]
await guardrail.filter_new_texts_for_session(texts=texts, request_data=request, cache=cache)
await guardrail.mark_texts_scanned(texts=texts, request_data=request, cache=cache)
again = await guardrail.filter_new_texts_for_session(texts=texts, request_data=request, cache=cache)
assert again == []
@pytest.mark.asyncio
async def test_mark_texts_scanned_disabled_does_not_persist(self):
from unittest.mock import AsyncMock
guardrail = self._guardrail(only_scan_new_messages=False)
cache = self._cache()
cache.async_set_cache = AsyncMock()
await guardrail.mark_texts_scanned(texts=["a"], request_data={"litellm_session_id": "s1"}, cache=cache)
cache.async_set_cache.assert_not_awaited()
@pytest.mark.asyncio
async def test_mark_texts_scanned_masking_does_not_persist(self):
from unittest.mock import AsyncMock
guardrail = self._guardrail(mask_request_content=True)
cache = self._cache()
cache.async_set_cache = AsyncMock()
await guardrail.mark_texts_scanned(texts=["a"], request_data={"litellm_session_id": "s1"}, cache=cache)
cache.async_set_cache.assert_not_awaited()
@pytest.mark.asyncio
async def test_mark_texts_scanned_without_session_does_not_persist(self):
from unittest.mock import AsyncMock
guardrail = self._guardrail()
cache = self._cache()
cache.async_set_cache = AsyncMock()
await guardrail.mark_texts_scanned(texts=["a"], request_data={"metadata": {}}, cache=cache)
cache.async_set_cache.assert_not_awaited()
@pytest.mark.asyncio
async def test_mark_texts_scanned_survives_cache_write_failure(self):
from unittest.mock import AsyncMock
guardrail = self._guardrail()
cache = self._cache()
cache.async_set_cache = AsyncMock(side_effect=RuntimeError("redis down"))
await guardrail.mark_texts_scanned(texts=["a"], request_data={"litellm_session_id": "s1"}, cache=cache)