#Pytest async help

61 messages · Page 1 of 1 (latest)

blissful sinew
#

This is tests/test_function.py

import pytest

@pytest.mark.asyncio
@pytest.mark.parametrize("test_input", [1])
async def test_random_number(session_setup, random_number, test_input):
    number = await random_number
    print(f"PRINTING X: {session_setup}")  # Directly use the value from session_setup
    assert isinstance(number, int)
    assert 1 <= number <= 100

@pytest.mark.asyncio
async def test_multiply_random(session_setup, random_number):
    number = await random_number
    print(f"PRINTING X: {session_setup}")  # Directly use the value from session_setup
    result = number * 10
    assert isinstance(result, int)
    assert 10 <= result <= 1000

This is tests/conftest.py

import pytest
import sys

@pytest.fixture(scope="session")
async def session_setup():
    # Any setup code that needs to run once before the test session
    sys.stdout.flush()
    print("Setting up test session...asdfjasdlkfajdsflkasdjflasdkjfadslkfjasdlkfajsdflkasjdfkadsljfalskdjafsdklfjasdlkfjasdlkfjasdlkfjasdlfkadsjflaskdjflakfjsdlkfjasdlkfasdjflk")
    sys.stdout.flush()
    return 1
    # Any teardown code that needs to run once after the test session
    sys.stdout.flush()
    print("Tearing down test session...")
    sys.stdout.flush()

@pytest.fixture(scope="function")
async def random_number():
    import random
    return random.randint(1, 100)

Writing more below on how I need help

# 
 python3.11 -m pytest tests/test_function.py  -vv --capture=no
====================================================================================== test session starts =======================================================================================
platform linux -- Python 3.11.8, pytest-7.4.4, pluggy-1.4.0 -- /usr/bin/python3.11
cachedir: .pytest_cache
rootdir: /root/My-Computer/My-Working/pytest
plugins: asyncio-0.23.5, anyio-3.7.1
asyncio: mode=Mode.STRICT
collected 2 items

tests/test_function.py::test_random_number[1] PRINTING X: <coroutine object session_setup at 0x7f3b54f5e500>
PASSED
tests/test_function.py::test_multiply_random PRINTING X: <coroutine object session_setup at 0x7f3b54f5e500>
PASSED

======================================================================================== warnings summary ========================================================================================
tests/test_function.py::test_multiply_random
  /usr/lib/python3/dist-packages/_pytest/runner.py:139: RuntimeWarning: coroutine 'session_setup' was never awaited
    item.funcargs = None  # type: ignore[attr-defined]
  Enable tracemalloc to get traceback where the object was allocated.
  See https://docs.pytest.org/en/stable/how-to/capture-warnings.html#resource-warnings for more info.

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
================================================================================== 2 passed, 1 warning in 0.01s ==================================================================================```
I want to session_setup to run only once per tests. And for every test it should return same value. I dont want to run session_setup after every function has been completed
#

if i do yield 1

#

second function test_multiply_random doesnt print PRINTING X

#

If i convert return 1 to yield 1

#

and convert test_function like this

# 
root@kali-linux:~/Temp/pytest# cat tests/test_function.py
import pytest

@pytest.mark.asyncio
@pytest.mark.parametrize("test_input", [1])
async def test_random_number(session_setup, random_number, test_input):
    number = await random_number
    async for x in session_setup:
        print(f"PRINTING X: {x}")
    assert isinstance(number, int)
    assert 1 <= number <= 100

@pytest.mark.asyncio
async def test_multiply_random(session_setup, random_number):
    number = await random_number
    async for x in session_setup:
        print(f"PRINTING X: {x}")
    result = number * 10
    assert isinstance(result, int)
    assert 10 <= result <= 1000
# 
 python3.11 -m pytest tests/test_function.py  -vv --capture=no
====================================================================================== test session starts =======================================================================================
platform linux -- Python 3.11.8, pytest-7.4.4, pluggy-1.4.0 -- /usr/bin/python3.11
cachedir: .pytest_cache
rootdir: /root/My-Computer/My-Working/pytest
plugins: asyncio-0.23.5, anyio-3.7.1
asyncio: mode=Mode.STRICT
collected 2 items

tests/test_function.py::test_random_number[1] Setting up test session...asdfjasdlkfajdsflkasdjflasdkjfadslkfjasdlkfajsdflkasjdfkadsljfalskdjafsdklfjasdlkfjasdlkfjasdlkfjasdlfkadsjflaskdjflakfjsdlkfjasdlkfasdjflk
PRINTING X: None
Tearing down test session...
PASSED
tests/test_function.py::test_multiply_random PASSED

======================================================================================= 2 passed in 0.01s ========================================================================================
root@kali-linux:~/Temp/pytest#
#

PRINTING X: None only prints 1 time

#

I want a session function that runs only once a session

#

And it returns something

#

That something should be available to all tests

#

And shouldnt be generator and run in async context

#

For example session_setup once reached return or yield shouldnt re start everything from start and just return what found when running first time

#

Think of a database doing some initialisation then cursor being returned. It shouldn't be re initaliser when first test is finished and second test is running

#

Just simply return what it returned when running first time

#

I don't know why it has to be return coroutine and not value

#

Unittest has bad async support while pytest is painful

#

Then once all tests are run

#

It should run some teardown per session

#

Probably the cursor it returned in context of database example it should run . close () only when all tests function is completed

#

Thing is after first test has been completed

#

It returns statements after yield 0

#

So i has to use hookimpl

#

Using pytest_sessionclose

#

And remove lines after yield

#

But damn why can't I return and have to do yield

#

If i yield then second function errors out using fixtures probably because generator is expired

#

But using
while True:
yield something

#

Causes test function to never run

#

While directly returning causes couroutine was never awaited

#

I am quite tired and my brain is fired

#

so i could be missing the simple details

# 
@pytest.fixture(scope="session")
async def session_setup():
    # Any setup code that needs to run once before the test session
    sys.stdout.flush()
    print("Setting up test session...asdfjasdlkfajdsflkasdjflasdkjfadslkfjasdlkfajsdflkasjdfkadsljfalskdjafsdklfjasdlkfjasdlkfjasdlkfjasdlfkadsjflaskdjflakfjsdlkfjasdlkfasdjflk")
    sys.stdout.flush()
    return 1
    # Any teardown code that needs to run once after the test session
    sys.stdout.flush()
    print("Tearing down test session...")
    sys.stdout.flush()

Just want setting up test session .... asfkladjflsdk to be printed only 1 but its return value I want to be used across multiple tests without causing test session.. askdfajdsklf to be printed again and again in async context then once all tests functions has been run I want to execute teardown function

@pytest.hookimpl(tryfirst=True)
def pytest_sessionfinish(session, exitstatus):
    print(f"Exit status: {exitstatus}")
    # run something/do something with session_setup return value
# 
import pytest
import sys

# Global variable to store the return value of session_setup
session_setup_value = None

@pytest.fixture(scope="session")
async def session_setup():
    global session_setup_value
    # Only execute setup code if session_setup_value is not set
    if session_setup_value is None:
        sys.stdout.flush()
        print("Setting up test session...asdfjasdlkfajdsflkasdjflasdkjfadslkfjasdlkfajsdflkasjdfkadsljfalskdjafsdklfjasdlkfjasdlkfjasdlkfjasdlfkadsjflaskdjflakfjsdlkfjasdlkfasdjflk")
        sys.stdout.flush()
        session_setup_value = 1
    return session_setup_value

@pytest.hookimpl(tryfirst=True)
def pytest_sessionfinish(session, exitstatus):
    global session_setup_value
    print(f"Exit status: {exitstatus}")
    # Run teardown code only if session_setup_value was set
    if session_setup_value is not None:
        sys.stdout.flush()
        print("Tearing down test session...")
        sys.stdout.flush()
        # Reset session_setup_value after teardown
        session_setup_value = None

This can be simple reproducer than code above

#

in test_functions.py

# 
import pytest

# Test function using session_setup_value
@pytest.mark.asyncio
async def test_using_session_setup_value_one(session_setup):
    # Await session_setup to get its return value
    session_setup_value = await session_setup
    # Access session_setup_value here
    assert session_setup_value == 1

# Another test function using session_setup_value
@pytest.mark.asyncio
async def test_using_session_setup_value_two(session_setup):
    # Await session_setup to get its return value
    session_setup_value = await session_setup
    # Access session_setup_value here
    assert session_setup_value == 1
# 
_______________________________________________________________________________ test_using_session_setup_value_two _______________________________________________________________________________

session_setup = <coroutine object session_setup at 0x7f815e25e440>

    @pytest.mark.asyncio
    async def test_using_session_setup_value_two(session_setup):
        # Await session_setup to get its return value
>       session_setup_value = await session_setup
E       RuntimeError: cannot reuse already awaited coroutine

tests/test_function.py:15: RuntimeError
==================================================================================== short test summary info =====================================================================================
FAILED tests/test_function.py::test_using_session_setup_value_two - RuntimeError: cannot reuse already awaited coroutine
================================================================================== 1 failed, 1 passed in 0.11s ===================================================================================
root@kali-linux:~/Temp/pytest#
#

However I am getting RunTimeError

blissful sinew
#

This looks impossible

blissful sinew
# 
import pytest
import asyncio

# Global variable to store the initialized object
something_important = None

async def async_initialization():
    print("Initializing")
    # Simulate async initialization
    await asyncio.sleep(1)  # Replace with your actual async initialization logic
    global something_important
    something_important = type('Important', (object,), {'x': 'Initialized Value'})()

def pytest_configure(config):
    # Ensure the async initialization runs once before any tests
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_initialization())

@pytest.fixture(scope="session")
def initialise_per_session():
    # This fixture now simply returns the already initialized object
    return something_important

@pytest.hookimpl(tryfirst=True)
def pytest_sessionfinish(session, exitstatus):
    # Teardown logic here, if necessary
    print("Teardown")
    # For example: something_important.close()  # Adjust according to your object's API```
#

this might potentially work

#

hmm putting in global and using pytest_configure to run once is good

#

it sucks native pytest doesnt have ability to just return and have reused fixtures

#

in async form but work in sync form

#

RuntimeError: Event loop is closed
The future belongs to a different loop than the one specified as the loop argument
Damn

blissful sinew
#

pytest error suppression is what causing lots of problems

#

literally gives 0 clue on where it went wrong

#

--full-trace is life saver; i dont know why it wasnt enabled by default

blissful sinew
#

Event loop seems getting closed in test_function.py

#

thats what causing some event loop closed problems

#

whatever issue i had was fixed by doing two thigns

#

pip install pytest-asyncio==0.21.1

# 
@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()
#

i am thinking whether scope=session or scope=package is better

#

also put this in init.py of tests/ folder

import pytest

pytestmark = pytest.mark.asyncio(scope="session")
blissful sinew