From 190f651b695efdea29c1c3208bc69cb076292f10 Mon Sep 17 00:00:00 2001
From: Igor Malinovskiy <u.glide@gmail.com>
Date: Fri, 14 Jun 2024 19:18:27 +0200
Subject: [PATCH 1/5] Add scenario tests for connection interruptions

---
 dev_requirements.txt                          |   2 +
 tests/conftest.py                             |   7 +
 tests/scenario/__init__.py                    |  54 ++++++
 tests/scenario/fake_app.py                    |  63 +++++++
 tests/scenario/fault_injection_client.py      |  39 +++++
 .../scenario/test_connection_interruptions.py | 157 ++++++++++++++++++
 tests/test_asyncio/scenario/__init__.py       |   0
 tests/test_asyncio/scenario/fake_app.py       |  63 +++++++
 .../scenario/fault_injection_client.py        |  40 +++++
 .../scenario/test_connection_interruptions.py | 144 ++++++++++++++++
 10 files changed, 569 insertions(+)
 create mode 100644 tests/scenario/__init__.py
 create mode 100644 tests/scenario/fake_app.py
 create mode 100644 tests/scenario/fault_injection_client.py
 create mode 100644 tests/scenario/test_connection_interruptions.py
 create mode 100644 tests/test_asyncio/scenario/__init__.py
 create mode 100644 tests/test_asyncio/scenario/fake_app.py
 create mode 100644 tests/test_asyncio/scenario/fault_injection_client.py
 create mode 100644 tests/test_asyncio/scenario/test_connection_interruptions.py

diff --git a/dev_requirements.txt b/dev_requirements.txt
index a8da4b49cd..eae75a3539 100644
--- a/dev_requirements.txt
+++ b/dev_requirements.txt
@@ -17,3 +17,5 @@ uvloop
 vulture>=2.3.0
 wheel>=0.30.0
 numpy>=1.24.0
+requests>=2.23.0
+aiohttp>=3.0.0
\ No newline at end of file
diff --git a/tests/conftest.py b/tests/conftest.py
index dd78bb6a2c..8c7164ba57 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -134,6 +134,13 @@ def pytest_addoption(parser):
         help="Name of the Redis master service that the sentinels are monitoring",
     )
 
+    parser.addoption(
+        "--endpoints-config",
+        action="store",
+        default="endpoints.json",
+        help="Path to the Redis endpoints configuration file",
+    )
+
 
 def _get_info(redis_url):
     client = redis.Redis.from_url(redis_url)
diff --git a/tests/scenario/__init__.py b/tests/scenario/__init__.py
new file mode 100644
index 0000000000..fe74bb1d55
--- /dev/null
+++ b/tests/scenario/__init__.py
@@ -0,0 +1,54 @@
+import json
+import os.path
+import dataclasses
+
+import pytest
+from urllib.parse import urlparse
+
+
+@dataclasses.dataclass
+class Endpoint:
+    bdb_id: int
+    username: str
+    password: str
+    tls: bool
+    endpoints: list[str]
+
+    @property
+    def url(self):
+        parsed_url = urlparse(self.endpoints[0])
+
+        if self.tls:
+            parsed_url = parsed_url._replace(scheme="rediss")
+
+        domain = parsed_url.netloc.split("@")[-1]
+        domain = f"{self.username}:{self.password}@{domain}"
+
+        parsed_url = parsed_url._replace(netloc=domain)
+
+        return parsed_url.geturl()
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        field_names = set(f.name for f in dataclasses.fields(cls))
+        return cls(**{k: v for k, v in data.items() if k in field_names})
+
+
+def get_endpoint(request: pytest.FixtureRequest, endpoint_name: str) -> Endpoint:
+    endpoints_config_path = request.config.getoption("--endpoints-config")
+
+    if not (endpoints_config_path and os.path.exists(endpoints_config_path)):
+        raise ValueError(f"Endpoints config file not found: {endpoints_config_path}")
+
+    try:
+        with open(endpoints_config_path, "r") as f:
+            endpoints_config = json.load(f)
+    except Exception as e:
+        raise ValueError(
+            f"Failed to load endpoints config file: {endpoints_config_path}"
+        ) from e
+
+    if not (isinstance(endpoints_config, dict) and endpoint_name in endpoints_config):
+        raise ValueError(f"Endpoint not found in config: {endpoint_name}")
+
+    return Endpoint.from_dict(endpoints_config.get(endpoint_name))
diff --git a/tests/scenario/fake_app.py b/tests/scenario/fake_app.py
new file mode 100644
index 0000000000..0528acc359
--- /dev/null
+++ b/tests/scenario/fake_app.py
@@ -0,0 +1,63 @@
+import multiprocessing
+import typing
+
+from threading import Thread, Event
+from multiprocessing import Process, Event as PEvent
+from unittest.mock import patch
+
+from redis import Redis
+
+
+class FakeApp:
+
+    def __init__(self, client: Redis, logic: typing.Callable[[Redis], None]):
+        self.client = client
+        self.logic = logic
+        self.disconnects = 0
+
+    def run(self) -> (Event, Thread):
+        e = Event()
+        t = Thread(target=self._run_logic, args=(e,))
+        t.start()
+        return e, t
+
+    def _run_logic(self, e: Event):
+        with patch.object(
+            self.client, "_disconnect_raise", wraps=self.client._disconnect_raise
+        ) as spy:
+            while not e.is_set():
+                self.logic(self.client)
+
+            self.disconnects = spy.call_count
+
+
+class FakeSubscriber:
+
+    def __init__(self, client: Redis, logic: typing.Callable[[dict], None]):
+        self.client = client
+        self.logic = logic
+        self.disconnects = multiprocessing.Value("i", 0)
+
+    def run(self, channel: str) -> (PEvent, Process):
+        e, started = PEvent(), PEvent()
+        p = Process(target=self._run_logic, args=(e, started, channel))
+        p.start()
+        return e, started, p
+
+    def _run_logic(self, should_stop: PEvent, started: PEvent, channel: str):
+        pubsub = self.client.pubsub()
+
+        with patch.object(
+            pubsub, "_disconnect_raise_connect", wraps=pubsub._disconnect_raise_connect
+        ) as spy_pubsub:
+            pubsub.subscribe(channel)
+
+            started.set()
+
+            while not should_stop.is_set():
+                message = pubsub.get_message(ignore_subscribe_messages=True, timeout=1)
+
+                if message:
+                    self.logic(message)
+
+            self.disconnects.value = spy_pubsub.call_count
diff --git a/tests/scenario/fault_injection_client.py b/tests/scenario/fault_injection_client.py
new file mode 100644
index 0000000000..750298779d
--- /dev/null
+++ b/tests/scenario/fault_injection_client.py
@@ -0,0 +1,39 @@
+import requests
+
+
+class TriggeredAction:
+
+    def __init__(self, client: "FaultInjectionClient", data: dict):
+        self.client = client
+        self.action_id = data["action_id"]
+        self.data = data
+
+    def refresh(self):
+        self.data = self.client.get_action(self.action_id)
+
+    @property
+    def status(self):
+        if "status" not in self.data:
+            return "pending"
+        return self.data["status"]
+
+    def wait_until_complete(self):
+        while self.status not in ("success", "failed"):
+            self.refresh()
+        return self.status
+
+
+class FaultInjectionClient:
+    def __init__(self, base_url: str = "http://127.0.0.1:20324"):
+        self.base_url = base_url
+
+    def trigger_action(self, action_type: str, parameters: dict):
+        response = requests.post(
+            f"{self.base_url}/action",
+            json={"type": action_type, "parameters": parameters},
+        )
+        return TriggeredAction(self, response.json())
+
+    def get_action(self, action_id: str):
+        response = requests.get(f"{self.base_url}/action/{action_id}")
+        return response.json()
diff --git a/tests/scenario/test_connection_interruptions.py b/tests/scenario/test_connection_interruptions.py
new file mode 100644
index 0000000000..1573483cc2
--- /dev/null
+++ b/tests/scenario/test_connection_interruptions.py
@@ -0,0 +1,157 @@
+import multiprocessing
+
+import socket
+import time
+
+import pytest
+
+
+from redis import Redis, BusyLoadingError
+from redis.backoff import ExponentialBackoff
+from redis.retry import Retry
+from redis.exceptions import (
+    ConnectionError as RedisConnectionError,
+    TimeoutError as RedisTimeoutError,
+)
+
+from ..conftest import _get_client
+from . import get_endpoint, Endpoint
+from .fake_app import FakeApp, FakeSubscriber
+from .fault_injection_client import FaultInjectionClient, TriggeredAction
+
+
+@pytest.fixture
+def endpoint_name():
+    return "re-standalone"
+
+
+@pytest.fixture
+def endpoint(request: pytest.FixtureRequest, endpoint_name: str):
+    return get_endpoint(request, endpoint_name)
+
+
+@pytest.fixture
+def clients(request: pytest.FixtureRequest, endpoint: Endpoint):
+    # Use Recommended settings
+    retry = Retry(ExponentialBackoff(base=1), 3)
+
+    clients = []
+
+    for _ in range(2):
+        r = _get_client(
+            Redis,
+            request,
+            decode_responses=True,
+            from_url=endpoint.url,
+            retry=retry,
+            retry_on_error=[
+                BusyLoadingError,
+                RedisConnectionError,
+                RedisTimeoutError,
+                # FIXME: This is a workaround for a bug in redis-py
+                # https://github.com/redis/redis-py/issues/3203
+                ConnectionError,
+                TimeoutError,
+            ],
+        )
+        r.flushdb()
+        clients.append(r)
+    return clients
+
+
+@pytest.fixture
+def fault_injection_client(request: pytest.FixtureRequest):
+    return FaultInjectionClient()
+
+
+@pytest.mark.parametrize("action", ("dmc_restart", "network_failure"))
+def test_connection_interruptions(
+    clients: list[Redis],
+    endpoint: Endpoint,
+    fault_injection_client: FaultInjectionClient,
+    action: str,
+):
+    client = clients.pop()
+    app = FakeApp(client, lambda c: c.set("foo", "bar"))
+
+    stop_app, thread = app.run()
+
+    triggered_action = fault_injection_client.trigger_action(
+        action, {"bdb_id": endpoint.bdb_id}
+    )
+
+    triggered_action.wait_until_complete()
+
+    stop_app.set()
+    thread.join()
+
+    if triggered_action.status == "failed":
+        pytest.fail(f"Action failed: {triggered_action.data['error']}")
+
+    assert app.disconnects > 0, "Client did not disconnect"
+
+
+@pytest.mark.parametrize("action", ("dmc_restart", "network_failure"))
+def test_pubsub_with_connection_interruptions(
+    clients: list[Redis],
+    endpoint: Endpoint,
+    fault_injection_client: FaultInjectionClient,
+    action: str,
+):
+    channel = "test"
+
+    # Subscriber is executed in a separate process to ensure it reacts
+    # to the disconnection at the same time as the publisher
+    with multiprocessing.Manager() as manager:
+        received_messages = manager.list()
+
+        def read_message(message):
+            nonlocal received_messages
+            if message and message["type"] == "message":
+                received_messages.append(message["data"])
+
+        subscriber_client = clients.pop()
+        subscriber = FakeSubscriber(subscriber_client, read_message)
+        stop_subscriber, subscriber_started, subscriber_t = subscriber.run(channel)
+
+        # Allow subscriber subscribe to the channel
+        subscriber_started.wait(timeout=5)
+
+        messages_sent = 0
+
+        def publish_message(c):
+            nonlocal messages_sent, channel
+            messages_sent += 1
+            c.publish(channel, messages_sent)
+
+        publisher_client = clients.pop()
+        publisher = FakeApp(publisher_client, publish_message)
+        stop_publisher, publisher_t = publisher.run()
+
+        triggered_action = fault_injection_client.trigger_action(
+            action, {"bdb_id": endpoint.bdb_id}
+        )
+
+        triggered_action.wait_until_complete()
+        last_message_sent_after_trigger = messages_sent
+
+        time.sleep(3)  # Wait for the publisher to send more messages
+
+        stop_publisher.set()
+        publisher_t.join()
+
+        stop_subscriber.set()
+        subscriber_t.join()
+
+        assert publisher.disconnects > 0
+        assert subscriber.disconnects.value > 0
+
+        if triggered_action.status == "failed":
+            pytest.fail(f"Action failed: {triggered_action.data['error']}")
+
+        assert (
+            last_message_sent_after_trigger < messages_sent
+        ), "No messages were sent after the failure"
+        assert (
+            int(received_messages[-1]) == messages_sent
+        ), "Not all messages were received"
diff --git a/tests/test_asyncio/scenario/__init__.py b/tests/test_asyncio/scenario/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/test_asyncio/scenario/fake_app.py b/tests/test_asyncio/scenario/fake_app.py
new file mode 100644
index 0000000000..598c47eaa8
--- /dev/null
+++ b/tests/test_asyncio/scenario/fake_app.py
@@ -0,0 +1,63 @@
+import typing
+
+from asyncio import create_task, Event, Task
+from unittest.mock import patch
+
+from redis.asyncio import Redis
+
+
+class AsyncFakeApp:
+
+    def __init__(
+        self, client: Redis, logic: typing.Callable[[Redis], typing.Awaitable[None]]
+    ):
+        self.client = client
+        self.logic = logic
+        self.disconnects = 0
+
+    async def run(self):
+        e = Event()
+        t = create_task(self._run_logic(e))
+        return e, t
+
+    async def _run_logic(self, e: Event):
+        with patch.object(
+            self.client, "_disconnect_raise", wraps=self.client._disconnect_raise
+        ) as spy:
+            while not e.is_set():
+                await self.logic(self.client)
+
+            self.disconnects = spy.call_count
+
+
+class AsyncFakeSubscriber:
+
+    def __init__(
+        self, client: Redis, logic: typing.Callable[[dict], typing.Awaitable[None]]
+    ):
+        self.client = client
+        self.logic = logic
+        self.disconnects = 0
+
+    async def run(self, channel: str) -> (Event, Task):
+        e = Event()
+        t = create_task(self._run_logic(e, channel))
+        return e, t
+
+    async def _run_logic(self, should_stop: Event, channel: str):
+        pubsub = self.client.pubsub()
+
+        with patch.object(
+            pubsub, "_disconnect_raise_connect", wraps=pubsub._disconnect_raise_connect
+        ) as spy_pubsub:
+            await pubsub.subscribe(channel)
+
+            while not should_stop.is_set():
+                message = await pubsub.get_message(
+                    ignore_subscribe_messages=True, timeout=1
+                )
+
+                if message:
+                    await self.logic(message)
+
+            self.disconnects = spy_pubsub.call_count
diff --git a/tests/test_asyncio/scenario/fault_injection_client.py b/tests/test_asyncio/scenario/fault_injection_client.py
new file mode 100644
index 0000000000..4fcc685ded
--- /dev/null
+++ b/tests/test_asyncio/scenario/fault_injection_client.py
@@ -0,0 +1,40 @@
+import aiohttp
+
+
+class TriggeredAction:
+
+    def __init__(self, client: "AsyncFaultInjectionClient", data: dict):
+        self.client = client
+        self.action_id = data["action_id"]
+        self.data = data
+
+    async def refresh(self):
+        self.data = await self.client.get_action(self.action_id)
+
+    @property
+    def status(self):
+        if "status" not in self.data:
+            return "pending"
+        return self.data["status"]
+
+    async def wait_until_complete(self):
+        while self.status not in ("success", "failed"):
+            await self.refresh()
+        return self.status
+
+
+class AsyncFaultInjectionClient:
+    def __init__(self, base_url: str = "http://127.0.0.1:20324"):
+        self.base_url = base_url
+        self.session = aiohttp.ClientSession()
+
+    async def trigger_action(self, action_type: str, parameters: dict):
+        async with self.session.post(
+            f"{self.base_url}/action",
+            json={"type": action_type, "parameters": parameters},
+        ) as response:
+            return TriggeredAction(self, await response.json())
+
+    async def get_action(self, action_id: str):
+        async with self.session.get(f"{self.base_url}/action/{action_id}") as response:
+            return await response.json()
diff --git a/tests/test_asyncio/scenario/test_connection_interruptions.py b/tests/test_asyncio/scenario/test_connection_interruptions.py
new file mode 100644
index 0000000000..f293ffe5df
--- /dev/null
+++ b/tests/test_asyncio/scenario/test_connection_interruptions.py
@@ -0,0 +1,144 @@
+import asyncio
+import pytest
+
+from redis.asyncio import Redis, BusyLoadingError
+from redis.backoff import ExponentialBackoff
+from redis.asyncio.retry import Retry
+from redis.exceptions import ConnectionError as RedisConnectionError, TimeoutError
+
+from ..conftest import create_redis
+from tests.scenario import get_endpoint, Endpoint
+from .fake_app import AsyncFakeApp, AsyncFakeSubscriber
+from .fault_injection_client import AsyncFaultInjectionClient
+
+
+@pytest.fixture
+async def endpoint_name():
+    return "re-standalone"
+
+
+@pytest.fixture
+async def endpoint(request: pytest.FixtureRequest, endpoint_name: str):
+    return get_endpoint(request, endpoint_name)
+
+
+@pytest.fixture
+async def clients(
+    request: pytest.FixtureRequest, endpoint: Endpoint, create_redis: callable
+):
+    # Use Recommended settings
+    retry = Retry(ExponentialBackoff(base=1), 5)
+
+    clients = []
+
+    for _ in range(2):
+        client = await create_redis(
+            endpoint.url,
+            decode_responses=True,
+            retry=retry,
+            retry_on_error=[
+                BusyLoadingError,
+                RedisConnectionError,
+                TimeoutError,
+                # FIXME: This is a workaround for a bug in redis-py
+                # https://github.com/redis/redis-py/issues/3203
+                ConnectionError,
+                OSError,
+            ],
+            retry_on_timeout=True,
+        )
+        await client.flushdb()
+        clients.append(client)
+
+    return clients
+
+
+@pytest.fixture
+async def fault_injection_client(request: pytest.FixtureRequest):
+    return AsyncFaultInjectionClient()
+
+
+@pytest.mark.parametrize("action", ("dmc_restart", "network_failure"))
+async def test_connection_interruptions(
+    clients: list[Redis],
+    endpoint: Endpoint,
+    fault_injection_client: AsyncFaultInjectionClient,
+    action: str,
+):
+    client = clients.pop()
+    app = AsyncFakeApp(client, lambda c: c.set("foo", "bar"))
+
+    stop_app, task = await app.run()
+
+    triggered_action = await fault_injection_client.trigger_action(
+        action, {"bdb_id": endpoint.bdb_id}
+    )
+
+    await triggered_action.wait_until_complete()
+
+    stop_app.set()
+    await task
+
+    if triggered_action.status == "failed":
+        pytest.fail(f"Action failed: {triggered_action.data['error']}")
+
+    assert app.disconnects > 0
+
+
+@pytest.mark.parametrize("action", ("dmc_restart",))  # "network_failure"))
+async def test_pubsub_with_connection_interruptions(
+    clients: list[Redis],
+    endpoint: Endpoint,
+    fault_injection_client: AsyncFaultInjectionClient,
+    action: str,
+):
+    channel = "test"
+
+    received_messages = []
+
+    async def read_message(message):
+        nonlocal received_messages
+        if message and message["type"] == "message":
+            received_messages.append(message["data"])
+
+    messages_sent = 0
+
+    async def publish_message(c):
+        nonlocal messages_sent, channel
+        messages_sent += 1
+        await c.publish(channel, messages_sent)
+
+    subscriber_client = clients.pop()
+    publisher_client = clients.pop()
+
+    subscriber = AsyncFakeSubscriber(subscriber_client, read_message)
+    stop_subscriber, subscriber_t = await subscriber.run(channel)
+
+    publisher = AsyncFakeApp(publisher_client, publish_message)
+    stop_publisher, publisher_t = await publisher.run()
+
+    triggered_action = await fault_injection_client.trigger_action(
+        action, {"bdb_id": endpoint.bdb_id}
+    )
+
+    await triggered_action.wait_until_complete()
+    last_message_sent_after_trigger = messages_sent
+
+    if triggered_action.status == "failed":
+        pytest.fail(f"Action failed: {triggered_action.data['error']}")
+
+    await asyncio.sleep(3)
+
+    stop_publisher.set()
+    await publisher_t
+
+    stop_subscriber.set()
+    await subscriber_t
+
+    assert publisher.disconnects > 0
+    assert subscriber.disconnects > 0
+
+    assert (
+        last_message_sent_after_trigger < messages_sent
+    ), "No messages were sent after the failure"
+    assert int(received_messages[-1]) == messages_sent, "Not all messages were received"

From 8006ed01e3bc15925a8d5f047c3fbdf1e2141ef0 Mon Sep 17 00:00:00 2001
From: Igor Malinovskiy <u.glide@gmail.com>
Date: Tue, 23 Jul 2024 14:37:55 +0200
Subject: [PATCH 2/5] Fix formatting and skip tests in default CI pipelines

---
 tests/scenario/__init__.py                    |  8 ++++---
 tests/scenario/fake_app.py                    |  6 ++---
 .../scenario/test_connection_interruptions.py | 23 +++++++++----------
 tests/test_asyncio/scenario/fake_app.py       |  3 +--
 .../scenario/test_connection_interruptions.py | 19 +++++++++------
 5 files changed, 32 insertions(+), 27 deletions(-)

diff --git a/tests/scenario/__init__.py b/tests/scenario/__init__.py
index fe74bb1d55..80f79e7524 100644
--- a/tests/scenario/__init__.py
+++ b/tests/scenario/__init__.py
@@ -1,9 +1,9 @@
+import dataclasses
 import json
 import os.path
-import dataclasses
+from urllib.parse import urlparse
 
 import pytest
-from urllib.parse import urlparse
 
 
 @dataclasses.dataclass
@@ -38,7 +38,9 @@ def get_endpoint(request: pytest.FixtureRequest, endpoint_name: str) -> Endpoint
     endpoints_config_path = request.config.getoption("--endpoints-config")
 
     if not (endpoints_config_path and os.path.exists(endpoints_config_path)):
-        raise ValueError(f"Endpoints config file not found: {endpoints_config_path}")
+        raise FileNotFoundError(
+            f"Endpoints config file not found: {endpoints_config_path}"
+        )
 
     try:
         with open(endpoints_config_path, "r") as f:
diff --git a/tests/scenario/fake_app.py b/tests/scenario/fake_app.py
index 0528acc359..e643f52591 100644
--- a/tests/scenario/fake_app.py
+++ b/tests/scenario/fake_app.py
@@ -1,8 +1,8 @@
 import multiprocessing
 import typing
-
-from threading import Thread, Event
-from multiprocessing import Process, Event as PEvent
+from multiprocessing import Event as PEvent
+from multiprocessing import Process
+from threading import Event, Thread
 from unittest.mock import patch
 
 from redis import Redis
diff --git a/tests/scenario/test_connection_interruptions.py b/tests/scenario/test_connection_interruptions.py
index 1573483cc2..18df7fe05b 100644
--- a/tests/scenario/test_connection_interruptions.py
+++ b/tests/scenario/test_connection_interruptions.py
@@ -1,23 +1,17 @@
 import multiprocessing
-
-import socket
 import time
 
 import pytest
-
-
-from redis import Redis, BusyLoadingError
+from redis import BusyLoadingError, Redis
 from redis.backoff import ExponentialBackoff
+from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import TimeoutError as RedisTimeoutError
 from redis.retry import Retry
-from redis.exceptions import (
-    ConnectionError as RedisConnectionError,
-    TimeoutError as RedisTimeoutError,
-)
 
 from ..conftest import _get_client
-from . import get_endpoint, Endpoint
+from . import Endpoint, get_endpoint
 from .fake_app import FakeApp, FakeSubscriber
-from .fault_injection_client import FaultInjectionClient, TriggeredAction
+from .fault_injection_client import FaultInjectionClient
 
 
 @pytest.fixture
@@ -27,7 +21,12 @@ def endpoint_name():
 
 @pytest.fixture
 def endpoint(request: pytest.FixtureRequest, endpoint_name: str):
-    return get_endpoint(request, endpoint_name)
+    try:
+        return get_endpoint(request, endpoint_name)
+    except FileNotFoundError as e:
+        pytest.skip(
+            f"Skipping scenario test because endpoints file is missing: {str(e)}"
+        )
 
 
 @pytest.fixture
diff --git a/tests/test_asyncio/scenario/fake_app.py b/tests/test_asyncio/scenario/fake_app.py
index 598c47eaa8..ea4d7fd22a 100644
--- a/tests/test_asyncio/scenario/fake_app.py
+++ b/tests/test_asyncio/scenario/fake_app.py
@@ -1,6 +1,5 @@
 import typing
-
-from asyncio import create_task, Event, Task
+from asyncio import Event, Task, create_task
 from unittest.mock import patch
 
 from redis.asyncio import Redis
diff --git a/tests/test_asyncio/scenario/test_connection_interruptions.py b/tests/test_asyncio/scenario/test_connection_interruptions.py
index f293ffe5df..ec3270e446 100644
--- a/tests/test_asyncio/scenario/test_connection_interruptions.py
+++ b/tests/test_asyncio/scenario/test_connection_interruptions.py
@@ -1,13 +1,13 @@
 import asyncio
-import pytest
 
-from redis.asyncio import Redis, BusyLoadingError
-from redis.backoff import ExponentialBackoff
+import pytest
+from redis.asyncio import BusyLoadingError, Redis
 from redis.asyncio.retry import Retry
-from redis.exceptions import ConnectionError as RedisConnectionError, TimeoutError
+from redis.backoff import ExponentialBackoff
+from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import TimeoutError
+from tests.scenario import Endpoint, get_endpoint
 
-from ..conftest import create_redis
-from tests.scenario import get_endpoint, Endpoint
 from .fake_app import AsyncFakeApp, AsyncFakeSubscriber
 from .fault_injection_client import AsyncFaultInjectionClient
 
@@ -19,7 +19,12 @@ async def endpoint_name():
 
 @pytest.fixture
 async def endpoint(request: pytest.FixtureRequest, endpoint_name: str):
-    return get_endpoint(request, endpoint_name)
+    try:
+        return get_endpoint(request, endpoint_name)
+    except FileNotFoundError as e:
+        pytest.skip(
+            f"Skipping scenario test because endpoints file is missing: {str(e)}"
+        )
 
 
 @pytest.fixture

From 880ba22fadc109f3ade324486e6235be3e87b24a Mon Sep 17 00:00:00 2001
From: Igor Malinovskiy <u.glide@gmail.com>
Date: Tue, 23 Jul 2024 16:03:48 +0200
Subject: [PATCH 3/5] Make typehints py 3.8 compatible

---
 tests/scenario/test_connection_interruptions.py              | 5 +++--
 tests/test_asyncio/scenario/test_connection_interruptions.py | 5 +++--
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/tests/scenario/test_connection_interruptions.py b/tests/scenario/test_connection_interruptions.py
index 18df7fe05b..a1ce067108 100644
--- a/tests/scenario/test_connection_interruptions.py
+++ b/tests/scenario/test_connection_interruptions.py
@@ -1,5 +1,6 @@
 import multiprocessing
 import time
+from typing import List
 
 import pytest
 from redis import BusyLoadingError, Redis
@@ -65,7 +66,7 @@ def fault_injection_client(request: pytest.FixtureRequest):
 
 @pytest.mark.parametrize("action", ("dmc_restart", "network_failure"))
 def test_connection_interruptions(
-    clients: list[Redis],
+    clients: List[Redis],
     endpoint: Endpoint,
     fault_injection_client: FaultInjectionClient,
     action: str,
@@ -92,7 +93,7 @@ def test_connection_interruptions(
 
 @pytest.mark.parametrize("action", ("dmc_restart", "network_failure"))
 def test_pubsub_with_connection_interruptions(
-    clients: list[Redis],
+    clients: List[Redis],
     endpoint: Endpoint,
     fault_injection_client: FaultInjectionClient,
     action: str,
diff --git a/tests/test_asyncio/scenario/test_connection_interruptions.py b/tests/test_asyncio/scenario/test_connection_interruptions.py
index ec3270e446..c18720cf40 100644
--- a/tests/test_asyncio/scenario/test_connection_interruptions.py
+++ b/tests/test_asyncio/scenario/test_connection_interruptions.py
@@ -1,4 +1,5 @@
 import asyncio
+from typing import List
 
 import pytest
 from redis.asyncio import BusyLoadingError, Redis
@@ -65,7 +66,7 @@ async def fault_injection_client(request: pytest.FixtureRequest):
 
 @pytest.mark.parametrize("action", ("dmc_restart", "network_failure"))
 async def test_connection_interruptions(
-    clients: list[Redis],
+    clients: List[Redis],
     endpoint: Endpoint,
     fault_injection_client: AsyncFaultInjectionClient,
     action: str,
@@ -92,7 +93,7 @@ async def test_connection_interruptions(
 
 @pytest.mark.parametrize("action", ("dmc_restart",))  # "network_failure"))
 async def test_pubsub_with_connection_interruptions(
-    clients: list[Redis],
+    clients: List[Redis],
     endpoint: Endpoint,
     fault_injection_client: AsyncFaultInjectionClient,
     action: str,

From c81959910df74fdfdcd9533cedae48f77e396706 Mon Sep 17 00:00:00 2001
From: Igor Malinovskiy <u.glide@gmail.com>
Date: Tue, 23 Jul 2024 16:32:07 +0200
Subject: [PATCH 4/5] Add missing fix

---
 tests/scenario/__init__.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/scenario/__init__.py b/tests/scenario/__init__.py
index 80f79e7524..d5655642f4 100644
--- a/tests/scenario/__init__.py
+++ b/tests/scenario/__init__.py
@@ -2,6 +2,7 @@
 import json
 import os.path
 from urllib.parse import urlparse
+from typing import List
 
 import pytest
 
@@ -12,7 +13,7 @@ class Endpoint:
     username: str
     password: str
     tls: bool
-    endpoints: list[str]
+    endpoints: List[str]
 
     @property
     def url(self):

From 83df75672123ae1a147061eb35d164c8d61a8a03 Mon Sep 17 00:00:00 2001
From: Igor Malinovskiy <u.glide@gmail.com>
Date: Tue, 23 Jul 2024 16:53:13 +0200
Subject: [PATCH 5/5] Fix formatting

---
 tests/scenario/__init__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/scenario/__init__.py b/tests/scenario/__init__.py
index d5655642f4..2b4bd0c423 100644
--- a/tests/scenario/__init__.py
+++ b/tests/scenario/__init__.py
@@ -1,8 +1,8 @@
 import dataclasses
 import json
 import os.path
-from urllib.parse import urlparse
 from typing import List
+from urllib.parse import urlparse
 
 import pytest