|
6 | 6 | from __future__ import annotations |
7 | 7 |
|
8 | 8 | from textwrap import dedent |
| 9 | +from typing import Literal |
9 | 10 |
|
| 11 | +import pytest |
10 | 12 | from pytest import Pytester |
11 | 13 |
|
12 | 14 | _prelude = dedent( |
@@ -213,3 +215,56 @@ async def test(same_var_fixture): |
213 | 215 | ) |
214 | 216 | result = pytester.runpytest("--asyncio-mode=strict") |
215 | 217 | result.assert_outcomes(passed=1) |
| 218 | + |
| 219 | + |
| 220 | +def test_no_isolation_against_context_changes_in_sync_tests(pytester: Pytester): |
| 221 | + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") |
| 222 | + pytester.makepyfile( |
| 223 | + dedent( |
| 224 | + """ |
| 225 | + import pytest |
| 226 | + import pytest_asyncio |
| 227 | + from contextvars import ContextVar |
| 228 | +
|
| 229 | + _context_var = ContextVar("my_var") |
| 230 | +
|
| 231 | + def test_sync(): |
| 232 | + _context_var.set("new_value") |
| 233 | +
|
| 234 | + @pytest.mark.asyncio |
| 235 | + async def test_async(): |
| 236 | + assert _context_var.get() == "new_value" |
| 237 | + """ |
| 238 | + ) |
| 239 | + ) |
| 240 | + result = pytester.runpytest("--asyncio-mode=strict") |
| 241 | + result.assert_outcomes(passed=2) |
| 242 | + |
| 243 | + |
| 244 | +@pytest.mark.parametrize("loop_scope", ("function", "module")) |
| 245 | +def test_isolation_against_context_changes_in_async_tests( |
| 246 | + pytester: Pytester, loop_scope: Literal["function", "module"] |
| 247 | +): |
| 248 | + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") |
| 249 | + pytester.makepyfile( |
| 250 | + dedent( |
| 251 | + f""" |
| 252 | + import pytest |
| 253 | + import pytest_asyncio |
| 254 | + from contextvars import ContextVar |
| 255 | +
|
| 256 | + _context_var = ContextVar("my_var") |
| 257 | +
|
| 258 | + @pytest.mark.asyncio(loop_scope="{loop_scope}") |
| 259 | + async def test_async_first(): |
| 260 | + _context_var.set("new_value") |
| 261 | +
|
| 262 | + @pytest.mark.asyncio(loop_scope="{loop_scope}") |
| 263 | + async def test_async_second(): |
| 264 | + with pytest.raises(LookupError): |
| 265 | + _context_var.get() |
| 266 | + """ |
| 267 | + ) |
| 268 | + ) |
| 269 | + result = pytester.runpytest("--asyncio-mode=strict") |
| 270 | + result.assert_outcomes(passed=2) |
0 commit comments