Support python-multipart 0.0.12 (#2708)

Author: Kludex

pyproject.toml

@@ -100,7 +100,6 @@ source_pkgs = ["starlette", "tests"]
 [tool.coverage.report]
 exclude_lines = [
     "pragma: no cover",
-    "pragma: nocover",
     "if typing.TYPE_CHECKING:",
     "@typing.overload",
     "raise NotImplementedError",

starlette/applications.py

@@ -113,7 +113,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         await self.middleware_stack(scope, receive, send)
 
     def on_event(self, event_type: str) -> typing.Callable:  # type: ignore[type-arg]
-        return self.router.on_event(event_type)  # pragma: nocover
+        return self.router.on_event(event_type)  # pragma: no cover
 
     def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:
         self.router.mount(path, app=app, name=name)  # pragma: no cover

starlette/formparsers.py

@@ -11,9 +11,12 @@
 try:
     import multipart
     from multipart.multipart import parse_options_header
-except ModuleNotFoundError:  # pragma: nocover
-    parse_options_header = None
-    multipart = None
+except ModuleNotFoundError:  # pragma: no cover
+    parse_options_header = None  # type: ignore
+    multipart = None  # type: ignore
+
+if typing.TYPE_CHECKING:
+    from multipart.multipart import MultipartCallbacks, QuerystringCallbacks
 
 
 class FormMessage(Enum):
@@ -74,7 +77,7 @@ def on_end(self) -> None:
 
     async def parse(self) -> FormData:
         # Callbacks dictionary.
-        callbacks = {
+        callbacks: QuerystringCallbacks = {
             "on_field_start": self.on_field_start,
             "on_field_name": self.on_field_name,
             "on_field_data": self.on_field_data,
@@ -220,7 +223,7 @@ async def parse(self) -> FormData:
             raise MultiPartException("Missing boundary in multipart.")
 
         # Callbacks dictionary.
-        callbacks = {
+        callbacks: MultipartCallbacks = {
             "on_part_begin": self.on_part_begin,
             "on_part_data": self.on_part_data,
             "on_part_end": self.on_part_end,

starlette/requests.py

@@ -14,8 +14,8 @@
 
 try:
     from multipart.multipart import parse_options_header
-except ModuleNotFoundError:  # pragma: nocover
-    parse_options_header = None
+except ModuleNotFoundError:  # pragma: no cover
+    parse_options_header = None  # type: ignore
 
 
 if typing.TYPE_CHECKING:

starlette/routing.py

@@ -767,7 +767,7 @@ async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
     def __eq__(self, other: typing.Any) -> bool:
         return isinstance(other, Router) and self.routes == other.routes
 
-    def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:  # pragma: nocover
+    def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:  # pragma: no cover
         route = Mount(path, app=app, name=name)
         self.routes.append(route)
 
@@ -782,7 +782,7 @@ def add_route(
         methods: list[str] | None = None,
         name: str | None = None,
         include_in_schema: bool = True,
-    ) -> None:  # pragma: nocover
+    ) -> None:  # pragma: no cover
         route = Route(
             path,
             endpoint=endpoint,

starlette/schemas.py

@@ -10,7 +10,7 @@
 
 try:
     import yaml
-except ModuleNotFoundError:  # pragma: nocover
+except ModuleNotFoundError:  # pragma: no cover
     yaml = None  # type: ignore[assignment]
 
 

starlette/templating.py

@@ -19,9 +19,9 @@
     # adding a type ignore for mypy to let us access an attribute that may not exist
     if hasattr(jinja2, "pass_context"):
         pass_context = jinja2.pass_context
-    else:  # pragma: nocover
+    else:  # pragma: no cover
         pass_context = jinja2.contextfunction  # type: ignore[attr-defined]
-except ModuleNotFoundError:  # pragma: nocover
+except ModuleNotFoundError:  # pragma: no cover
     jinja2 = None  # type: ignore[assignment]
 
 

tests/middleware/test_errors.py

@@ -77,7 +77,7 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None:
     with pytest.raises(RuntimeError):
         client = test_client_factory(app)
         with client.websocket_connect("/"):
-            pass  # pragma: nocover
+            pass  # pragma: no cover
 
 
 def test_background_task(test_client_factory: TestClientFactory) -> None:

tests/test_authentication.py

@@ -211,7 +211,7 @@ def test_invalid_decorator_usage() -> None:
 
         @requires("authenticated")
         def foo() -> None:
-            pass  # pragma: nocover
+            pass  # pragma: no cover
 
 
 def test_user_interface(test_client_factory: TestClientFactory) -> None:
@@ -281,23 +281,23 @@ def test_websocket_authentication_required(
     with test_client_factory(app) as client:
         with pytest.raises(WebSocketDisconnect):
             with client.websocket_connect("/ws"):
-                pass  # pragma: nocover
+                pass  # pragma: no cover
 
         with pytest.raises(WebSocketDisconnect):
             with client.websocket_connect("/ws", headers={"Authorization": "basic foobar"}):
-                pass  # pragma: nocover
+                pass  # pragma: no cover
 
         with client.websocket_connect("/ws", auth=("tomchristie", "example")) as websocket:
             data = websocket.receive_json()
             assert data == {"authenticated": True, "user": "tomchristie"}
 
         with pytest.raises(WebSocketDisconnect):
             with client.websocket_connect("/ws/decorated"):
-                pass  # pragma: nocover
+                pass  # pragma: no cover
 
         with pytest.raises(WebSocketDisconnect):
             with client.websocket_connect("/ws/decorated", headers={"Authorization": "basic foobar"}):
-                pass  # pragma: nocover
+                pass  # pragma: no cover
 
         with client.websocket_connect("/ws/decorated", auth=("tomchristie", "example")) as websocket:
             data = websocket.receive_json()

tests/test_concurrency.py

@@ -22,8 +22,8 @@ async def task1() -> None:
 
     async def task2() -> None:
         await task1_finished.wait()
-        await anyio.sleep(0)  # pragma: nocover
-        task2_finished.set()  # pragma: nocover
+        await anyio.sleep(0)  # pragma: no cover
+        task2_finished.set()  # pragma: no cover
 
     await run_until_first_complete((task1, {}), (task2, {}))
     assert task1_finished.is_set()