Use provided redis address. Use DelayProxy over local ipv4 only.

Author: kristjanvalur

tests/test_asyncio/test_cwe_404.py

@@ -1,13 +1,20 @@
 import asyncio
 import contextlib
-import sys
 
 import pytest
 
 from redis.asyncio import Redis
 from redis.asyncio.cluster import RedisCluster
 
 
+@pytest.fixture
+def redis_addr(request):
+    redis_url = request.config.getoption("--redis-url")
+    assert redis_url.startswith("redis://")
+    host, port = redis_url[8:].split(":")
+    return host, int(port)
+
+
 async def pipe(
     reader: asyncio.StreamReader,
     writer: asyncio.StreamWriter,
@@ -68,18 +75,16 @@ async def stop(self):
 
 @pytest.mark.onlynoncluster
 @pytest.mark.parametrize("delay", argvalues=[0.05, 0.5, 1, 2])
-async def test_standalone(delay):
+async def test_standalone(delay, redis_addr):
 
     # create a tcp socket proxy that relays data to Redis and back,
     # inserting 0.1 seconds of delay
-    dp = DelayProxy(
-        addr=("localhost", 5380), redis_addr=("localhost", 6379), delay=delay * 2
-    )
+    dp = DelayProxy(addr=("127.0.0.1", 5380), redis_addr=redis_addr, delay=delay * 2)
     await dp.start()
 
     for b in [True, False]:
         # note that we connect to proxy, rather than to Redis directly
-        async with Redis(host="localhost", port=5380, single_connection_client=b) as r:
+        async with Redis(host="127.0.0.1", port=5380, single_connection_client=b) as r:
 
             with dp.override():
                 await r.set("foo", "foo")
@@ -107,13 +112,11 @@ async def test_standalone(delay):
 
 @pytest.mark.onlynoncluster
 @pytest.mark.parametrize("delay", argvalues=[0.05, 0.5, 1, 2])
-async def test_standalone_pipeline(delay):
-    dp = DelayProxy(
-        addr=("localhost", 5380), redis_addr=("localhost", 6379), delay=delay * 2
-    )
+async def test_standalone_pipeline(delay, redis_addr):
+    dp = DelayProxy(addr=("127.0.0.1", 5380), redis_addr=redis_addr, delay=delay * 2)
     await dp.start()
     for b in [True, False]:
-        async with Redis(host="localhost", port=5380, single_connection_client=b) as r:
+        async with Redis(host="127.0.0.1", port=5380, single_connection_client=b) as r:
             with dp.override():
                 await r.set("foo", "foo")
                 await r.set("bar", "bar")
@@ -156,12 +159,13 @@ async def test_standalone_pipeline(delay):
 
 
 @pytest.mark.onlycluster
-async def test_cluster(request):
+async def test_cluster(request, redis_addr):
 
-    dp = DelayProxy(addr=("localhost", 5381), redis_addr=("localhost", 6372), delay=0.1)
+    redis_addr = redis_addr[0], 6372  # use the cluster port
+    dp = DelayProxy(addr=("127.0.0.1", 5381), redis_addr=redis_addr, delay=0.1)
     await dp.start()
 
-    r = RedisCluster.from_url("redis://localhost:5381")
+    r = RedisCluster.from_url("redis://127.0.0.1:5381")
     await r.initialize()
     with dp.override():
         await r.set("foo", "foo")