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

Code before:
from redis import Redis
from redis.connection import ConnectionInterface
from redis.maintenance_events import (
    EndpointType,
    MaintenanceEventsConfig,
    MaintenanceState,
)
from tests.test_scenario.conftest import (
    CLIENT_TIMEOUT,
    RELAX_TIMEOUT,
    _get_client_maint_events,
)
from tests.test_scenario.fault_injector_client import (
    FaultInjectorClient,
)
from tests.test_scenario.hitless_upgrade_helpers import (
    ClientValidations,
    ClusterOperations,
)

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s %(message)s",
    filemode="w",
    filename="./test_hitless_upgrade.log",
)

BIND_TIMEOUT = 30
MIGRATE_TIMEOUT = 60
FAILOVER_TIMEOUT = 15


class TestPushNotifications:
    """
    Test Redis Enterprise maintenance push notifications with real cluster
    operations.
    """

    @pytest.fixture(autouse=True)
    def setup_and_cleanup(
        self,
        client_maint_events: Redis,
        fault_injector_client: FaultInjectorClient,
        endpoints_config: Dict[str, Any],
        endpoint_name: str,
    ):
        # Initialize cleanup flags first to ensure they exist even if setup fails
        self._failover_executed = False
        self._migration_executed = False
        self._bind_executed = False
        self.target_node = None

Review comment (bug):
[nitpick] The timeout values have been significantly reduced (BIND_TIMEOUT from 60 to 30, MIGRATE_TIMEOUT from 120 to 60). Consider documenting the rationale for these changes or ensuring they provide sufficient time for operations to complete in slower environments.
```suggestion

# Timeout values have been reduced based on recent empirical test runs showing
# that bind, migrate, and failover operations consistently complete well within these limits.
# If running in a slower or less performant environment, consider increasing these values.
```

Code after:

from redis import Redis
from redis.connection import ConnectionInterface
from redis.maintenance_events import (
    EndpointType,
    MaintenanceEventsConfig,
    MaintenanceState,
)
from tests.test_scenario.conftest import (
    CLIENT_TIMEOUT,
    RELAX_TIMEOUT,
    _get_client_maint_events,
)
from tests.test_scenario.fault_injector_client import (
    FaultInjectorClient,
)
from tests.test_scenario.hitless_upgrade_helpers import (
    ClientValidations,
    ClusterOperations,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

BIND_TIMEOUT = 30
MIGRATE_TIMEOUT = 60
FAILOVER_TIMEOUT = 15

DEFAULT_BIND_TTL = 15


class TestPushNotifications:
    """
    Test Redis Enterprise maintenance push notifications with real cluster
    operations.
    """

    @pytest.fixture(autouse=True)
    def setup_and_cleanup(
        self,
        client_maint_events: Redis,
        fault_injector_client: FaultInjectorClient,
        endpoints_config: Dict[str, Any],
        endpoint_name: str,
    ):
        # Initialize cleanup flags first to ensure they exist even if setup fails
        self._failover_executed = False
        self._migration_executed = False
        self._bind_executed = False