test(proxy): cover control field hardening branches

tests/proxy_unit_tests/test_audit_logs_proxy.py

@@ -90,6 +90,23 @@ def test_get_audit_log_changed_by_honors_header_with_admin_opt_in():
     )
 
 
+def test_get_audit_log_changed_by_honors_header_with_team_opt_in():
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="test-key",
+        user_id="service-account",
+        team_metadata={"allow_litellm_changed_by_header": True},
+    )
+
+    assert (
+        get_audit_log_changed_by(
+            litellm_changed_by="delegated-user",
+            user_api_key_dict=user_api_key_dict,
+            litellm_proxy_admin_name="proxy-admin",
+        )
+        == "delegated-user"
+    )
+
+
 def test_get_audit_log_changed_by_falls_back_to_header_when_user_id_missing():
     user_api_key_dict = UserAPIKeyAuth(api_key="test-key")
 
@@ -103,6 +120,40 @@ def test_get_audit_log_changed_by_falls_back_to_header_when_user_id_missing():
     )
 
 
+@pytest.mark.asyncio
+async def test_create_internal_user_audit_log_uses_changed_by_helper():
+    from litellm.proxy.hooks.user_management_event_hooks import UserManagementEventHooks
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="test-key",
+        user_id="service-account",
+        metadata={"allow_litellm_changed_by_header": True},
+    )
+
+    with (
+        patch("litellm.store_audit_logs", True),
+        patch(
+            "litellm.proxy.hooks.user_management_event_hooks.create_audit_log_for_update",
+            new_callable=AsyncMock,
+        ) as mock_create_audit_log_for_update,
+    ):
+        await UserManagementEventHooks.create_internal_user_audit_log(
+            user_id="target-user",
+            action="updated",
+            litellm_changed_by="delegated-user",
+            user_api_key_dict=user_api_key_dict,
+            litellm_proxy_admin_name="proxy-admin",
+            before_value='{"before": true}',
+            after_value='{"after": true}',
+        )
+
+    request_data = mock_create_audit_log_for_update.await_args.kwargs["request_data"]
+    assert request_data.changed_by == "delegated-user"
+    assert request_data.changed_by_api_key == "test-key"
+    assert request_data.object_id == "target-user"
+    assert request_data.action == "updated"
+
+
 @pytest.mark.asyncio
 async def test_create_audit_log_for_update_premium_user():
     """

tests/test_litellm/litellm_core_utils/test_redact_messages.py

@@ -5,10 +5,13 @@ Covers the proxy flow where headers arrive in litellm_params["metadata"]["header
 but litellm_params["litellm_metadata"] is None.
 """
 
+from types import SimpleNamespace
+
 import pytest
 
 import litellm
 from litellm.litellm_core_utils.redact_messages import (
+    _redact_responses_api_output,
     perform_redaction,
     should_redact_message_logging,
 )
@@ -255,6 +258,49 @@ class TestPerformRedaction:
         assert delta["thinking_blocks"] is None
         assert delta["audio"] is None
 
+    def test_redacts_object_choices_inside_model_response_dict(self):
+        result = {
+            "choices": [
+                litellm.Choices(
+                    message=litellm.Message(
+                        content="message content",
+                        role="assistant",
+                        reasoning_content="message reasoning",
+                    )
+                )
+            ]
+        }
+
+        redacted = perform_redaction({}, result)
+
+        choice = redacted["choices"][0]
+        assert choice.message.content == "redacted-by-litellm"
+        assert choice.message.reasoning_content == "redacted-by-litellm"
+
+    def test_redacts_response_output_objects_with_top_level_text(self):
+        output_items = [
+            SimpleNamespace(text="top-level output"),
+            "non-dict output item",
+        ]
+
+        _redact_responses_api_output(output_items)
+
+        assert output_items[0].text == "redacted-by-litellm"
+        assert output_items[1] == "non-dict output item"
+
+    def test_skips_non_dict_response_output_items(self):
+        result = {
+            "output": [
+                "non-dict output item",
+                {"content": [{"text": "nested result"}]},
+            ]
+        }
+
+        redacted = perform_redaction({}, result)
+
+        assert redacted["output"][0] == "non-dict output item"
+        assert redacted["output"][1]["content"][0]["text"] == "redacted-by-litellm"
+
     def test_redacts_responses_api_response_object(self):
         response = mock_responses_api_response("sensitive output")