[CI] Add unit tests for srt/utils/auth.py (#21400)

2026-04-06 10:30:46 +08:00
parent 30f5b87608
commit 682ddf6416
1 changed files with 325 additions and 0 deletions
@@ -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()