alvaro
abd03ed07d
Rename the project directories, Python distribution and import package, and npm package. Update launcher, Docker, tooling, tests, lockfiles, and documentation while preserving existing API contracts. Make AGENTS.md self-contained and remove external skill and workflow references.
98 lines
3.9 KiB
Python
98 lines
3.9 KiB
Python
"""Validate owned fields without tightening or rewriting upstream payloads."""
|
|
|
|
import pytest
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import ValidationError
|
|
|
|
from vocero_api.api.responses import validated_json_response
|
|
from vocero_api.models.requests import ApipilotExecuteRequest, QueueEndPayload, SearchRequest
|
|
from vocero_api.models.responses import (
|
|
AllocationResponse,
|
|
QueueCancellationResponse,
|
|
ReasoningResponse,
|
|
ReportsResponse,
|
|
SearchResponse,
|
|
ToolExecutionResponse,
|
|
UserProfileResponse,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("model", "payload"),
|
|
[
|
|
(AllocationResponse, {}),
|
|
(AllocationResponse, {"state": [1, None], "connectUrl": "wss://host", "future": {"x": 1.0}}),
|
|
(ToolExecutionResponse, {"success": True, "unknown": {"nested": [None, 1.0, False]}}),
|
|
(ToolExecutionResponse, {"success": False, "data": 42, "error": "loose error", "future": []}),
|
|
(ReasoningResponse, {"answer": "á", "usage_pass2": [1, None], "future": True}),
|
|
(SearchResponse, {"query": "q", "answer": {"x": 2}, "results": [{"title": [], "snippet": 7, "url": None}]}),
|
|
(
|
|
ReportsResponse,
|
|
{"count": 1, "wrappedLen": 2, "hints": [], "entries": [{"sourceScheduleLabel": 42, "date": None}]},
|
|
),
|
|
*[(UserProfileResponse, payload) for payload in [{}, [], 42, False, None, "profile"]],
|
|
],
|
|
)
|
|
def test_upstream_values_and_missing_fields_keep_exact_wire_bytes(model, payload):
|
|
original = JSONResponse(payload, status_code=202)
|
|
validated = validated_json_response(payload, model, status_code=202)
|
|
assert validated.body == original.body
|
|
assert validated.raw_headers == original.raw_headers
|
|
assert validated.status_code == original.status_code
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [{}, {"ok": "true"}, {"ok": 1}, {"ok": None}])
|
|
def test_owned_response_fields_are_actually_validated(payload):
|
|
with pytest.raises(ValidationError):
|
|
validated_json_response(payload, QueueCancellationResponse)
|
|
|
|
|
|
def test_request_models_keep_legacy_extra_field_and_argument_behavior():
|
|
assert SearchRequest.model_validate({"query": "q", "future": True}).model_dump() == {"query": "q", "key": None}
|
|
arguments = {"arbitrary": {"nested": [1, None]}, "unknown": True}
|
|
assert ApipilotExecuteRequest.model_validate({"name": "tool", "arguments": arguments}).arguments == arguments
|
|
for value in [None, "id", 42, False, [], {}, {"future": True}]:
|
|
assert QueueEndPayload.model_validate({"queueId": value, "future": 1}).queueId == value
|
|
assert QueueEndPayload.model_validate({}).queueId == ""
|
|
|
|
|
|
def test_http_routes_publish_response_schemas():
|
|
# Import the app only after the compatibility suite's isolated configuration.
|
|
from vocero_api.main import app
|
|
|
|
spec = app.openapi()
|
|
for path, operations in spec["paths"].items():
|
|
if path.startswith("/api/"):
|
|
for operation in operations.values():
|
|
schema = operation["responses"]["200"]["content"]["application/json"]["schema"]
|
|
assert "$ref" in schema, path
|
|
allocation = spec["components"]["schemas"]["AllocationResponse"]
|
|
assert not allocation.get("required")
|
|
assert allocation["additionalProperties"] is True
|
|
config = spec["components"]["schemas"]["ConfigResponse"]
|
|
assert config["properties"]["reasoningType"]["enum"] == ["PROXY", "OPENAI"]
|
|
|
|
|
|
def test_legacy_entry_point_exports_the_renamed_application():
|
|
import server
|
|
from vocero_api.main import app
|
|
|
|
assert server.app is app
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "expected"),
|
|
[
|
|
("OPENAI", "OPENAI"),
|
|
(" 'openai' ", "OPENAI"),
|
|
('"OPENAI"', "OPENAI"),
|
|
("PROXY", "PROXY"),
|
|
("", "PROXY"),
|
|
("UNSUPPORTED", "PROXY"),
|
|
],
|
|
)
|
|
def test_reasoning_configuration_preserves_normalization_and_native_fallback(value, expected):
|
|
from vocero_api.core.config import _normalize_reasoning_type
|
|
|
|
assert _normalize_reasoning_type(value) == expected
|