diff --git a/docs/changelog.rst b/docs/changelog.rst index 1d5ede8..959f064 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -6,6 +6,10 @@ Changelog UNRELEASED ---------- +* Fix live server startup on POSIX with non-fork multiprocessing defaults, + including Python 3.14, without changing the application's start method + (`#206 `_). + * Add support for Python 3.10, 3.11, and 3.12. * Drop support for EOL Python 3.7. * Add type hints. diff --git a/src/pytest_flask/live_server.py b/src/pytest_flask/live_server.py index 5f2150e..d3ce382 100644 --- a/src/pytest_flask/live_server.py +++ b/src/pytest_flask/live_server.py @@ -1,7 +1,6 @@ import logging import multiprocessing import os -import platform import signal import socket import time @@ -25,8 +24,9 @@ def run( ) -> None: ... -# force 'fork' on macOS -if platform.system() == "Darwin": +# Flask applications and local worker functions cannot be pickled. +# Use a private context without changing the application's start method. +if "fork" in multiprocessing.get_all_start_methods(): multiprocessing = multiprocessing.get_context("fork") # type: ignore[assignment] diff --git a/tests/test_live_server.py b/tests/test_live_server.py index 3f365aa..3229079 100755 --- a/tests/test_live_server.py +++ b/tests/test_live_server.py @@ -1,3 +1,4 @@ +import multiprocessing import os import pytest @@ -77,6 +78,38 @@ def test_a(live_server): result.stdout.fnmatch_lines(["*passed*"]) assert result.ret == 0 + def test_non_fork_start_method(self, appdir): + appdir.create_test_module( + """ + import multiprocessing + from urllib.request import urlopen + + import pytest + + @pytest.mark.parametrize("start_method", [ + method for method in multiprocessing.get_all_start_methods() + if method != "fork" + ]) + def test_a(live_server, start_method): + multiprocessing.set_start_method(start_method, force=True) + message = "response from a local route" + + @live_server.app.route("/" + start_method, endpoint=start_method) + def closure(): + return message + + live_server.start() + url = live_server.url("/" + start_method) + with urlopen(url, timeout=5) as response: + assert response.read() == message.encode() + assert multiprocessing.get_start_method() == start_method + """ + ) + result = appdir.runpytest_subprocess( + "--no-start-live-server", "-o", "live_server_scope=function" + ) + result.assert_outcomes(passed=len(multiprocessing.get_all_start_methods()) - 1) + def test_stop_cleanly_join_exception(self, appdir, live_server, caplog): # timeout = 'a' here to force an exception when # attempting to self._process.join()