mirror of https://github.com/tiangolo/fastapi.git
⬆️ Add tests, fix issues and update Pydantic
This commit is contained in:
parent
0125ea4f83
commit
804ec460fc
|
|
@ -18,6 +18,6 @@ async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
|
|||
response = {}
|
||||
if commons.q:
|
||||
response.update({"q": commons.q})
|
||||
items = fake_items_db[commons.skip : commons.limit]
|
||||
items = fake_items_db[commons.skip : commons.skip + commons.limit]
|
||||
response.update({"items": items})
|
||||
return response
|
||||
|
|
|
|||
|
|
@ -18,6 +18,6 @@ async def read_items(commons=Depends(CommonQueryParams)):
|
|||
response = {}
|
||||
if commons.q:
|
||||
response.update({"q": commons.q})
|
||||
items = fake_items_db[commons.skip : commons.limit]
|
||||
items = fake_items_db[commons.skip : commons.skip + commons.limit]
|
||||
response.update({"items": items})
|
||||
return response
|
||||
|
|
|
|||
|
|
@ -18,6 +18,6 @@ async def read_items(commons: CommonQueryParams = Depends()):
|
|||
response = {}
|
||||
if commons.q:
|
||||
response.update({"q": commons.q})
|
||||
items = fake_items_db[commons.skip : commons.limit]
|
||||
items = fake_items_db[commons.skip : commons.skip + commons.limit]
|
||||
response.update({"items": items})
|
||||
return response
|
||||
|
|
|
|||
|
|
@ -7,4 +7,4 @@ fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"
|
|||
|
||||
@app.get("/items/")
|
||||
async def read_item(skip: int = 0, limit: int = 100):
|
||||
return fake_items_db[skip:limit]
|
||||
return fake_items_db[skip : skip + limit]
|
||||
|
|
|
|||
|
|
@ -58,8 +58,6 @@ def get_flat_dependant(dependant: Dependant) -> Dependant:
|
|||
security_schemes=dependant.security_requirements.copy(),
|
||||
)
|
||||
for sub_dependant in dependant.dependencies:
|
||||
if sub_dependant is dependant:
|
||||
raise ValueError("recursion", dependant.dependencies)
|
||||
flat_sub = get_flat_dependant(sub_dependant)
|
||||
flat_dependant.path_params.extend(flat_sub.path_params)
|
||||
flat_dependant.query_params.extend(flat_sub.query_params)
|
||||
|
|
@ -197,16 +195,12 @@ def add_param_to_body_fields(*, param: inspect.Parameter, dependant: Dependant)
|
|||
dependant.body_params.append(field)
|
||||
|
||||
|
||||
def is_coroutine_callable(call: Callable = None) -> bool:
|
||||
if not call:
|
||||
return False
|
||||
def is_coroutine_callable(call: Callable) -> bool:
|
||||
if inspect.isfunction(call):
|
||||
return asyncio.iscoroutinefunction(call)
|
||||
if inspect.isclass(call):
|
||||
return False
|
||||
call = getattr(call, "__call__", None)
|
||||
if not call:
|
||||
return False
|
||||
return asyncio.iscoroutinefunction(call)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -147,61 +147,65 @@ def get_openapi_path(
|
|||
security_schemes: Dict[str, Any] = {}
|
||||
definitions: Dict[str, Any] = {}
|
||||
assert route.methods is not None, "Methods must be a list"
|
||||
for method in route.methods:
|
||||
operation = get_openapi_operation_metadata(route=route, method=method)
|
||||
parameters: List[Dict] = []
|
||||
flat_dependant = get_flat_dependant(route.dependant)
|
||||
security_definitions, operation_security = get_openapi_security_definitions(
|
||||
flat_dependant=flat_dependant
|
||||
)
|
||||
if operation_security:
|
||||
operation.setdefault("security", []).extend(operation_security)
|
||||
if security_definitions:
|
||||
security_schemes.update(security_definitions)
|
||||
all_route_params = get_openapi_params(route.dependant)
|
||||
validation_definitions, operation_parameters = get_openapi_operation_parameters(
|
||||
all_route_params=all_route_params
|
||||
)
|
||||
definitions.update(validation_definitions)
|
||||
parameters.extend(operation_parameters)
|
||||
if parameters:
|
||||
operation["parameters"] = parameters
|
||||
if method in METHODS_WITH_BODY:
|
||||
request_body_oai = get_openapi_operation_request_body(
|
||||
body_field=route.body_field, model_name_map=model_name_map
|
||||
if route.include_in_schema:
|
||||
for method in route.methods:
|
||||
operation = get_openapi_operation_metadata(route=route, method=method)
|
||||
parameters: List[Dict] = []
|
||||
flat_dependant = get_flat_dependant(route.dependant)
|
||||
security_definitions, operation_security = get_openapi_security_definitions(
|
||||
flat_dependant=flat_dependant
|
||||
)
|
||||
if request_body_oai:
|
||||
operation["requestBody"] = request_body_oai
|
||||
if "ValidationError" not in definitions:
|
||||
definitions["ValidationError"] = validation_error_definition
|
||||
definitions[
|
||||
"HTTPValidationError"
|
||||
] = validation_error_response_definition
|
||||
status_code = str(route.status_code)
|
||||
response_schema = {"type": "string"}
|
||||
if lenient_issubclass(route.content_type, JSONResponse):
|
||||
if route.response_field:
|
||||
response_schema, _ = field_schema(
|
||||
route.response_field,
|
||||
model_name_map=model_name_map,
|
||||
ref_prefix=REF_PREFIX,
|
||||
if operation_security:
|
||||
operation.setdefault("security", []).extend(operation_security)
|
||||
if security_definitions:
|
||||
security_schemes.update(security_definitions)
|
||||
all_route_params = get_openapi_params(route.dependant)
|
||||
validation_definitions, operation_parameters = get_openapi_operation_parameters(
|
||||
all_route_params=all_route_params
|
||||
)
|
||||
definitions.update(validation_definitions)
|
||||
parameters.extend(operation_parameters)
|
||||
if parameters:
|
||||
operation["parameters"] = parameters
|
||||
if method in METHODS_WITH_BODY:
|
||||
request_body_oai = get_openapi_operation_request_body(
|
||||
body_field=route.body_field, model_name_map=model_name_map
|
||||
)
|
||||
else:
|
||||
response_schema = {}
|
||||
content = {route.content_type.media_type: {"schema": response_schema}}
|
||||
operation["responses"] = {
|
||||
status_code: {"description": route.response_description, "content": content}
|
||||
}
|
||||
if all_route_params or route.body_field:
|
||||
operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
|
||||
"description": "Validation Error",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
|
||||
}
|
||||
},
|
||||
if request_body_oai:
|
||||
operation["requestBody"] = request_body_oai
|
||||
if "ValidationError" not in definitions:
|
||||
definitions["ValidationError"] = validation_error_definition
|
||||
definitions[
|
||||
"HTTPValidationError"
|
||||
] = validation_error_response_definition
|
||||
status_code = str(route.status_code)
|
||||
response_schema = {"type": "string"}
|
||||
if lenient_issubclass(route.content_type, JSONResponse):
|
||||
if route.response_field:
|
||||
response_schema, _ = field_schema(
|
||||
route.response_field,
|
||||
model_name_map=model_name_map,
|
||||
ref_prefix=REF_PREFIX,
|
||||
)
|
||||
else:
|
||||
response_schema = {}
|
||||
content = {route.content_type.media_type: {"schema": response_schema}}
|
||||
operation["responses"] = {
|
||||
status_code: {
|
||||
"description": route.response_description,
|
||||
"content": content,
|
||||
}
|
||||
}
|
||||
path[method.lower()] = operation
|
||||
if all_route_params or route.body_field:
|
||||
operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
|
||||
"description": "Validation Error",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
|
||||
}
|
||||
},
|
||||
}
|
||||
path[method.lower()] = operation
|
||||
return path, security_schemes, definitions
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ classifiers = [
|
|||
]
|
||||
requires = [
|
||||
"starlette >=0.9.7",
|
||||
"pydantic >=0.16"
|
||||
"pydantic >=0.17"
|
||||
]
|
||||
description-file = "README.md"
|
||||
requires-python = ">=3.6"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,25 @@
|
|||
from fastapi import FastAPI
|
||||
from fastapi.params import Param
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
@app.get("/items/")
|
||||
def read_items(q: str = Param(None)):
|
||||
return {"q": q}
|
||||
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def test_default_param_query_none():
|
||||
response = client.get("/items/")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"q": None}
|
||||
|
||||
|
||||
def test_default_param_query():
|
||||
response = client.get("/items/?q=foo")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"q": "foo"}
|
||||
|
|
@ -28,7 +28,13 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
def test_items():
|
||||
response = client.get("/items/")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == [{"name": "Foo"}]
|
||||
|
|
|
|||
|
|
@ -83,7 +83,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -101,7 +101,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ html_contents = """
|
|||
"""
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -0,0 +1,144 @@
|
|||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from dependencies.tutorial004 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
openapi_schema = {
|
||||
"openapi": "3.0.2",
|
||||
"info": {"title": "Fast API", "version": "0.1.0"},
|
||||
"paths": {
|
||||
"/items/": {
|
||||
"get": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
},
|
||||
"422": {
|
||||
"description": "Validation Error",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/HTTPValidationError"
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
"summary": "Read Items Get",
|
||||
"operationId": "read_items_items__get",
|
||||
"parameters": [
|
||||
{
|
||||
"required": False,
|
||||
"schema": {"title": "Q", "type": "string"},
|
||||
"name": "q",
|
||||
"in": "query",
|
||||
},
|
||||
{
|
||||
"required": False,
|
||||
"schema": {"title": "Skip", "type": "integer", "default": 0},
|
||||
"name": "skip",
|
||||
"in": "query",
|
||||
},
|
||||
{
|
||||
"required": False,
|
||||
"schema": {"title": "Limit", "type": "integer", "default": 100},
|
||||
"name": "limit",
|
||||
"in": "query",
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"ValidationError": {
|
||||
"title": "ValidationError",
|
||||
"required": ["loc", "msg", "type"],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"loc": {
|
||||
"title": "Location",
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"msg": {"title": "Message", "type": "string"},
|
||||
"type": {"title": "Error Type", "type": "string"},
|
||||
},
|
||||
},
|
||||
"HTTPValidationError": {
|
||||
"title": "HTTPValidationError",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"detail": {
|
||||
"title": "Detail",
|
||||
"type": "array",
|
||||
"items": {"$ref": "#/components/schemas/ValidationError"},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path,expected_status,expected_response",
|
||||
[
|
||||
(
|
||||
"/items",
|
||||
200,
|
||||
{
|
||||
"items": [
|
||||
{"item_name": "Foo"},
|
||||
{"item_name": "Bar"},
|
||||
{"item_name": "Baz"},
|
||||
]
|
||||
},
|
||||
),
|
||||
(
|
||||
"/items?q=foo",
|
||||
200,
|
||||
{
|
||||
"items": [
|
||||
{"item_name": "Foo"},
|
||||
{"item_name": "Bar"},
|
||||
{"item_name": "Baz"},
|
||||
],
|
||||
"q": "foo",
|
||||
},
|
||||
),
|
||||
(
|
||||
"/items?q=foo&skip=1",
|
||||
200,
|
||||
{"items": [{"item_name": "Bar"}, {"item_name": "Baz"}], "q": "foo"},
|
||||
),
|
||||
(
|
||||
"/items?q=bar&limit=2",
|
||||
200,
|
||||
{"items": [{"item_name": "Foo"}, {"item_name": "Bar"}], "q": "bar"},
|
||||
),
|
||||
(
|
||||
"/items?q=bar&skip=1&limit=1",
|
||||
200,
|
||||
{"items": [{"item_name": "Bar"}], "q": "bar"},
|
||||
),
|
||||
(
|
||||
"/items?limit=1&q=bar&skip=1",
|
||||
200,
|
||||
{"items": [{"item_name": "Bar"}], "q": "bar"},
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_get(path, expected_status, expected_response):
|
||||
response = client.get(path)
|
||||
assert response.status_code == expected_status
|
||||
assert response.json() == expected_response
|
||||
|
|
@ -74,7 +74,7 @@ openapi_schema = {
|
|||
},
|
||||
"process_after": {
|
||||
"title": "Process_After",
|
||||
"type": "string",
|
||||
"type": "number",
|
||||
"format": "time-delta",
|
||||
},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -0,0 +1,36 @@
|
|||
from starlette.testclient import TestClient
|
||||
|
||||
from path_operation_advanced_configuration.tutorial001 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
openapi_schema = {
|
||||
"openapi": "3.0.2",
|
||||
"info": {"title": "Fast API", "version": "0.1.0"},
|
||||
"paths": {
|
||||
"/items/": {
|
||||
"get": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
}
|
||||
},
|
||||
"summary": "Read Items Get",
|
||||
"operationId": "some_specific_id_you_define",
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
def test_get():
|
||||
response = client.get("/items/")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == [{"item_id": "Foo"}]
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
from starlette.testclient import TestClient
|
||||
|
||||
from path_operation_advanced_configuration.tutorial002 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
openapi_schema = {
|
||||
"openapi": "3.0.2",
|
||||
"info": {"title": "Fast API", "version": "0.1.0"},
|
||||
"paths": {},
|
||||
}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
def test_get():
|
||||
response = client.get("/items/")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == [{"item_id": "Foo"}]
|
||||
|
|
@ -0,0 +1,112 @@
|
|||
from starlette.testclient import TestClient
|
||||
|
||||
from path_operation_configuration.tutorial005 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
openapi_schema = {
|
||||
"openapi": "3.0.2",
|
||||
"info": {"title": "Fast API", "version": "0.1.0"},
|
||||
"paths": {
|
||||
"/items/": {
|
||||
"post": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "The created item",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Item"}
|
||||
}
|
||||
},
|
||||
},
|
||||
"422": {
|
||||
"description": "Validation Error",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/HTTPValidationError"
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
"summary": "Create an item",
|
||||
"description": "\n Create an item with all the information:\n \n * name: each item must have a name\n * description: a long description\n * price: required\n * tax: if the item doesn't have tax, you can omit this\n * tags: a set of unique tag strings for this item\n ",
|
||||
"operationId": "create_item_items__post",
|
||||
"requestBody": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Item"}
|
||||
}
|
||||
},
|
||||
"required": True,
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"Item": {
|
||||
"title": "Item",
|
||||
"required": ["name", "price"],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
"price": {"title": "Price", "type": "number"},
|
||||
"description": {"title": "Description", "type": "string"},
|
||||
"tax": {"title": "Tax", "type": "number"},
|
||||
"tags": {
|
||||
"title": "Tags",
|
||||
"uniqueItems": True,
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"default": [],
|
||||
},
|
||||
},
|
||||
},
|
||||
"ValidationError": {
|
||||
"title": "ValidationError",
|
||||
"required": ["loc", "msg", "type"],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"loc": {
|
||||
"title": "Location",
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"msg": {"title": "Message", "type": "string"},
|
||||
"type": {"title": "Error Type", "type": "string"},
|
||||
},
|
||||
},
|
||||
"HTTPValidationError": {
|
||||
"title": "HTTPValidationError",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"detail": {
|
||||
"title": "Detail",
|
||||
"type": "array",
|
||||
"items": {"$ref": "#/components/schemas/ValidationError"},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
def test_query_params_str_validations():
|
||||
response = client.post("/items/", json={"name": "Foo", "price": 42})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
"name": "Foo",
|
||||
"price": 42,
|
||||
"description": None,
|
||||
"tax": None,
|
||||
"tags": [],
|
||||
}
|
||||
|
|
@ -0,0 +1,73 @@
|
|||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from path_operation_configuration.tutorial006 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
openapi_schema = {
|
||||
"openapi": "3.0.2",
|
||||
"info": {"title": "Fast API", "version": "0.1.0"},
|
||||
"paths": {
|
||||
"/items/": {
|
||||
"get": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
}
|
||||
},
|
||||
"tags": ["items"],
|
||||
"summary": "Read Items Get",
|
||||
"operationId": "read_items_items__get",
|
||||
}
|
||||
},
|
||||
"/users/": {
|
||||
"get": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
}
|
||||
},
|
||||
"tags": ["users"],
|
||||
"summary": "Read Users Get",
|
||||
"operationId": "read_users_users__get",
|
||||
}
|
||||
},
|
||||
"/elements/": {
|
||||
"get": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
}
|
||||
},
|
||||
"tags": ["items"],
|
||||
"summary": "Read Elements Get",
|
||||
"operationId": "read_elements_elements__get",
|
||||
"deprecated": True,
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path,expected_status,expected_response",
|
||||
[
|
||||
("/items/", 200, [{"name": "Foo", "price": 42}]),
|
||||
("/users/", 200, [{"username": "johndoe"}]),
|
||||
("/elements/", 200, [{"item_id": "Foo"}]),
|
||||
],
|
||||
)
|
||||
def test_query_params_str_validations(path, expected_status, expected_response):
|
||||
response = client.get(path)
|
||||
assert response.status_code == expected_status
|
||||
assert response.json() == expected_response
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from query_params_str_validations.tutorial010 import app
|
||||
|
|
@ -80,7 +81,42 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
regex_error = {
|
||||
"detail": [
|
||||
{
|
||||
"ctx": {"pattern": "^fixedquery$"},
|
||||
"loc": ["query", "item-query"],
|
||||
"msg": 'string does not match regex "^fixedquery$"',
|
||||
"type": "value_error.str.regex",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"q_name,q,expected_status,expected_response",
|
||||
[
|
||||
(None, None, 200, {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}),
|
||||
(
|
||||
"item-query",
|
||||
"fixedquery",
|
||||
200,
|
||||
{"items": [{"item_id": "Foo"}, {"item_id": "Bar"}], "q": "fixedquery"},
|
||||
),
|
||||
("q", "fixedquery", 200, {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}),
|
||||
("item-query", "nonregexquery", 422, regex_error),
|
||||
],
|
||||
)
|
||||
def test_query_params_str_validations(q_name, q, expected_status, expected_response):
|
||||
url = "/items/"
|
||||
if q_name and q:
|
||||
url = f"{url}?{q_name}={q}"
|
||||
response = client.get(url)
|
||||
assert response.status_code == expected_status
|
||||
assert response.json() == expected_response
|
||||
|
|
|
|||
|
|
@ -81,7 +81,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -81,7 +81,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -82,7 +82,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -96,7 +96,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ openapi_schema = {
|
|||
}
|
||||
|
||||
|
||||
def test_openapi_scheme():
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
|
|
|||
|
|
@ -0,0 +1,167 @@
|
|||
from starlette.testclient import TestClient
|
||||
|
||||
from security.tutorial003 import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
openapi_schema = {
|
||||
"openapi": "3.0.2",
|
||||
"info": {"title": "Fast API", "version": "0.1.0"},
|
||||
"paths": {
|
||||
"/token": {
|
||||
"post": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
},
|
||||
"422": {
|
||||
"description": "Validation Error",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/HTTPValidationError"
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
"summary": "Login Post",
|
||||
"operationId": "login_token_post",
|
||||
"requestBody": {
|
||||
"content": {
|
||||
"application/x-www-form-urlencoded": {
|
||||
"schema": {"$ref": "#/components/schemas/Body_login"}
|
||||
}
|
||||
},
|
||||
"required": True,
|
||||
},
|
||||
}
|
||||
},
|
||||
"/users/me": {
|
||||
"get": {
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Successful Response",
|
||||
"content": {"application/json": {"schema": {}}},
|
||||
}
|
||||
},
|
||||
"summary": "Read Users Me Get",
|
||||
"operationId": "read_users_me_users_me_get",
|
||||
"security": [{"OAuth2PasswordBearer": []}],
|
||||
}
|
||||
},
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"Body_login": {
|
||||
"title": "Body_login",
|
||||
"required": ["username", "password"],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"grant_type": {
|
||||
"title": "Grant_Type",
|
||||
"pattern": "password",
|
||||
"type": "string",
|
||||
},
|
||||
"username": {"title": "Username", "type": "string"},
|
||||
"password": {"title": "Password", "type": "string"},
|
||||
"scope": {"title": "Scope", "type": "string", "default": ""},
|
||||
"client_id": {"title": "Client_Id", "type": "string"},
|
||||
"client_secret": {"title": "Client_Secret", "type": "string"},
|
||||
},
|
||||
},
|
||||
"ValidationError": {
|
||||
"title": "ValidationError",
|
||||
"required": ["loc", "msg", "type"],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"loc": {
|
||||
"title": "Location",
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"msg": {"title": "Message", "type": "string"},
|
||||
"type": {"title": "Error Type", "type": "string"},
|
||||
},
|
||||
},
|
||||
"HTTPValidationError": {
|
||||
"title": "HTTPValidationError",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"detail": {
|
||||
"title": "Detail",
|
||||
"type": "array",
|
||||
"items": {"$ref": "#/components/schemas/ValidationError"},
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
"securitySchemes": {
|
||||
"OAuth2PasswordBearer": {
|
||||
"type": "oauth2",
|
||||
"flows": {"password": {"scopes": {}, "tokenUrl": "/token"}},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_openapi_schema():
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == openapi_schema
|
||||
|
||||
|
||||
def test_login():
|
||||
response = client.post("/token", data={"username": "johndoe", "password": "secret"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"access_token": "johndoe", "token_type": "bearer"}
|
||||
|
||||
|
||||
def test_login_incorrect_password():
|
||||
response = client.post("/token", data={"username": "johndoe", "password": "incorrect"})
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Incorrect username or password"}
|
||||
|
||||
|
||||
def test_login_incorrect_username():
|
||||
response = client.post("/token", data={"username": "foo", "password": "secret"})
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Incorrect username or password"}
|
||||
|
||||
|
||||
def test_no_token():
|
||||
response = client.get("/users/me")
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "Not authenticated"}
|
||||
|
||||
|
||||
def test_token():
|
||||
response = client.get("/users/me", headers={"Authorization": "Bearer johndoe"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "fakehashedsecret",
|
||||
"disabled": False,
|
||||
}
|
||||
|
||||
|
||||
def test_incorrect_token():
|
||||
response = client.get("/users/me", headers={"Authorization": "Bearer nonexistent"})
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Invalid authentication credentials"}
|
||||
|
||||
def test_incorrect_token_type():
|
||||
response = client.get(
|
||||
"/users/me", headers={"Authorization": "Notexistent testtoken"}
|
||||
)
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "Not authenticated"}
|
||||
|
||||
def test_inactive_user():
|
||||
response = client.get("/users/me", headers={"Authorization": "Bearer alice"})
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Inactive user"}
|
||||
Loading…
Reference in New Issue