The fix has a bug (subscribe happens too frequently)
#
#
#
from __future__ import annotations
from unittest.mock import MagicMock, call, patch
from airflow.models.dag import DAG
from airflow.providers.redis.sensors.redis_pub_sub import RedisPubSubSensor
from airflow.utils import timezone
DEFAULT_DATE = timezone.datetime(2017, 1, 1)
class TestRedisPubSubSensor:
    def setup_method(self):
        args = {"owner": "airflow", "start_date": DEFAULT_DATE}
        self.dag = DAG("test_dag_id", default_args=args)
        self.mock_context = MagicMock()
    @patch("airflow.providers.redis.hooks.redis.RedisHook.get_conn")
    def test_poke_mock_true(self, mock_redis_conn):