hello-agents/Co-creation-projects/senming666-api_test_assistant/tests/test_parser_agent.py

"""ParserAgent 解析 OpenAPI 文档的单元测试"""
import json
from pathlib import Path

from src.agents.parser_agent import ParserAgent


def _parser():
    return ParserAgent()


# --- parse_text：空输入 / JSON / YAML / 非 dict ---

def test_parse_text_empty():
    assert _parser().parse_text("") == []
    assert _parser().parse_text("   ") == []


def test_parse_text_json():
    doc = '{"openapi": "3.0.0", "paths": {"/users": {"get": {"summary": "list", "responses": {"200": {"description": "ok"}}}}}}'
    endpoints = _parser().parse_text(doc)
    assert len(endpoints) == 1
    assert endpoints[0]["path"] == "/users"
    assert endpoints[0]["method"] == "GET"


def test_parse_text_yaml():
    doc = """
openapi: 3.0.0
paths:
  /users:
    get:
      summary: list users
      responses:
        '200':
          description: ok
"""
    endpoints = _parser().parse_text(doc)
    assert len(endpoints) == 1
    assert endpoints[0]["method"] == "GET"


def test_parse_text_non_dict():
    # "null" 会被解析成 None，应返回空列表而不是崩溃
    assert _parser().parse_text("null") == []


# --- extract_endpoints：类型容错 + requestBody ---

def test_extract_endpoints_non_dict_input():
    p = _parser()
    assert p.extract_endpoints(None) == []
    assert p.extract_endpoints("not a dict") == []
    assert p.extract_endpoints([]) == []


def test_extract_endpoints_request_body_and_content_type():
    doc = {
        "paths": {
            "/create": {
                "post": {
                    "requestBody": {
                        "content": {
                            "application/json": {
                                "schema": {
                                    "type": "object",
                                    "required": ["name"],
                                    "properties": {"name": {"type": "string"}},
                                }
                            }
                        }
                    },
                    "responses": {"200": {}},
                }
            }
        }
    }
    endpoints = _parser().extract_endpoints(doc)
    assert len(endpoints) == 1
    ep = endpoints[0]
    assert ep["method"] == "POST"
    # 请求体被解析成真正的 schema，而不是原始 requestBody 对象
    assert ep["request_body"] == {
        "type": "object",
        "required": ["name"],
        "properties": {"name": {"type": "string"}},
    }
    assert ep["request_content_type"] == "application/json"


# --- $ref 展开 ---

def _ref_doc():
    """一个带 components/schemas 和 $ref 的最小文档"""
    return {
        "components": {
            "schemas": {
                "User": {
                    "type": "object",
                    "required": ["username", "password"],
                    "properties": {
                        "username": {"type": "string"},
                        "password": {"type": "string"},
                    },
                },
                "Nested": {
                    "type": "object",
                    "properties": {"user": {"$ref": "#/components/schemas/User"}},
                },
            }
        },
        "paths": {
            "/register": {
                "post": {
                    "requestBody": {
                        "content": {
                            "application/json": {
                                "schema": {"$ref": "#/components/schemas/User"},
                            }
                        }
                    },
                    "responses": {
                        "201": {
                            "content": {
                                "application/json": {
                                    "schema": {"$ref": "#/components/schemas/Nested"},
                                }
                            }
                        }
                    },
                }
            }
        },
    }


def test_request_body_ref_resolved():
    endpoints = _parser().extract_endpoints(_ref_doc())
    ep = endpoints[0]
    # $ref 被展开，字段直接可见
    assert ep["request_body"]["required"] == ["username", "password"]
    assert "username" in ep["request_body"]["properties"]


def test_response_schema_ref_resolved():
    endpoints = _parser().extract_endpoints(_ref_doc())
    ep = endpoints[0]
    # 主成功响应（201）的 schema 被展开，且嵌套 $ref 也递归展开
    assert ep["response_schema"]["properties"]["user"]["type"] == "object"
    assert "username" in ep["response_schema"]["properties"]["user"]["properties"]


def test_get_response_schema_by_status():
    endpoints = _parser().extract_endpoints(_ref_doc())
    ep = endpoints[0]
    assert _parser().get_response_schema(ep, 201)["properties"]["user"]["type"] == "object"
    # 没有声明的状态码返回 None
    assert _parser().get_response_schema(ep, 404) is None


# --- get_expected_status ---

def test_expected_status_normal_prefers_2xx():
    ep = {"responses": {"200": {}, "400": {}}}
    assert _parser().get_expected_status(ep, "normal") == 200


def test_expected_status_boundary_prefers_2xx():
    ep = {"responses": {"201": {}, "422": {}}}
    assert _parser().get_expected_status(ep, "boundary") == 201


def test_expected_status_error_prefers_4xx():
    ep = {"responses": {"200": {}, "400": {}}}
    assert _parser().get_expected_status(ep, "error") == 400


def test_expected_status_error_with_body_validation():
    # 有必填请求体字段 → 缺字段 → 校验失败 422
    ep = {
        "responses": {"201": {}, "422": {}},
        "request_body": {"required": ["name"], "properties": {"name": {"type": "string"}}},
        "parameters": [],
    }
    assert _parser().get_expected_status(ep, "error") == 422


def test_expected_status_error_with_only_path_param():
    # 只有必填路径参数 → 缺路径参数 → 路由不匹配 404（而非 422）
    ep = {
        "responses": {"200": {}, "422": {}},
        "request_body": None,
        "parameters": [{"name": "id", "in": "path", "required": True, "schema": {"type": "string"}}],
    }
    assert _parser().get_expected_status(ep, "error") == 404


def test_expected_status_default_200_when_no_responses():
    assert _parser().get_expected_status({"responses": {}}, "normal") == 200


# --- 结构判断辅助 ---

def test_has_testable_inputs():
    # 没参数没请求体 → 无可测输入
    assert _parser().has_testable_inputs({"parameters": [], "request_body": None}) is False
    # 有请求体 → 可测
    assert _parser().has_testable_inputs({"parameters": [], "request_body": {"type": "object"}}) is True
    # 有 query 参数 → 可测
    assert _parser().has_testable_inputs(
        {"parameters": [{"in": "query", "name": "limit"}], "request_body": None}
    ) is True


def test_has_validation_input():
    assert _parser().has_validation_input(
        {"parameters": [], "request_body": {"required": ["name"]}}
    ) is True
    assert _parser().has_validation_input({"parameters": [], "request_body": None}) is False


def test_has_required_path_param():
    assert _parser().has_required_path_param(
        {"parameters": [{"in": "path", "name": "id", "required": True}]}
    ) is True
    assert _parser().has_required_path_param(
        {"parameters": [{"in": "query", "name": "limit"}]}
    ) is False


# --- parse_file ---

def test_parse_file(tmp_path):
    f = tmp_path / "api.yaml"
    f.write_text(
        "openapi: 3.0.0\npaths:\n  /a:\n    get:\n      responses:\n        '200': {}\n",
        encoding="utf-8",
    )
    endpoints = _parser().parse_file(str(f))
    assert len(endpoints) == 1
    assert endpoints[0]["path"] == "/a"


# --- 真实 Chat openapi.json 集成解析 ---

def _chat_endpoints():
    fixture = Path(__file__).parent / "fixtures" / "chat_openapi.json"
    doc = json.loads(fixture.read_text(encoding="utf-8"))
    return _parser().extract_endpoints(doc)


def test_chat_register_body_ref_resolved():
    ep = next(e for e in _chat_endpoints() if e["path"] == "/api/auth/register")
    # 注册请求体应能看到真实字段，而不是 $ref 字符串
    assert ep["request_body"]["required"] == [
        "username", "display_name", "password", "confirm_password"
    ]
    assert set(ep["request_body"]["properties"]) == {
        "username", "display_name", "password", "confirm_password"
    }


def test_chat_login_response_schema_resolved():
    ep = next(e for e in _chat_endpoints() if e["path"] == "/api/auth/login")
    # 登录成功响应的 schema 展开后应包含 access_token 字段
    assert "access_token" in ep["response_schema"]["properties"]


def test_chat_health_has_no_testable_inputs():
    ep = next(e for e in _chat_endpoints() if e["path"] == "/api/health")
    assert _parser().has_testable_inputs(ep) is False
    assert _parser().has_validation_input(ep) is False
    assert _parser().has_required_path_param(ep) is False


def test_chat_files_images_is_multipart():
    ep = next(e for e in _chat_endpoints() if e["path"] == "/api/files/images")
    assert ep["request_content_type"] == "multipart/form-data"
    # 文件字段用 contentMediaType 标识
    assert "contentMediaType" in ep["request_body"]["properties"]["file"]