app = FastAPI()

security = HTTPBasic(realm="simple")


@app.get("/users/me")
def read_current_user(credentials: HTTPBasicCredentials = Security(security)):
    return {"username": credentials.username, "password": credentials.password}


client = TestClient(app)


def test_security_http_basic():
    response = client.get("/users/me", auth=("john", "secret"))

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

router_a = APIRouter(route_class=APIRouteA)
router_b = APIRouter(route_class=APIRouteB)
router_c = APIRouter(route_class=APIRouteC)


@router_a.get("/")
def get_a():
    return {"msg": "A"}


@router_b.get("/")
def get_b():
    return {"msg": "B"}


@router_c.get("/")
def get_c():
    return {"msg": "C"}


router_b.include_router(router=router_c, prefix="/c")
router_a.include_router(router=router_b, prefix="/b")

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(

swagger_ui_init_oauth = {"clientId": "the-foo-clients", "appName": "The Predendapp"}

app = FastAPI(swagger_ui_init_oauth=swagger_ui_init_oauth)


@app.get("/items/")
async def read_items():
    return {"id": "foo"}


client = TestClient(app)


def test_swagger_ui():
    response = client.get("/docs")
    assert response.status_code == 200, response.text
    print(response.text)
    assert "ui.initOAuth" in response.text

app = FastAPI()


@app.get("/", openapi_extra={"x-custom-extension": "value"})
def route_with_extras():
    return {}


client = TestClient(app)


def test_get_route():
    response = client.get("/")
    assert response.status_code == 200, response.text
    assert response.json() == {}


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

from fastapi import FastAPI, File
from fastapi.testclient import TestClient

app = FastAPI()


@app.post("/files")
async def upload_files(files: list[bytes] | None = File(None)):
    if files is None:
        return {"files_count": 0}
    return {"files_count": len(files), "sizes": [len(f) for f in files]}


def test_optional_bytes_list():
    client = TestClient(app)
    response = client.post(
        "/files",

from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)

from fastapi.testclient import TestClient
from inline_snapshot import snapshot

from .utils import needs_py310


def get_client():
    app = FastAPI()
    with pytest.warns(FastAPIDeprecationWarning):

        @app.get("/items/")
        async def read_items(
            q: Annotated[str | None, Query(regex="^fixedquery$")] = None,
        ):
            if q:
                return f"Hello {q}"

    }

    def "#member change should not be reported"() {
        expect:
        noViolation(rule)

        where:
        member << [Mock(JApiMethod), Mock(JApiField), Mock(JApiImplementedInterface), Mock(JApiConstructor)]
    }

    def "nothing should be reported if no changes"() {
        expect:
        noViolation(rule)
    }