# Without aliases


@app.get("/required-str")
async def read_required_str(p: Annotated[str, Header()]):
    return {"p": p}


class HeaderModelRequiredStr(BaseModel):
    p: str


@app.get("/model-required-str")
async def read_model_required_str(p: Annotated[HeaderModelRequiredStr, Header()]):
    return {"p": p.p}


@pytest.mark.parametrize(

@app.get(
    "/a",
    response_class=Response,
    responses={500: {"description": "Error", "model": JsonApiError}},
)
async def a():
    pass  # pragma: no cover


@app.get("/b", responses={500: {"description": "Error", "model": Error}})
async def b():
    pass  # pragma: no cover


client = TestClient(app)


def test_openapi_schema():
    response = client.get("/openapi.json")

//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package dsync

//go:generate msgp -file $GOFILE

// LockArgs is minimal required values for any dsync compatible lock operation.
type LockArgs struct {
	// Unique ID of lock/unlock request.
	UID string

	// Resources contains single or multiple entries to be locked/unlocked.

/// tip | 提示

注意测试函数是普通的 `def`，不是 `async def`。

还有client的调用也是普通的调用，不是用 `await`。

这让你可以直接使用 `pytest` 而不会遇到麻烦。

///

/// note | 技术细节

你也可以用 `from starlette.testclient import TestClient`。

**FastAPI** 提供了和 `starlette.testclient` 一样的 `fastapi.testclient`，只是为了方便开发者。但它直接来自Starlette。

///

/// tip | 提示

# Without aliases


@app.post("/list-bytes", operation_id="list_bytes")
async def read_list_bytes(p: Annotated[list[bytes], File()]):
    return {"file_size": [len(file) for file in p]}


@app.post("/list-uploadfile", operation_id="list_uploadfile")
async def read_list_uploadfile(p: Annotated[list[UploadFile], File()]):
    return {"file_size": [file.size for file in p]}

# Alias


@app.get("/required-list-alias")
async def read_required_list_alias(p: Annotated[list[str], Header(alias="p_alias")]):
    return {"p": p}


class HeaderModelRequiredListAlias(BaseModel):
    p: list[str] = Field(alias="p_alias")


@app.get("/model-required-list-alias")
async def read_model_required_list_alias(
    p: Annotated[HeaderModelRequiredListAlias, Header()],
):

@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/")

from fastapi.testclient import TestClient

items = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}
    yield
    # clean up items
    items.clear()


app = FastAPI(lifespan=lifespan)


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]


def test_read_items():

@app.get(
    "/a",
    status_code=204,
    response_class=JsonApiResponse,
    responses={500: {"description": "Error", "model": JsonApiError}},
)
async def a():
    pass


@app.get("/b", responses={204: {"description": "No Content"}})
async def b():
    pass  # pragma: no cover


client = TestClient(app)


def test_get_response():
    response = client.get("/a")

        self.app = app
        self.max_content_size = max_content_size

    def receive_wrapper(self, receive):
        received = 0

        async def inner():
            nonlocal received
            message = await receive()
            if message["type"] != "http.request":
                return message  # pragma: no cover