swh_storage = <swh.storage.proxies.tenacious.TenaciousProxyStorage object at 0x7f0244eb1e48> @pytest.fixture def swh_contents(swh_storage: StorageInterface) -> Iterable[BaseContent]: contents = [BaseContent.from_dict(c) for c in gen_contents(n=20)] swh_storage.content_add([c for c in contents if c.status != "absent"]) > swh_storage.skipped_content_add([c for c in contents if c.status == "absent"]) .tox/py3/lib/python3.7/site-packages/swh/storage/tests/conftest.py:65: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <swh.storage.proxies.tenacious.TenaciousProxyStorage object at 0x7f0244eb1e48> func_name = 'skipped_content_add' objects = [SkippedContent(sha1=b'9\xbd2\xa4F\x05\x8b\xf3\x12R\x1b)\x83\xb7\xa5\xe0\x88\xe78k', sha1_git=b'\xb73\x8f:\x98\xfdn\xd...e=datetime.datetime(2024, 1, 21, 8, 54, 9, 808481, tzinfo=<DstTzInfo 'America/Halifax' AST-1 day, 20:00:00 STD>)), ...] def _tenacious_add(self, func_name, objects: Iterable[BaseModel]) -> Dict[str, int]: """Enqueue objects to write to the storage. This checks if the queue's threshold is hit. If it is actually write those to the storage. """ add_function = getattr(self.storage, func_name) object_type = self.tenacious_methods[func_name] # list of lists of objects; note this to_add list is consumed from the tail to_add: List[List[BaseModel]] = [list(objects)] n_objs: int = len(to_add[0]) results: CounterT[str] = Counter() retries: int = self._single_object_retries while to_add: if self.rate_queue.limit_reached(): logging.error( "Too many insertion errors have been detected; " "disabling insertions" ) raise RuntimeError( > "Too many insertion errors have been detected; " "disabling insertions" ) E RuntimeError: Too many insertion errors have been detected; disabling insertions .tox/py3/lib/python3.7/site-packages/swh/storage/proxies/tenacious.py:134: RuntimeError