(self, r, timeout=10)
| 94 | |
| 95 | class TestRedisCommands: |
| 96 | def _wait_for_bgsave(self, r, timeout=10): |
| 97 | deadline = time.monotonic() + timeout |
| 98 | while True: |
| 99 | info = r.info("persistence") |
| 100 | if int(info.get("rdb_bgsave_in_progress", 0)) == 0: |
| 101 | return |
| 102 | if time.monotonic() > deadline: |
| 103 | pytest.fail("Timed out waiting for BGSAVE to finish") |
| 104 | time.sleep(0.05) |
| 105 | |
| 106 | @pytest.mark.onlynoncluster |
| 107 | @skip_if_redis_enterprise() |
no test coverage detected