C:\Users\nkiselev\PycharmProjects\etl-framework-dev\env\Scripts\python.exe "C:/Program Files/JetBrains/PyCharm Community Edition 2022.3.3/plugins/python-ce/helpers/pycharm/_jb_pytest_runner.py" --target test_monitoring.py::test_get_failed_tis_with_specific_tags
Testing started at 13:10 ...
Launching pytest with arguments test_monitoring.py::test_get_failed_tis_with_specific_tags --no-header --no-summary -q in C:\Users\nkiselev\PycharmProjects\etl-framework-dev\main\modules\airflow\dags\tests\unit_tests\services

============================= test session starts =============================
collecting ... collected 1 item

test_monitoring.py::test_get_failed_tis_with_specific_tags FAILED        [100%]
main\modules\airflow\dags\tests\unit_tests\services\test_monitoring.py:46 (test_get_failed_tis_with_specific_tags)
mock_get_ti_log_lines = <MagicMock name='get_ti_log_lines_with_specific_log_levels' id='2553320663840'>
mock_get_failed_tis = <MagicMock name='_get_failed_tis_by_period_with_specific_tags' id='2553316808496'>
mock_get_data_interval = <MagicMock name='get_data_interval' id='2553320707552'>
mock_get_context = <MagicMock name='get_current_context' id='2553320719600'>

    @patch('airflow.operators.python.get_current_context')
    @patch('dependencies.services.utils.get_data_interval')
    @patch('dependencies.services.monitoring.monitoring._get_failed_tis_by_period_with_specific_tags')
    @patch('dependencies.services.monitoring.monitoring.get_ti_log_lines_with_specific_log_levels')
    def test_get_failed_tis_with_specific_tags(mock_get_ti_log_lines, mock_get_failed_tis, mock_get_data_interval, mock_get_context):
        # Настройка моков
        mock_get_context.return_value = MagicMock()
        mock_get_data_interval.return_value = (MagicMock(), MagicMock())
        mock_get_failed_tis.return_value = [MagicMock()]
        mock_get_ti_log_lines.return_value = ['Log line 1', 'Log line 2']

        # Вызов тестируемой функции
>       result = get_failed_tis_with_specific_tags(['tag1', 'tag2'], ['INFO', 'WARNING'])

test_monitoring.py:59:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\..\..\dependencies\services\monitoring\monitoring.py:52: in get_failed_tis_with_specific_tags
    context = get_current_context()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    def get_current_context() -> Context:
        """
        Obtain the execution context for the currently executing operator without
        altering user method's signature.
        This is the simplest method of retrieving the execution context dictionary.

        **Old style:**

        .. code:: python

            def my_task(**context):
                ti = context["ti"]

        **New style:**

        .. code:: python

            from airflow.operators.python import get_current_context


            def my_task():
                context = get_current_context()
                ti = context["ti"]

        Current context will only have value if this method was called after an operator
        was starting to execute.
        """
        if not _CURRENT_CONTEXT:
>           raise AirflowException(
                "Current context was requested but no context was found! "
                "Are you running within an airflow task?"
            )
E           airflow.exceptions.AirflowException: Current context was requested but no context was found! Are you running within an airflow task?

..\..\..\..\..\..\..\env\lib\site-packages\airflow\operators\python.py:748: AirflowException
















======================== 1 failed, 1 warning in 1.75s =========================

Process finished with exit code 1