🔒 PyCharm won't stop in a particular breakpoint during tests

[hollow bough]

For context, I have a CustomLogger class that has a method using the @classmethod decorator. When this method is invoked, it creates a session to connect with the database and then writes the logs to the database.

I also have a test written for this method where I'm mocking the content of the log and mocking the database connection.

Tests are executed on a Docker Container that is raised on conftest.py and I have a function that provides Session objects connecting to that test database. With all my other tests, this works and I can debug them, even see the data going in on the database running inside the Docker Container.

But with the tests for the CustomLogger class, I can't debug them if I use unittest.mock.patch to mock the database connection.

(will provide code samples below)

[tawdry token]

@hollow bough

Python help channel opened

Remember to:

• Ask your Python question, not if you can ask or if there's an expert who can help.
• Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
• Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.

[hollow bough]

The CustomLogger class is roughly defined like this:

class CustomLogger:
    @classmethod
    def save_communication_log(cls, request: dict = None, httpx_response: httpx.Response = None):
        response_json_data = {}
        status_code = 0

        if httpx_response is not None:
            response_json_data.update({
                "headers": dict(httpx_response.headers),
                "response_body": json.loads(httpx_response.text)
            })

            status_code = httpx_response.status_code

        log_id = str(uuid.uuid4())

        insert_statement = insert(CommunicationLog).values(
            log_id=log_id,
            request=request,
            status_code=status_code,
            response=json.dumps(response_json_data, indent=4, ensure_ascii=False)
        )

        # THIS IS WHERE THE METHOD CREATES THE CONNECTION WITH THE DATABASE
        db_connection: Session = next(database.get_db_session())

        result = db_connection.execute(insert_statement)
        db_connection.commit()

        return result.fetchone()[0]

This is the test that the breakpoint just won't hit:

# THIS IS WHERE I'M IMPORTING THE DATABASE MODULE TO BE ABLE TO MOCK THE FUNCTION THAT CONNECTS WITH THE DATABASE
from src.log_utils.custom_logger import database as custom_logger_database, CustomLogger

# IF I REMOVE THIS INSTRUCTION BELOW, THE BREAKPOINT HITS
@patch.object(target=custom_logger_database, attribute="get_db_session")
@patch.object(target=httpx, attribute="Response", spec=True)
def test_should_save_communication_log(mock_httpx_response, mocked_database_session, setup_database_container):

    request_json_data = {...}

    response_json_data = {...}

    mock_httpx_response.headers = response_json_data["headers"]
    mock_httpx_response.text = json.dumps(response_json_data["content"])
    mock_httpx_response.status_code = 200

# INSERT INTO THE TEST DATABASE
db_session_insert = aux_get_db_session(setup_database_container)
    with db_session_insert as db_conn:

        supplier = aux_create_supplier(db_conn=db_conn)

        created_log_id = CustomLogger.save_communication_log(
            request=request_json_data
            httpx_response=mock_httpx_response,
        )

    # THIS IS WHERE I'M MOCKING THE CONNECTION WITH THE DATABASE, MAKING THE TEST CONNECT WITH THE TEST_DATABASE
    test_db_session_select = aux_get_db_session(setup_database_container)
    mock_db_session_instance = mocked_database_session.return_value
    mock_db_session_instance.get_db_session.return_value = yield test_db_session_select

    created_communication_log = CustomLogger.get_communication_log_by_id(communication_log_id=created_log_id)

    assert created_communication_log is not None

[granite dust]

you cannot use yield like that

it's turning your entire def test_should_save_communication_log function into a generator instead of setting it as the return value of mock_db_session_instance.get_db_session

!e ```py
def foo():
example = 42
value = yield example
print("value:", value)

gen = foo()
print("gen next()", next(gen))
print("gen send()", gen.send(10))

[tawdry token]

@granite dust :x: Your 3.12 eval job has completed with return code 1.

001 | gen next() 42
002 | value: 10
003 | Traceback (most recent call last):
004 |   File "/home/main.py", line 8, in <module>
005 |     print("gen send()", gen.send(10))
006 |                         ^^^^^^^^^^^^
007 | StopIteration

[granite dust]

judging by https://docs.python.org/3/library/unittest.mock-examples.html try ```py
mock_db_session_instance.get_db_session.return_value = iter([test_db_session_select])

[hollow bough]

@granite dust thank you so much. I had no idea.
This gives me a new angle to investigate!

Probably the way I'm creating the Database Session inside the actual CustomLogger method isn't the right way as well.

[granite dust]

yeah... calling next() on that seems weird to say the least

usually you would rather use a context manager than a generator for that sort of things

[hollow bough]

Please bear in mind I'm a newbie at Python, migrating from .NET C# development.

I only did it that way because after all the tutorials and courses on FastAPI, I managed to make it work like that.

I see... so instead of using next() I should write using context manager like with ?

[granite dust]

are you defining get_session yourself or is it coming from some library?

[hollow bough]

Im doing it myself using SQLAlchemy as the ORM.

This is the method that generates the Session and when used with Dependency Injection in FastAPI, it resolves the iterator and returns the actual Session object.

def get_db_session() -> Session:
    db_connection = DB_Session_Pool()

    # TODO: No caso de dar um erro na tentativa de fechar/usar a conexão, precisamos logar o erro de alguma forma
    try:
        yield db_connection
    finally:
        db_connection.close()

[granite dust]

when used with Dependency Injection in FastAPI
Usually you should not be calling dependencies yourself 🙃

[hollow bough]

Inside the CustomLogger methods I don't have dependency injection so I used the next() solution to access that resource.

I will change it to use with and see how it goes. Your hint seems pretty spot on.

[granite dust]

You have to change the function for it to actually support with

!d contextlib

[tawdry token]

contextlib

Source code: Lib/contextlib.py

This module provides utilities for common tasks involving the with statement. For more information see also Context Manager Types and With Statement Context Managers.

[granite dust]

needless to say, the same function will not work for both yield dependencies and with non-dependencies

[hollow bough]

That makes sense but I don't know of any other way to generate a connection with the database inside the CustomLogger because I can't rely on the session injected by FastAPI there.

We had a case where a Session had a problem with a database statement and that rendered the whole session unusable, so we started to create a fresh one inside the logger.

[hollow bough]

Thank you for the suggestion. Now that I have something new to investigate and study, I'm sure I'll find a way. I might have two functions that will generate Session.

One to be used with FastAPI's DI and the other to be used by the CustomLogger.

[granite dust]

you can try just doing something like```py
import contextlib

get_db_context = contextlib.contextmanager(get_db_session)
later onpy
with get_db_context() as connection:
...

[hollow bough]

Ok. Will do it now and will post back here once I'm done.
Thank you very much for this.

[hollow bough]

@granite dust thank you.
I've been fighting with this for 2 days now and no one thought about sharing me that bit on the yield keyword.

This made everything work. You're great.

[tawdry token]

Python help channel closed

This help channel has been closed and it's no longer possible to send messages here. If your question wasn't answered, feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.