### _clone_function_fixture Internal Usage Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Example showing internal usage of _clone_function_fixture during fixture setup to ensure independent fixture instances. ```python # Internal usage during fixture setup # Ensures each test in a group gets its own fixture instance original_fixture = pytest_session.config.pluginmanager.get_plugin("funcmanage").getfixturedefs("my_fixture", test_item) independent_fixture = _clone_function_fixture(original_fixture[0]) # Now each test can have different cached values ``` -------------------------------- ### setup method Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md No-op setup for the group node. Actual setup is handled by pytest_runtest_setup_async_group hook. ```python def setup(self) -> None ``` -------------------------------- ### Configuration File Options: pyproject.toml (Full Example) Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md A more comprehensive example of pyproject.toml configuration. ```toml [tool.pytest.ini_options] default_group_strategy = "parent" asyncio_mode = "strict" asyncio_default_fixture_loop_scope = "function" ``` -------------------------------- ### Valid Configuration Examples Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Examples of valid command-line arguments for setting the default group strategy. ```bash pytest --default-group-strategy self # Valid pytest --default-group-strategy parent # Valid ``` -------------------------------- ### AsyncioConcurrentGroup Initialization Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example of instantiating AsyncioConcurrentGroup. ```python # Generally created internally by the plugin, but can be instantiated as: group = AsyncioConcurrentGroup(parent=parent_node, originalname="AsyncioConcurrentGroup[my_group]") ``` -------------------------------- ### Example for _wrap_asyncgen_fixture Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Example usage of an async generator fixture. ```python @pytest.fixture async def temporary_file(): # Setup: create a temporary file file = await create_temp_file() yield file # Teardown: delete the file await delete_file(file) @pytest.mark.asyncio_concurrent async def test_file_operations(temporary_file): # The file is created before this test # And deleted after this test content = await temporary_file.read() assert len(content) >= 0 ``` -------------------------------- ### Configuration File Options: pytest.ini (Full Example) Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md A more comprehensive example of pytest.ini configuration. ```ini [pytest] default_group_strategy = parent asyncio_mode = strict asyncio_default_fixture_loop_scope = function ``` -------------------------------- ### Configuration File Options: setup.cfg Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of setting default_group_strategy in setup.cfg. ```ini [tool:pytest] default_group_strategy = self ``` -------------------------------- ### Timeout Tuning Examples Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Examples of setting timeouts for tests based on expected operation duration. ```python # For operations that typically take < 1 second @pytest.mark.asyncio_concurrent(timeout=5) async def test_fast(): pass # For operations that typically take 1-5 seconds @pytest.mark.asyncio_concurrent(timeout=30) async def test_normal(): pass # For operations that typically take 5+ seconds @pytest.mark.asyncio_concurrent(timeout=120) async def test_slow(): pass ``` -------------------------------- ### Invalid Configuration Examples Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Examples of invalid command-line arguments for the default group strategy, showing expected error output. ```bash pytest --default-group-strategy invalid # Error: invalid choice pytest --default-group-strategy Self # Error: case-sensitive ``` ```text usage: pytest [options] [file_or_dir] [file_or_dir] ... pytest: error: argument --default-group-strategy: invalid choice: 'invalid' (choose from 'self', 'parent') ``` -------------------------------- ### Example for pytest_fixture_setup Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md User code - fixtures are wrapped automatically ```python # User code - fixtures are wrapped automatically @pytest.fixture async def async_database(): db = await connect_to_database() yield db await db.close() @pytest.mark.asyncio_concurrent async def test_with_async_fixture(async_database): result = await async_database.query("SELECT * FROM users") assert len(result) > 0 ``` -------------------------------- ### Example pytest_runtest_setup_async_group Hook Implementation Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/hooks.md Example hook implementation for pytest_runtest_setup_async_group. ```python @pytest.hookimpl(specname="pytest_runtest_setup_async_group") def my_group_setup(item): # Custom setup for async group print(f"Setting up group: {item.name}") # Perform initialization ``` -------------------------------- ### Example of configuration precedence Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Demonstrates how command-line options override configuration file settings. ```bash # Config file says: default_group_strategy = parent # Command line says: --default-group-strategy self # Result: Uses "self" (command line wins) pytest --default-group-strategy self ``` -------------------------------- ### Function-scoped fixture behavior Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Example showing function-scoped fixtures, where each test gets its own instance. ```python # Function-scoped fixture - each test gets its own instance @pytest.fixture async def connection(): return await Database.connect() class TestSuite: @pytest.mark.asyncio_concurrent(group="tests") async def test_one(self, connection): # Gets its own connection pass @pytest.mark.asyncio_concurrent(group="tests") async def test_two(self, connection): # Gets its own connection pass ``` -------------------------------- ### Example Access of pytest.Mark Attributes Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Example code demonstrating how to retrieve a pytest marker and access its `kwargs` to get custom attributes like `group` and `timeout`. ```python mark = item.get_closest_marker("asyncio_concurrent") if mark: group = mark.kwargs.get("group") timeout = mark.kwargs.get("timeout") ``` -------------------------------- ### pytest_runtest_setup Hook Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Redirect setup for async test items to their group setup mechanism. ```python @pytest.hookimpl(specname="pytest_runtest_setup") def pytest_runtest_setup_handle_async_function(item: pytest.Item) -> None ``` -------------------------------- ### BaseExceptionGroup Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Example of raising a BaseExceptionGroup when multiple exceptions occur. ```python if len(exceptions) > 1: msg = f"errors while tearing down {item!r}" raise BaseExceptionGroup(msg, exceptions) ``` -------------------------------- ### With Timeout Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example of a test with an explicit group and a timeout. ```python @pytest.mark.asyncio_concurrent(group="slow_tests", timeout=30) async def test_with_timeout(): result = await long_running_operation() assert result is not None ``` -------------------------------- ### pytest.mark.asyncio_concurrent marker example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/API-INDEX.md Example usage of the @pytest.mark.asyncio_concurrent marker. ```python import pytest @pytest.mark.asyncio_concurrent(group="api", timeout=30) async def test_example(): pass ``` -------------------------------- ### asyncio_concurrent Marker Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Example demonstrating concurrent execution of tests marked with @pytest.mark.asyncio_concurrent. ```python import pytest import asyncio @pytest.mark.asyncio_concurrent(group="api_tests", timeout=30) async def test_api_call_one(): result = await fetch_data() assert result.status == 200 @pytest.mark.asyncio_concurrent(group="api_tests", timeout=30) async def test_api_call_two(): result = await fetch_data() assert result.status == 200 # These two tests run concurrently within the same event loop ``` -------------------------------- ### Same Parent Rule - Invalid Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/README.md Demonstrates an invalid test setup where tests in the same group have different direct parents (classes). ```python class ClassA: @pytest.mark.asyncio_concurrent(group="shared") async def test_a(): pass class ClassB: @pytest.mark.asyncio_concurrent(group="shared") async def test_b(): # ERROR: Different parent! pass ``` -------------------------------- ### Async Function Fixture Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Example demonstrating the usage of an async function fixture with pytest.mark.asyncio_concurrent. ```python @pytest.fixture async def api_client(): # Setup: create and initialize API client client = await AsyncAPIClient.create() return client # No teardown here (see async_generator for teardown) @pytest.mark.asyncio_concurrent async def test_api(api_client): response = await api_client.get("/users") assert response.status == 200 ``` -------------------------------- ### Accessing Command-Line Options Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Example of how to access command-line options and ini/toml settings using the config object. ```python config.getoption("--default-group-strategy") # Get CLI option value config.getini("default_group_strategy") # Get ini/toml setting ``` -------------------------------- ### Configuration File Options: pyproject.toml Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of setting default_group_strategy in pyproject.toml. ```toml [tool.pytest.ini_options] default_group_strategy = "self" ``` -------------------------------- ### Using Async Fixtures Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Demonstrates how to define and use async fixtures within concurrent tests. ```python @pytest.fixture async def database_connection(): # Setup db = await Database.connect("localhost") yield db # Teardown await db.disconnect() @pytest.fixture async def cache(): cache = await Cache.create() return cache # No teardown class TestDatabase: @pytest.mark.asyncio_concurrent(group="db_tests") async def test_query(self, database_connection): result = await database_connection.execute("SELECT 1") assert result == [1] @pytest.mark.asyncio_concurrent(group="db_tests") async def test_insert(self, database_connection): await database_connection.execute("INSERT INTO ...") result = await database_connection.execute("SELECT COUNT(*)") assert result[0] > 0 ``` -------------------------------- ### Example Usage Patterns Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Illustrates various ways to use the `pytest.mark.asyncio_concurrent` decorator with different parameter combinations. ```python # No parameters - uses default strategy @pytest.mark.asyncio_concurrent async def test_independent(): pass # Only group specified @pytest.mark.asyncio_concurrent(group="api_tests") async def test_grouped(): pass # Only timeout specified @pytest.mark.asyncio_concurrent(timeout=60) async def test_slow_operation(): pass # Both parameters @pytest.mark.asyncio_concurrent(group="network", timeout=30) async def test_api_call(): pass ``` -------------------------------- ### Example Usage of GroupStrategy Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Example function demonstrating how to retrieve and cast the group strategy configuration from pytest's configuration object. ```python def _get_group_strategy(config: pytest.Config) -> GroupStrategy: # Returns either "self" or "parent" strategy = ( config.getoption("--default-group-strategy") or config.getini("default_group_strategy") or "self" ).lower() return cast(GroupStrategy, strategy) ``` -------------------------------- ### Accessing Stash Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Example of how to access the stash dictionary for group information. ```python config.stash[asyncio_concurrent_group_key] # Dict of groups ``` -------------------------------- ### Command Line with Config File Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Examples of running pytest from the command line, showing how to use config file defaults and override them. ```bash # Use config file defaults pytest # Override config file pytest --default-group-strategy self # Run specific tests with custom strategy pytest tests/test_api.py --default-group-strategy parent ``` -------------------------------- ### Basic Grouping Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example of tests without explicit group, which run sequentially. ```python import pytest import asyncio # Tests without explicit group (default strategy = "self") @pytest.mark.asyncio_concurrent async def test_independent_one(): await asyncio.sleep(0.1) assert True @pytest.mark.asyncio_concurrent async def test_independent_two(): await asyncio.sleep(0.1) assert True # These run sequentially (each in its own group) ``` -------------------------------- ### Example pytest_runtest_call_async Hook Implementation Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/hooks.md Example hook implementation for pytest_runtest_call_async. ```python @pytest.hookimpl(specname="pytest_runtest_call_async") async def my_async_call_hook(item): # Custom test execution logic print(f"Running: {item.name}") # Delegate to the default implementation or create your own coroutine return None # Implement your own coroutine or return None to delegate ``` -------------------------------- ### Concurrent Group Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example of tests with an explicit group, which run concurrently. ```python # Tests with explicit group run concurrently @pytest.mark.asyncio_concurrent(group="io_operations") async def test_fetch_user(): data = await fetch_from_api("/users/1") assert data["id"] == 1 @pytest.mark.asyncio_concurrent(group="io_operations") async def test_fetch_posts(): data = await fetch_from_api("/posts/1") assert data["author"] is not None # These run concurrently ``` -------------------------------- ### Parametrized Tests Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example of parametrized tests running concurrently within a group. ```python @pytest.mark.asyncio_concurrent(group="parametrized") @pytest.mark.parametrize("user_id", [1, 2, 3]) async def test_fetch_users(user_id): user = await fetch_user(user_id) assert user["id"] == user_id # All 3 parametrized instances run concurrently # Each gets its own fixture instances due to _refresh_function_scoped_fixture ``` -------------------------------- ### Example Usage for pytest_runtest_protocol_async_group Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Example demonstrating how to mark tests for concurrent execution within a group. ```python # This hook is called automatically by pytest # For tests marked like this: @pytest.mark.asyncio_concurrent(group="my_group") async def test_one(): await asyncio.sleep(0.1) assert True @pytest.mark.asyncio_concurrent(group="my_group") async def test_two(): await asyncio.sleep(0.1) assert True # Both tests run concurrently within the same event loop ``` -------------------------------- ### Example pytest_runtest_protocol_async_group Hook Implementation Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/hooks.md Example hook implementation for pytest_runtest_protocol_async_group. ```python @pytest.hookimpl(specname="pytest_runtest_protocol_async_group") def my_async_group_hook(group, nextgroup): # Custom logging or setup print(f"Starting group: {group.name}") # Let the default implementation run return None # Falls through to other implementations ``` -------------------------------- ### Function Scope Fixture Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Illustrates how a function-scoped fixture provides an independent instance for each concurrent test. ```python @pytest.fixture async def database(): db = await Database.connect() yield db await db.close() # Each test gets its own instance @pytest.mark.asyncio_concurrent(group="tests") async def test_one(database): pass @pytest.mark.asyncio_concurrent(group="tests") async def test_two(database): pass # test_one gets database instance A # test_two gets database instance B ``` -------------------------------- ### Prevent Bytecode Caching Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Demonstrates how to prevent Python bytecode caching using the `PYTHONDONTWRITEBYTECODE` environment variable. ```bash PYTHONDONTWRITEBYTECODE=1 pytest ``` -------------------------------- ### Combined Parameters Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Demonstrates using both `group` and `timeout` parameters with the `pytest.mark.asyncio_concurrent` decorator. ```python @pytest.mark.asyncio_concurrent(group="my_group", timeout=30) async def test_with_group_and_timeout(): await operation() ``` -------------------------------- ### asyncio_concurrent Marker Syntax Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Syntax examples for the @pytest.mark.asyncio_concurrent marker. ```python @pytest.mark.asyncio_concurrent async def test_example(): pass ``` ```python @pytest.mark.asyncio_concurrent(group="my_group") async def test_grouped(): pass ``` ```python @pytest.mark.asyncio_concurrent(timeout=10) async def test_with_timeout(): pass ``` ```python @pytest.mark.asyncio_concurrent(group="my_group", timeout=10) async def test_grouped_with_timeout(): pass ``` -------------------------------- ### PytestAsyncioConcurrentInvalidMarkWarning Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example demonstrating when PytestAsyncioConcurrentInvalidMarkWarning is raised for a synchronous test marked with @pytest.mark.asyncio_concurrent. ```python @pytest.mark.asyncio_concurrent def test_sync(): # This is sync, not async! assert True # This raises PytestAsyncioConcurrentInvalidMarkWarning # Because test_sync is not a coroutine function ``` -------------------------------- ### Module Scope Fixture Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Illustrates how a module-scoped fixture shares a single instance among concurrent tests within the same module. ```python @pytest.fixture(scope="module") async def module_resource(): return await Resource.create() @pytest.mark.asyncio_concurrent(group="module") async def test_one(module_resource): pass @pytest.mark.asyncio_concurrent(group="module") async def test_two(module_resource): pass # Both tests share the same module_resource instance ``` -------------------------------- ### Configuration File Options: pytest.ini Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of setting default_group_strategy in pytest.ini. ```ini [pytest] default_group_strategy = self ``` -------------------------------- ### Basic Configuration (pytest.ini) Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md A basic `pytest.ini` configuration example for `pytest-asyncio-concurrent` and related pytest settings. ```ini [pytest] default_group_strategy = self asyncio_mode = strict asyncio_default_fixture_loop_scope = function markers = asyncio_concurrent: mark test for concurrent async execution ``` -------------------------------- ### Class Scope Fixture Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Demonstrates how a class-scoped fixture shares a single instance among concurrent tests within the same class. ```python @pytest.fixture(scope="class") async def shared_service(): return await Service.create() class TestClass: @pytest.mark.asyncio_concurrent(group="tests") async def test_one(self, shared_service): pass @pytest.mark.asyncio_concurrent(group="tests") async def test_two(self, shared_service): pass # Both tests share the same shared_service instance ``` -------------------------------- ### PytestAsyncioConcurrentGroupingWarning Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example demonstrating when PytestAsyncioConcurrentGroupingWarning is raised due to tests from different parent classes sharing a group. ```python class TestClass1: @pytest.mark.asyncio_concurrent(group="shared") async def test_one(): pass class TestClass2: @pytest.mark.asyncio_concurrent(group="shared") async def test_two(): pass # This raises PytestAsyncioConcurrentGroupingWarning # Because tests from TestClass1 and TestClass2 (different parents) share a group ``` -------------------------------- ### Example Usage of MakeItemResult Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Example pytest hook implementation demonstrating the use of the MakeItemResult type for returning collected items or collectors. ```python @pytest.hookimpl(specname="pytest_pycollect_makeitem", wrapper=True, trylast=True) def pytest_pycollect_makeitem_make_group_and_member( collector: pytest.Collector, name: str, obj: object ) -> Generator[None, MakeItemResult, MakeItemResult]: ori_result = yield # MakeItemResult from previous implementations # Process and return MakeItemResult return result # Return type is MakeItemResult ``` -------------------------------- ### Fixture Sharing Examples Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/README.md Illustrates the difference between shared group-scoped fixtures and isolated function-scoped fixtures. ```python # Shared across all tests in group @pytest.fixture(scope="class") async def shared_resource(): return await create_resource() # Isolated per test @pytest.fixture async def independent_resource(): return await create_resource() ``` -------------------------------- ### --default-group-strategy Config File Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Configuration example for --default-group-strategy in pytest.ini or pyproject.toml. ```ini [tool.pytest.ini_options] default_group_strategy = "parent" ``` -------------------------------- ### Parametrized Tests Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/README.rst Examples showing how to handle parametrized tests for sequential and concurrent execution with pytest-asyncio-concurrent. ```python # the parametrized tests below will run sequential @pytest.mark.asyncio_concurrent @pytest.parametrize("p", [0, 1, 2]) async def test_parametrize_sequential(p): res = await wait_for_something_async() assert result.is_valid() # the parametrized tests below will run concurrently ``` -------------------------------- ### Specify group explicitly Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of specifying a group explicitly for a test. ```python # Either specify group explicitly @pytest.mark.asyncio_concurrent(group="my_group") async def test_one(): pass ``` -------------------------------- ### Example Code That Triggers PytestAsyncioConcurrentInvalidMarkWarning Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md Example code demonstrating how to trigger PytestAsyncioConcurrentInvalidMarkWarning by marking a sync function. ```python # WRONG: sync function marked as async concurrent @pytest.mark.asyncio_concurrent def test_sync_function(): assert True ``` -------------------------------- ### --default-group-strategy Usage Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Usage example for the --default-group-strategy command-line option. ```bash pytest --default-group-strategy self ``` ```bash pytest --default-group-strategy parent ``` -------------------------------- ### Increase timeout Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of increasing the timeout for a test that might take longer. ```python # Increase timeout @pytest.mark.asyncio_concurrent(timeout=60) # Increased from 30 async def test_slow_operation(): pass ``` -------------------------------- ### Pytest Marker: asyncio_concurrent - timeout parameter example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md A practical example combining 'group' and 'timeout' parameters for a test. ```python @pytest.mark.asyncio_concurrent(group="api", timeout=30) async def test_fetch_user(): user = await api.get_user(1) assert user["id"] == 1 ``` -------------------------------- ### Invalid Grouping Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/grouping.md Example demonstrating invalid grouping where tests from different classes share a group, leading to them being skipped. ```python class TestClassA: @pytest.mark.asyncio_concurrent(group="cross_class") async def test_method(): pass class TestClassB: @pytest.mark.asyncio_concurrent(group="cross_class") async def test_method(): pass # Both tests will be skipped with a warning # because they're from different parent classes ``` -------------------------------- ### Run tests Concurrently Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/README.rst Examples demonstrating how to run tests concurrently using async groups and timeouts with pytest-asyncio-concurrent. ```python # the test below will run by itself @pytest.mark.asyncio_concurrent async def test_by_itself(): res = await wait_for_something_async() assert result.is_valid() # the two tests below will run concurrently @pytest.mark.asyncio_concurrent(group="my_group") async def test_groupA(): res = await wait_for_something_async() assert result.is_valid() # this one will have a 10s timeout @pytest.mark.asyncio_concurrent(group="my_group", timeout=10) async def test_groupB(): res = await wait_for_something_async() assert result.is_valid() ``` -------------------------------- ### Concurrent Test Count Tuning Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of how to control the number of concurrent tests within a group. ```python # Few concurrent tests per group @pytest.mark.asyncio_concurrent(group="small") async def test_one(): pass # Many concurrent tests in one group (use with caution) @pytest.mark.asyncio_concurrent(group="large") async def test_many_1(): pass # ... 100 more tests with same group ... ``` -------------------------------- ### Complex Configuration (pyproject.toml) Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md An example of a complex configuration using `pyproject.toml` for `pytest-asyncio-concurrent` and standard pytest options. ```toml [tool.pytest.ini_options] # AsyncIO concurrent settings default_group_strategy = "parent" asyncio_mode = "strict" asyncio_default_fixture_loop_scope = "function" # Standard pytest settings testpaths = ["tests"] python_files = ["test_*.py"] python_classes = ["Test*"] python_functions = ["test_*"] # Markers markers = [ "asyncio_concurrent: mark test for concurrent async execution", "slow: mark test as slow", "integration: mark test as integration test", ] # Output addopts = "-v --tb=short" ``` -------------------------------- ### _wrap_asyncgen_fixture Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Wrap an async generator fixture to provide setup and teardown within a synchronous context. ```python def _wrap_asyncgen_fixture(fixturedef: pytest.FixtureDef) -> None ``` -------------------------------- ### Run test Sequentially Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/README.rst Example of how to mark async tests to run sequentially using pytest-asyncio-concurrent. ```python import pytest @pytest.mark.asyncio_concurrent async def async_test_A(): res = await wait_for_something_async() assert result.is_valid() @pytest.mark.asyncio_concurrent async def async_test_B(): res = await wait_for_something_async() assert result.is_valid() ``` -------------------------------- ### Implementing Custom Hooks Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/hooks.md Example implementations for custom hooks in pytest-asyncio-concurrent. ```python import pytest @pytest.hookimpl(specname="pytest_runtest_setup_async_group") def pytest_runtest_setup_async_group(item): # item is an AsyncioConcurrentGroup print(f"Setting up group: {item.name}") @pytest.hookimpl(specname="pytest_runtest_teardown_async_group") def pytest_runtest_teardown_async_group(item, nextitem): # item is an AsyncioConcurrentGroup print(f"Tearing down group: {item.name}") @pytest.hookimpl(specname="pytest_runtest_call_async") async def pytest_runtest_call_async(item): # item is an AsyncioConcurrentGroupMember # Must return a coroutine # Return None to use default implementation pass ``` -------------------------------- ### Use different group names per class Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of using different group names for tests in different classes to avoid conflicts. ```python # Use different group names per class class TestClassA: @pytest.mark.asyncio_concurrent(group="class_a") async def test_method(): pass class TestClassB: @pytest.mark.asyncio_concurrent(group="class_b") async def test_method(): pass ``` -------------------------------- ### Disable Plugin Autoloading Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Demonstrates how to disable automatic plugin loading using the `PYTEST_DISABLE_PLUGIN_AUTOLOAD` environment variable. ```bash PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 pytest ``` -------------------------------- ### Pytest Marker: asyncio_concurrent - Example with Multiple Groups Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Demonstrates using the 'group' parameter to define multiple concurrency groups within a test class. ```python class TestAPI: # These run together @pytest.mark.asyncio_concurrent(group="api_calls") async def test_get_user(): pass @pytest.mark.asyncio_concurrent(group="api_calls") async def test_get_posts(): pass # These also run together (different group) @pytest.mark.asyncio_concurrent(group="database") async def test_db_query(): pass @pytest.mark.asyncio_concurrent(group="database") async def test_db_insert(): pass ``` -------------------------------- ### Module-scoped fixture behavior Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Example showing a module-scoped fixture shared across concurrent tests. ```python # Module-scoped fixture - shared across ALL concurrent tests @pytest.fixture(scope="module") async def shared_service(): return await Service.create() class TestSuite: @pytest.mark.asyncio_concurrent(group="tests") async def test_one(self, shared_service): # Uses the same service instance pass @pytest.mark.asyncio_concurrent(group="tests") async def test_two(self, shared_service): # Uses the same service instance pass ``` -------------------------------- ### Show configuration values Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Command to display configuration values. ```bash pytest -v --setup-show ``` -------------------------------- ### _get_group_strategy Helper Function Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Get the default grouping strategy from command-line or config file. ```python @functools.lru_cache(maxsize=1) def _get_group_strategy(config: pytest.Config) -> GroupStrategy ``` -------------------------------- ### Multiple Exception Handling Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md A try-except block demonstrating how to handle a BaseExceptionGroup, printing the main message and individual exceptions. ```python try: # Group teardown with multiple failures pass except BaseExceptionGroup as eg: print(f"Errors occurred: {eg.message}") for exc in eg.exceptions: print(f" - {exc}") ``` -------------------------------- ### Example Code Triggering PytestAsyncioConcurrentGroupingWarning Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md Demonstrates how marking tests from different classes with the same group name triggers the PytestAsyncioConcurrentGroupingWarning and skips the tests. ```python class TestClassA: @pytest.mark.asyncio_concurrent(group="shared_group") async def test_one(): pass class TestClassB: @pytest.mark.asyncio_concurrent(group="shared_group") async def test_two(): pass # WARNING: PytestAsyncioConcurrentGroupingWarning is issued # Both tests are skipped ``` -------------------------------- ### Show markers Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Command to display available markers, filtered for asyncio_concurrent. ```bash pytest --markers | grep asyncio_concurrent ``` -------------------------------- ### Example Code That Triggers PytestAsyncioConcurrentGroupingWarning Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md Example code demonstrating how to trigger PytestAsyncioConcurrentGroupingWarning. ```python # conftest.py or test file class TestClassA: @pytest.mark.asyncio_concurrent(group="shared_group") async def test_method(): assert True class TestClassB: @pytest.mark.asyncio_concurrent(group="shared_group") async def test_method(): assert True ``` -------------------------------- ### Example hook implementation for pytest_runtest_teardown_async_group Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/hooks.md An example of how to implement a custom teardown hook for an async group. ```python @pytest.hookimpl(specname="pytest_runtest_teardown_async_group") def my_group_teardown(item, nextitem): # Custom teardown for async group print(f"Tearing down group: {item.name}") # Perform cleanup ``` -------------------------------- ### pytest_fixture_setup Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Automatically wrap async fixtures to make them work with the plugin's concurrent execution model. ```python @pytest.hookimpl(specname="pytest_fixture_setup", tryfirst=True) def pytest_fixture_setup_wrap_async( fixturedef: pytest.FixtureDef, request: pytest.FixtureRequest ) -> None ``` -------------------------------- ### Catching Timeout Errors Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md Example of how to catch asyncio.TimeoutError in a test marked with @pytest.mark.asyncio_concurrent(timeout=5). ```python @pytest.mark.asyncio_concurrent(timeout=5) async def test_network_call(): try: result = await fetch_with_timeout() assert result is not None except asyncio.TimeoutError: pytest.fail("Network call timed out") ``` -------------------------------- ### Output of Example Code Triggering PytestAsyncioConcurrentGroupingWarning Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md The output produced by the example code that triggers PytestAsyncioConcurrentGroupingWarning. ```text PytestAsyncioConcurrentGroupingWarning: Asyncio Concurrent Group [shared_group] has children from different parents, skipping all of it's children. ``` -------------------------------- ### Example Usage with asyncio.TimeoutError Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md An example of using the @pytest.mark.asyncio_concurrent mark with a timeout to control test execution time. ```python @pytest.mark.asyncio_concurrent(group="api_tests", timeout=30) async def test_api_call_with_timeout(): # Must complete within 30 seconds response = await fetch_data() assert response.status == 200 ``` -------------------------------- ### Pattern 4: Fixture-Based Grouping Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Configuration and test code demonstrating grouping tests by resource requirements using fixtures. ```ini [pytest] default_group_strategy = self ``` ```python @pytest.fixture async def database(): return await Database.connect() @pytest.fixture async def cache(): return await Cache.connect() class TestDatabaseOperations: @pytest.mark.asyncio_concurrent(group="db") async def test_read(self, database): pass @pytest.mark.asyncio_concurrent(group="db") async def test_write(self, database): pass class TestCacheOperations: @pytest.mark.asyncio_concurrent(group="cache") async def test_get(self, cache): pass @pytest.mark.asyncio_concurrent(group="cache") async def test_set(self, cache): pass ``` -------------------------------- ### pytest_runtest_setup_async_group Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Set up fixtures and state for an entire async group before any individual test runs. ```python @pytest.hookimpl(specname="pytest_runtest_setup_async_group") def pytest_runtest_setup_async_group(item: AsyncioConcurrentGroup) -> None: pass ``` -------------------------------- ### Recommended asyncio_mode Setting Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Shows the recommended setting for `asyncio_mode` in `pytest.ini` for optimal async test execution. ```ini [pytest] asyncio_mode = strict ``` -------------------------------- ### Normal pytest (sequential) fixture lifecycle Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Illustrates the standard fixture lifecycle in sequential pytest execution. ```text For each test: 1. Setup fixtures (retrieve from cache or create) 2. Run test 3. Run finalizers and teardown fixtures ``` -------------------------------- ### Hypothetical Error Trigger Example Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md A hypothetical Python code example that would trigger the PytestAysncioGroupInvokeError, emphasizing that this should not happen in normal usage. ```python # This would trigger the error (should never happen): group = AsyncioConcurrentGroup(parent=parent_node, originalname="test_group") group.runtest() # Raises PytestAysncioGroupInvokeError ``` -------------------------------- ### Parametrized concurrent test Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/README.rst Example of a parametrized test that runs concurrently. ```python @pytest.mark.asyncio_concurrent(group="my_group") @pytest.parametrize("p", [0, 1, 2]) async def test_parametrize_concurrent(): res = await wait_for_something_async() assert result.is_valid() ``` -------------------------------- ### Behavior Differences: --default-group-strategy self Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Illustrates sequential test execution with --default-group-strategy self. ```python class TestClass: @pytest.mark.asyncio_concurrent async def test_one(): pass @pytest.mark.asyncio_concurrent async def test_two(): pass # test_one and test_two run sequentially (separate groups) ``` -------------------------------- ### Pattern 3: Mixed Modules with Timeouts Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Configuration and test code for scenarios with different modules having different concurrency requirements and timeouts. ```toml [tool.pytest.ini_options] default_group_strategy = "parent" ``` ```python # tests/test_fast.py class TestQuickOperations: @pytest.mark.asyncio_concurrent(timeout=5) async def test_cache_get(): pass # tests/test_api.py class TestAPIOperations: @pytest.mark.asyncio_concurrent(timeout=30) async def test_api_call(): pass # tests/test_slow.py class TestSlowOperations: @pytest.mark.asyncio_concurrent(timeout=120) async def test_complex_operation(): pass ``` -------------------------------- ### Debugging Configuration Command Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Command to display all ini options, useful for debugging configuration. ```bash pytest --co -q ``` -------------------------------- ### Behavior Differences: --default-group-strategy parent Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Illustrates concurrent test execution with --default-group-strategy parent. ```python class TestClass: @pytest.mark.asyncio_concurrent async def test_one(): pass @pytest.mark.asyncio_concurrent async def test_two(): pass # test_one and test_two run concurrently (same group) ``` -------------------------------- ### pytest_sessionstart Hook Signature Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/fixture_async.md Signature for the pytest_sessionstart hook implementation that initializes fixture caching. ```python @pytest.hookimpl(specname="pytest_sessionstart", trylast=True) def pytest_sessionstart_cache_fixture(session: pytest.Session) -> None ``` -------------------------------- ### _get_asyncio_concurrent_mark Helper Function Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/plugin.md Get the `asyncio_concurrent` marker from a test item if present. ```python def _get_asyncio_concurrent_mark(item: pytest.Item) -> Optional[pytest.Mark] ``` -------------------------------- ### Configuration file setting for default group strategy Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Setting the default group strategy in pytest.ini, pyproject.toml, or setup.cfg. ```ini [pytest] default_group_strategy = parent ``` -------------------------------- ### Recommended asyncio_default_fixture_loop_scope Setting Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Shows the recommended setting for `asyncio_default_fixture_loop_scope` in `pytest.ini` to ensure independent event loops per test function. ```ini [pytest] asyncio_default_fixture_loop_scope = function ``` -------------------------------- ### Registering custom hooks in conftest.py Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/api-reference/hooks.md Example of registering custom hooks within a conftest.py file. ```python # conftest.py import pytest @pytest.hookimpl(specname="pytest_runtest_setup_async_group") def pytest_runtest_setup_async_group(item): print(f"Custom hook: Setting up {item.name}") ``` -------------------------------- ### Code Triggering asyncio.TimeoutError Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md An example test function that is expected to exceed its timeout, triggering an asyncio.TimeoutError. ```python @pytest.mark.asyncio_concurrent(timeout=5) async def test_with_timeout(): await asyncio.sleep(10) # Exceeds 5 second timeout ``` -------------------------------- ### Pytest Plugin Entry Point Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/API-INDEX.md Defines the entry point for the pytest plugin. ```ini [project.entry-points.pytest11] asyncio-concurrent = "pytest_asyncio_concurrent.plugin" ``` -------------------------------- ### Pattern 2: Parent-Grouped All Concurrent Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Configuration and test code where all tests in a class run together automatically. ```ini [pytest] default_group_strategy = parent ``` ```python class TestAPI: # All tests in this class run concurrently # No @asyncio_concurrent group parameter needed @pytest.mark.asyncio_concurrent async def test_one(): pass @pytest.mark.asyncio_concurrent async def test_two(): pass @pytest.mark.asyncio_concurrent async def test_three(): pass ``` -------------------------------- ### Pytest Marker: asyncio_concurrent - timeout parameter usage Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/configuration.md Example of setting a timeout for a test using the 'timeout' parameter. ```python @pytest.mark.asyncio_concurrent(timeout=30) async def test_with_timeout(): await some_operation() ``` -------------------------------- ### Correct Usage of Async Generator Fixture Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/errors.md An example of a correctly implemented async generator fixture that yields only once. ```python @pytest.fixture async def correct_fixture(): resource = await setup() yield resource # Single yield await teardown() ``` -------------------------------- ### Type Relationships Summary Source: https://github.com/czl9707/pytest-asyncio-concurrent/blob/main/_autodocs/types.md A summary of type relationships within pytest and the pytest-asyncio-concurrent library. ```text pytest.Item (stdlib) ├── pytest.Function (stdlib) │ ├── AsyncioConcurrentGroup (our class) │ │ └── has many AsyncioConcurrentGroupMember children │ └── AsyncioConcurrentGroupMember (our class) │ └── has reference to parent AsyncioConcurrentGroup pytest.Mark (stdlib) └── used by @pytest.mark.asyncio_concurrent GroupStrategy = Literal["self", "parent"] PytestAsyncioConcurrentGroupingWarning ├── PytestAsyncioConcurrentInvalidMarkWarning └── PytestAysncioGroupInvokeError ```