异步感知模拟/覆盖环境变量？

Question

是否有一种异步感知方式来模拟环境变量？

我想要的是一个上下文管理器，它可以让我模拟 with 块内的代码，以便在执行该代码时应用模拟，但在执行其他代码时不应用模拟。

类似这样，但通过了：

import pytest
import os
import asyncio
import mock


async def aaa():
    with mock.patch.dict(os.environ, {"FAVORITE_LETTER": "A"}, clear=True):
        for _ in range(50):
            assert os.environ["FAVORITE_LETTER"] == "A"
            await asyncio.sleep(0)


async def bbb():
    with mock.patch.dict(os.environ, {"FAVORITE_LETTER": "B"}, clear=True):
        for _ in range(50):
            assert os.environ["FAVORITE_LETTER"] == "B"
            await asyncio.sleep(0)


@pytest.mark.asyncio
async def test_mock_env():
    await asyncio.gather(aaa(), bbb())

目前失败：

    async def aaa():
        with mock.patch.dict(os.environ, {"FAVORITE_LETTER": "A"}, clear=True):
            for _ in range(50):
>               assert os.environ["FAVORITE_LETTER"] == "A"
E               AssertionError: assert 'B' == 'A'
E                 - A
E                 + B

test_mock_env.py:9: AssertionError

=======

编辑：

接受的答案解决了我提出的问题，但没有解决我的问题，因为我在制作示例时简化了太多。我的真正问题不会成为粘贴在这里的好例子，但这更接近我的实际情况：

from my_library import some_component_i_control
from someone_elses_library import not_my_code

async def wrap_not_my_code():
    with patch.dict(os.environ, {"ENV_VARS_NEEDED_FOR_COMPONENT_I_DONT_CONTROL": "A"}, clear=True):
        await not_my_code()  # this runs forever but does not block the event loop


@pytest.mark.asyncio
async def test_mock_env():
    task_for_not_my_code = asyncio.create_task(wrap_not_my_code())
    task_for_my_component = asyncio.create_task(some_component_i_control())
    await asyncio.wait([task_for_my_component, task_for_not_my_code], return_when=asyncio.FIRST_COMPLETED) 
    # ...
    # Make some assertions about the results

我正在尝试测试我控制的库与我不控制的库之间的交互，我无法控制的组件看到环境的修改版本。

实际上更糟糕的是：我需要多个我无法控制的组件的任务实例，它们都可以看到不同的环境，并且都与我自己的代码同时运行。这可能吗？

Answer 1

您将需要一些东西来允许对状态的独占访问，在这种情况下，asyncio.Lock 对象可以工作。这就需要你把原来的代码修改成如下。

import pytest
import os
import asyncio
from unittest.mock import patch


async def aaa(lock):
    for _ in range(50):
        async with lock:
            with patch.dict(os.environ, {"FAVORITE_LETTER": "A"}, clear=True):
                assert os.environ["FAVORITE_LETTER"] == "A"
        await asyncio.sleep(0)


async def bbb(lock):
    for _ in range(50):
        async with lock:
            with patch.dict(os.environ, {"FAVORITE_LETTER": "B"}, clear=True):
                assert os.environ["FAVORITE_LETTER"] == "B"
        await asyncio.sleep(0)


@pytest.mark.asyncio
async def test_mock_env():
    lock = asyncio.Lock()
    await asyncio.gather(aaa(lock), bbb(lock))