diff mrjunejune/inference/litellm_gateway_test.py @ 260:1f9877b637e9

Add Copilot-powered cyberpunk JRPG chat Integrate the production JRPG chat with Seobeo streaming, Deita persistence, and a Bazel-managed Copilot SDK and LiteLLM inference stack. Co-authored-by: Copilot <[email protected]>
author MrJuneJune <mrjunejune@users.noreply.github.com>
date Wed, 05 Aug 2026 09:19:41 -0700
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/mrjunejune/inference/litellm_gateway_test.py	Wed Aug 05 09:19:41 2026 -0700
@@ -0,0 +1,114 @@
+import os
+import pathlib
+import unittest
+from unittest import mock
+
+from fastapi.testclient import TestClient
+
+from mrjunejune.inference.litellm_proxy import (
+    create_app,
+    resolve_token_directory,
+)
+
+
+class Chunk:
+    def __init__(self, content):
+        self.content = content
+
+    def model_dump(self, exclude_none=True):
+        del exclude_none
+        return {
+            "choices": [
+                {"index": 0, "delta": {"content": self.content}}
+            ]
+        }
+
+
+class Stream:
+    def __init__(self):
+        self.chunks = iter([Chunk("hello "), Chunk("traveler")])
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        try:
+            return next(self.chunks)
+        except StopIteration as error:
+            raise StopAsyncIteration from error
+
+
+class LiteLlmGatewayTest(unittest.TestCase):
+    def setUp(self):
+        os.environ["LITELLM_MASTER_KEY"] = "test-key"
+        self.calls = []
+
+        async def completion(**payload):
+            self.calls.append(payload)
+            return Stream()
+
+        self.client = TestClient(create_app(completion))
+
+    def test_health_does_not_require_provider_auth(self):
+        response = self.client.get("/health/liveliness")
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json(), {"status": "ready"})
+
+    def test_stream_requires_key_and_maps_alias(self):
+        denied = self.client.post(
+            "/v1/chat/completions",
+            json={"model": "jrpg-copilot", "messages": [], "stream": True},
+        )
+        self.assertEqual(denied.status_code, 401)
+
+        with self.client.stream(
+            "POST",
+            "/v1/chat/completions",
+            headers={"Authorization": "Bearer test-key"},
+            json={
+                "model": "jrpg-copilot",
+                "messages": [{"role": "user", "content": "hello"}],
+                "stream": True,
+                "max_tokens": 99999,
+            },
+        ) as response:
+            body = "".join(response.iter_text())
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("hello ", body)
+        self.assertIn("traveler", body)
+        self.assertIn("data: [DONE]", body)
+        self.assertEqual(self.calls[0]["model"], "github_copilot/gpt-4")
+        self.assertEqual(self.calls[0]["max_tokens"], 1024)
+
+    def test_token_directory_precedence(self):
+        with mock.patch.dict(
+            os.environ,
+            {
+                "GITHUB_COPILOT_TOKEN_DIR": "/tmp/copilot-env",
+                "MRJUNEJUNE_INFERENCE_STATE": "/tmp/inference-state",
+            },
+            clear=True,
+        ):
+            self.assertEqual(
+                resolve_token_directory("/tmp/copilot-argument"),
+                pathlib.Path("/tmp/copilot-argument"),
+            )
+            self.assertEqual(
+                resolve_token_directory(),
+                pathlib.Path("/tmp/copilot-env"),
+            )
+
+    def test_token_directory_uses_persistent_state_root(self):
+        with mock.patch.dict(
+            os.environ,
+            {"MRJUNEJUNE_INFERENCE_STATE": "/tmp/inference-state"},
+            clear=True,
+        ):
+            self.assertEqual(
+                resolve_token_directory(),
+                pathlib.Path("/tmp/inference-state/litellm-copilot"),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()