异步感知 mock/override 环境变量？

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))

异步感知 mock/override 环境变量？

async-aware mock/override environment variables?

python

mocking

async-await