diff --git a/tests/test_config.py b/tests/test_config.py index b66de7e15..2208780d1 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,4 +1,5 @@ import os +import platform import socket import sys @@ -164,12 +165,18 @@ def test_config_file_descriptor(): assert config.fd == 1 +@pytest.mark.skipif( + sys.platform.startswith("win") or platform.python_implementation() == "PyPy", + reason="Skipping uds test on Windows", +) def test_config_rebind_socket(): sock = socket.socket() config = Config(asgi_app) - with pytest.raises(SystemExit) as pytest_wrapped_e: + try: sock.bind((config.host, config.port)) - config.bind_socket() - assert pytest_wrapped_e.type == SystemExit - assert pytest_wrapped_e.value.code == 1 - sock.close() + with pytest.raises(SystemExit) as pytest_wrapped_e: + config.bind_socket() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 + finally: + sock.close() diff --git a/tests/test_main.py b/tests/test_main.py index dcddf1536..0219e337c 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,4 +1,5 @@ import asyncio +import platform import socket import sys import threading @@ -127,7 +128,8 @@ def safe_run(): @pytest.mark.skipif( - sys.platform.startswith("win"), reason="Skipping uds test on Windows" + sys.platform.startswith("win") or platform.python_implementation() == "PyPy", + reason="Skipping uds test on Windows and pypy", ) def test_run_uds(tmp_path): class App: @@ -161,3 +163,45 @@ def install_signal_handlers(self): finally: sock_client.close() thread.join() + + +@pytest.mark.skipif( + sys.platform.startswith("win") or platform.python_implementation() == "PyPy", + reason="Skipping fd test on Windows and pypy", +) +def test_run_fd(tmp_path): + class App: + def __init__(self, scope): + if scope["type"] != "http": + raise Exception() + + async def __call__(self, receive, send): + await send({"type": "http.response.start", "status": 204, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + class CustomServer(Server): + def install_signal_handlers(self): + pass + + uds = str(tmp_path / "socket") + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + fd = sock.fileno() + sock.bind(uds) + config = Config(app=App, loop="asyncio", limit_max_requests=1, fd=fd) + server = CustomServer(config=config) + thread = threading.Thread(target=server.run) + thread.start() + while not server.started: + time.sleep(0.01) + data = b"GET / HTTP/1.1\r\n\r\n" + sock_client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + sock_client.connect(uds) + r = sock_client.sendall(data) + assert r is None + except Exception as e: + print(e) + finally: + sock_client.close() + sock.close() + thread.join()