Coverage for tests / test_stream_json_validation_error.py: 100%

1from collections.abc import AsyncIterable, Iterable 4 ctx1adbc

2 

3import pytest 4 ctx1adbc

4from fastapi import FastAPI 4 ctx1adbc

5from fastapi.exceptions import ResponseValidationError 4 ctx1adbc

6from fastapi.testclient import TestClient 4 ctx1adbc

7from pydantic import BaseModel 4 ctx1adbc

8 

9 

10class Item(BaseModel): 4 ctx1adbc

11 name: str 3 ctx1abc

12 price: float 3 ctx1abc

13 

14 

15app = FastAPI() 4 ctx1adbc

16 

17 

18@app.get("/items/stream-invalid") 4 ctx1adbc

19async def stream_items_invalid() -> AsyncIterable[Item]: 4 ctx1adbc

20 yield {"name": "valid", "price": 1.0} 4 ctx1efgh

21 yield {"name": "invalid", "price": "not-a-number"} 4 ctx1efgh

22 

23 

24@app.get("/items/stream-invalid-sync") 4 ctx1adbc

25def stream_items_invalid_sync() -> Iterable[Item]: 4 ctx1adbc

26 yield {"name": "valid", "price": 1.0} 4 ctx1ijkl

27 yield {"name": "invalid", "price": "not-a-number"} 4 ctx1ijkl

28 

29 

30client = TestClient(app) 4 ctx1adbc

31 

32 

33def test_stream_json_validation_error_async(): 4 ctx1adbc

34 with pytest.raises(ResponseValidationError): 4 ctx1efgh

35 client.get("/items/stream-invalid") 4 ctx1efgh

36 

37 

38def test_stream_json_validation_error_sync(): 4 ctx1adbc

39 with pytest.raises(ResponseValidationError): 4 ctx1ijkl

40 client.get("/items/stream-invalid-sync") 4 ctx1ijkl