Spaces:
Sleeping
Sleeping
File size: 6,028 Bytes
cfd3735 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 |
"""Test the APIOperation class."""
import json
import os
from pathlib import Path
from typing import Iterable, List, Tuple
import pytest
import yaml
from openapi_schema_pydantic import (
Components,
Info,
MediaType,
Reference,
RequestBody,
Schema,
)
from langchain.tools.openapi.utils.api_models import (
APIOperation,
APIRequestBody,
APIRequestBodyProperty,
)
from langchain.tools.openapi.utils.openapi_utils import HTTPVerb, OpenAPISpec
_DIR = Path(__file__).parent
def _get_test_specs() -> Iterable[Path]:
"""Walk the test_specs directory and collect all files with the name 'apispec'
in them.
"""
test_specs_dir = _DIR / "test_specs"
return (
Path(root) / file
for root, _, files in os.walk(test_specs_dir)
for file in files
if file.startswith("apispec")
)
def _get_paths_and_methods_from_spec_dictionary(
spec: dict,
) -> Iterable[Tuple[str, str]]:
"""Return a tuple (paths, methods) for every path in spec."""
valid_methods = [verb.value for verb in HTTPVerb]
for path_name, path_item in spec["paths"].items():
for method in valid_methods:
if method in path_item:
yield (path_name, method)
def http_paths_and_methods() -> List[Tuple[str, OpenAPISpec, str, str]]:
"""Return a args for every method in cached OpenAPI spec in test_specs."""
http_paths_and_methods = []
for test_spec in _get_test_specs():
spec_name = test_spec.parent.name
if test_spec.suffix == ".json":
with test_spec.open("r") as f:
spec = json.load(f)
else:
with test_spec.open("r") as f:
spec = yaml.safe_load(f.read())
parsed_spec = OpenAPISpec.from_file(test_spec)
for path, method in _get_paths_and_methods_from_spec_dictionary(spec):
http_paths_and_methods.append(
(
spec_name,
parsed_spec,
path,
method,
)
)
return http_paths_and_methods
@pytest.mark.parametrize(
"spec_name, spec, path, method",
http_paths_and_methods(),
)
def test_parse_api_operations(
spec_name: str, spec: OpenAPISpec, path: str, method: str
) -> None:
"""Test the APIOperation class."""
try:
APIOperation.from_openapi_spec(spec, path, method)
except Exception as e:
raise AssertionError(f"Error processong {spec_name}: {e} ") from e
@pytest.fixture
def raw_spec() -> OpenAPISpec:
"""Return a raw OpenAPI spec."""
return OpenAPISpec(
info=Info(title="Test API", version="1.0.0"),
)
def test_api_request_body_from_request_body_with_ref(raw_spec: OpenAPISpec) -> None:
"""Test instantiating APIRequestBody from RequestBody with a reference."""
raw_spec.components = Components(
schemas={
"Foo": Schema(
type="object",
properties={
"foo": Schema(type="string"),
"bar": Schema(type="number"),
},
required=["foo"],
)
}
)
media_type = MediaType(
schema=Reference(
ref="#/components/schemas/Foo",
)
)
request_body = RequestBody(content={"application/json": media_type})
api_request_body = APIRequestBody.from_request_body(request_body, raw_spec)
assert api_request_body.description is None
assert len(api_request_body.properties) == 2
foo_prop = api_request_body.properties[0]
assert foo_prop.name == "foo"
assert foo_prop.required is True
bar_prop = api_request_body.properties[1]
assert bar_prop.name == "bar"
assert bar_prop.required is False
assert api_request_body.media_type == "application/json"
def test_api_request_body_from_request_body_with_schema(raw_spec: OpenAPISpec) -> None:
"""Test instantiating APIRequestBody from RequestBody with a schema."""
request_body = RequestBody(
content={
"application/json": MediaType(
schema=Schema(type="object", properties={"foo": Schema(type="string")})
)
}
)
api_request_body = APIRequestBody.from_request_body(request_body, raw_spec)
assert api_request_body.properties == [
APIRequestBodyProperty(
name="foo",
required=False,
type="string",
default=None,
description=None,
properties=[],
references_used=[],
)
]
assert api_request_body.media_type == "application/json"
def test_api_request_body_property_from_schema(raw_spec: OpenAPISpec) -> None:
raw_spec.components = Components(
schemas={
"Bar": Schema(
type="number",
)
}
)
schema = Schema(
type="object",
properties={
"foo": Schema(type="string"),
"bar": Reference(ref="#/components/schemas/Bar"),
},
required=["bar"],
)
api_request_body_property = APIRequestBodyProperty.from_schema(
schema, "test", required=True, spec=raw_spec
)
expected_sub_properties = [
APIRequestBodyProperty(
name="foo",
required=False,
type="string",
default=None,
description=None,
properties=[],
references_used=[],
),
APIRequestBodyProperty(
name="bar",
required=True,
type="number",
default=None,
description=None,
properties=[],
references_used=["Bar"],
),
]
assert api_request_body_property.properties[0] == expected_sub_properties[0]
assert api_request_body_property.properties[1] == expected_sub_properties[1]
assert api_request_body_property.type == "object"
assert api_request_body_property.properties[1].references_used == ["Bar"]
|