From 0ca3d50170aa34a29567da2ee46b28514fc45a67 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebasti=C3=A1n=20Ram=C3=ADrez?= Date: Sat, 1 Nov 2025 15:02:39 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=85=20Add=20more=20tests=20for=20dependen?= =?UTF-8?q?cies=20with=20scopes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_dependency_yield_scope.py | 108 ++++++++++++++++++++++++++- 1 file changed, 107 insertions(+), 1 deletion(-) diff --git a/tests/test_dependency_yield_scope.py b/tests/test_dependency_yield_scope.py index e0bc7d5d5..108d812a7 100644 --- a/tests/test_dependency_yield_scope.py +++ b/tests/test_dependency_yield_scope.py @@ -1,7 +1,9 @@ import json -from typing import Any +from typing import Any, Tuple +import pytest from fastapi import Depends, FastAPI +from fastapi.exceptions import FastAPIError from fastapi.responses import StreamingResponse from fastapi.testclient import TestClient from typing_extensions import Annotated @@ -20,8 +22,47 @@ def dep_session() -> Any: SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] +SessionDefaultDep = Annotated[Session, Depends(dep_session)] +class NamedSession: + def __init__(self, name: str = "default") -> None: + self.name = name + self.open = True + + +def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: + assert session is session_b + named_session = NamedSession(name="named") + yield named_session, session_b + named_session.open = False + + +NamedSessionsDep = Annotated[Tuple[NamedSession, Session], Depends(get_named_session)] + + +def get_named_func_session(session: SessionFuncDep) -> Any: + named_session = NamedSession(name="named") + yield named_session, session + named_session.open = False + + +def get_named_regular_func_session(session: SessionFuncDep) -> Any: + named_session = NamedSession(name="named") + return named_session, session + + +BrokenSessionsDep = Annotated[ + Tuple[NamedSession, Session], Depends(get_named_func_session) +] +NamedSessionsFuncDep = Annotated[ + Tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") +] + +RegularSessionsDep = Annotated[ + Tuple[NamedSession, Session], Depends(get_named_regular_func_session) +] + app = FastAPI() @@ -53,6 +94,36 @@ def get_stream_session( return StreamingResponse(iter_data()) +@app.get("/sub") +def get_sub(sessions: NamedSessionsDep) -> Any: + def iter_data(): + yield json.dumps( + {"named_session_open": sessions[0].open, "session_open": sessions[1].open} + ) + + return StreamingResponse(iter_data()) + + +@app.get("/named-function-scope") +def get_named_function_scope(sessions: NamedSessionsFuncDep) -> Any: + def iter_data(): + yield json.dumps( + {"named_session_open": sessions[0].open, "session_open": sessions[1].open} + ) + + return StreamingResponse(iter_data()) + + +@app.get("/regular-function-scope") +def get_regular_function_scope(sessions: RegularSessionsDep) -> Any: + def iter_data(): + yield json.dumps( + {"named_session_open": sessions[0].open, "session_open": sessions[1].open} + ) + + return StreamingResponse(iter_data()) + + client = TestClient(app) @@ -76,3 +147,38 @@ def test_two_scopes() -> None: data = json.loads(response.content) assert data["func_is_open"] is False assert data["req_is_open"] is True + + +def test_sub() -> None: + response = client.get("/sub") + assert response.status_code == 200 + data = json.loads(response.content) + assert data["named_session_open"] is True + assert data["session_open"] is True + + +def test_broken_scope() -> None: + with pytest.raises( + FastAPIError, + match='The dependency "get_named_func_session" has a scope of "request", it cannot depend on dependencies with scope "function"', + ): + + @app.get("/broken-scope") + def get_broken(sessions: BrokenSessionsDep) -> Any: # pragma: no cover + pass + + +def test_named_function_scope() -> None: + response = client.get("/named-function-scope") + assert response.status_code == 200 + data = json.loads(response.content) + assert data["named_session_open"] is False + assert data["session_open"] is False + + +def test_regular_function_scope() -> None: + response = client.get("/regular-function-scope") + assert response.status_code == 200 + data = json.loads(response.content) + assert data["named_session_open"] is True + assert data["session_open"] is False