mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-28 06:23:07 +00:00
34319dff7d
Transcription reaches every provider as on main again, sends stream only when asked, forwards language and languages together, keeps form key order, bills provider seconds only when above zero, honors base_model for the new flags, keeps main's Azure api_version on the proxy path, and streams carry chat's headers, an error event on a mid-stream failure, and a spend row when the stream ends early. Translation cost reads both per-second rates from one source and spend rows carry the billed seconds. Realtime transcription WebSockets keep model in the upstream URL outside OpenAI's hosts, client secrets forward explicit nulls, accept conversation and delay, and answer 400 on a malformed body, the /openai translation paths are back to main, translation refuses only for guardrails realtime sessions run, nested aliases resolve and later session.update models are authorized, translation client secrets run per-model auth, beta clients bridged to GA get beta-shaped session events, and transcription sessions send the deployment model in both shapes.
72 lines
3.2 KiB
Python
72 lines
3.2 KiB
Python
"""Tests for scripts/check_openapi_docs.py, the scan behind scripts/openapi_docs_gate.py."""
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Final
|
|
|
|
import pytest
|
|
|
|
import check_openapi_docs as checker
|
|
|
|
_BODY: Final = {"content": {"application/json": {"schema": {"type": "object"}}}}
|
|
_RESPONSES: Final = {"200": {"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Out"}}}}}
|
|
_DOCUMENTED: Final = {"description": "Does a thing", "requestBody": _BODY, "responses": _RESPONSES}
|
|
|
|
|
|
def _post_with_media_types(request_media_type: str, response_media_type: str) -> Mapping[str, object]:
|
|
return {
|
|
**_DOCUMENTED,
|
|
"requestBody": {"content": {request_media_type: {"schema": {"type": "string"}}}},
|
|
"responses": {"200": {"content": {response_media_type: {"schema": {"type": "string"}}}}},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("media_type", ("application/json", "application/sdp"))
|
|
def test_a_fully_documented_post_has_no_findings(media_type: str) -> None:
|
|
operation: Final = _post_with_media_types(media_type, media_type)
|
|
assert checker.violations({"paths": {"/v1/thing": {"post": operation}}}) == ()
|
|
|
|
|
|
def test_a_schema_under_another_media_type_still_counts_as_undocumented() -> None:
|
|
operation: Final = _post_with_media_types("multipart/form-data", "audio/wav")
|
|
assert {v.code for v in checker.violations({"paths": {"/v1/thing": {"post": operation}}})} == {"OA001", "OA003"}
|
|
|
|
|
|
def test_each_missing_piece_is_its_own_rule() -> None:
|
|
bare: Final = {"summary": "Thing", "responses": {"200": {"description": "Successful Response"}}}
|
|
codes: Final = {v.code for v in checker.violations({"paths": {"/v1/thing": {"post": bare}}})}
|
|
assert codes == {"OA001", "OA002", "OA003"}
|
|
|
|
|
|
def test_only_the_missing_piece_is_reported() -> None:
|
|
no_body: Final = {**_DOCUMENTED, "requestBody": {}}
|
|
blank_description: Final = {**_DOCUMENTED, "description": " "}
|
|
untyped_200: Final = {**_DOCUMENTED, "responses": {"200": {"content": {"application/json": {"schema": {}}}}}}
|
|
spec: Final = {
|
|
"paths": {
|
|
"/a": {"post": no_body},
|
|
"/b": {"post": blank_description},
|
|
"/c": {"post": untyped_200},
|
|
}
|
|
}
|
|
assert checker.violations(spec) == (
|
|
checker.Violation("POST", "/a", "OA001"),
|
|
checker.Violation("POST", "/b", "OA002"),
|
|
checker.Violation("POST", "/c", "OA003"),
|
|
)
|
|
|
|
|
|
def test_non_post_operations_are_not_gated() -> None:
|
|
spec: Final = {"paths": {"/v1/thing": {"get": {}, "delete": {}, "put": {}, "post": _DOCUMENTED}}}
|
|
assert checker.violations(spec) == ()
|
|
|
|
|
|
def test_findings_are_sorted_by_path_so_counts_are_stable() -> None:
|
|
spec: Final = {"paths": {"/z": {"post": {}}, "/a": {"post": {}}}}
|
|
assert [v.path for v in checker.violations(spec)] == ["/a", "/a", "/a", "/z", "/z", "/z"]
|
|
|
|
|
|
def test_main_reads_a_spec_file_and_prints_one_line_per_finding(tmp_path, capsys) -> None:
|
|
spec_file: Final = tmp_path / "openapi.json"
|
|
spec_file.write_text('{"paths": {"/v1/thing": {"post": {"description": "x", "requestBody": {}}}}}')
|
|
assert checker.main([str(spec_file)]) == 0
|
|
assert capsys.readouterr().out.splitlines() == ["POST /v1/thing OA001", "POST /v1/thing OA003"]
|