1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
from __future__ import annotations
import asyncio
import gc
import socket
import sys
import time
import uuid
from typing import AsyncIterator, Iterator, TypedDict
import aiomcache
import pytest
from docker import DockerClient, from_env as docker_from_env, models as docker_models
from redis import asyncio as aioredis
class _ContainerInfo(TypedDict):
host: str
port: int
container: docker_models.containers.Container
class _MemcachedParams(TypedDict):
host: str
port: int
def unused_port() -> int: # pragma: no cover
"""Only used for people testing on OS X."""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
s.listen(1)
port: int = s.getsockname()[1]
s.close()
return port
@pytest.fixture(scope="session")
def session_id() -> str:
"""Unique session identifier, random string."""
return str(uuid.uuid4())
@pytest.fixture(scope="session")
def docker() -> DockerClient: # type: ignore[misc] # No docker types.
client = docker_from_env(version="auto")
yield client
client.close()
@pytest.fixture(scope="session")
async def redis_server( # type: ignore[misc] # No docker types.
docker: DockerClient,
session_id: str,
) -> Iterator[_ContainerInfo]:
image = "redis:{}".format("latest")
if sys.platform.startswith("darwin"): # pragma: no cover
port = unused_port()
else:
port = None
container = docker.containers.run(
image=image,
detach=True,
name="redis-test-server-{}-{}".format("latest", session_id),
ports={
"6379/tcp": port,
},
environment={
"http.host": "0.0.0.0",
"transport.host": "127.0.0.1",
},
)
if sys.platform.startswith("darwin"): # pragma: no cover
host = "0.0.0.0"
else:
inspection = docker.api.inspect_container(container.id)
host = inspection["NetworkSettings"]["IPAddress"]
port = 6379
delay = 0.1
for _i in range(20): # pragma: no cover
try:
conn = aioredis.from_url(f"redis://{host}:{port}") # type: ignore[no-untyped-call]
await conn.set("foo", "bar")
break
except aioredis.ConnectionError:
time.sleep(delay)
delay *= 2
finally:
await conn.aclose()
else: # pragma: no cover
pytest.fail("Cannot start redis server")
yield {"host": host, "port": port, "container": container}
container.kill(signal=9)
container.remove(force=True)
@pytest.fixture
def redis_url(redis_server: _ContainerInfo) -> str: # type: ignore[misc]
return "redis://{}:{}".format(redis_server["host"], redis_server["port"])
@pytest.fixture
async def redis(
redis_url: str,
) -> AsyncIterator[aioredis.Redis]:
async def start(pool: aioredis.ConnectionPool) -> aioredis.Redis:
return aioredis.Redis(connection_pool=pool)
pool = aioredis.ConnectionPool.from_url(redis_url)
redis = await start(pool)
yield redis
await redis.aclose()
await pool.disconnect()
@pytest.fixture(scope="session")
async def memcached_server( # type: ignore[misc] # No docker types.
docker: DockerClient,
session_id: str,
) -> AsyncIterator[_ContainerInfo]:
image = "memcached:{}".format("latest")
if sys.platform.startswith("darwin"): # pragma: no cover
port = unused_port()
else:
port = None
container = docker.containers.run(
image=image,
detach=True,
name="memcached-test-server-{}-{}".format("latest", session_id),
ports={
"11211/tcp": port,
},
environment={
"http.host": "0.0.0.0",
"transport.host": "127.0.0.1",
},
)
if sys.platform.startswith("darwin"): # pragma: no cover
host = "0.0.0.0"
else:
inspection = docker.api.inspect_container(container.id)
host = inspection["NetworkSettings"]["IPAddress"]
port = 11211
delay = 0.1
for _i in range(20): # pragma: no cover
try:
conn = aiomcache.Client(host, port)
await conn.set(b"foo", b"bar")
break
except ConnectionRefusedError:
time.sleep(delay)
delay *= 2
finally:
await conn.close()
else: # pragma: no cover
pytest.fail("Cannot start memcached server")
yield {"host": host, "port": port, "container": container}
container.kill(signal=9)
container.remove(force=True)
@pytest.fixture
def memcached_params( # type: ignore[misc]
memcached_server: _ContainerInfo,
) -> _MemcachedParams:
return dict(host=memcached_server["host"], port=memcached_server["port"])
@pytest.fixture
async def memcached(memcached_params: _MemcachedParams) -> AsyncIterator[aiomcache.Client]:
conn = aiomcache.Client(**memcached_params)
yield conn
await conn.close()
|
