Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/changelog.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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 <https://github.com/pytest-dev/pytest-flask/issues/206>`_).

* Add support for Python 3.10, 3.11, and 3.12.
* Drop support for EOL Python 3.7.
* Add type hints.
Expand Down
6 changes: 3 additions & 3 deletions src/pytest_flask/live_server.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import logging
import multiprocessing
import os
import platform
import signal
import socket
import time
Expand All @@ -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]


Expand Down
33 changes: 33 additions & 0 deletions tests/test_live_server.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import multiprocessing
import os

import pytest
Expand Down Expand Up @@ -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()
Expand Down