From 6c9e7f7d8942c470ec7c9aa3f7ebb50203bb2f62 Mon Sep 17 00:00:00 2001
From: Marcelo Trylesinski <marcelotryle@gmail.com>
Date: Mon, 9 May 2022 20:39:28 +0200
Subject: [PATCH] Add tests for `_subprocess.py` (#1477)

* Add tests for `_subprocess.py`

* Ignore coverage on windows

* Add another ignore
---
 tests/conftest.py            |  2 +-
 tests/protocols/test_http.py |  2 +-
 tests/test_subprocess.py     | 49 ++++++++++++++++++++++++++++++++++++
 uvicorn/protocols/utils.py   |  2 +-
 4 files changed, 52 insertions(+), 3 deletions(-)
 create mode 100644 tests/test_subprocess.py

diff --git a/tests/conftest.py b/tests/conftest.py
index d02eb14ab8..9c81a58b26 100644
--- a/tests/conftest.py
+++ b/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
diff --git a/tests/protocols/test_http.py b/tests/protocols/test_http.py
index 9dff4b0674..876fcf076d 100644
--- a/tests/protocols/test_http.py
+++ b/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
diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py
new file mode 100644
index 0000000000..330a98b2e3
--- /dev/null
+++ b/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()
diff --git a/uvicorn/protocols/utils.py b/uvicorn/protocols/utils.py
index 1da733600a..ce71dbb610 100644
--- a/uvicorn/protocols/utils.py
+++ b/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