Co-authored-by: -LAN- <laipz8200@outlook.com>
This commit is contained in:
@@ -7,20 +7,17 @@ from _pytest.monkeypatch import MonkeyPatch
|
||||
|
||||
|
||||
class MockedHttp:
|
||||
def httpx_request(method: Literal['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD'],
|
||||
url: str, **kwargs) -> httpx.Response:
|
||||
def httpx_request(
|
||||
method: Literal["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD"], url: str, **kwargs
|
||||
) -> httpx.Response:
|
||||
"""
|
||||
Mocked httpx.request
|
||||
"""
|
||||
request = httpx.Request(
|
||||
method,
|
||||
url,
|
||||
params=kwargs.get('params'),
|
||||
headers=kwargs.get('headers'),
|
||||
cookies=kwargs.get('cookies')
|
||||
method, url, params=kwargs.get("params"), headers=kwargs.get("headers"), cookies=kwargs.get("cookies")
|
||||
)
|
||||
data = kwargs.get('data', None)
|
||||
resp = json.dumps(data).encode('utf-8') if data else b'OK'
|
||||
data = kwargs.get("data", None)
|
||||
resp = json.dumps(data).encode("utf-8") if data else b"OK"
|
||||
response = httpx.Response(
|
||||
status_code=200,
|
||||
request=request,
|
||||
|
||||
@@ -10,6 +10,7 @@ todos_data = {
|
||||
"user1": ["Go for a run", "Read a book"],
|
||||
}
|
||||
|
||||
|
||||
class TodosResource(Resource):
|
||||
def get(self, username):
|
||||
todos = todos_data.get(username, [])
|
||||
@@ -32,7 +33,8 @@ class TodosResource(Resource):
|
||||
|
||||
return {"error": "Invalid todo index"}, 400
|
||||
|
||||
api.add_resource(TodosResource, '/todos/<string:username>')
|
||||
|
||||
if __name__ == '__main__':
|
||||
api.add_resource(TodosResource, "/todos/<string:username>")
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(port=5003, debug=True)
|
||||
|
||||
@@ -3,37 +3,40 @@ from core.tools.tool.tool import Tool
|
||||
from tests.integration_tests.tools.__mock.http import setup_http_mock
|
||||
|
||||
tool_bundle = {
|
||||
'server_url': 'http://www.example.com/{path_param}',
|
||||
'method': 'post',
|
||||
'author': '',
|
||||
'openapi': {'parameters': [{'in': 'path', 'name': 'path_param'},
|
||||
{'in': 'query', 'name': 'query_param'},
|
||||
{'in': 'cookie', 'name': 'cookie_param'},
|
||||
{'in': 'header', 'name': 'header_param'},
|
||||
],
|
||||
'requestBody': {
|
||||
'content': {'application/json': {'schema': {'properties': {'body_param': {'type': 'string'}}}}}}
|
||||
},
|
||||
'parameters': []
|
||||
"server_url": "http://www.example.com/{path_param}",
|
||||
"method": "post",
|
||||
"author": "",
|
||||
"openapi": {
|
||||
"parameters": [
|
||||
{"in": "path", "name": "path_param"},
|
||||
{"in": "query", "name": "query_param"},
|
||||
{"in": "cookie", "name": "cookie_param"},
|
||||
{"in": "header", "name": "header_param"},
|
||||
],
|
||||
"requestBody": {
|
||||
"content": {"application/json": {"schema": {"properties": {"body_param": {"type": "string"}}}}}
|
||||
},
|
||||
},
|
||||
"parameters": [],
|
||||
}
|
||||
parameters = {
|
||||
'path_param': 'p_param',
|
||||
'query_param': 'q_param',
|
||||
'cookie_param': 'c_param',
|
||||
'header_param': 'h_param',
|
||||
'body_param': 'b_param',
|
||||
"path_param": "p_param",
|
||||
"query_param": "q_param",
|
||||
"cookie_param": "c_param",
|
||||
"header_param": "h_param",
|
||||
"body_param": "b_param",
|
||||
}
|
||||
|
||||
|
||||
def test_api_tool(setup_http_mock):
|
||||
tool = ApiTool(api_bundle=tool_bundle, runtime=Tool.Runtime(credentials={'auth_type': 'none'}))
|
||||
tool = ApiTool(api_bundle=tool_bundle, runtime=Tool.Runtime(credentials={"auth_type": "none"}))
|
||||
headers = tool.assembling_request(parameters)
|
||||
response = tool.do_http_request(tool.api_bundle.server_url, tool.api_bundle.method, headers, parameters)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert '/p_param' == response.request.url.path
|
||||
assert b'query_param=q_param' == response.request.url.query
|
||||
assert 'h_param' == response.request.headers.get('header_param')
|
||||
assert 'application/json' == response.request.headers.get('content-type')
|
||||
assert 'cookie_param=c_param' == response.request.headers.get('cookie')
|
||||
assert 'b_param' in response.content.decode()
|
||||
assert "/p_param" == response.request.url.path
|
||||
assert b"query_param=q_param" == response.request.url.query
|
||||
assert "h_param" == response.request.headers.get("header_param")
|
||||
assert "application/json" == response.request.headers.get("content-type")
|
||||
assert "cookie_param=c_param" == response.request.headers.get("cookie")
|
||||
assert "b_param" in response.content.decode()
|
||||
|
||||
@@ -7,16 +7,17 @@ provider_names = [provider.identity.name for provider in provider_generator]
|
||||
ToolManager.clear_builtin_providers_cache()
|
||||
provider_generator = ToolManager.list_builtin_providers()
|
||||
|
||||
@pytest.mark.parametrize('name', provider_names)
|
||||
|
||||
@pytest.mark.parametrize("name", provider_names)
|
||||
def test_tool_providers(benchmark, name):
|
||||
"""
|
||||
Test that all tool providers can be loaded
|
||||
"""
|
||||
|
||||
|
||||
def test(generator):
|
||||
try:
|
||||
return next(generator)
|
||||
except StopIteration:
|
||||
return None
|
||||
|
||||
benchmark.pedantic(test, args=(provider_generator,), iterations=1, rounds=1)
|
||||
|
||||
benchmark.pedantic(test, args=(provider_generator,), iterations=1, rounds=1)
|
||||
|
||||
Reference in New Issue
Block a user