import asyncio

import multiprocessing

import sys

from typing import AsyncGenerator

import pytest

import pytest_asyncio

from temporalio.client import Client

from temporalio.testing import WorkflowEnvironment

# Due to https://github.com/python/cpython/issues/77906, multiprocessing on

# macOS starting with Python 3.8 has changed from "fork" to "spawn". For

# pre-3.8, we are changing it for them.

if sys.version_info < (3, 8) and sys.platform.startswith("darwin"):

multiprocessing.set_start_method("spawn", True)

def pytest_addoption(parser):

parser.addoption(

"--workflow-environment",

default="local",

help="Which workflow environment to use ('local', 'time-skipping', or target to existing server)",

)

@pytest.fixture(scope="session")

def event_loop():

# See https://github.com/pytest-dev/pytest-asyncio/issues/68

# See https://github.com/pytest-dev/pytest-asyncio/issues/257

# Also need ProactorEventLoop on older versions of Python with Windows so

# that asyncio subprocess works properly

if sys.version_info < (3, 8) and sys.platform == "win32":

loop = asyncio.ProactorEventLoop()

else:

loop = asyncio.get_event_loop_policy().new_event_loop()

yield loop

loop.close()

@pytest_asyncio.fixture(scope="session")

async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]:

env_type = request.config.getoption("--workflow-environment")

if env_type == "local":

env = await WorkflowEnvironment.start_local()

elif env_type == "time-skipping":

env = await WorkflowEnvironment.start_time_skipping()

else:

env = WorkflowEnvironment.from_client(await Client.connect(env_type))

yield env

await env.shutdown()

@pytest_asyncio.fixture

async def client(env: WorkflowEnvironment) -> Client:

return env.client