From 682ddf641602ff5be12a835ecf30665f2792804b Mon Sep 17 00:00:00 2001 From: Lidang Jiang <119769478+Lidang-Jiang@users.noreply.github.com> Date: Mon, 6 Apr 2026 10:30:46 +0800 Subject: [PATCH] [CI] Add unit tests for srt/utils/auth.py (#21400) --- test/registered/unit/utils/test_auth.py | 325 ++++++++++++++++++++++++ 1 file changed, 325 insertions(+) create mode 100644 test/registered/unit/utils/test_auth.py diff --git a/test/registered/unit/utils/test_auth.py b/test/registered/unit/utils/test_auth.py new file mode 100644 index 000000000..ac4b9b8e8 --- /dev/null +++ b/test/registered/unit/utils/test_auth.py @@ -0,0 +1,325 @@ +"""Unit tests for srt/utils/auth.py — no server, no model loading.""" + +import unittest + +from sglang.srt.utils.auth import ( + AuthDecision, + AuthLevel, + auth_level, + decide_request_auth, +) +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(1.0, "stage-a-test-cpu") + + +class TestAuthDecision(CustomTestCase): + def test_allowed_default(self): + decision = AuthDecision(allowed=True) + self.assertTrue(decision.allowed) + self.assertEqual(decision.error_status_code, 401) + + def test_not_allowed_with_custom_status(self): + decision = AuthDecision(allowed=False, error_status_code=403) + self.assertFalse(decision.allowed) + self.assertEqual(decision.error_status_code, 403) + + def test_frozen(self): + decision = AuthDecision(allowed=True) + with self.assertRaises(AttributeError): + decision.allowed = False + + +class TestAuthLevel(CustomTestCase): + def test_enum_values(self): + self.assertEqual(AuthLevel.NORMAL.value, "normal") + self.assertEqual(AuthLevel.ADMIN_OPTIONAL.value, "admin_optional") + self.assertEqual(AuthLevel.ADMIN_FORCE.value, "admin_force") + + def test_is_string_enum(self): + self.assertIsInstance(AuthLevel.NORMAL, str) + # str mixin allows direct comparison with string values + self.assertEqual(AuthLevel.NORMAL, "normal") + + +class TestAuthLevelDecorator(CustomTestCase): + def test_decorator_sets_auth_level(self): + @auth_level(AuthLevel.ADMIN_FORCE) + def my_endpoint(): + pass + + self.assertEqual(my_endpoint._auth_level, AuthLevel.ADMIN_FORCE) + + def test_decorator_preserves_function(self): + @auth_level(AuthLevel.NORMAL) + def my_endpoint(): + return 42 + + self.assertEqual(my_endpoint(), 42) + + +class TestDecideRequestAuth(CustomTestCase): + """Tests for the pure decide_request_auth function.""" + + # ==================== Always-Allowed Paths ==================== + + def test_options_method_always_allowed(self): + decision = decide_request_auth( + method="OPTIONS", + path="/v1/chat/completions", + authorization_header=None, + api_key="secret", + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_FORCE, + ) + self.assertTrue(decision.allowed) + + def test_health_path_always_allowed(self): + decision = decide_request_auth( + method="GET", + path="/health", + authorization_header=None, + api_key="secret", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + def test_health_subpath_always_allowed(self): + decision = decide_request_auth( + method="GET", + path="/health_generate", + authorization_header=None, + api_key="secret", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + def test_metrics_path_always_allowed(self): + decision = decide_request_auth( + method="GET", + path="/metrics", + authorization_header=None, + api_key="secret", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + # ==================== NORMAL Auth Level ==================== + + def test_normal_no_keys_configured(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header=None, + api_key=None, + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + def test_normal_with_api_key_correct(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header="Bearer my-api-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + def test_normal_with_api_key_wrong(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header="Bearer wrong-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertFalse(decision.allowed) + + def test_normal_with_api_key_missing_header(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header=None, + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertFalse(decision.allowed) + + def test_normal_only_admin_key_configured(self): + """When only admin_api_key is configured, normal endpoints allow all.""" + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header=None, + api_key=None, + admin_api_key="admin-secret", + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + # ==================== ADMIN_FORCE Auth Level ==================== + + def test_admin_force_no_admin_key_configured(self): + """ADMIN_FORCE without admin_api_key configured returns 403.""" + decision = decide_request_auth( + method="POST", + path="/admin/endpoint", + authorization_header="Bearer my-api-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.ADMIN_FORCE, + ) + self.assertFalse(decision.allowed) + self.assertEqual(decision.error_status_code, 403) + + def test_admin_force_correct_admin_key(self): + decision = decide_request_auth( + method="POST", + path="/admin/endpoint", + authorization_header="Bearer admin-secret", + api_key="my-api-key", + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_FORCE, + ) + self.assertTrue(decision.allowed) + + def test_admin_force_wrong_admin_key(self): + decision = decide_request_auth( + method="POST", + path="/admin/endpoint", + authorization_header="Bearer wrong-key", + api_key="my-api-key", + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_FORCE, + ) + self.assertFalse(decision.allowed) + self.assertEqual(decision.error_status_code, 401) + + def test_admin_force_api_key_not_accepted(self): + """ADMIN_FORCE rejects api_key, only accepts admin_api_key.""" + decision = decide_request_auth( + method="POST", + path="/admin/endpoint", + authorization_header="Bearer my-api-key", + api_key="my-api-key", + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_FORCE, + ) + self.assertFalse(decision.allowed) + + # ==================== ADMIN_OPTIONAL Auth Level ==================== + + def test_admin_optional_no_keys_configured(self): + decision = decide_request_auth( + method="POST", + path="/admin/optional", + authorization_header=None, + api_key=None, + admin_api_key=None, + auth_level=AuthLevel.ADMIN_OPTIONAL, + ) + self.assertTrue(decision.allowed) + + def test_admin_optional_only_api_key_correct(self): + decision = decide_request_auth( + method="POST", + path="/admin/optional", + authorization_header="Bearer my-api-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.ADMIN_OPTIONAL, + ) + self.assertTrue(decision.allowed) + + def test_admin_optional_only_api_key_wrong(self): + decision = decide_request_auth( + method="POST", + path="/admin/optional", + authorization_header="Bearer wrong-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.ADMIN_OPTIONAL, + ) + self.assertFalse(decision.allowed) + + def test_admin_optional_only_admin_key_correct(self): + decision = decide_request_auth( + method="POST", + path="/admin/optional", + authorization_header="Bearer admin-secret", + api_key=None, + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_OPTIONAL, + ) + self.assertTrue(decision.allowed) + + def test_admin_optional_both_keys_requires_admin(self): + """When both keys configured, ADMIN_OPTIONAL requires admin_api_key.""" + decision = decide_request_auth( + method="POST", + path="/admin/optional", + authorization_header="Bearer my-api-key", + api_key="my-api-key", + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_OPTIONAL, + ) + self.assertFalse(decision.allowed) + + def test_admin_optional_both_keys_admin_accepted(self): + decision = decide_request_auth( + method="POST", + path="/admin/optional", + authorization_header="Bearer admin-secret", + api_key="my-api-key", + admin_api_key="admin-secret", + auth_level=AuthLevel.ADMIN_OPTIONAL, + ) + self.assertTrue(decision.allowed) + + # ==================== Bearer Token Edge Cases ==================== + + def test_malformed_authorization_header(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header="NotBearer my-api-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertFalse(decision.allowed) + + def test_empty_authorization_header(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header="", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertFalse(decision.allowed) + + def test_bearer_case_insensitive(self): + decision = decide_request_auth( + method="POST", + path="/v1/chat/completions", + authorization_header="BEARER my-api-key", + api_key="my-api-key", + admin_api_key=None, + auth_level=AuthLevel.NORMAL, + ) + self.assertTrue(decision.allowed) + + +if __name__ == "__main__": + unittest.main()