mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-08 22:21:35 +00:00
Merge 92c4c828ca into ae7e50f096
This commit is contained in:
Jacky Lam
GitHub
commit
b96ef17194
20 changed files with 688 additions and 315 deletions
|
|
@ -27,7 +27,7 @@ RUST_KWARG_KEY: Final = "rust"
|
|||
# Keys `completion()` forwards from its own kwargs into `get_litellm_params`,
|
||||
# which are otherwise invisible to it because that call site passes explicit
|
||||
# named arguments rather than `**kwargs`.
|
||||
FORWARDED_KWARGS_KEYS: Final = AWS_CREDENTIAL_KWARGS_KEYS | frozenset({RUST_KWARG_KEY})
|
||||
FORWARDED_KWARGS_KEYS: Final = AWS_CREDENTIAL_KWARGS_KEYS | frozenset({RUST_KWARG_KEY, "github_copilot_token_dir"})
|
||||
|
||||
# Pre-define optional kwargs keys as frozenset for O(1) lookups
|
||||
# These are extracted from kwargs only if present, avoiding unnecessary .get() calls
|
||||
|
|
@ -55,6 +55,7 @@ OPTIONAL_KWARGS_KEYS: Final = (
|
|||
"itpm",
|
||||
"otpm",
|
||||
"use_xai_oauth",
|
||||
"github_copilot_token_dir",
|
||||
# The per-deployment Rust opt-in. `all_litellm_params` keeps it out
|
||||
# of the provider body; this keeps it *in* litellm_params, which is
|
||||
# where the chat completions handlers read it from.
|
||||
|
|
|
|||
|
|
@ -746,7 +746,11 @@ def _get_openai_compatible_provider_info(
|
|||
dynamic_api_key,
|
||||
custom_llm_provider,
|
||||
) = litellm.GithubCopilotConfig()._get_openai_compatible_provider_info(
|
||||
model, api_base, api_key, custom_llm_provider
|
||||
model,
|
||||
api_base,
|
||||
api_key,
|
||||
custom_llm_provider,
|
||||
litellm_params=(dict(litellm_params) if litellm_params is not None else None),
|
||||
)
|
||||
elif custom_llm_provider == "chatgpt":
|
||||
(
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime
|
||||
from typing import Any, Final
|
||||
|
||||
|
|
@ -22,22 +24,20 @@ DEFAULT_GITHUB_CLIENT_ID: Final = "Iv1.b507a08c87ecfe98"
|
|||
DEFAULT_GITHUB_DEVICE_CODE_URL: Final = "https://github.com/login/device/code"
|
||||
DEFAULT_GITHUB_ACCESS_TOKEN_URL: Final = "https://github.com/login/oauth/access_token"
|
||||
DEFAULT_GITHUB_API_KEY_URL: Final = "https://api.github.com/copilot_internal/v2/token"
|
||||
GITHUB_COPILOT_TOKEN_DIR_PARAM: Final = "github_copilot_token_dir"
|
||||
|
||||
|
||||
class Authenticator:
|
||||
def __init__(self) -> None:
|
||||
def __init__(self, token_dir: str | None = None) -> None:
|
||||
"""Initialize the GitHub Copilot authenticator with configurable token paths."""
|
||||
# Token storage paths
|
||||
self.token_dir = os.getenv(
|
||||
"GITHUB_COPILOT_TOKEN_DIR",
|
||||
os.path.expanduser("~/.config/litellm/github_copilot"),
|
||||
self.token_dir = os.path.expanduser(
|
||||
token_dir or os.getenv("GITHUB_COPILOT_TOKEN_DIR") or "~/.config/litellm/github_copilot"
|
||||
)
|
||||
self.access_token_file = os.path.join(
|
||||
self.token_dir,
|
||||
os.getenv("GITHUB_COPILOT_ACCESS_TOKEN_FILE", "access-token"),
|
||||
)
|
||||
self.api_key_file = os.path.join(self.token_dir, os.getenv("GITHUB_COPILOT_API_KEY_FILE", "api-key.json"))
|
||||
self._ensure_token_dir()
|
||||
|
||||
def get_access_token(self) -> str:
|
||||
"""
|
||||
|
|
@ -62,8 +62,7 @@ class Authenticator:
|
|||
try:
|
||||
access_token = self._login()
|
||||
try:
|
||||
with open(self.access_token_file, "w") as f:
|
||||
f.write(access_token)
|
||||
self._write_private_text(self.access_token_file, access_token)
|
||||
except OSError:
|
||||
verbose_logger.error("Error saving access token to file")
|
||||
return access_token
|
||||
|
|
@ -106,8 +105,7 @@ class Authenticator:
|
|||
|
||||
try:
|
||||
api_key_info = self._refresh_api_key()
|
||||
with open(self.api_key_file, "w") as f:
|
||||
json.dump(api_key_info, f)
|
||||
self._write_private_text(self.api_key_file, json.dumps(api_key_info))
|
||||
token: Final = api_key_info.get("token")
|
||||
if token:
|
||||
return token
|
||||
|
|
@ -141,6 +139,11 @@ class Authenticator:
|
|||
endpoints: Final = api_key_info.get("endpoints", {})
|
||||
api_endpoint: Final = endpoints.get("api")
|
||||
return api_endpoint
|
||||
except FileNotFoundError:
|
||||
# A global token file is intentionally absent when all Copilot
|
||||
# deployments use their own token directories.
|
||||
verbose_logger.debug("No API endpoint file found at %s", self.api_key_file)
|
||||
return None
|
||||
except (OSError, json.JSONDecodeError, KeyError) as e:
|
||||
verbose_logger.warning("Error reading API endpoint from file: %s", e)
|
||||
return None
|
||||
|
|
@ -185,7 +188,21 @@ class Authenticator:
|
|||
def _ensure_token_dir(self) -> None:
|
||||
"""Ensure the token directory exists."""
|
||||
if not os.path.exists(self.token_dir):
|
||||
os.makedirs(self.token_dir, exist_ok=True)
|
||||
os.makedirs(self.token_dir, mode=0o700, exist_ok=True)
|
||||
|
||||
def _write_private_text(self, file_path: str, value: str) -> None:
|
||||
self._ensure_token_dir()
|
||||
file_descriptor, temporary_path = tempfile.mkstemp(prefix=".litellm-", dir=self.token_dir, text=True)
|
||||
try:
|
||||
with os.fdopen(file_descriptor, "w") as temporary_file:
|
||||
temporary_file.write(value)
|
||||
temporary_file.flush()
|
||||
os.fsync(temporary_file.fileno())
|
||||
os.chmod(temporary_path, 0o600)
|
||||
os.replace(temporary_path, file_path)
|
||||
finally:
|
||||
if os.path.exists(temporary_path):
|
||||
os.unlink(temporary_path)
|
||||
|
||||
def _get_github_headers(self, access_token: str | None = None) -> dict[str, str]:
|
||||
"""
|
||||
|
|
@ -354,3 +371,18 @@ class Authenticator:
|
|||
)
|
||||
|
||||
return self._poll_for_access_token(device_code)
|
||||
|
||||
|
||||
def get_authenticator_for_litellm_params(
|
||||
default_authenticator: Authenticator,
|
||||
litellm_params: Mapping[str, object] | None,
|
||||
) -> Authenticator:
|
||||
if litellm_params is None:
|
||||
return default_authenticator
|
||||
token_dir: Final = litellm_params.get(GITHUB_COPILOT_TOKEN_DIR_PARAM)
|
||||
if not isinstance(token_dir, str) or not token_dir.strip():
|
||||
return default_authenticator
|
||||
expanded_token_dir: Final = os.path.expanduser(token_dir.strip())
|
||||
if expanded_token_dir == default_authenticator.token_dir:
|
||||
return default_authenticator
|
||||
return Authenticator(token_dir=expanded_token_dir)
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import json
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, Final
|
||||
|
||||
import httpx
|
||||
|
|
@ -10,7 +11,11 @@ from litellm.llms.openai.openai import OpenAIConfig
|
|||
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolCallChunk
|
||||
from litellm.types.utils import ModelResponse
|
||||
|
||||
from ..authenticator import Authenticator
|
||||
from ..authenticator import (
|
||||
GITHUB_COPILOT_TOKEN_DIR_PARAM,
|
||||
Authenticator,
|
||||
get_authenticator_for_litellm_params,
|
||||
)
|
||||
from ..common_utils import (
|
||||
DEFAULT_GITHUB_COPILOT_API_BASE,
|
||||
GetAPIKeyError,
|
||||
|
|
@ -34,23 +39,55 @@ class GithubCopilotConfig(OpenAIConfig):
|
|||
api_base: str | None,
|
||||
api_key: str | None,
|
||||
custom_llm_provider: str,
|
||||
litellm_params: Mapping[str, object] | None = None,
|
||||
) -> tuple[str | None, str | None, str]:
|
||||
dynamic_api_base: Final = (
|
||||
api_base
|
||||
or self.authenticator.get_api_base()
|
||||
or os.getenv("GITHUB_COPILOT_API_BASE")
|
||||
or DEFAULT_GITHUB_COPILOT_API_BASE
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
configured_token_dir: Final = (
|
||||
litellm_params.get(GITHUB_COPILOT_TOKEN_DIR_PARAM) if litellm_params is not None else None
|
||||
)
|
||||
try:
|
||||
dynamic_api_key: Final = self.authenticator.get_api_key()
|
||||
dynamic_api_key: Final = (
|
||||
api_key
|
||||
if api_key is not None
|
||||
else authenticator.get_api_key()
|
||||
if isinstance(configured_token_dir, str) and configured_token_dir.strip()
|
||||
else None
|
||||
)
|
||||
except GetAPIKeyError as e:
|
||||
raise AuthenticationError(
|
||||
model=model,
|
||||
llm_provider=custom_llm_provider,
|
||||
message=str(e),
|
||||
)
|
||||
dynamic_api_base: Final = (
|
||||
authenticator.get_api_base() or os.getenv("GITHUB_COPILOT_API_BASE") or DEFAULT_GITHUB_COPILOT_API_BASE
|
||||
)
|
||||
return dynamic_api_base, dynamic_api_key, custom_llm_provider
|
||||
|
||||
def resolve_request_credentials(
|
||||
self,
|
||||
model: str,
|
||||
api_key: str | None,
|
||||
headers: Mapping[str, str] | None,
|
||||
litellm_params: Mapping[str, object] | None,
|
||||
) -> tuple[str, dict[str, str]]: # mutable-ok: OpenAI request headers are merged by callers
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
try:
|
||||
resolved_api_key: Final = api_key or authenticator.get_api_key()
|
||||
except GetAPIKeyError as e:
|
||||
raise AuthenticationError(
|
||||
model=model,
|
||||
llm_provider="github_copilot",
|
||||
message=str(e),
|
||||
)
|
||||
return resolved_api_key, {**get_copilot_default_headers(resolved_api_key), **dict(headers or {})}
|
||||
|
||||
def _transform_messages(
|
||||
self,
|
||||
messages,
|
||||
|
|
@ -83,22 +120,25 @@ class GithubCopilotConfig(OpenAIConfig):
|
|||
model: str,
|
||||
messages: list[AllMessageValues],
|
||||
optional_params: dict,
|
||||
litellm_params: dict,
|
||||
litellm_params: dict[str, object],
|
||||
api_key: str | None = None,
|
||||
api_base: str | None = None,
|
||||
) -> dict:
|
||||
# Get base headers from parent
|
||||
validated_headers = super().validate_environment(
|
||||
headers, model, messages, optional_params, litellm_params, api_key, api_base
|
||||
copilot_api_key, copilot_headers = self.resolve_request_credentials(
|
||||
model=model,
|
||||
api_key=api_key,
|
||||
headers=headers,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
validated_headers = super().validate_environment(
|
||||
copilot_headers,
|
||||
model,
|
||||
messages,
|
||||
optional_params,
|
||||
litellm_params,
|
||||
copilot_api_key,
|
||||
api_base,
|
||||
)
|
||||
|
||||
# Add Copilot-specific headers (editor-version, user-agent, etc.)
|
||||
try:
|
||||
copilot_api_key: Final = self.authenticator.get_api_key()
|
||||
copilot_headers: Final = get_copilot_default_headers(copilot_api_key)
|
||||
validated_headers = {**copilot_headers, **validated_headers}
|
||||
except GetAPIKeyError:
|
||||
pass # Will be handled later in the request flow
|
||||
|
||||
# Add X-Initiator header based on message roles
|
||||
initiator: Final = self._determine_initiator(messages)
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class GetAPIKeyError(GithubCopilotError):
|
|||
pass
|
||||
|
||||
|
||||
def get_copilot_default_headers(api_key: str) -> dict:
|
||||
def get_copilot_default_headers(api_key: str) -> dict[str, str]:
|
||||
"""
|
||||
Get default headers for GitHub Copilot Responses API.
|
||||
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ from litellm.types.llms.openai import AllEmbeddingInputValues
|
|||
from litellm.types.utils import EmbeddingResponse
|
||||
from litellm.utils import convert_to_model_response_object
|
||||
|
||||
from ..authenticator import Authenticator
|
||||
from ..authenticator import Authenticator, get_authenticator_for_litellm_params
|
||||
from ..common_utils import (
|
||||
DEFAULT_GITHUB_COPILOT_API_BASE,
|
||||
GetAPIKeyError,
|
||||
|
|
@ -51,7 +51,7 @@ class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
|
|||
model: str,
|
||||
messages: list,
|
||||
optional_params: dict,
|
||||
litellm_params: dict,
|
||||
litellm_params: dict[str, object],
|
||||
api_key: str | None = None,
|
||||
api_base: str | None = None,
|
||||
) -> dict:
|
||||
|
|
@ -59,10 +59,13 @@ class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
|
|||
Validate environment and set up headers for GitHub Copilot API.
|
||||
"""
|
||||
try:
|
||||
# Get GitHub Copilot API key via OAuth
|
||||
api_key = self.authenticator.get_api_key()
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
resolved_api_key: Final = api_key or authenticator.get_api_key()
|
||||
|
||||
if not api_key:
|
||||
if not resolved_api_key:
|
||||
raise AuthenticationError(
|
||||
model=model,
|
||||
llm_provider="github_copilot",
|
||||
|
|
@ -70,7 +73,7 @@ class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
|
|||
)
|
||||
|
||||
# Get default headers
|
||||
default_headers: Final = get_copilot_default_headers(api_key)
|
||||
default_headers: Final = get_copilot_default_headers(resolved_api_key)
|
||||
|
||||
# Merge with existing headers (user's extra_headers take priority)
|
||||
merged_headers: Final = {**default_headers, **headers}
|
||||
|
|
@ -92,22 +95,19 @@ class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
|
|||
api_key: str | None,
|
||||
model: str,
|
||||
optional_params: dict,
|
||||
litellm_params: dict,
|
||||
litellm_params: dict[str, object],
|
||||
stream: bool | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
Get the complete URL for GitHub Copilot Embedding API endpoint.
|
||||
"""
|
||||
# Use provided api_base or fall back to authenticator's base or default
|
||||
effective_api_base = (
|
||||
api_base
|
||||
or self.authenticator.get_api_base()
|
||||
or os.getenv("GITHUB_COPILOT_API_BASE")
|
||||
or DEFAULT_GITHUB_COPILOT_API_BASE
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
|
||||
# Remove trailing slashes
|
||||
effective_api_base = effective_api_base.rstrip("/")
|
||||
effective_api_base: Final = (
|
||||
authenticator.get_api_base() or os.getenv("GITHUB_COPILOT_API_BASE") or DEFAULT_GITHUB_COPILOT_API_BASE
|
||||
).rstrip("/")
|
||||
|
||||
# Return the embeddings endpoint
|
||||
return f"{effective_api_base}/embeddings"
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from litellm.llms.anthropic.experimental_pass_through.messages.transformation im
|
|||
AnthropicMessagesConfig,
|
||||
)
|
||||
|
||||
from ..authenticator import Authenticator
|
||||
from ..authenticator import Authenticator, get_authenticator_for_litellm_params
|
||||
from ..common_utils import (
|
||||
DEFAULT_GITHUB_COPILOT_API_BASE,
|
||||
GetAPIKeyError,
|
||||
|
|
@ -53,7 +53,7 @@ class GithubCopilotAnthropicMessagesConfig(AnthropicMessagesConfig):
|
|||
model: str,
|
||||
messages: list[Any],
|
||||
optional_params: dict,
|
||||
litellm_params: dict,
|
||||
litellm_params: dict[str, object],
|
||||
api_key: str | None = None,
|
||||
api_base: str | None = None,
|
||||
) -> tuple[dict, str | None]:
|
||||
|
|
@ -68,15 +68,19 @@ class GithubCopilotAnthropicMessagesConfig(AnthropicMessagesConfig):
|
|||
# session, never the caller-supplied api_base. rstrip so a
|
||||
# tenant-specific base with a trailing slash does not yield a
|
||||
# double-slash URL once "/v1/messages" is appended downstream.
|
||||
dynamic_api_base: Final = (self.authenticator.get_api_base() or DEFAULT_GITHUB_COPILOT_API_BASE).rstrip("/")
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
try:
|
||||
dynamic_api_key: Final = self.authenticator.get_api_key()
|
||||
dynamic_api_key: Final = api_key or authenticator.get_api_key()
|
||||
except GetAPIKeyError as e:
|
||||
raise AuthenticationError(
|
||||
model=model,
|
||||
llm_provider="github_copilot",
|
||||
message=str(e),
|
||||
)
|
||||
dynamic_api_base: Final = (authenticator.get_api_base() or DEFAULT_GITHUB_COPILOT_API_BASE).rstrip("/")
|
||||
|
||||
# Merge Copilot headers with provided headers
|
||||
copilot_headers: Final = get_copilot_default_headers(dynamic_api_key)
|
||||
|
|
@ -103,7 +107,7 @@ class GithubCopilotAnthropicMessagesConfig(AnthropicMessagesConfig):
|
|||
api_key: str | None,
|
||||
model: str,
|
||||
optional_params: dict,
|
||||
litellm_params: dict,
|
||||
litellm_params: dict[str, object],
|
||||
stream: bool | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
|
|
@ -116,7 +120,11 @@ class GithubCopilotAnthropicMessagesConfig(AnthropicMessagesConfig):
|
|||
reuse it to avoid a second authenticator read, falling back to a fresh
|
||||
resolution only if it was not provided.
|
||||
"""
|
||||
resolved = (api_base or self.authenticator.get_api_base() or DEFAULT_GITHUB_COPILOT_API_BASE).rstrip("/")
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
resolved = (api_base or authenticator.get_api_base() or DEFAULT_GITHUB_COPILOT_API_BASE).rstrip("/")
|
||||
if not resolved.endswith("/v1/messages"):
|
||||
resolved = f"{resolved}/v1/messages"
|
||||
return resolved
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ https://github.com/caozhiyuan/copilot-api
|
|||
"""
|
||||
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
import litellm
|
||||
|
|
@ -24,7 +25,7 @@ from litellm.types.router import GenericLiteLLMParams
|
|||
from litellm.types.utils import LlmProviders
|
||||
from litellm.utils import _cached_get_model_info_helper
|
||||
|
||||
from ..authenticator import Authenticator
|
||||
from ..authenticator import Authenticator, get_authenticator_for_litellm_params
|
||||
from ..common_utils import (
|
||||
DEFAULT_GITHUB_COPILOT_API_BASE,
|
||||
GetAPIKeyError,
|
||||
|
|
@ -183,7 +184,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
|
|||
self,
|
||||
headers: dict,
|
||||
model: str,
|
||||
litellm_params: GenericLiteLLMParams | None,
|
||||
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None,
|
||||
) -> dict:
|
||||
"""
|
||||
Validate environment and set up headers for GitHub Copilot API.
|
||||
|
|
@ -199,8 +200,18 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
|
|||
- User-provided extra_headers (merged with priority)
|
||||
"""
|
||||
try:
|
||||
# Get GitHub Copilot API key via OAuth
|
||||
api_key: Final = self.authenticator.get_api_key()
|
||||
params_dict: Final[Mapping[str, object]] = (
|
||||
litellm_params.model_dump(exclude_none=True)
|
||||
if isinstance(litellm_params, GenericLiteLLMParams)
|
||||
else litellm_params or {}
|
||||
)
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=params_dict,
|
||||
)
|
||||
configured_api_key: Final = params_dict.get("api_key")
|
||||
explicit_api_key: Final = configured_api_key if isinstance(configured_api_key, str) else None
|
||||
api_key: Final = explicit_api_key or authenticator.get_api_key()
|
||||
|
||||
if not api_key:
|
||||
raise AuthenticationError(
|
||||
|
|
@ -243,21 +254,18 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
|
|||
def get_complete_url(
|
||||
self,
|
||||
api_base: str | None,
|
||||
litellm_params: dict,
|
||||
litellm_params: dict[str, object],
|
||||
) -> str:
|
||||
"""
|
||||
Get the complete URL for GitHub Copilot Responses API endpoint.
|
||||
"""
|
||||
# Use provided api_base or fall back to authenticator's base or default
|
||||
effective_api_base = (
|
||||
api_base
|
||||
or self.authenticator.get_api_base()
|
||||
or os.getenv("GITHUB_COPILOT_API_BASE")
|
||||
or DEFAULT_GITHUB_COPILOT_API_BASE
|
||||
authenticator: Final = get_authenticator_for_litellm_params(
|
||||
default_authenticator=self.authenticator,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
|
||||
# Remove trailing slashes
|
||||
effective_api_base = effective_api_base.rstrip("/")
|
||||
effective_api_base: Final = (
|
||||
authenticator.get_api_base() or os.getenv("GITHUB_COPILOT_API_BASE") or DEFAULT_GITHUB_COPILOT_API_BASE
|
||||
).rstrip("/")
|
||||
|
||||
# Return the responses endpoint
|
||||
return f"{effective_api_base}/responses"
|
||||
|
|
@ -303,7 +311,10 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
|
|||
|
||||
# ==================== Helper Methods ====================
|
||||
|
||||
def _get_input_from_params(self, litellm_params: GenericLiteLLMParams | None) -> str | ResponseInputParam | None:
|
||||
def _get_input_from_params(
|
||||
self,
|
||||
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None,
|
||||
) -> str | ResponseInputParam | None:
|
||||
"""
|
||||
Extract input parameter from litellm_params.
|
||||
|
||||
|
|
|
|||
|
|
@ -2536,17 +2536,17 @@ def _complete_custom_openai(
|
|||
|
||||
# Add GitHub Copilot headers (same as /responses endpoint does)
|
||||
if custom_llm_provider == "github_copilot":
|
||||
from litellm.llms.github_copilot.authenticator import Authenticator
|
||||
from litellm.llms.github_copilot.common_utils import (
|
||||
get_copilot_default_headers,
|
||||
)
|
||||
from litellm.llms.github_copilot.chat.transformation import GithubCopilotConfig
|
||||
|
||||
copilot_auth: Final = Authenticator()
|
||||
copilot_api_key: Final = copilot_auth.get_api_key()
|
||||
copilot_headers: Final = get_copilot_default_headers(copilot_api_key)
|
||||
if extra_headers:
|
||||
copilot_headers.update(extra_headers)
|
||||
extra_headers = copilot_headers
|
||||
copilot_config: Final = (
|
||||
provider_config if isinstance(provider_config, GithubCopilotConfig) else GithubCopilotConfig()
|
||||
)
|
||||
api_key, extra_headers = copilot_config.resolve_request_credentials(
|
||||
model=model,
|
||||
api_key=ctx.api_key,
|
||||
headers=extra_headers,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
|
||||
if extra_headers is not None:
|
||||
optional_params["extra_headers"] = extra_headers
|
||||
|
|
@ -6163,12 +6163,21 @@ def embedding(
|
|||
non_default_params: Final = {
|
||||
k: v for k, v in kwargs.items() if k not in default_params
|
||||
} # model-specific params - pass them straight to the model/provider
|
||||
github_copilot_token_dir: Final = (
|
||||
kwargs["github_copilot_token_dir"]
|
||||
if "github_copilot_token_dir" in kwargs and isinstance(kwargs["github_copilot_token_dir"], str)
|
||||
else None
|
||||
)
|
||||
provider_litellm_params: Final = GenericLiteLLMParams.model_validate(
|
||||
{"github_copilot_token_dir": github_copilot_token_dir} if github_copilot_token_dir is not None else {}
|
||||
)
|
||||
|
||||
model, custom_llm_provider, dynamic_api_key, api_base = get_llm_provider(
|
||||
model=model,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
api_base=api_base,
|
||||
api_key=api_key,
|
||||
litellm_params=provider_litellm_params,
|
||||
)
|
||||
|
||||
if dynamic_api_key is not None:
|
||||
|
|
@ -6249,7 +6258,6 @@ def embedding(
|
|||
litellm_params=litellm_params_dict,
|
||||
)
|
||||
elif custom_llm_provider == "github_copilot":
|
||||
api_key = api_key or litellm.api_key
|
||||
response = base_llm_http_handler.embedding(
|
||||
model=model,
|
||||
input=input,
|
||||
|
|
|
|||
|
|
@ -1441,6 +1441,22 @@ class ModelManagementAuthChecks:
|
|||
Common auth checks for model management endpoints
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def require_proxy_admin_for_github_copilot_token_dir(
|
||||
model_params: Deployment | updateDeployment,
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
) -> Literal[True]:
|
||||
litellm_params: Final = model_params.litellm_params
|
||||
has_token_dir: Final = (
|
||||
litellm_params is not None and "github_copilot_token_dir" in litellm_params.model_fields_set
|
||||
)
|
||||
if has_token_dir and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail={"error": "Only proxy admins can configure GitHub Copilot token directories."},
|
||||
)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def can_user_make_team_model_call(
|
||||
team_id: str,
|
||||
|
|
@ -1471,6 +1487,10 @@ class ModelManagementAuthChecks:
|
|||
prisma_client: PrismaClient,
|
||||
premium_user: bool,
|
||||
) -> Literal[True]:
|
||||
ModelManagementAuthChecks.require_proxy_admin_for_github_copilot_token_dir(
|
||||
model_params=model_params,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
)
|
||||
if model_params.model_info is None or model_params.model_info.team_id is None:
|
||||
return True
|
||||
if model_params.model_info.team_id is not None and premium_user is not True:
|
||||
|
|
@ -1506,6 +1526,10 @@ class ModelManagementAuthChecks:
|
|||
premium_user: bool,
|
||||
allow_missing_team: bool = False,
|
||||
) -> Literal[True]:
|
||||
ModelManagementAuthChecks.require_proxy_admin_for_github_copilot_token_dir(
|
||||
model_params=model_params,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
)
|
||||
## Check team model auth
|
||||
if model_params.model_info is not None and model_params.model_info.team_id is not None:
|
||||
team_obj_row: Final = await _repo_team_table(prisma_client).find_unique(
|
||||
|
|
|
|||
|
|
@ -3435,6 +3435,7 @@ class Router:
|
|||
- Adds default litellm params to kwargs, if set.
|
||||
- Merges tools from deployment with request (proxy-configured tools + request tools).
|
||||
"""
|
||||
kwargs.pop("github_copilot_token_dir", None)
|
||||
for key in self._forwarded_alias_marker_keys_the_deployment_sets(
|
||||
deployment=deployment, forwarded_keys=kwargs.pop(_ALIAS_MARKER_FORWARDED_PARAMS_KWARG, ())
|
||||
):
|
||||
|
|
|
|||
|
|
@ -3544,6 +3544,7 @@ all_litellm_params = (
|
|||
"mock_timeout",
|
||||
"disable_add_transform_inline_image_block",
|
||||
"api_key",
|
||||
"github_copilot_token_dir",
|
||||
"api_version",
|
||||
"prompt_id",
|
||||
"prompt_variables",
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
"limit": 733
|
||||
},
|
||||
"TQ002": {
|
||||
"limit": 742
|
||||
"limit": 741
|
||||
},
|
||||
"TQ003": {
|
||||
"limit": 62
|
||||
|
|
|
|||
|
|
@ -11,6 +11,9 @@ from litellm.litellm_core_utils.get_litellm_params import (
|
|||
_get_base_model_from_litellm_call_metadata,
|
||||
get_litellm_params,
|
||||
)
|
||||
from litellm.types.router import CredentialLiteLLMParams, GenericLiteLLMParams
|
||||
from litellm.types.utils import all_litellm_params
|
||||
from litellm.utils import get_non_default_completion_params
|
||||
|
||||
|
||||
class TestGetBaseModelFromLitellmCallMetadata:
|
||||
|
|
@ -30,9 +33,7 @@ class TestGetBaseModelFromLitellmCallMetadata:
|
|||
assert _get_base_model_from_litellm_call_metadata({"model_info": {}}) is None
|
||||
|
||||
def test_returns_base_model(self):
|
||||
result = _get_base_model_from_litellm_call_metadata(
|
||||
{"model_info": {"base_model": "gpt-4"}}
|
||||
)
|
||||
result = _get_base_model_from_litellm_call_metadata({"model_info": {"base_model": "gpt-4"}})
|
||||
assert result == "gpt-4"
|
||||
|
||||
|
||||
|
|
@ -73,6 +74,19 @@ class TestGetLitellmParamsKwargsExtraction:
|
|||
for key in _OPTIONAL_KWARGS_KEYS:
|
||||
assert result[key] == f"val_{key}"
|
||||
|
||||
def test_github_copilot_token_dir_is_internal_credential_config(self):
|
||||
token_dir = "/var/lib/litellm/copilot/account-a"
|
||||
|
||||
assert get_non_default_completion_params({"github_copilot_token_dir": token_dir}) == {}
|
||||
assert "github_copilot_token_dir" in all_litellm_params
|
||||
internal_params = GenericLiteLLMParams.model_validate(
|
||||
{"github_copilot_token_dir": token_dir}
|
||||
).model_dump(exclude_none=True)
|
||||
public_credentials = CredentialLiteLLMParams.model_validate(internal_params).model_dump(exclude_none=True)
|
||||
|
||||
assert internal_params["github_copilot_token_dir"] == token_dir
|
||||
assert "github_copilot_token_dir" not in public_credentials
|
||||
|
||||
|
||||
class TestGetLitellmParamsBaseModel:
|
||||
"""Verify base_model resolution precedence."""
|
||||
|
|
@ -85,9 +99,7 @@ class TestGetLitellmParamsBaseModel:
|
|||
assert result["base_model"] == "explicit"
|
||||
|
||||
def test_falls_back_to_metadata(self):
|
||||
result = get_litellm_params(
|
||||
metadata={"model_info": {"base_model": "from-metadata"}}
|
||||
)
|
||||
result = get_litellm_params(metadata={"model_info": {"base_model": "from-metadata"}})
|
||||
assert result["base_model"] == "from-metadata"
|
||||
|
||||
def test_none_when_no_source(self):
|
||||
|
|
|
|||
|
|
@ -1,13 +1,17 @@
|
|||
from unittest.mock import MagicMock, patch
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
|
||||
import litellm
|
||||
from litellm.exceptions import AuthenticationError
|
||||
from litellm.llms.github_copilot.common_utils import GetAPIKeyError
|
||||
from litellm.llms.github_copilot.embedding.transformation import (
|
||||
GithubCopilotEmbeddingConfig,
|
||||
)
|
||||
from litellm.llms.github_copilot.common_utils import GetAPIKeyError
|
||||
|
||||
|
||||
def test_github_copilot_embedding_config_validate_environment():
|
||||
|
|
@ -64,7 +68,7 @@ def test_github_copilot_embedding_config_get_complete_url():
|
|||
# Test with default API base
|
||||
config.authenticator.get_api_base.return_value = None
|
||||
url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_base="https://api.githubcopilot.com/",
|
||||
api_key=None,
|
||||
model="github_copilot/text-embedding-3-small",
|
||||
optional_params={},
|
||||
|
|
@ -73,9 +77,7 @@ def test_github_copilot_embedding_config_get_complete_url():
|
|||
assert url == "https://api.githubcopilot.com/embeddings"
|
||||
|
||||
# Test with custom API base from authenticator
|
||||
config.authenticator.get_api_base.return_value = (
|
||||
"https://api.enterprise.githubcopilot.com"
|
||||
)
|
||||
config.authenticator.get_api_base.return_value = "https://api.enterprise.githubcopilot.com"
|
||||
url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
|
|
@ -85,8 +87,8 @@ def test_github_copilot_embedding_config_get_complete_url():
|
|||
)
|
||||
assert url == "https://api.enterprise.githubcopilot.com/embeddings"
|
||||
|
||||
# Test with custom API base from params
|
||||
config.authenticator.get_api_base.return_value = None
|
||||
# Caller-controlled bases must not receive the Copilot bearer token.
|
||||
config.authenticator.get_api_base.return_value = "https://api.enterprise.githubcopilot.com"
|
||||
url = config.get_complete_url(
|
||||
api_base="https://custom.api.com",
|
||||
api_key=None,
|
||||
|
|
@ -94,7 +96,77 @@ def test_github_copilot_embedding_config_get_complete_url():
|
|||
optional_params={},
|
||||
litellm_params={},
|
||||
)
|
||||
assert url == "https://custom.api.com/embeddings"
|
||||
assert url == "https://api.enterprise.githubcopilot.com/embeddings"
|
||||
|
||||
|
||||
def test_github_copilot_embedding_uses_per_deployment_token_directory(tmp_path):
|
||||
token_dir = tmp_path / "embedding-account"
|
||||
token_dir.mkdir()
|
||||
(token_dir / "api-key.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"token": "embedding-account-key",
|
||||
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
|
||||
"endpoints": {"api": "https://embedding-account.example"},
|
||||
}
|
||||
)
|
||||
)
|
||||
params = {"github_copilot_token_dir": str(token_dir)}
|
||||
config = GithubCopilotEmbeddingConfig()
|
||||
|
||||
headers = config.validate_environment(
|
||||
headers={},
|
||||
model="github_copilot/text-embedding-3-small",
|
||||
messages=[],
|
||||
optional_params={},
|
||||
litellm_params=params,
|
||||
)
|
||||
url = config.get_complete_url(
|
||||
api_base="https://attacker.example",
|
||||
api_key=None,
|
||||
model="github_copilot/text-embedding-3-small",
|
||||
optional_params={},
|
||||
litellm_params=params,
|
||||
)
|
||||
|
||||
assert headers["Authorization"] == "Bearer embedding-account-key"
|
||||
assert url == "https://embedding-account.example/embeddings"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_embedding_forwards_selected_account_credentials(tmp_path):
|
||||
token_dir = tmp_path / "embedding-account"
|
||||
token_dir.mkdir()
|
||||
(token_dir / "api-key.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"token": "embedding-account-key",
|
||||
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
|
||||
"endpoints": {"api": "https://embedding-account.example"},
|
||||
}
|
||||
)
|
||||
)
|
||||
route = respx.post("https://embedding-account.example/embeddings").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"object": "list",
|
||||
"data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}],
|
||||
"model": "text-embedding-3-small",
|
||||
"usage": {"prompt_tokens": 1, "total_tokens": 1},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
response = litellm.embedding(
|
||||
model="github_copilot/text-embedding-3-small",
|
||||
input=["hello"],
|
||||
github_copilot_token_dir=str(token_dir),
|
||||
)
|
||||
|
||||
assert route.called
|
||||
assert route.calls.last.request.headers["authorization"] == "Bearer embedding-account-key"
|
||||
assert response.data[0]["embedding"] == [0.1, 0.2]
|
||||
|
||||
|
||||
def test_github_copilot_embedding_config_transform_request():
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
from litellm.exceptions import AuthenticationError
|
||||
from litellm.llms.github_copilot.common_utils import GetAPIKeyError
|
||||
from litellm.llms.github_copilot.messages.transformation import (
|
||||
|
|
@ -124,6 +125,35 @@ def test_github_copilot_anthropic_messages_validate_environment():
|
|||
assert api_base == "https://api.githubcopilot.com"
|
||||
|
||||
|
||||
def test_github_copilot_messages_uses_per_deployment_token_directory(tmp_path):
|
||||
token_dir = tmp_path / "messages-account"
|
||||
token_dir.mkdir()
|
||||
(token_dir / "api-key.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"token": "messages-account-key",
|
||||
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
|
||||
"endpoints": {"api": "https://messages-account.example"},
|
||||
}
|
||||
)
|
||||
)
|
||||
params = {"github_copilot_token_dir": str(token_dir)}
|
||||
config = GithubCopilotAnthropicMessagesConfig()
|
||||
|
||||
headers, api_base = config.validate_anthropic_messages_environment(
|
||||
headers={},
|
||||
model="github_copilot/claude-haiku-4.5",
|
||||
messages=[{"role": "user", "content": "Hello"}],
|
||||
optional_params={},
|
||||
litellm_params=params,
|
||||
api_key=None,
|
||||
api_base="https://attacker.example.com",
|
||||
)
|
||||
|
||||
assert headers["Authorization"] == "Bearer messages-account-key"
|
||||
assert api_base == "https://messages-account.example"
|
||||
|
||||
|
||||
def test_github_copilot_anthropic_messages_validate_environment_injects_beta_headers():
|
||||
"""Anthropic-beta headers must be auto-injected for advanced features
|
||||
(context_management, output_format, etc.) — matches the parent
|
||||
|
|
|
|||
|
|
@ -7,18 +7,21 @@ transformations for the Responses API.
|
|||
Source: litellm/llms/github_copilot/responses/transformation.py
|
||||
"""
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.get_model_cost_map import get_model_cost_map
|
||||
from litellm.types.utils import LlmProviders
|
||||
from litellm.utils import ProviderConfigManager
|
||||
from litellm.llms.github_copilot.responses.transformation import (
|
||||
GithubCopilotResponsesAPIConfig,
|
||||
)
|
||||
from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.utils import LlmProviders
|
||||
from litellm.utils import ProviderConfigManager
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
|
@ -26,9 +29,7 @@ def use_local_model_cost_map(monkeypatch: pytest.MonkeyPatch):
|
|||
"""Pin litellm.model_cost to the bundled local backup so tests don't depend
|
||||
on remote catalog fetches (and don't change behavior across remote refreshes)."""
|
||||
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
monkeypatch.setattr(
|
||||
litellm, "model_cost", get_model_cost_map(url=litellm.model_cost_map_url)
|
||||
)
|
||||
monkeypatch.setattr(litellm, "model_cost", get_model_cost_map(url=litellm.model_cost_map_url))
|
||||
litellm.add_known_models(model_cost_map=litellm.model_cost)
|
||||
|
||||
|
||||
|
|
@ -44,49 +45,38 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
provider=LlmProviders.GITHUB_COPILOT,
|
||||
)
|
||||
|
||||
assert (
|
||||
config is not None
|
||||
), "Config should not be None for GitHub Copilot provider"
|
||||
assert isinstance(
|
||||
config, GithubCopilotResponsesAPIConfig
|
||||
), f"Expected GithubCopilotResponsesAPIConfig, got {type(config)}"
|
||||
assert (
|
||||
config.custom_llm_provider == LlmProviders.GITHUB_COPILOT
|
||||
), "custom_llm_provider should be GITHUB_COPILOT"
|
||||
assert config is not None, "Config should not be None for GitHub Copilot provider"
|
||||
assert isinstance(config, GithubCopilotResponsesAPIConfig), (
|
||||
f"Expected GithubCopilotResponsesAPIConfig, got {type(config)}"
|
||||
)
|
||||
assert config.custom_llm_provider == LlmProviders.GITHUB_COPILOT, "custom_llm_provider should be GITHUB_COPILOT"
|
||||
|
||||
@patch("litellm.llms.github_copilot.responses.transformation.Authenticator")
|
||||
def test_github_copilot_responses_endpoint_url(self, mock_authenticator_class):
|
||||
"""Test that get_complete_url returns correct GitHub Copilot endpoint"""
|
||||
# Mock authenticator to return default base
|
||||
mock_auth_instance = MagicMock()
|
||||
mock_auth_instance.get_api_base.return_value = (
|
||||
"https://api.individual.githubcopilot.com"
|
||||
)
|
||||
mock_auth_instance.get_api_base.return_value = "https://api.individual.githubcopilot.com"
|
||||
mock_authenticator_class.return_value = mock_auth_instance
|
||||
|
||||
config = GithubCopilotResponsesAPIConfig()
|
||||
|
||||
# Test with default GitHub Copilot API base (from authenticator)
|
||||
url = config.get_complete_url(api_base=None, litellm_params={})
|
||||
assert (
|
||||
url == "https://api.individual.githubcopilot.com/responses"
|
||||
), f"Expected GitHub Copilot responses endpoint, got {url}"
|
||||
|
||||
# Test with custom api_base (overrides authenticator)
|
||||
custom_url = config.get_complete_url(
|
||||
api_base="https://custom.githubcopilot.com", litellm_params={}
|
||||
assert url == "https://api.individual.githubcopilot.com/responses", (
|
||||
f"Expected GitHub Copilot responses endpoint, got {url}"
|
||||
)
|
||||
assert (
|
||||
custom_url == "https://custom.githubcopilot.com/responses"
|
||||
), f"Expected custom endpoint, got {custom_url}"
|
||||
|
||||
# Test with trailing slash
|
||||
url_with_slash = config.get_complete_url(
|
||||
api_base="https://api.githubcopilot.com/", litellm_params={}
|
||||
# Caller-controlled bases must not receive the Copilot bearer token.
|
||||
custom_url = config.get_complete_url(api_base="https://custom.githubcopilot.com", litellm_params={})
|
||||
assert custom_url == "https://api.individual.githubcopilot.com/responses"
|
||||
|
||||
# The generic base injected by responses() must not override the base
|
||||
# bound to the selected account's token.
|
||||
url_with_slash = config.get_complete_url(api_base="https://api.githubcopilot.com/", litellm_params={})
|
||||
assert url_with_slash == "https://api.individual.githubcopilot.com/responses", (
|
||||
"Should prefer the authenticated account endpoint"
|
||||
)
|
||||
assert (
|
||||
url_with_slash == "https://api.githubcopilot.com/responses"
|
||||
), "Should handle trailing slash"
|
||||
|
||||
@patch("litellm.llms.github_copilot.responses.transformation.Authenticator")
|
||||
def test_validate_environment_default_headers(self, mock_authenticator_class):
|
||||
|
|
@ -98,9 +88,7 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
|
||||
config = GithubCopilotResponsesAPIConfig()
|
||||
|
||||
headers = config.validate_environment(
|
||||
headers={}, model="gpt-5.1-codex", litellm_params={}
|
||||
)
|
||||
headers = config.validate_environment(headers={}, model="gpt-5.1-codex", litellm_params={})
|
||||
|
||||
# Check required headers
|
||||
assert headers["Authorization"] == "Bearer test-api-key-123"
|
||||
|
|
@ -127,9 +115,7 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
"custom-header": "custom-value",
|
||||
}
|
||||
|
||||
headers = config.validate_environment(
|
||||
headers=custom_headers, model="gpt-5.1-codex", litellm_params={}
|
||||
)
|
||||
headers = config.validate_environment(headers=custom_headers, model="gpt-5.1-codex", litellm_params={})
|
||||
|
||||
# User header should override default
|
||||
assert headers["editor-version"] == "custom/2.0.0"
|
||||
|
|
@ -138,6 +124,30 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
# Default headers should still be present
|
||||
assert headers["Authorization"] == "Bearer test-api-key-123"
|
||||
|
||||
def test_responses_uses_per_deployment_token_directory(self, tmp_path):
|
||||
token_dir = tmp_path / "responses-account"
|
||||
token_dir.mkdir()
|
||||
(token_dir / "api-key.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"token": "responses-account-key",
|
||||
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
|
||||
"endpoints": {"api": "https://responses-account.example"},
|
||||
}
|
||||
)
|
||||
)
|
||||
params = GenericLiteLLMParams(github_copilot_token_dir=str(token_dir))
|
||||
config = GithubCopilotResponsesAPIConfig()
|
||||
|
||||
headers = config.validate_environment(headers={}, model="gpt-5.3-codex", litellm_params=params)
|
||||
url = config.get_complete_url(
|
||||
api_base="https://attacker.example",
|
||||
litellm_params=dict(params),
|
||||
)
|
||||
|
||||
assert headers["Authorization"] == "Bearer responses-account-key"
|
||||
assert url == "https://responses-account.example/responses"
|
||||
|
||||
def test_get_initiator_with_assistant_role(self):
|
||||
"""Test _get_initiator returns 'agent' for assistant role"""
|
||||
config = GithubCopilotResponsesAPIConfig()
|
||||
|
|
@ -182,9 +192,7 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
"""Test _has_vision_input detects input_image type"""
|
||||
config = GithubCopilotResponsesAPIConfig()
|
||||
|
||||
input_with_vision = [
|
||||
{"role": "user", "content": [{"type": "input_image", "data": "base64..."}]}
|
||||
]
|
||||
input_with_vision = [{"role": "user", "content": [{"type": "input_image", "data": "base64..."}]}]
|
||||
|
||||
has_vision = config._has_vision_input(input_with_vision)
|
||||
assert has_vision is True, "Should detect input_image type"
|
||||
|
|
@ -246,13 +254,11 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
}
|
||||
]
|
||||
|
||||
headers = config.validate_environment(
|
||||
headers={}, model="gpt-5.1-codex", litellm_params=mock_litellm_params
|
||||
)
|
||||
headers = config.validate_environment(headers={}, model="gpt-5.1-codex", litellm_params=mock_litellm_params)
|
||||
|
||||
assert (
|
||||
headers.get("copilot-vision-request") == "true"
|
||||
), "Should add copilot-vision-request header for vision input"
|
||||
assert headers.get("copilot-vision-request") == "true", (
|
||||
"Should add copilot-vision-request header for vision input"
|
||||
)
|
||||
|
||||
@patch("litellm.llms.github_copilot.responses.transformation.Authenticator")
|
||||
def test_validate_environment_with_x_initiator(self, mock_authenticator_class):
|
||||
|
|
@ -270,21 +276,15 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
{"role": "assistant", "content": "Hi"},
|
||||
]
|
||||
|
||||
headers = config.validate_environment(
|
||||
headers={}, model="gpt-5.1-codex", litellm_params=mock_litellm_params
|
||||
)
|
||||
headers = config.validate_environment(headers={}, model="gpt-5.1-codex", litellm_params=mock_litellm_params)
|
||||
|
||||
assert (
|
||||
headers.get("X-Initiator") == "agent"
|
||||
), "Should set X-Initiator to 'agent' for assistant role"
|
||||
assert headers.get("X-Initiator") == "agent", "Should set X-Initiator to 'agent' for assistant role"
|
||||
|
||||
def test_map_openai_params_no_transformation(self):
|
||||
"""Test that map_openai_params passes through parameters unchanged"""
|
||||
config = GithubCopilotResponsesAPIConfig()
|
||||
|
||||
params = ResponsesAPIOptionalRequestParams(
|
||||
temperature=0.7, max_output_tokens=1000, stream=False
|
||||
)
|
||||
params = ResponsesAPIOptionalRequestParams(temperature=0.7, max_output_tokens=1000, stream=False)
|
||||
|
||||
result = config.map_openai_params(
|
||||
response_api_optional_params=params,
|
||||
|
|
@ -338,9 +338,9 @@ class TestGithubCopilotResponsesAPITransformation:
|
|||
result = config._handle_reasoning_item(reasoning_item)
|
||||
|
||||
# encrypted_content should be preserved
|
||||
assert (
|
||||
result.get("encrypted_content") == "encrypted-blob-abc123"
|
||||
), "encrypted_content must be preserved for GitHub Copilot multi-turn conversations"
|
||||
assert result.get("encrypted_content") == "encrypted-blob-abc123", (
|
||||
"encrypted_content must be preserved for GitHub Copilot multi-turn conversations"
|
||||
)
|
||||
# status=None should be filtered out
|
||||
assert "status" not in result, "status=None should be filtered out"
|
||||
# content=None should be filtered out
|
||||
|
|
@ -393,9 +393,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
in the (already-merged) model info; otherwise returns None so the dispatcher
|
||||
routes through the chat-completions translation bridge."""
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_returns_config_when_mode_is_responses(self, mock_get_info):
|
||||
"""``mode=responses`` returns native config."""
|
||||
mock_get_info.return_value = {"mode": "responses"}
|
||||
|
|
@ -405,9 +403,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
)
|
||||
assert isinstance(config, GithubCopilotResponsesAPIConfig)
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_returns_none_when_mode_is_chat(self, mock_get_info):
|
||||
"""``mode=chat`` returns None so dispatcher uses bridge."""
|
||||
mock_get_info.return_value = {"mode": "chat"}
|
||||
|
|
@ -417,9 +413,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
)
|
||||
assert config is None
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_returns_none_when_mode_is_unset_and_no_endpoints(self, mock_get_info):
|
||||
"""Entry without ``mode`` and without ``supported_endpoints`` returns None
|
||||
(conservative default)."""
|
||||
|
|
@ -499,9 +493,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
)
|
||||
assert isinstance(config, GithubCopilotResponsesAPIConfig)
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_returns_none_when_get_model_info_raises(self, mock_get_info):
|
||||
"""Catalog lookup failure (model not registered) returns None
|
||||
(conservative default; bridge handles unknown models safely)."""
|
||||
|
|
@ -512,9 +504,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
)
|
||||
assert config is None
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_user_override_via_register_model(self, mock_get_info):
|
||||
"""User-supplied per-deployment ``model_info`` flows through
|
||||
``litellm.register_model`` (called by the router) into the merged
|
||||
|
|
@ -528,9 +518,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
)
|
||||
assert isinstance(config, GithubCopilotResponsesAPIConfig)
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_realistic_chat_only_entry_returns_none(self, mock_get_info):
|
||||
"""Realistic ``model_prices_and_context_window.json`` shape for a
|
||||
chat-only Copilot model (e.g. github_copilot/gemini-3.1-pro-preview)
|
||||
|
|
@ -554,9 +542,7 @@ class TestGithubCopilotResponsesAPIRouting:
|
|||
)
|
||||
assert config is None
|
||||
|
||||
@patch(
|
||||
"litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper"
|
||||
)
|
||||
@patch("litellm.llms.github_copilot.responses.transformation._cached_get_model_info_helper")
|
||||
def test_realistic_responses_only_entry_returns_config(self, mock_get_info):
|
||||
"""Realistic catalog entry for a Responses-only Copilot model
|
||||
(e.g. github_copilot/gpt-5.5) returns the native config."""
|
||||
|
|
@ -592,9 +578,7 @@ class TestGithubCopilotReasoningStreamItemIdNormalization:
|
|||
output_index group to the id from its output_item.added."""
|
||||
|
||||
def _config(self):
|
||||
with patch(
|
||||
"litellm.llms.github_copilot.responses.transformation.Authenticator"
|
||||
):
|
||||
with patch("litellm.llms.github_copilot.responses.transformation.Authenticator"):
|
||||
return GithubCopilotResponsesAPIConfig()
|
||||
|
||||
def _transform(self, config, chunk):
|
||||
|
|
|
|||
|
|
@ -1,16 +1,17 @@
|
|||
import json
|
||||
import os
|
||||
import time
|
||||
import stat
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.llms.github_copilot.authenticator import Authenticator
|
||||
from litellm.llms.github_copilot.authenticator import (
|
||||
Authenticator,
|
||||
get_authenticator_for_litellm_params,
|
||||
)
|
||||
from litellm.llms.github_copilot.common_utils import (
|
||||
APIKeyExpiredError,
|
||||
GetAccessTokenError,
|
||||
GetAPIKeyError,
|
||||
GetDeviceCodeError,
|
||||
RefreshAPIKeyError,
|
||||
)
|
||||
|
|
@ -18,14 +19,8 @@ from litellm.llms.github_copilot.common_utils import (
|
|||
|
||||
class TestGitHubCopilotAuthenticator:
|
||||
@pytest.fixture
|
||||
def authenticator(self):
|
||||
with (
|
||||
patch("os.path.exists", return_value=False),
|
||||
patch("os.makedirs") as mock_makedirs,
|
||||
):
|
||||
auth = Authenticator()
|
||||
mock_makedirs.assert_called_once()
|
||||
return auth
|
||||
def authenticator(self, tmp_path):
|
||||
return Authenticator(token_dir=str(tmp_path))
|
||||
|
||||
@pytest.fixture
|
||||
def mock_http_client(self):
|
||||
|
|
@ -36,26 +31,39 @@ class TestGitHubCopilotAuthenticator:
|
|||
mock_response.raise_for_status.return_value = None
|
||||
return mock_client, mock_response
|
||||
|
||||
def test_init(self):
|
||||
def test_init(self, tmp_path, monkeypatch):
|
||||
"""Test the initialization of the authenticator."""
|
||||
with (
|
||||
patch("os.path.exists", return_value=False),
|
||||
patch("os.makedirs") as mock_makedirs,
|
||||
):
|
||||
auth = Authenticator()
|
||||
assert auth.token_dir.endswith("/github_copilot")
|
||||
assert auth.access_token_file.endswith("/access-token")
|
||||
assert auth.api_key_file.endswith("/api-key.json")
|
||||
mock_makedirs.assert_called_once()
|
||||
env_token_dir = tmp_path / "env-account"
|
||||
explicit_token_dir = tmp_path / "explicit-account"
|
||||
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(env_token_dir))
|
||||
|
||||
def test_ensure_token_dir(self):
|
||||
with patch("os.makedirs") as mock_makedirs:
|
||||
auth = Authenticator(token_dir=str(explicit_token_dir))
|
||||
|
||||
assert auth.token_dir == str(explicit_token_dir)
|
||||
assert auth.access_token_file.endswith("/access-token")
|
||||
assert auth.api_key_file.endswith("/api-key.json")
|
||||
mock_makedirs.assert_not_called()
|
||||
|
||||
def test_ensure_token_dir(self, tmp_path):
|
||||
"""Test that the token directory is created if it doesn't exist."""
|
||||
with (
|
||||
patch("os.path.exists", return_value=False),
|
||||
patch("os.makedirs") as mock_makedirs,
|
||||
):
|
||||
auth = Authenticator()
|
||||
mock_makedirs.assert_called_once_with(auth.token_dir, exist_ok=True)
|
||||
token_dir = tmp_path / "nested" / "account"
|
||||
auth = Authenticator(token_dir=str(token_dir))
|
||||
|
||||
auth._ensure_token_dir()
|
||||
|
||||
assert token_dir.is_dir()
|
||||
assert stat.S_IMODE(token_dir.stat().st_mode) == 0o700
|
||||
|
||||
def test_private_write_is_atomic_and_owner_only(self, tmp_path):
|
||||
token_dir = tmp_path / "account"
|
||||
auth = Authenticator(token_dir=str(token_dir))
|
||||
|
||||
auth._write_private_text(auth.access_token_file, "access-token-value")
|
||||
|
||||
assert (token_dir / "access-token").read_text() == "access-token-value"
|
||||
assert stat.S_IMODE((token_dir / "access-token").stat().st_mode) == 0o600
|
||||
assert list(token_dir.glob(".litellm-*")) == []
|
||||
|
||||
def test_get_github_headers(self, authenticator):
|
||||
"""Test that GitHub headers are correctly generated."""
|
||||
|
|
@ -82,8 +90,7 @@ class TestGitHubCopilotAuthenticator:
|
|||
|
||||
with (
|
||||
patch.object(authenticator, "_login", return_value=mock_token),
|
||||
patch("builtins.open", mock_open()),
|
||||
patch("builtins.open", side_effect=IOError) as mock_read,
|
||||
patch("builtins.open", side_effect=IOError),
|
||||
):
|
||||
token = authenticator.get_access_token()
|
||||
assert token == mock_token
|
||||
|
|
@ -106,9 +113,7 @@ class TestGitHubCopilotAuthenticator:
|
|||
def test_get_api_key_from_file(self, authenticator):
|
||||
"""Test retrieving an API key from a file."""
|
||||
future_time = (datetime.now() + timedelta(hours=1)).timestamp()
|
||||
mock_api_key_data = json.dumps(
|
||||
{"token": "mock-api-key", "expires_at": future_time}
|
||||
)
|
||||
mock_api_key_data = json.dumps({"token": "mock-api-key", "expires_at": future_time})
|
||||
|
||||
with patch("builtins.open", mock_open(read_data=mock_api_key_data)):
|
||||
api_key = authenticator.get_api_key()
|
||||
|
|
@ -117,9 +122,7 @@ class TestGitHubCopilotAuthenticator:
|
|||
def test_get_api_key_expired(self, authenticator):
|
||||
"""Test refreshing an expired API key."""
|
||||
past_time = (datetime.now() - timedelta(hours=1)).timestamp()
|
||||
mock_expired_data = json.dumps(
|
||||
{"token": "expired-api-key", "expires_at": past_time}
|
||||
)
|
||||
mock_expired_data = json.dumps({"token": "expired-api-key", "expires_at": past_time})
|
||||
mock_new_data = {
|
||||
"token": "new-api-key",
|
||||
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
|
||||
|
|
@ -128,7 +131,6 @@ class TestGitHubCopilotAuthenticator:
|
|||
with (
|
||||
patch("builtins.open", mock_open(read_data=mock_expired_data)),
|
||||
patch.object(authenticator, "_refresh_api_key", return_value=mock_new_data),
|
||||
patch("json.dump") as mock_json_dump,
|
||||
):
|
||||
api_key = authenticator.get_api_key()
|
||||
assert api_key == "new-api-key"
|
||||
|
|
@ -217,20 +219,14 @@ class TestGitHubCopilotAuthenticator:
|
|||
mock_token = "mock-access-token"
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
authenticator, "_get_device_code", return_value=mock_device_code_data
|
||||
),
|
||||
patch.object(
|
||||
authenticator, "_poll_for_access_token", return_value=mock_token
|
||||
),
|
||||
patch.object(authenticator, "_get_device_code", return_value=mock_device_code_data),
|
||||
patch.object(authenticator, "_poll_for_access_token", return_value=mock_token),
|
||||
patch("builtins.print") as mock_print,
|
||||
):
|
||||
result = authenticator._login()
|
||||
assert result == mock_token
|
||||
authenticator._get_device_code.assert_called_once()
|
||||
authenticator._poll_for_access_token.assert_called_once_with(
|
||||
"mock-device-code"
|
||||
)
|
||||
authenticator._poll_for_access_token.assert_called_once_with("mock-device-code")
|
||||
mock_print.assert_called_once()
|
||||
|
||||
def test_get_api_base_from_file(self, authenticator):
|
||||
|
|
@ -246,6 +242,45 @@ class TestGitHubCopilotAuthenticator:
|
|||
api_base = authenticator.get_api_base()
|
||||
assert api_base == "https://api.enterprise.githubcopilot.com"
|
||||
|
||||
def test_get_api_base_without_global_token_file(self, authenticator):
|
||||
"""Per-deployment auth does not require a global endpoint file."""
|
||||
assert authenticator.get_api_base() is None
|
||||
|
||||
def test_get_api_base_with_invalid_json(self, authenticator):
|
||||
authenticator._write_private_text(authenticator.api_key_file, "not-json")
|
||||
|
||||
assert authenticator.get_api_base() is None
|
||||
|
||||
def test_authenticator_selection_reuses_default_without_override(self, authenticator):
|
||||
assert get_authenticator_for_litellm_params(authenticator, None) is authenticator
|
||||
|
||||
def test_authenticator_selection_reuses_matching_directory(self, authenticator):
|
||||
selected = get_authenticator_for_litellm_params(
|
||||
authenticator,
|
||||
{"github_copilot_token_dir": authenticator.token_dir},
|
||||
)
|
||||
|
||||
assert selected is authenticator
|
||||
|
||||
def test_authenticator_selection_ignores_empty_override(self, authenticator):
|
||||
selected = get_authenticator_for_litellm_params(
|
||||
authenticator,
|
||||
{"github_copilot_token_dir": " "},
|
||||
)
|
||||
|
||||
assert selected is authenticator
|
||||
|
||||
def test_authenticator_selection_uses_distinct_directory(self, authenticator, tmp_path):
|
||||
token_dir = tmp_path / "second-account"
|
||||
|
||||
selected = get_authenticator_for_litellm_params(
|
||||
authenticator,
|
||||
{"github_copilot_token_dir": str(token_dir)},
|
||||
)
|
||||
|
||||
assert selected is not authenticator
|
||||
assert selected.token_dir == str(token_dir)
|
||||
|
||||
def test_get_device_code_with_custom_url(self, authenticator, mock_http_client):
|
||||
"""GITHUB_COPILOT_DEVICE_CODE_URL env var must be used by _get_device_code at call time."""
|
||||
mock_client, mock_response = mock_http_client
|
||||
|
|
@ -255,8 +290,10 @@ class TestGitHubCopilotAuthenticator:
|
|||
"user_code": "UC",
|
||||
"verification_uri": "https://example.com",
|
||||
}
|
||||
with patch.dict(os.environ, {"GITHUB_COPILOT_DEVICE_CODE_URL": custom_url}), \
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client):
|
||||
with (
|
||||
patch.dict(os.environ, {"GITHUB_COPILOT_DEVICE_CODE_URL": custom_url}),
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client),
|
||||
):
|
||||
authenticator._get_device_code()
|
||||
assert mock_client.post.call_args[0][0] == custom_url
|
||||
|
||||
|
|
@ -269,8 +306,10 @@ class TestGitHubCopilotAuthenticator:
|
|||
"user_code": "UC",
|
||||
"verification_uri": "https://example.com",
|
||||
}
|
||||
with patch.dict(os.environ, {"GITHUB_COPILOT_CLIENT_ID": custom_id}), \
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client):
|
||||
with (
|
||||
patch.dict(os.environ, {"GITHUB_COPILOT_CLIENT_ID": custom_id}),
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client),
|
||||
):
|
||||
authenticator._get_device_code()
|
||||
assert mock_client.post.call_args[1]["json"]["client_id"] == custom_id
|
||||
|
||||
|
|
@ -279,9 +318,11 @@ class TestGitHubCopilotAuthenticator:
|
|||
mock_client, mock_response = mock_http_client
|
||||
custom_url = "https://custom.example.com/token"
|
||||
mock_response.json.return_value = {"access_token": "tok"}
|
||||
with patch.dict(os.environ, {"GITHUB_COPILOT_ACCESS_TOKEN_URL": custom_url}), \
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client), \
|
||||
patch("time.sleep"):
|
||||
with (
|
||||
patch.dict(os.environ, {"GITHUB_COPILOT_ACCESS_TOKEN_URL": custom_url}),
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client),
|
||||
patch("time.sleep"),
|
||||
):
|
||||
authenticator._poll_for_access_token("dc")
|
||||
assert mock_client.post.call_args[0][0] == custom_url
|
||||
|
||||
|
|
@ -290,9 +331,11 @@ class TestGitHubCopilotAuthenticator:
|
|||
mock_client, mock_response = mock_http_client
|
||||
custom_id = "custom_client_id"
|
||||
mock_response.json.return_value = {"access_token": "tok"}
|
||||
with patch.dict(os.environ, {"GITHUB_COPILOT_CLIENT_ID": custom_id}), \
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client), \
|
||||
patch("time.sleep"):
|
||||
with (
|
||||
patch.dict(os.environ, {"GITHUB_COPILOT_CLIENT_ID": custom_id}),
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client),
|
||||
patch("time.sleep"),
|
||||
):
|
||||
authenticator._poll_for_access_token("dc")
|
||||
assert mock_client.post.call_args[1]["json"]["client_id"] == custom_id
|
||||
|
||||
|
|
@ -301,9 +344,10 @@ class TestGitHubCopilotAuthenticator:
|
|||
mock_client, mock_response = mock_http_client
|
||||
custom_url = "https://custom.example.com/api-key"
|
||||
mock_response.json.return_value = {"token": "api-tok", "expires_at": 9999999999}
|
||||
with patch.dict(os.environ, {"GITHUB_COPILOT_API_KEY_URL": custom_url}), \
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client), \
|
||||
patch.object(authenticator, "get_access_token", return_value="access-tok"):
|
||||
with (
|
||||
patch.dict(os.environ, {"GITHUB_COPILOT_API_KEY_URL": custom_url}),
|
||||
patch("litellm.llms.github_copilot.authenticator._get_httpx_client", return_value=mock_client),
|
||||
patch.object(authenticator, "get_access_token", return_value="access-tok"),
|
||||
):
|
||||
authenticator._refresh_api_key()
|
||||
assert mock_client.get.call_args[0][0] == custom_url
|
||||
|
||||
|
|
|
|||
|
|
@ -1,30 +1,17 @@
|
|||
import asyncio
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from typing import AsyncGenerator
|
||||
from unittest.mock import AsyncMock, MagicMock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
from respx import MockRouter
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
import litellm
|
||||
|
||||
# Import at the top to make the patch work correctly
|
||||
import litellm.llms.github_copilot.chat.transformation
|
||||
from litellm import Choices, Message, ModelResponse, Usage, acompletion, completion
|
||||
from litellm.exceptions import AuthenticationError
|
||||
from litellm.llms.github_copilot.authenticator import Authenticator
|
||||
from litellm import ModelResponse, completion
|
||||
from litellm.llms.github_copilot.chat.transformation import GithubCopilotConfig
|
||||
from litellm.llms.github_copilot.common_utils import (
|
||||
APIKeyExpiredError,
|
||||
GetAccessTokenError,
|
||||
GetAPIKeyError,
|
||||
GetDeviceCodeError,
|
||||
RefreshAPIKeyError,
|
||||
)
|
||||
|
||||
|
||||
def test_github_copilot_config_get_openai_compatible_provider_info():
|
||||
|
|
@ -32,14 +19,8 @@ def test_github_copilot_config_get_openai_compatible_provider_info():
|
|||
|
||||
config = GithubCopilotConfig()
|
||||
|
||||
# Mock the authenticator to avoid actual API calls
|
||||
mock_api_key = "gh.test-key-123456789"
|
||||
config.authenticator = MagicMock()
|
||||
config.authenticator.get_api_key.return_value = mock_api_key
|
||||
# Test with dynamic endpoint
|
||||
config.authenticator.get_api_base.return_value = (
|
||||
"https://api.enterprise.githubcopilot.com"
|
||||
)
|
||||
config.authenticator.get_api_base.return_value = "https://api.enterprise.githubcopilot.com"
|
||||
|
||||
# Test with default values
|
||||
model = "github_copilot/gpt-4"
|
||||
|
|
@ -55,8 +36,9 @@ def test_github_copilot_config_get_openai_compatible_provider_info():
|
|||
)
|
||||
|
||||
assert api_base == "https://api.enterprise.githubcopilot.com"
|
||||
assert dynamic_api_key == mock_api_key
|
||||
assert dynamic_api_key is None
|
||||
assert custom_llm_provider == "github_copilot"
|
||||
config.authenticator.get_api_key.assert_not_called()
|
||||
|
||||
# Test fallback to default if no dynamic endpoint
|
||||
config.authenticator.get_api_base.return_value = None
|
||||
|
|
@ -71,22 +53,82 @@ def test_github_copilot_config_get_openai_compatible_provider_info():
|
|||
custom_llm_provider="github_copilot",
|
||||
)
|
||||
assert api_base == "https://api.githubcopilot.com"
|
||||
assert dynamic_api_key is None
|
||||
|
||||
# Test with authentication failure
|
||||
config.authenticator.get_api_key.side_effect = GetAPIKeyError(
|
||||
message="Failed to get API key",
|
||||
status_code=401,
|
||||
|
||||
def test_github_copilot_config_resolves_per_deployment_token_directory(tmp_path):
|
||||
account_a = tmp_path / "account-a"
|
||||
account_b = tmp_path / "account-b"
|
||||
account_a.mkdir()
|
||||
account_b.mkdir()
|
||||
expires_at = (datetime.now() + timedelta(hours=1)).timestamp()
|
||||
(account_a / "api-key.json").write_text(
|
||||
json.dumps({"token": "token-a", "expires_at": expires_at, "endpoints": {"api": "https://api-a.example"}})
|
||||
)
|
||||
(account_b / "api-key.json").write_text(
|
||||
json.dumps({"token": "token-b", "expires_at": expires_at, "endpoints": {"api": "https://api-b.example"}})
|
||||
)
|
||||
|
||||
with pytest.raises(AuthenticationError) as excinfo:
|
||||
config._get_openai_compatible_provider_info(
|
||||
model=model,
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
custom_llm_provider="github_copilot",
|
||||
)
|
||||
config = GithubCopilotConfig()
|
||||
resolved_a = config._get_openai_compatible_provider_info(
|
||||
model="github_copilot/gpt-4",
|
||||
api_base="https://attacker.example",
|
||||
api_key=None,
|
||||
custom_llm_provider="github_copilot",
|
||||
litellm_params={"github_copilot_token_dir": str(account_a)},
|
||||
)
|
||||
resolved_b = config._get_openai_compatible_provider_info(
|
||||
model="github_copilot/gpt-4",
|
||||
api_base="https://attacker.example",
|
||||
api_key=None,
|
||||
custom_llm_provider="github_copilot",
|
||||
litellm_params={"github_copilot_token_dir": str(account_b)},
|
||||
)
|
||||
|
||||
assert "Failed to get API key" in str(excinfo.value)
|
||||
assert resolved_a == ("https://api-a.example", "token-a", "github_copilot")
|
||||
assert resolved_b == ("https://api-b.example", "token-b", "github_copilot")
|
||||
|
||||
|
||||
def test_router_registers_multiple_copilot_accounts_without_authenticating(tmp_path, monkeypatch):
|
||||
global_token_dir = tmp_path / "global"
|
||||
account_a = tmp_path / "account-a"
|
||||
account_b = tmp_path / "account-b"
|
||||
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(global_token_dir))
|
||||
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "copilot-pool",
|
||||
"litellm_params": {
|
||||
"model": "github_copilot/gpt-4",
|
||||
"github_copilot_token_dir": str(account_a),
|
||||
},
|
||||
},
|
||||
{
|
||||
"model_name": "copilot-pool",
|
||||
"litellm_params": {
|
||||
"model": "github_copilot/gpt-4",
|
||||
"github_copilot_token_dir": str(account_b),
|
||||
},
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
deployments = router.get_model_list(model_name="copilot-pool")
|
||||
assert deployments is not None
|
||||
assert len(deployments) == 2
|
||||
assert len({deployment["model_info"]["id"] for deployment in deployments}) == 2
|
||||
assert {deployment["litellm_params"]["github_copilot_token_dir"] for deployment in deployments} == {
|
||||
str(account_a),
|
||||
str(account_b),
|
||||
}
|
||||
assert not global_token_dir.exists()
|
||||
assert not account_a.exists()
|
||||
assert not account_b.exists()
|
||||
|
||||
request_kwargs = {"github_copilot_token_dir": str(tmp_path / "request-controlled")}
|
||||
router._update_kwargs_with_deployment(deployment=deployments[0], kwargs=request_kwargs)
|
||||
assert "github_copilot_token_dir" not in request_kwargs
|
||||
|
||||
|
||||
@patch("litellm.llms.github_copilot.authenticator.Authenticator.get_api_key")
|
||||
|
|
@ -143,6 +185,51 @@ def test_completion_github_copilot_mock_response(
|
|||
assert kwargs.get("messages") == messages
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_completion_github_copilot_uses_selected_account_directory(tmp_path, monkeypatch):
|
||||
monkeypatch.delenv("EXPERIMENTAL_OPENAI_BASE_LLM_HTTP_HANDLER", raising=False)
|
||||
token_dir = tmp_path / "selected-account"
|
||||
token_dir.mkdir()
|
||||
(token_dir / "api-key.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"token": "selected-account-key",
|
||||
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
|
||||
"endpoints": {"api": "https://selected-account.example"},
|
||||
}
|
||||
)
|
||||
)
|
||||
route = respx.post("https://selected-account.example/chat/completions").mock(
|
||||
return_value=httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"id": "chatcmpl-selected-account",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4",
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"message": {"role": "assistant", "content": "selected-account-ok"},
|
||||
"finish_reason": "stop",
|
||||
}
|
||||
],
|
||||
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
response = completion(
|
||||
model="github_copilot/gpt-4",
|
||||
messages=[{"role": "user", "content": "Hello"}],
|
||||
github_copilot_token_dir=str(token_dir),
|
||||
)
|
||||
|
||||
assert route.called
|
||||
assert route.calls.last.request.headers["authorization"] == "Bearer selected-account-key"
|
||||
assert response.choices[0].message.content == "selected-account-ok"
|
||||
|
||||
|
||||
def test_transform_messages_disable_copilot_system_to_assistant(monkeypatch):
|
||||
"""Test that system messages are converted to assistant unless disable_copilot_system_to_assistant is True."""
|
||||
import litellm
|
||||
|
|
@ -158,25 +245,19 @@ def test_transform_messages_disable_copilot_system_to_assistant(monkeypatch):
|
|||
{"role": "system", "content": "System message."},
|
||||
{"role": "user", "content": "User message."},
|
||||
]
|
||||
out = config._transform_messages(
|
||||
[m.copy() for m in messages], model="github_copilot/gpt-4"
|
||||
)
|
||||
out = config._transform_messages([m.copy() for m in messages], model="github_copilot/gpt-4")
|
||||
assert out[0]["role"] == "assistant"
|
||||
assert out[1]["role"] == "user"
|
||||
|
||||
# Case 2: Flag is True (conversion does not happen)
|
||||
litellm.disable_copilot_system_to_assistant = True
|
||||
out = config._transform_messages(
|
||||
[m.copy() for m in messages], model="github_copilot/gpt-4"
|
||||
)
|
||||
out = config._transform_messages([m.copy() for m in messages], model="github_copilot/gpt-4")
|
||||
assert out[0]["role"] == "system"
|
||||
assert out[1]["role"] == "user"
|
||||
|
||||
# Case 3: Flag is False again (conversion happens)
|
||||
litellm.disable_copilot_system_to_assistant = False
|
||||
out = config._transform_messages(
|
||||
[m.copy() for m in messages], model="github_copilot/gpt-4"
|
||||
)
|
||||
out = config._transform_messages([m.copy() for m in messages], model="github_copilot/gpt-4")
|
||||
assert out[0]["role"] == "assistant"
|
||||
assert out[1]["role"] == "user"
|
||||
finally:
|
||||
|
|
@ -381,9 +462,7 @@ def test_get_supported_openai_params_claude_model():
|
|||
assert "reasoning_effort" in supported_params
|
||||
|
||||
# Test Claude 3-7 model supports thinking and reasoning_effort parameters
|
||||
supported_params_claude37 = config.get_supported_openai_params(
|
||||
"claude-3-7-sonnet-20250219"
|
||||
)
|
||||
supported_params_claude37 = config.get_supported_openai_params("claude-3-7-sonnet-20250219")
|
||||
assert "thinking" in supported_params_claude37
|
||||
assert "reasoning_effort" in supported_params_claude37
|
||||
|
||||
|
|
@ -410,16 +489,12 @@ def test_get_supported_openai_params_case_insensitive():
|
|||
config = GithubCopilotConfig()
|
||||
|
||||
# Test uppercase Claude 4 model with full model name
|
||||
supported_params_upper = config.get_supported_openai_params(
|
||||
"CLAUDE-SONNET-4-20250514"
|
||||
)
|
||||
supported_params_upper = config.get_supported_openai_params("CLAUDE-SONNET-4-20250514")
|
||||
assert "thinking" in supported_params_upper
|
||||
assert "reasoning_effort" in supported_params_upper
|
||||
|
||||
# Test mixed case Claude 3-7 model (has extended thinking) with full model name
|
||||
supported_params_mixed = config.get_supported_openai_params(
|
||||
"Claude-3-7-Sonnet-20250219"
|
||||
)
|
||||
supported_params_mixed = config.get_supported_openai_params("Claude-3-7-Sonnet-20250219")
|
||||
assert "thinking" in supported_params_mixed
|
||||
assert "reasoning_effort" in supported_params_mixed
|
||||
|
||||
|
|
@ -753,13 +828,8 @@ class TestGithubCopilotTransformResponse:
|
|||
assert result.choices[0].message.tool_calls is not None
|
||||
assert len(result.choices[0].message.tool_calls) == 1
|
||||
assert result.choices[0].message.tool_calls[0]["id"] == "toolu_01ABC"
|
||||
assert (
|
||||
result.choices[0].message.tool_calls[0]["function"]["name"] == "get_weather"
|
||||
)
|
||||
assert (
|
||||
'"Boston, MA"'
|
||||
in result.choices[0].message.tool_calls[0]["function"]["arguments"]
|
||||
)
|
||||
assert result.choices[0].message.tool_calls[0]["function"]["name"] == "get_weather"
|
||||
assert '"Boston, MA"' in result.choices[0].message.tool_calls[0]["function"]["arguments"]
|
||||
|
||||
def test_transform_response_anthropic_native_multiple_text_blocks(self):
|
||||
"""All text blocks must be concatenated, not only the first."""
|
||||
|
|
@ -927,12 +997,8 @@ class TestGithubCopilotTransformParsedResponseDict:
|
|||
|
||||
|
||||
@patch("litellm.llms.openai.openai.OpenAIChatCompletion._get_openai_client")
|
||||
@patch(
|
||||
"litellm.llms.openai.openai.OpenAIChatCompletion.make_sync_openai_chat_completion_request"
|
||||
)
|
||||
def test_openai_handler_repairs_github_copilot_empty_choices(
|
||||
mock_request, mock_get_client
|
||||
):
|
||||
@patch("litellm.llms.openai.openai.OpenAIChatCompletion.make_sync_openai_chat_completion_request")
|
||||
def test_openai_handler_repairs_github_copilot_empty_choices(mock_request, mock_get_client):
|
||||
"""
|
||||
The OpenAI SDK handler calls convert_to_model_response_object directly on the
|
||||
SDK's parsed output, bypassing transform_response. convert raises APIError on
|
||||
|
|
|
|||
|
|
@ -182,6 +182,40 @@ class TestModelManagementAuthChecks:
|
|||
)
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_team_admin_cannot_add_github_copilot_token_directory(self):
|
||||
model_params = Deployment(
|
||||
model_name="copilot-pool",
|
||||
litellm_params=LiteLLM_Params(
|
||||
model="github_copilot/gpt-5.3-codex",
|
||||
github_copilot_token_dir="/server/copilot/account-a",
|
||||
),
|
||||
model_info={"team_id": "test_team"},
|
||||
)
|
||||
|
||||
with pytest.raises(Exception, match="Only proxy admins can configure GitHub Copilot token directories"):
|
||||
await ModelManagementAuthChecks.can_user_make_model_call(
|
||||
model_params=model_params,
|
||||
user_api_key_dict=self.team_admin_user,
|
||||
prisma_client=MockPrismaClient(team_exists=True),
|
||||
premium_user=True,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_team_admin_cannot_patch_github_copilot_token_directory(self):
|
||||
model_params = updateDeployment(
|
||||
litellm_params={"github_copilot_token_dir": "/server/copilot/account-a"},
|
||||
model_info={"team_id": "test_team"},
|
||||
)
|
||||
|
||||
with pytest.raises(Exception, match="Only proxy admins can configure GitHub Copilot token directories"):
|
||||
await ModelManagementAuthChecks.allow_team_model_action(
|
||||
model_params=model_params,
|
||||
user_api_key_dict=self.team_admin_user,
|
||||
prisma_client=MockPrismaClient(team_exists=True),
|
||||
premium_user=True,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allow_team_model_action_non_premium_fails(self):
|
||||
"""Test team model action fails for non-premium users"""
|
||||
|
|
@ -228,7 +262,8 @@ class TestModelManagementAuthChecks:
|
|||
model_params = Deployment(
|
||||
model_name="test_model",
|
||||
litellm_params=LiteLLM_Params(
|
||||
model="test_model",
|
||||
model="github_copilot/gpt-5.3-codex",
|
||||
github_copilot_token_dir="/server/copilot/account-a",
|
||||
),
|
||||
model_info={"team_id": "test_team"},
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue