93 lines
3.7 KiB
Python
93 lines
3.7 KiB
Python
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import SecretStr
|
|
|
|
from ai_platform.bootstrap.app import create_app
|
|
from ai_platform.infrastructure.config.settings import Settings
|
|
from ai_platform.infrastructure.persistence.models import AIProfile
|
|
|
|
FIXTURE = Path(__file__).parent / "fixtures" / "mock_omni_run.json"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("api_key", "selected_url"),
|
|
[
|
|
(None, "http://127.0.0.1:8317"),
|
|
("test-local-proxy-key", "http://127.0.0.1:8317"),
|
|
("test-local-proxy-key", "http://127.0.0.1:8318"),
|
|
],
|
|
)
|
|
def test_local_provider_ignores_request_vkey_and_runs_hybrid(
|
|
aihub_test_environment: dict,
|
|
wait_for_terminal,
|
|
api_key: str | None,
|
|
selected_url: str,
|
|
):
|
|
requests: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
assert request.url == f"{selected_url}/v1/chat/completions"
|
|
assert json.loads(request.content)["model"] == "gemini-3.5-flash-lite"
|
|
assert "caller-vkey-must-not-reach-local-provider" not in request.content.decode()
|
|
if api_key and selected_url == "http://127.0.0.1:8317":
|
|
assert request.headers["authorization"] == f"Bearer {api_key}"
|
|
else:
|
|
assert "authorization" not in request.headers
|
|
content = (
|
|
'{"items":[{"signal":"ý định","detail":"Đang tìm hiểu","source":"message-1"}]}'
|
|
if len(requests) == 1
|
|
else '{"score":80,"confidence":0.91,"reasoning":"Quan tâm rõ ràng."}'
|
|
)
|
|
return httpx.Response(200, json={"choices": [{"message": {"content": content}}]})
|
|
|
|
loaded = Settings.load()
|
|
settings = Settings(
|
|
database_url=loaded.database_url,
|
|
environment="development",
|
|
tenant_provisioning_secret=loaded.tenant_provisioning_secret,
|
|
openai_base_url="http://127.0.0.1:8317",
|
|
openai_api_key=SecretStr(api_key) if api_key else None,
|
|
)
|
|
|
|
async def select_local_profile():
|
|
try:
|
|
async with aihub_test_environment["factory"]() as db:
|
|
profile = await db.get(AIProfile, aihub_test_environment["profile_version"])
|
|
assert profile is not None
|
|
profile.provider = "openai_compatible"
|
|
profile.base_url = selected_url
|
|
profile.model = "gemini-3.5-flash-lite"
|
|
await db.commit()
|
|
finally:
|
|
await aihub_test_environment["factory"].kw["bind"].dispose()
|
|
|
|
asyncio.run(select_local_profile())
|
|
payload = json.loads(FIXTURE.read_text())
|
|
payload["request_id"] = str(uuid4())
|
|
payload["ai_profile"] = "medium"
|
|
payload["scoring_config"].update(rule_weight=0.5, ai_weight=0.5)
|
|
headers = {
|
|
"Authorization": f"Bearer {aihub_test_environment['tenant_key']}",
|
|
"X-Tenant-VKey": "caller-vkey-must-not-reach-local-provider",
|
|
"Idempotency-Key": payload["request_id"],
|
|
}
|
|
with TestClient(create_app(settings, openai_transport=httpx.MockTransport(handler))) as client:
|
|
created = client.post("/v1/lead-scoring/runs", headers=headers, json=payload)
|
|
assert created.status_code == 202
|
|
result = wait_for_terminal(client, created.json()["run_id"], headers)
|
|
assert result["status"] == "completed"
|
|
assert result["result_mode"] == "hybrid"
|
|
assert result["ai_score"] == 80
|
|
assert "caller-vkey-must-not-reach-local-provider" not in json.dumps(result)
|
|
assert len(requests) == 2
|
|
if api_key:
|
|
assert api_key not in json.dumps(result)
|
|
assert all(api_key not in request.content.decode() for request in requests)
|