120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
"""历史接入调用形态的真实冒烟；不能替代缺失下游的现行配置验收。"""
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from dotenv import dotenv_values
|
||
|
||
from polygateway import Effort, GatewayClient, GatewaySettings
|
||
from tests.e2e.conftest import (
|
||
LiveCapture,
|
||
captured_chat_round,
|
||
chat_expectations,
|
||
enforce_verdict,
|
||
observed_client,
|
||
source_controls,
|
||
)
|
||
from tests.live_evidence import write_live_round
|
||
|
||
_REPO = Path(__file__).resolve().parents[2]
|
||
_ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
|
||
_HAS_SOURCE = any(k.startswith("LLM__") and k.endswith("__API_KEY") for k in _ENV)
|
||
pytestmark = [
|
||
pytest.mark.slow,
|
||
pytest.mark.skipif(not _HAS_SOURCE, reason="缺少矩阵必需凭据，未覆盖"),
|
||
]
|
||
_OUT = Path("tests/outputs/134/live")
|
||
|
||
|
||
async def _call_shape(matrix, **kwargs):
|
||
"""session/parent 总由逐轮 UUID 传入；保留 cache_salt 调用形态。"""
|
||
settings = GatewaySettings.from_env("LLM", env=_ENV)
|
||
messages = [{"role": "user", "content": "Reply with exactly: compatibility-ok"}]
|
||
capture = LiveCapture(
|
||
expectations=chat_expectations(
|
||
settings, messages=messages, stream=True, controls=source_controls(settings)
|
||
)
|
||
)
|
||
|
||
def validate(response):
|
||
assert response.content.strip() and response.call_id
|
||
|
||
async with observed_client(settings, capture) as client:
|
||
_, verdict = await captured_chat_round(
|
||
client,
|
||
capture,
|
||
run_id=uuid4().hex,
|
||
matrix_id=matrix,
|
||
round_index=1,
|
||
output_dir=_OUT,
|
||
messages=messages,
|
||
models={s.name: s.model for s in settings.sources},
|
||
providers={s.name: s.provider for s in settings.sources},
|
||
source_efforts={
|
||
s.name: s.reasoning_effort
|
||
if s.reasoning_effort is not None
|
||
else (Effort.AUTO if s.enable_thinking else Effort.NONE)
|
||
if s.enable_thinking is not None
|
||
else None
|
||
for s in settings.sources
|
||
},
|
||
aliases={},
|
||
validate=validate,
|
||
**kwargs,
|
||
)
|
||
enforce_verdict(verdict)
|
||
|
||
|
||
class TestGovDocOnboarding:
|
||
"""历史 session_id＋parent_call_id 调用点契约。"""
|
||
|
||
async def test_call_site_shape_runs_governed(self):
|
||
await _call_shape("compat-parent")
|
||
|
||
async def test_structural_protocol_match(self):
|
||
"""外部 Protocol 缺包单列未覆盖；合成契约另在 unit 跑。"""
|
||
run_id = uuid4().hex
|
||
sys.path.insert(0, str(_REPO / "reference/GovDoc-SaaS/packages/docagent-core/src"))
|
||
try:
|
||
from docagent_core.protocols import LLMProvider
|
||
except ImportError:
|
||
write_live_round(
|
||
_OUT,
|
||
run_id=run_id,
|
||
matrix_id="external-protocol",
|
||
round_index=0,
|
||
safe_fields={
|
||
"status": "UNCOVERED",
|
||
"reason": "外部 Protocol 包缺失；未验证真实下游",
|
||
},
|
||
)
|
||
pytest.skip("外部 Protocol 包缺失，未覆盖")
|
||
finally:
|
||
sys.path.pop(0)
|
||
status = "FAIL"
|
||
client = None
|
||
try:
|
||
client = GatewayClient.from_env("LLM", env=_ENV)
|
||
assert isinstance(client, LLMProvider)
|
||
status = "PASS"
|
||
finally:
|
||
if client is not None:
|
||
await client.aclose()
|
||
write_live_round(
|
||
_OUT,
|
||
run_id=run_id,
|
||
matrix_id="external-protocol",
|
||
round_index=0,
|
||
safe_fields={"status": status, "reason": "外部 Protocol 结构契约，不是模型能力"},
|
||
)
|
||
|
||
|
||
class TestVideoTreeOnboarding:
|
||
"""历史 cache_salt 调用点契约，平铺键装配已移至 unit。"""
|
||
|
||
async def test_call_site_shape_with_cache_salt(self):
|
||
await _call_shape("compat-salt", cache_salt="epoch-1")
|