study / test /test_image_api_adapter.py
xiaolongmr
整合功能更新并排除本地恢复产物
2dc6fe4
Raw History Blame Contribute Delete
7.42 kB
from __future__ import annotations
import unittest
from services.image_api_adapter import (
adapter_system_prompt,
document_to_text,
infer_document_auth_type,
normalize_adapter_result,
normalize_adapter_template,
parse_model_json,
redact_document_secrets,
)
class ImageApiAdapterTests(unittest.TestCase):
def test_html_document_is_readable_and_secrets_are_redacted(self) -> None:
source = """
<html><body><h1>Pear API</h1><pre>{"key":"pear-secret-123456", "prompt":"cat"}</pre>
<p>Authorization: Bearer abcdefghijklmnop</p><script>const key = 'do-not-keep';</script></body></html>
"""
readable = document_to_text(source, "pear.html")
redacted = redact_document_secrets(readable)
self.assertIn("Pear API", redacted)
self.assertIn("[REDACTED]", redacted)
self.assertNotIn("pear-secret-123456", redacted)
self.assertNotIn("abcdefghijklmnop", redacted)
self.assertNotIn("do-not-keep", redacted)
def test_normalizes_safe_pear_style_template(self) -> None:
result = normalize_adapter_result({
"summary": "PearAPI 同步生图",
"models": ["grok-4-image", "grok-4-image"],
"template": {
"version": 2,
"method": "POST",
"auth_type": "body",
"auth_name": "key",
"generate": {
"endpoint": "https://api.pearapi.ai/api/image_generate",
"content_type": "json",
"body": {
"prompt": "{{prompt}}",
"model": "{{model}}",
"size": "{{size}}",
"task_type": "sync",
},
"image_field": "",
"response": {
"image_paths": ["data.image_urls"],
"error_paths": ["detail", "msg"],
"success_path": "code",
"success_values": [200],
},
},
"edit": {
"endpoint": "https://api.pearapi.ai/api/image_generate",
"content_type": "multipart",
"body": {"prompt": "{{prompt}}", "model": "{{model}}", "task_type": "sync"},
"image_field": "images[]",
"response": {"image_paths": ["data.image_urls"]},
},
},
}, base_url="https://api.pearapi.ai")
self.assertEqual(result["models"], ["grok-4-image"])
self.assertEqual(result["template"]["auth_type"], "body")
self.assertEqual(result["template"]["generate"]["response"]["image_paths"], ["data.image_urls"])
self.assertEqual(result["template"]["generate"]["body"]["size"], "{{size}}")
self.assertEqual(result["template"]["edit"]["image_field"], "images[]")
def test_rejects_static_secret_and_cross_domain_endpoint(self) -> None:
common = {
"version": 2,
"method": "POST",
"auth_type": "body",
"auth_name": "key",
"generate": {
"endpoint": "/generate",
"content_type": "json",
"body": {"api_key": "real-secret-value"},
"image_field": "",
"response": {"image_paths": ["data.images"]},
},
"edit": None,
}
with self.assertRaisesRegex(ValueError, "静态密钥"):
normalize_adapter_template(common, base_url="https://safe.example")
cross_domain = {
**common,
"generate": {**common["generate"], "endpoint": "https://evil.example/generate", "body": {"api_key": "{{api_key}}"}},
}
with self.assertRaisesRegex(ValueError, "相同域名"):
normalize_adapter_template(cross_domain, base_url="https://safe.example")
def test_parses_json_from_fenced_model_output(self) -> None:
parsed = parse_model_json('```json\n{"summary":"ok","template":{}}\n```')
self.assertEqual(parsed["summary"], "ok")
def test_infers_no_auth_from_complete_curl_examples_without_credentials(self) -> None:
document = """
POST /v1/generate
curl -X POST https://imagefree.example/v1/generate \\
-H "Content-Type: application/json" \\
-d '{"prompt":"cat","aspect_ratio":"1:1"}'
"""
self.assertEqual(infer_document_auth_type(document), "none")
self.assertEqual(
infer_document_auth_type("Authorization: Bearer YOUR_API_KEY\nPOST /v1/generate"),
"required",
)
def test_no_auth_hint_and_async_poll_are_normalized(self) -> None:
result = normalize_adapter_result({
"summary": "无鉴权异步编辑 API",
"models": ["default"],
"template": {
"version": 2,
"method": "POST",
"auth_type": "bearer",
"auth_name": "",
"generate": {
"endpoint": "/v1/generate",
"content_type": "json",
"body": {"prompt": "{{prompt}}", "aspect_ratio": "{{size}}"},
"image_field": "",
"poll": None,
"response": {
"image_paths": ["image_url", "image_base64"],
"error_paths": ["error"],
"success_path": "status",
"success_values": ["completed"],
"failure_values": ["error"],
},
},
"edit": {
"endpoint": "/v1/edit",
"content_type": "json",
"body": {"prompt": "{{prompt}}", "images": "{{images_data_urls}}"},
"image_field": "",
"poll": {
"endpoint": "/v1/edit/tasks/{{task_id}}",
"id_path": "id",
"interval_seconds": 3,
"max_attempts": 100,
},
"response": {
"image_paths": ["image_url", "image_base64"],
"error_paths": ["error"],
"success_path": "status",
"success_values": ["completed"],
"failure_values": ["error"],
},
},
"models": {
"endpoint": "/v1/models",
"response_path": "items",
"id_path": "id",
},
},
}, base_url="https://imagefree.example", auth_hint="none")
self.assertEqual(result["template"]["auth_type"], "none")
self.assertEqual(result["template"]["auth_name"], "")
self.assertIsNone(result["template"]["generate"]["poll"])
self.assertEqual(
result["template"]["edit"]["poll"]["endpoint"],
"/v1/edit/tasks/{{task_id}}",
)
self.assertEqual(result["template"]["models"]["response_path"], "items")
self.assertIn("asynchronous", adapter_system_prompt().lower())
self.assertIn("model-list endpoint", adapter_system_prompt().lower())
if __name__ == "__main__":
unittest.main()