from base64 import b64encode
from typing import Optional

from fastapi import FastAPI, Security
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.testclient import TestClient
from requests.auth import HTTPBasicAuth

app = FastAPI()

            

Reported by Pylint.

              

@app.get("/users/me")
def read_current_user(credentials: Optional[HTTPBasicCredentials] = Security(security)):
    if credentials is None:
        return {"msg": "Create an account first"}
    return {"username": credentials.username, "password": credentials.password}



            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


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

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


            

Reported by Pylint.

              

def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}



            

Reported by Bandit.

              def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():
    response = client.get("/users/me")

            

Reported by Bandit.

                  auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():
    response = client.get("/users/me")
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():
    response = client.get("/users/me")
    assert response.status_code == 200, response.text
    assert response.json() == {"msg": "Create an account first"}



            

Reported by Pylint.

              from base64 import b64encode

from fastapi.testclient import TestClient
from requests.auth import HTTPBasicAuth

from docs_src.security.tutorial006 import app

client = TestClient(app)


            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


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

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


            

Reported by Pylint.

              

def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}



            

Reported by Bandit.

              def test_security_http_basic():
    auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():
    response = client.get("/users/me")

            

Reported by Bandit.

                  auth = HTTPBasicAuth(username="john", password="secret")
    response = client.get("/users/me", auth=auth)
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():
    response = client.get("/users/me")
    assert response.json() == {"detail": "Not authenticated"}

            

Reported by Bandit.

                  assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():
    response = client.get("/users/me")
    assert response.json() == {"detail": "Not authenticated"}
    assert response.status_code == 401, response.text
    assert response.headers["WWW-Authenticate"] == "Basic"


            

Reported by Pylint.

              
def test_security_http_basic_no_credentials():
    response = client.get("/users/me")
    assert response.json() == {"detail": "Not authenticated"}
    assert response.status_code == 401, response.text
    assert response.headers["WWW-Authenticate"] == "Basic"


def test_security_http_basic_invalid_credentials():

            

Reported by Bandit.

              import time
from typing import List

from fastapi import Depends, FastAPI, HTTPException

from . import crud, database, models, schemas
from .database import db_state_default

database.db.connect()

            

Reported by Pylint.

              
from fastapi import Depends, FastAPI, HTTPException

from . import crud, database, models, schemas
from .database import db_state_default

database.db.connect()
database.db.create_tables([models.User, models.Item])
database.db.close()

            

Reported by Pylint.

              from fastapi import Depends, FastAPI, HTTPException

from . import crud, database, models, schemas
from .database import db_state_default

database.db.connect()
database.db.create_tables([models.User, models.Item])
database.db.close()


            

Reported by Pylint.

              

async def reset_db_state():
    database.db._state._state.set(db_state_default.copy())
    database.db._state.reset()


def get_db(db_state=Depends(reset_db_state)):
    try:

            

Reported by Pylint.

              

async def reset_db_state():
    database.db._state._state.set(db_state_default.copy())
    database.db._state.reset()


def get_db(db_state=Depends(reset_db_state)):
    try:

            

Reported by Pylint.

              
async def reset_db_state():
    database.db._state._state.set(db_state_default.copy())
    database.db._state.reset()


def get_db(db_state=Depends(reset_db_state)):
    try:
        database.db.connect()

            

Reported by Pylint.

                  database.db._state.reset()


def get_db(db_state=Depends(reset_db_state)):
    try:
        database.db.connect()
        yield
    finally:
        if not database.db.is_closed():

            

Reported by Pylint.

                  "/slowusers/", response_model=List[schemas.User], dependencies=[Depends(get_db)]
)
def read_slow_users(skip: int = 0, limit: int = 100):
    global sleep_time
    sleep_time = max(0, sleep_time - 1)
    time.sleep(sleep_time)  # Fake long processing request
    users = crud.get_users(skip=skip, limit=limit)
    return users

            

Reported by Pylint.

              import time
from typing import List

from fastapi import Depends, FastAPI, HTTPException

from . import crud, database, models, schemas
from .database import db_state_default

database.db.connect()

            

Reported by Pylint.

              
app = FastAPI()

sleep_time = 10


async def reset_db_state():
    database.db._state._state.set(db_state_default.copy())
    database.db._state.reset()

            

Reported by Pylint.

              from fastapi import APIRouter, Depends, FastAPI, WebSocket
from fastapi.testclient import TestClient

router = APIRouter()
prefix_router = APIRouter()
app = FastAPI()


@app.websocket_route("/")

            

Reported by Pylint.

              

@app.websocket_route("/")
async def index(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Hello, world!")
    await websocket.close()



            

Reported by Pylint.

              

@router.websocket_route("/router")
async def routerindex(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Hello, router!")
    await websocket.close()



            

Reported by Pylint.

              

@prefix_router.websocket_route("/")
async def routerprefixindex(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Hello, router with prefix!")
    await websocket.close()



            

Reported by Pylint.

              

@router.websocket("/router2")
async def routerindex2(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Hello, router!")
    await websocket.close()



            

Reported by Pylint.

                  await websocket.close()


async def ws_dependency():
    return "Socket Dependency"


@router.websocket("/router-ws-depends/")
async def router_ws_decorator_depends(

            

Reported by Pylint.

              

@router.websocket("/router-ws-depends/")
async def router_ws_decorator_depends(
    websocket: WebSocket, data=Depends(ws_dependency)
):
    await websocket.accept()
    await websocket.send_text(data)
    await websocket.close()

            

Reported by Pylint.

              app.include_router(prefix_router, prefix="/prefix")


def test_app():
    client = TestClient(app)
    with client.websocket_connect("/") as websocket:
        data = websocket.receive_text()
        assert data == "Hello, world!"


            

Reported by Pylint.

                  client = TestClient(app)
    with client.websocket_connect("/") as websocket:
        data = websocket.receive_text()
        assert data == "Hello, world!"


def test_router():
    client = TestClient(app)
    with client.websocket_connect("/router") as websocket:

            

Reported by Bandit.

                      assert data == "Hello, world!"


def test_router():
    client = TestClient(app)
    with client.websocket_connect("/router") as websocket:
        data = websocket.receive_text()
        assert data == "Hello, router!"


            

Reported by Pylint.

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

app = FastAPI()
router = APIRouter()


@router.get("/a", responses={501: {"description": "Error 1"}})
async def a():

            

Reported by Pylint.

              

@router.get("/a", responses={501: {"description": "Error 1"}})
async def a():
    return "a"


@router.get(
    "/b",

            

Reported by Pylint.

              

@router.get("/a", responses={501: {"description": "Error 1"}})
async def a():
    return "a"


@router.get(
    "/b",

            

Reported by Pylint.

                  responses={
        502: {"description": "Error 2"},
        "4XX": {"description": "Error with range, upper"},
    },
)
async def b():
    return "b"



            

Reported by Pylint.

                  responses={
        502: {"description": "Error 2"},
        "4XX": {"description": "Error with range, upper"},
    },
)
async def b():
    return "b"



            

Reported by Pylint.

                      "400": {"description": "Error with str"},
        "5xx": {"description": "Error with range, lower"},
        "default": {"description": "A default response"},
    },
)
async def c():
    return "c"



            

Reported by Pylint.

                      "400": {"description": "Error with str"},
        "5xx": {"description": "Error with range, lower"},
        "default": {"description": "A default response"},
    },
)
async def c():
    return "c"



            

Reported by Pylint.

              client = TestClient(app)


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


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

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_a():
    response = client.get("/a")
    assert response.status_code == 200, response.text

            

Reported by Bandit.

              import inspect

from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html


def test_strings_in_generated_swagger():
    sig = inspect.signature(get_swagger_ui_html)
    swagger_js_url = sig.parameters.get("swagger_js_url").default  # type: ignore
    swagger_css_url = sig.parameters.get("swagger_css_url").default  # type: ignore

            

Reported by Pylint.

              from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html


def test_strings_in_generated_swagger():
    sig = inspect.signature(get_swagger_ui_html)
    swagger_js_url = sig.parameters.get("swagger_js_url").default  # type: ignore
    swagger_css_url = sig.parameters.get("swagger_css_url").default  # type: ignore
    swagger_favicon_url = sig.parameters.get("swagger_favicon_url").default  # type: ignore
    html = get_swagger_ui_html(openapi_url="/docs", title="title")

            

Reported by Pylint.

                  swagger_favicon_url = sig.parameters.get("swagger_favicon_url").default  # type: ignore
    html = get_swagger_ui_html(openapi_url="/docs", title="title")
    body_content = html.body.decode()
    assert swagger_js_url in body_content
    assert swagger_css_url in body_content
    assert swagger_favicon_url in body_content


def test_strings_in_custom_swagger():

            

Reported by Bandit.

                  html = get_swagger_ui_html(openapi_url="/docs", title="title")
    body_content = html.body.decode()
    assert swagger_js_url in body_content
    assert swagger_css_url in body_content
    assert swagger_favicon_url in body_content


def test_strings_in_custom_swagger():
    swagger_js_url = "swagger_fake_file.js"

            

Reported by Bandit.

                  body_content = html.body.decode()
    assert swagger_js_url in body_content
    assert swagger_css_url in body_content
    assert swagger_favicon_url in body_content


def test_strings_in_custom_swagger():
    swagger_js_url = "swagger_fake_file.js"
    swagger_css_url = "swagger_fake_file.css"

            

Reported by Bandit.

                  assert swagger_favicon_url in body_content


def test_strings_in_custom_swagger():
    swagger_js_url = "swagger_fake_file.js"
    swagger_css_url = "swagger_fake_file.css"
    swagger_favicon_url = "swagger_fake_file.png"
    html = get_swagger_ui_html(
        openapi_url="/docs",

            

Reported by Pylint.

                      swagger_favicon_url=swagger_favicon_url,
    )
    body_content = html.body.decode()
    assert swagger_js_url in body_content
    assert swagger_css_url in body_content
    assert swagger_favicon_url in body_content


def test_strings_in_generated_redoc():

            

Reported by Bandit.

                  )
    body_content = html.body.decode()
    assert swagger_js_url in body_content
    assert swagger_css_url in body_content
    assert swagger_favicon_url in body_content


def test_strings_in_generated_redoc():
    sig = inspect.signature(get_redoc_html)

            

Reported by Bandit.

                  body_content = html.body.decode()
    assert swagger_js_url in body_content
    assert swagger_css_url in body_content
    assert swagger_favicon_url in body_content


def test_strings_in_generated_redoc():
    sig = inspect.signature(get_redoc_html)
    redoc_js_url = sig.parameters.get("redoc_js_url").default  # type: ignore

            

Reported by Bandit.

                  assert swagger_favicon_url in body_content


def test_strings_in_generated_redoc():
    sig = inspect.signature(get_redoc_html)
    redoc_js_url = sig.parameters.get("redoc_js_url").default  # type: ignore
    redoc_favicon_url = sig.parameters.get("redoc_favicon_url").default  # type: ignore
    html = get_redoc_html(openapi_url="/docs", title="title")
    body_content = html.body.decode()

            

Reported by Pylint.

              from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None


            

Reported by Pylint.

              from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None


            

Reported by Pylint.

              from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None


class ItemCreate(ItemBase):

            

Reported by Pylint.

              from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None


class ItemCreate(ItemBase):

            

Reported by Pylint.

                  description: Optional[str] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int

            

Reported by Pylint.

                  description: Optional[str] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int

            

Reported by Pylint.

                  pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

            

Reported by Pylint.

                  pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

            

Reported by Pylint.

                  id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str

            

Reported by Pylint.

                  id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str

            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(openIdConnectUrl="/openid", auto_error=False)


            

Reported by Pylint.

              from typing import Optional

from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


            

Reported by Pylint.

              oid = OpenIdConnect(openIdConnectUrl="/openid", auto_error=False)


class User(BaseModel):
    username: str


def get_current_user(oauth_header: Optional[str] = Security(oid)):
    if oauth_header is None:

            

Reported by Pylint.

              oid = OpenIdConnect(openIdConnectUrl="/openid", auto_error=False)


class User(BaseModel):
    username: str


def get_current_user(oauth_header: Optional[str] = Security(oid)):
    if oauth_header is None:

            

Reported by Pylint.

                  username: str


def get_current_user(oauth_header: Optional[str] = Security(oid)):
    if oauth_header is None:
        return None
    user = User(username=oauth_header)
    return user


            

Reported by Pylint.

              

@app.get("/users/me")
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
    if current_user is None:
        return {"msg": "Create an account first"}
    return current_user



            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "Bearer footokenbar"}



            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(openIdConnectUrl="/openid")


            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(openIdConnectUrl="/openid")


            

Reported by Pylint.

              oid = OpenIdConnect(openIdConnectUrl="/openid")


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)

            

Reported by Pylint.

              oid = OpenIdConnect(openIdConnectUrl="/openid")


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)

            

Reported by Pylint.

                  username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)
    return user


@app.get("/users/me")

            

Reported by Pylint.

              

@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user


client = TestClient(app)


            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "Bearer footokenbar"}



            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(
    openIdConnectUrl="/openid", description="OpenIdConnect security scheme"

            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(
    openIdConnectUrl="/openid", description="OpenIdConnect security scheme"

            

Reported by Pylint.

              )


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)

            

Reported by Pylint.

              )


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)

            

Reported by Pylint.

                  username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)
    return user


@app.get("/users/me")

            

Reported by Pylint.

              

@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user


client = TestClient(app)


            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "Bearer footokenbar"}



            

Reported by Pylint.