Test that concurrent requests on a single client are synchronised.
()
| 264 | |
| 265 | @pytest.mark.fixed_client |
| 266 | async def test_single_connection(): |
| 267 | """Test that concurrent requests on a single client are synchronised.""" |
| 268 | r = Redis(single_connection_client=True) |
| 269 | |
| 270 | init_call_count = 0 |
| 271 | command_call_count = 0 |
| 272 | in_use = False |
| 273 | |
| 274 | class Retry_: |
| 275 | async def call_with_retry(self, _, __, with_failure_count=False): |
| 276 | # If we remove the single-client lock, this error gets raised as two |
| 277 | # coroutines will be vying for the `in_use` flag due to the two |
| 278 | # asymmetric sleep calls |
| 279 | nonlocal command_call_count |
| 280 | nonlocal in_use |
| 281 | if in_use is True: |
| 282 | raise ValueError("Commands should be executed one at a time.") |
| 283 | in_use = True |
| 284 | await asyncio.sleep(0.01) |
| 285 | command_call_count += 1 |
| 286 | await asyncio.sleep(0.03) |
| 287 | in_use = False |
| 288 | return "foo" |
| 289 | |
| 290 | mock_conn = mock.AsyncMock(spec=Connection) |
| 291 | mock_conn.retry = Retry_() |
| 292 | mock_conn.host = "localhost" |
| 293 | mock_conn.port = 6379 |
| 294 | |
| 295 | async def get_conn(): |
| 296 | # Validate only one client is created in single-client mode when |
| 297 | # concurrent requests are made |
| 298 | nonlocal init_call_count |
| 299 | await asyncio.sleep(0.01) |
| 300 | init_call_count += 1 |
| 301 | return mock_conn |
| 302 | |
| 303 | with mock.patch.object(r.connection_pool, "get_connection", get_conn): |
| 304 | with mock.patch.object(r.connection_pool, "release"): |
| 305 | await asyncio.gather(r.set("a", "b"), r.set("c", "d")) |
| 306 | |
| 307 | assert init_call_count == 1 |
| 308 | assert command_call_count == 2 |
| 309 | r.connection = None # it was a Mock |
| 310 | await r.aclose() |
| 311 | |
| 312 | |
| 313 | @skip_if_server_version_lt("4.0.0") |