[测试、性能、安全、验收与归档]:完成全量测试验收与 OpenSpec 归档,包含 CLI 验收、NewAPI smoke、安全扫描和主 spec 更新
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
from owner_voice_pet.cli import main
|
||||
|
||||
|
||||
class CliAcceptanceTests(unittest.TestCase):
|
||||
def call(self, *args: str) -> tuple[int, dict[str, object]]:
|
||||
buffer = io.StringIO()
|
||||
with redirect_stdout(buffer):
|
||||
code = main(list(args))
|
||||
return code, json.loads(buffer.getvalue())
|
||||
|
||||
def test_acceptance_command_runs_end_to_end(self) -> None:
|
||||
code, data = self.call("acceptance")
|
||||
self.assertEqual(code, 0)
|
||||
self.assertTrue(data["success"])
|
||||
self.assertEqual(data["transcript"], "你是谁")
|
||||
self.assertGreaterEqual(data["played_segments"], 1)
|
||||
|
||||
def test_validate_assets_command(self) -> None:
|
||||
code, data = self.call("validate-assets")
|
||||
self.assertEqual(code, 0)
|
||||
self.assertTrue(data["valid"])
|
||||
|
||||
def test_security_check_command_has_no_leaks(self) -> None:
|
||||
code, data = self.call("security-check")
|
||||
self.assertEqual(code, 0)
|
||||
self.assertEqual(data["secret_leaks"], [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -62,6 +62,7 @@ class ModelsConfigTests(unittest.TestCase):
|
||||
self.assertEqual(config.llm_base_url, "https://newapi.mkbk.shop")
|
||||
self.assertEqual(config.llm_api_key, "secret-value")
|
||||
self.assertEqual(config.llm_model, "test-model")
|
||||
self.assertTrue(config.llm_stream)
|
||||
self.assertEqual(config.validate_basic(), [])
|
||||
finally:
|
||||
os.environ.clear()
|
||||
|
||||
@@ -120,12 +120,14 @@ class PipelineLlmTtsTests(unittest.TestCase):
|
||||
def __exit__(self, *args: object) -> None:
|
||||
return None
|
||||
|
||||
def __iter__(self):
|
||||
def read(self):
|
||||
event = {"choices": [{"delta": {"content": "你好。"}, "finish_reason": None}]}
|
||||
done = {"choices": [{"delta": {}, "finish_reason": "stop"}]}
|
||||
yield f"data: {json.dumps(event, ensure_ascii=False)}\n".encode()
|
||||
yield f"data: {json.dumps(done, ensure_ascii=False)}\n".encode()
|
||||
yield b"data: [DONE]\n"
|
||||
return (
|
||||
f"data: {json.dumps(event, ensure_ascii=False)}\n"
|
||||
f"data: {json.dumps(done, ensure_ascii=False)}\n"
|
||||
"data: [DONE]\n"
|
||||
).encode()
|
||||
|
||||
requests = []
|
||||
|
||||
@@ -146,6 +148,27 @@ class PipelineLlmTtsTests(unittest.TestCase):
|
||||
self.assertTrue(body["stream"])
|
||||
self.assertEqual(body["model"], "test-model")
|
||||
|
||||
def test_openai_non_stream_json_parser(self) -> None:
|
||||
class FakeResponse:
|
||||
def __enter__(self) -> "FakeResponse":
|
||||
return self
|
||||
|
||||
def __exit__(self, *args: object) -> None:
|
||||
return None
|
||||
|
||||
def read(self):
|
||||
return json.dumps({"choices": [{"message": {"content": "非流式回复。"}}]}).encode()
|
||||
|
||||
config = AppConfig(
|
||||
llm_base_url="https://newapi.mkbk.shop",
|
||||
llm_api_key="secret",
|
||||
llm_model="test-model",
|
||||
llm_api_style="chat_completions",
|
||||
llm_stream=False,
|
||||
)
|
||||
provider = OpenAICompatibleLlmProvider(config, urlopen=lambda request, timeout: FakeResponse())
|
||||
self.assertEqual(list(provider.stream_reply([Message("user", "你好", 1.0)]))[0].text_delta, "非流式回复。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user