}
}

val Project.sourceSets: SourceSetContainer
  get() = (this as ExtensionAware).extensions["sourceSets"] as SourceSetContainer

private val Project.kotlinOsgi: MinimalExternalModuleDependency
  get() =
    extensions
      .getByType(VersionCatalogsExtension::class.java)
      .named("libs")
      .findLibrary("kotlin.stdlib.osgi")
      .get()

@callback_router.get(
    "{$callback_url}/callback/", responses={400: {"model": CustomModel}}
)
def callback_route():
    pass  # pragma: no cover


@app.post("/", callbacks=callback_router.routes)
def main_route(callback_url: HttpUrl):
    pass  # pragma: no cover


client = TestClient(app)


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

        this.acceptedApiChanges = acceptedApiChanges ? AcceptedApiChanges.fromAcceptedChangesMap(acceptedApiChanges.get()) : [:]

        // Tests will not supply these
        this.mainApiChangesJsonFile = params.get("mainApiChangesJsonFile") ? new File(params.get("mainApiChangesJsonFile") as String) : null

def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")
    session.close()


def generate_stream(query: str):
    for ch in query:
        yield ch
        time.sleep(0.1)


@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):

    def get_db():
        call_counter["count"] += 1
        return f"db_{call_counter['count']}"

    def get_user(db: Annotated[str, Depends(get_db)]):
        return "user"

    app = FastAPI()

    @app.get("/")
    def endpoint(
        db: Annotated[str, Depends(get_db)],
        user: Annotated[str, Security(get_user, scopes=["read"])],
    ):
        return {"db": db}

    return app

):
    return {"dep1": dep1, "dep2": dep2}


app = FastAPI()


@app.get("/scopes")
def get_scopes(
    dep3: Annotated[dict[str, Any], Security(dep3, scopes=["scope3"])],
):
    return dep3


client = TestClient(app)


def test_security_scopes_dont_propagate():
    response = client.get("/scopes")
    assert response.status_code == 200
    assert response.json() == {

): Cache = Cache(fileSystem, file, maxSize)

internal var RealConnection.idleAtNsAccessor: Long
  get() = idleAtNs
  set(value) {
    idleAtNs = value
  }

internal val Response.exchangeAccessor: Exchange?
  get() = this.exchange

internal val Exchange.connectionAccessor: RealConnection
  get() = this.connection

class JsonApiError(BaseModel):
    errors: list[Error]


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


client = TestClient(app)


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

                new DefaultLookup(mockedPlexusContainer));

        assertEquals("clean", dl.getLifeCycles().get(0).getId());
        assertEquals("default", dl.getLifeCycles().get(1).getId());
        assertEquals("site", dl.getLifeCycles().get(2).getId());
        assertEquals("etl", dl.getLifeCycles().get(3).getId());
    }

    private Lifecycle getLifeCycleById(String id) {

    )
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):