Add tests for _subprocess.py

tests/conftest.py

@@ -172,7 +172,7 @@ def make_tmp_dir(base_dir):
             sock_path = str(tmpd / socket_filename)
             sock_path_len = len(sock_path.encode())
             if sock_path_len <= max_sock_len:
-                if max_sock_len - sock_path_len >= identifier_len:
+                if max_sock_len - sock_path_len >= identifier_len:  # pragma: no cover
                     sock_path = str(tmpd / "".join((identifier, socket_filename)))
                 yield sock_path
                 return

tests/protocols/test_http.py

@@ -778,7 +778,7 @@ def send_fragmented_req(path):
         # we skip the error on bsd systems if python is too slow
         try:
             sock.shutdown(socket.SHUT_RDWR)
-        except Exception:  # pragma: py-linux
+        except Exception:  # pragma: no cover
             pass
         sock.close()
         return resp

tests/test_subprocess.py

@@ -0,0 +1,49 @@
+import socket
+import sys
+from typing import List
+from unittest.mock import patch
+
+import pytest
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope
+
+from uvicorn._subprocess import SpawnProcess, get_subprocess, subprocess_started
+from uvicorn.config import Config
+
+
+def server_run(sockets: List[socket.socket]):  # pragma: no cover
+    ...
+
+
+async def app(
+    scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+) -> None:  # pragma: no cover
+    ...
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+def test_get_subprocess() -> None:  # pragma: py-win32
+    fdsock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    fd = fdsock.fileno()
+    config = Config(app=app, fd=fd)
+    config.load()
+
+    process = get_subprocess(config, server_run, [fdsock])
+    assert isinstance(process, SpawnProcess)
+
+    fdsock.close()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+def test_subprocess_started() -> None:  # pragma: py-win32
+    fdsock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    fd = fdsock.fileno()
+    config = Config(app=app, fd=fd)
+    config.load()
+
+    with patch("tests.test_subprocess.server_run") as mock_run:
+        with patch.object(config, "configure_logging") as mock_config_logging:
+            subprocess_started(config, server_run, [fdsock], None)
+            mock_run.assert_called_once()
+            mock_config_logging.assert_called_once()
+
+    fdsock.close()

uvicorn/protocols/utils.py

@@ -11,7 +11,7 @@ def get_remote_addr(transport: asyncio.Transport) -> Optional[Tuple[str, int]]:
         try:
             info = socket_info.getpeername()
             return (str(info[0]), int(info[1])) if isinstance(info, tuple) else None
-        except OSError:
+        except OSError:  # pragma: no cover
             # This case appears to inconsistently occur with uvloop
             # bound to a unix domain socket.
             return None