how to mock a class with metaclass inheritance decorator wrapt_timeout_decorator

130 Views Asked by At

In this test it only simulates the call to a class method that in turn calls another class, and both inherit a metaclass that uses the wrapt_timeout_decorator library, so that all methods of a class are decorated with this wrapper, for control of waiting times.However, when doing this test it fails, because call_count is 0. However, when "eliminating" the inheritance of the FirstService class the test works, does anyone know what having the metaclass influences inheritance? Since when you remove it the test works correctly.

from wrapt_timeout_decorator import timeout
import inspect
import unittest
from unittest.mock import patch

class TimeoutManager(type):
    """
    A metaclass used to wrap all methods of external classes with a timeout decorator
    for controlling timeouts in the backend.

    Usage:
    class YourClass(metaclass=TimeoutManager):
        ...
    """

    def __new__(cls, name, bases, dct):
        """
        Overrides the creation of a new class.

        Args:
        - name (str): The name of the class.
        - bases (tuple): The base classes of the class.
        - dct (dict): The class dictionary.

        Returns:
        - type: The new class.
        """

        for key, value in dct.items():
            # Check if method is not a magic method
            if callable(value) and not key.startswith("__"):
                # Check if method has the @timeout decorator
                validation: bool = cls.has_timeout_decorator(value)
                if not validation:
                    dct[key] = cls.default_timeout_wrapper()(value)
        return super().__new__(cls, name, bases, dct)

    @classmethod
    def default_timeout_wrapper(cls) -> callable:
        """
        Returns a default timeout wrapper.

        Returns:
        - callable: Timeout wrapper function.
        """
        return timeout(
            45,
            use_signals=False,
            timeout_exception=TimeoutError,
            exception_message="Tiempo de espera agotado , favor intente nuevamente.",
        )

    @classmethod
    def has_timeout_decorator(cls, method: callable) -> bool:
        """
        Checks if a method has the @timeout decorator.

        Args:
        - method (callable): The method to check.

        Returns:
        - bool: True if the @timeout decorator is present, False otherwise.
        """
        decorators_method: list = cls.get_decorators(method)
        if len(decorators_method) > 0:
            if "timeout" in decorators_method:
                return True
        return False

    @classmethod
    def get_decorators(cls, method: callable) -> list:
        """
        Extracts decorators from the source code of a method.

        Args:
        - method (callable): The method to extract decorators from.

        Returns:
        - list: List of decorator names.
        """
        decorators: list = []
        source_lines, _ = inspect.getsourcelines(method)
        source_lines: list = source_lines
        for line in source_lines:
            stripped_line: str = line.strip()
            if stripped_line.startswith("@"):
                decorator_name: list = stripped_line.split("(")[0][1:]
                decorators.append(decorator_name)
        return decorators




class FirstService(metaclass=TimeoutManager):
    """Service class for performing inference operations."""

    @classmethod
    def persist_metadata(resource_info) :
        """Process the query and generate a response.

        Args:
            interaction (InteractionModel): The user's interaction

        Returns:
            PathResponse: Processed response object.
        """
       
        return resource_info
    
    @classmethod
    def get_sample_sql(setup_conversation) :
        return setup_conversation

# Clase base con métodos
class SecondService(metaclass=TimeoutManager):
    @staticmethod
    def setup_chat(setup_conversation):
        metadata_response: dict = FirstService().persist_metadata(
            setup_conversation
        )
        sql_code_sample: str = FirstService().get_sample_sql(setup_conversation)
        response: dict = {
            "metadata": metadata_response,
            "sql_code_sample": sql_code_sample,
        }
        return response


class TestBaseServices:

    def test_setup_conversation_chat(
        self
    ):
        mock_setup_conversation_chat = "HELLO"
        metadata_mock_response: dict = {"response": "value", "is_saved_succes": True}
        sql_code_sample_mock: str = "SELECT * FROM table;"

        with patch.object(
            FirstService, "persist_metadata", return_value=metadata_mock_response
        ) as mock_persist:
            with patch.object(
                FirstService, "get_sample_sql", return_value=sql_code_sample_mock
            ) as mock_get_sql:
                response = SecondService.setup_chat(
                    mock_setup_conversation_chat
                )

                print("Persist Metadata Calls:", mock_persist.call_count)
                print("Get Sample SQL Calls:", mock_get_sql.call_count)

                mock_persist.assert_called_once_with(mock_setup_conversation_chat)
                mock_get_sql.assert_called_once_with(mock_setup_conversation_chat)

                assert "metadata" in response and "sql_code_sample" in response
                assert response["metadata"] == metadata_mock_response
                assert response["sql_code_sample"] == sql_code_sample_mock

if __name__ == "__main__":
    unittest.main()


Because by removing class SecondService(metaclass=TimeoutManager) and leaving only class SecondService, that is, removing the inheritance of the metaclass in SecondService, does the test work? How does metaclass inheritance affect the test's failure?

0

There are 0 best solutions below