From b5ce6464e69d753b2db4588f5fd53965238589d5 Mon Sep 17 00:00:00 2001 From: Shalom Jamil Date: Mon, 7 Sep 2026 10:50:08 +0300 Subject: [PATCH] chore(ovalix): satisfy the tightened lint gates on the current base The base moved on and lowered the LIT002 ceiling, so bring the guardrail back under it. _validate_config now returns the event hooks it resolves instead of appending to a list the caller owns, which lets __init__ take supported_event_hooks as a real parameter rather than digging it out of kwargs and rebuilding the dict. The checkpoint routing forms and the tool dedup map are read-only, so they are MappingProxyType now; the payload itself stays a plain dict because httpx has to json-encode it. Also set match= on the _call_checkpoint raises assertion for PT011. --- .../guardrail_hooks/ovalix/ovalix.py | 34 +++++++++++-------- .../guardrails/guardrail_hooks/test_ovalix.py | 2 +- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py b/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py index ee041faf19e..6a0643d8e2d 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py +++ b/litellm/proxy/guardrails/guardrail_hooks/ovalix/ovalix.py @@ -163,6 +163,7 @@ class OvalixGuardrail(CustomGuardrail): file_checkpoint_id: str | None = None, enable_routing_cache: bool | None = None, fail_if_no_application: bool | None = None, + supported_event_hooks: list[GuardrailEventHooks] | None = None, **kwargs: Any, ): self._tracker_api_base = tracker_api_base or os.environ.get("OVALIX_TRACKER_API_BASE") @@ -188,9 +189,7 @@ class OvalixGuardrail(CustomGuardrail): self._routing_cache: OrderedDict[str, tuple[float, ResolvedRouting | None]] = OrderedDict() self._app_name_regex: re.Pattern[str] | None = None - supported_event_hooks: Final[list[GuardrailEventHooks]] = list(kwargs.get("supported_event_hooks") or ()) - - self._validate_config(supported_event_hooks) + event_hooks: Final = self._validated_event_hooks(supported_event_hooks or ()) self._tracker_headers = dict( httpx.Headers( @@ -207,7 +206,7 @@ class OvalixGuardrail(CustomGuardrail): self._async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback) - super().__init__(**{**kwargs, "supported_event_hooks": supported_event_hooks}) + super().__init__(supported_event_hooks=event_hooks, **kwargs) verbose_proxy_logger.debug( "Ovalix Guardrail initialized: tracker=%s, application_id=%s, pre_checkpoint_id=%s, post_checkpoint_id=%s", self._tracker_api_base, @@ -216,8 +215,8 @@ class OvalixGuardrail(CustomGuardrail): self._post_checkpoint_id, ) - def _validate_config(self, supported_event_hooks: list[GuardrailEventHooks]) -> None: - """Ensure required Tracker secrets are set; register the pre/post hooks this config can serve (both in discovery mode; only configured-checkpoint directions in static mode).""" + def _validated_event_hooks(self, requested: Sequence[GuardrailEventHooks]) -> list[GuardrailEventHooks]: + """Ensure required Tracker secrets are set; return the pre/post hooks this config can serve (both in discovery mode; only configured-checkpoint directions in static mode).""" errors: Final = tuple( message for present, message in ( @@ -236,10 +235,15 @@ class OvalixGuardrail(CustomGuardrail): supports_pre: Final = not self._application_id or bool(self._pre_checkpoint_id) supports_post: Final = not self._application_id or bool(self._post_checkpoint_id) - if supports_pre and GuardrailEventHooks.pre_call not in supported_event_hooks: - supported_event_hooks.append(GuardrailEventHooks.pre_call) - if supports_post and GuardrailEventHooks.post_call not in supported_event_hooks: - supported_event_hooks.append(GuardrailEventHooks.post_call) + auto_added: Final = tuple( + hook + for supported, hook in ( + (supports_pre, GuardrailEventHooks.pre_call), + (supports_post, GuardrailEventHooks.post_call), + ) + if supported and hook not in requested + ) + return [*requested, *auto_added] def _get_actor(self, data: Mapping[str, Any]) -> str: """Return a stable actor identifier from request metadata (e.g. user email or id).""" @@ -285,9 +289,9 @@ class OvalixGuardrail(CustomGuardrail): route: Final = "file_checkpoint" if data_type == "FILE" else "checkpoint" routing: Final = ( - {"application_name": target.application_name, "input_type": target.input_type} + MappingProxyType({"application_name": target.application_name, "input_type": target.input_type}) if target.application_name - else {"application_id": target.application_id, "checkpoint_id": checkpoint_id} + else MappingProxyType({"application_id": target.application_id, "checkpoint_id": checkpoint_id}) ) payload: Final = { "actor": actor, @@ -426,9 +430,9 @@ class OvalixGuardrail(CustomGuardrail): *(inputs.get("tool_calls") or ()), *extract_tool_calls_from_messages(structured_messages), ) - unique_tool_data: Final = { - tool_data_key(data): data for data in (tool_call_to_tool_data(tc) for tc in tool_calls) if data - } + unique_tool_data: Final = MappingProxyType( + {tool_data_key(data): data for data in (tool_call_to_tool_data(tc) for tc in tool_calls) if data} + ) tool_call_items: Final = tuple(("TOOL", data) for data in unique_tool_data.values()) tool_block: Final = await self._check_items_block_only( tool_call_items, diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_ovalix.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_ovalix.py index 928bdefa477..b0d66c39286 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_ovalix.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_ovalix.py @@ -1261,7 +1261,7 @@ def test_enable_routing_cache_from_env_string(monkeypatch): @pytest.mark.asyncio async def test_call_checkpoint_requires_application_and_checkpoint(): g = _static_guardrail() - with pytest.raises(ValueError): + with pytest.raises(ValueError, match="application_id or checkpoint_id not resolved"): await g._call_checkpoint("TEXT", {"content": "x"}, "", "actor", "sess", CheckpointTarget("app-1", "request"))