Repository: redis/redis-py
File: tests/test_scenario/test_active_active.py (Python)

Code before:
            target=trigger_network_failure_action,
            daemon=True,
            args=(fault_injector_client,config,event)
        )

        def callback(pipe: Pipeline):
            pipe.set('{hash}key1', 'value1')
            pipe.set('{hash}key2', 'value2')
            pipe.set('{hash}key3', 'value3')
            pipe.get('{hash}key1')
            pipe.get('{hash}key2')
            pipe.get('{hash}key3')

        # Client initialized on first transaction execution.
        r_multi_db.transaction(callback)
        thread.start()

        # Execute pipeline before network failure
        while not event.is_set():
            r_multi_db.transaction(callback)
            sleep(0.1)

        # Execute pipeline until database failover
        while not listener.is_changed_flag:
            r_multi_db.transaction(callback)
            sleep(0.1)

    @pytest.mark.parametrize(
        "r_multi_db",
        [
            {"client_class": Redis, "failure_threshold": 2},
            {"client_class": RedisCluster, "failure_threshold": 2},
        ],
        ids=["standalone", "cluster"],
        indirect=True
    )
    @pytest.mark.timeout(50)
    def test_pubsub_failover_to_another_db(self, r_multi_db, fault_injector_client):
        r_multi_db, listener, config = r_multi_db

        event = threading.Event()
        thread = threading.Thread(
            target=trigger_network_failure_action,
            daemon=True,
            args=(fault_injector_client,config,event)
        )
        data = json.dumps({'message': 'test'})
        messages_count = 0

        def handler(message):
            nonlocal messages_count

Review comment (suggestion):
0.1 might be too spamy on the logs, I usually do 0.5 or 1-3 seconds, but Idk if in your usecase its not 100% needed to be 0.1

Code after:
            target=trigger_network_failure_action,
            daemon=True,
            args=(fault_injector_client,config,event)
        )

        def callback(pipe: Pipeline):
            pipe.set('{hash}key1', 'value1')
            pipe.set('{hash}key2', 'value2')
            pipe.set('{hash}key3', 'value3')
            pipe.get('{hash}key1')
            pipe.get('{hash}key2')
            pipe.get('{hash}key3')

        # Client initialized on first transaction execution.
        r_multi_db.transaction(callback)
        thread.start()

        # Execute transaction before network failure
        while not event.is_set():
            r_multi_db.transaction(callback)
            sleep(0.5)

        # Execute transaction until database failover
        while not listener.is_changed_flag:
            r_multi_db.transaction(callback)
            sleep(0.5)

    @pytest.mark.parametrize(
        "r_multi_db",
        [
            {"client_class": Redis, "failure_threshold": 2},
            {"client_class": RedisCluster, "failure_threshold": 2},
        ],
        ids=["standalone", "cluster"],
        indirect=True
    )
    @pytest.mark.timeout(50)
    def test_pubsub_failover_to_another_db(self, r_multi_db, fault_injector_client):
        r_multi_db, listener, config = r_multi_db

        event = threading.Event()
        thread = threading.Thread(
            target=trigger_network_failure_action,
            daemon=True,
            args=(fault_injector_client,config,event)
        )
        data = json.dumps({'message': 'test'})
        messages_count = 0

        def handler(message):
            nonlocal messages_count