phoenix-oss/llama-stack-mirror
0
0
Fork 1
mirror of https://github.com/meta-llama/llama-stack.git synced 2025-12-12 12:06:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

test: suppress expected error logs in SSE test

Browse source 
Use pytest's caplog fixture to suppress ERROR logs when deliberately
triggering errors in test_sse_generator_error_before_response_starts.
This keeps test output clean while still validating error handling behavior.
This commit is contained in:
Ashwin Bharambe 2025-10-21 12:00:59 -07:00
parent cb2185b936
commit 62626d4d94
5 changed files with 60 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

32
tests/unit/server/test_auth.py
View file

@ -6,6 +6,7 @@
import base64
import json
import logging # allow-direct-logging
from unittest.mock import AsyncMock, Mock, patch
import pytest
@ -27,6 +28,13 @@ from llama_stack.core.server.auth_providers import (
)
@pytest.fixture
def suppress_auth_errors(caplog):
"""Suppress expected ERROR/WARNING logs for tests that deliberately trigger authentication errors"""
caplog.set_level(logging.CRITICAL, logger="llama_stack.core.server.auth")
caplog.set_level(logging.CRITICAL, logger="llama_stack.core.server.auth_providers")
class MockResponse:
def __init__(self, status_code, json_data):
self.status_code = status_code
@ -237,20 +245,20 @@ def test_valid_http_authentication(http_client, valid_api_key):
@patch("httpx.AsyncClient.post", new=mock_post_failure)
def test_invalid_http_authentication(http_client, invalid_api_key):
def test_invalid_http_authentication(http_client, invalid_api_key, suppress_auth_errors):
response = http_client.get("/test", headers={"Authorization": f"Bearer {invalid_api_key}"})
assert response.status_code == 401
assert "Authentication failed" in response.json()["error"]["message"]
@patch("httpx.AsyncClient.post", new=mock_post_exception)
def test_http_auth_service_error(http_client, valid_api_key):
def test_http_auth_service_error(http_client, valid_api_key, suppress_auth_errors):
response = http_client.get("/test", headers={"Authorization": f"Bearer {valid_api_key}"})
assert response.status_code == 401
assert "Authentication service error" in response.json()["error"]["message"]
def test_http_auth_request_payload(http_client, valid_api_key, mock_auth_endpoint):
def test_http_auth_request_payload(http_client, valid_api_key, mock_auth_endpoint, suppress_auth_errors):
with patch("httpx.AsyncClient.post") as mock_post:
mock_response = MockResponse(200, {"message": "Authentication successful"})
mock_post.return_value = mock_response
@ -420,7 +428,7 @@ def test_valid_oauth2_authentication(oauth2_client, jwt_token_valid, mock_jwks_u
@patch("httpx.AsyncClient.get", new=mock_jwks_response)
def test_invalid_oauth2_authentication(oauth2_client, invalid_token):
def test_invalid_oauth2_authentication(oauth2_client, invalid_token, suppress_auth_errors):
response = oauth2_client.get("/test", headers={"Authorization": f"Bearer {invalid_token}"})
assert response.status_code == 401
assert "Invalid JWT token" in response.json()["error"]["message"]
@ -465,7 +473,7 @@ def oauth2_client_with_jwks_token(oauth2_app_with_jwks_token):
@patch("httpx.AsyncClient.get", new=mock_auth_jwks_response)
def test_oauth2_with_jwks_token_expected(oauth2_client, jwt_token_valid):
def test_oauth2_with_jwks_token_expected(oauth2_client, jwt_token_valid, suppress_auth_errors):
response = oauth2_client.get("/test", headers={"Authorization": f"Bearer {jwt_token_valid}"})
assert response.status_code == 401
@ -726,21 +734,21 @@ def test_valid_introspection_authentication(introspection_client, valid_api_key)
@patch("httpx.AsyncClient.post", new=mock_introspection_inactive)
def test_inactive_introspection_authentication(introspection_client, invalid_api_key):
def test_inactive_introspection_authentication(introspection_client, invalid_api_key, suppress_auth_errors):
response = introspection_client.get("/test", headers={"Authorization": f"Bearer {invalid_api_key}"})
assert response.status_code == 401
assert "Token not active" in response.json()["error"]["message"]
@patch("httpx.AsyncClient.post", new=mock_introspection_invalid)
def test_invalid_introspection_authentication(introspection_client, invalid_api_key):
def test_invalid_introspection_authentication(introspection_client, invalid_api_key, suppress_auth_errors):
response = introspection_client.get("/test", headers={"Authorization": f"Bearer {invalid_api_key}"})
assert response.status_code == 401
assert "Not JSON" in response.json()["error"]["message"]
@patch("httpx.AsyncClient.post", new=mock_introspection_failed)
def test_failed_introspection_authentication(introspection_client, invalid_api_key):
def test_failed_introspection_authentication(introspection_client, invalid_api_key, suppress_auth_errors):
response = introspection_client.get("/test", headers={"Authorization": f"Bearer {invalid_api_key}"})
assert response.status_code == 401
assert "Token introspection failed: 500" in response.json()["error"]["message"]
@ -957,20 +965,22 @@ def test_valid_kubernetes_auth_authentication(kubernetes_auth_client, valid_toke
@patch("httpx.AsyncClient.post", new=mock_kubernetes_selfsubjectreview_failure)
def test_invalid_kubernetes_auth_authentication(kubernetes_auth_client, invalid_token):
def test_invalid_kubernetes_auth_authentication(kubernetes_auth_client, invalid_token, suppress_auth_errors):
response = kubernetes_auth_client.get("/test", headers={"Authorization": f"Bearer {invalid_token}"})
assert response.status_code == 401
assert "Invalid token" in response.json()["error"]["message"]
@patch("httpx.AsyncClient.post", new=mock_kubernetes_selfsubjectreview_http_error)
def test_kubernetes_auth_http_error(kubernetes_auth_client, valid_token):
def test_kubernetes_auth_http_error(kubernetes_auth_client, valid_token, suppress_auth_errors):
response = kubernetes_auth_client.get("/test", headers={"Authorization": f"Bearer {valid_token}"})
assert response.status_code == 401
assert "Token validation failed" in response.json()["error"]["message"]
def test_kubernetes_auth_request_payload(kubernetes_auth_client, valid_token, mock_kubernetes_api_server):
def test_kubernetes_auth_request_payload(
kubernetes_auth_client, valid_token, mock_kubernetes_api_server, suppress_auth_errors
):
with patch("httpx.AsyncClient.post") as mock_post:
mock_response = MockResponse(
200,