      athenad.Params = MockParams
      athenad.Api = MockApi
      athenad.LOCAL_PORT_WHITELIST = set([cls.SOCKET_PORT])
  
    def test_echo(self):
      assert dispatcher["echo"]("bob") == "bob"
  
    def test_getMessage(self):
 
</s> remove UploadItem = namedtuple('UploadItem', ['path', 'url', 'headers', 'created_at', 'id'])
 </s> add UploadItem = namedtuple('UploadItem', ['path', 'url', 'headers', 'created_at', 'id', 'retry_count'], defaults=(0,)) </s> remove       time.sleep(1) # give it time to process to prevent shutdown before upload completes
      now = time.time()
      while time.time() - now < 5:
        if athenad.upload_queue.qsize() == 0:
          break
 </s> add       self.wait_for_upload()
      time.sleep(0.1)

      # TODO: verify that upload actually succeeded
      self.assertEqual(athenad.upload_queue.qsize(), 0)
    finally:
      end_event.set()
      athenad.upload_queue = queue.Queue()
      os.unlink(fn)

  def test_upload_handler_timeout(self):
    """When an upload times out or fails to connect it should be placed back in the queue"""
    fn = os.path.join(athenad.ROOT, 'qlog.bz2')
    Path(fn).touch()
    item = athenad.UploadItem(path=fn, url="http://localhost:44444/qlog.bz2", headers={}, created_at=int(time.time()*1000), id='')
    item_no_retry = item._replace(retry_count=MAX_RETRY_COUNT)

    end_event = threading.Event()
    thread = threading.Thread(target=athenad.upload_handler, args=(end_event,))
    thread.start()

    try:
      athenad.upload_queue.put_nowait(item_no_retry)
      self.wait_for_upload()
      time.sleep(0.1)

      # Check that upload with retry count exceeded is not put back </s> remove       now = time.time()
      while time.time() - now < 5:
        if athenad.upload_queue.qsize() == 0 and len(athenad.cancelled_uploads) == 0:
          break
 </s> add       self.wait_for_upload()
      time.sleep(0.1)
 </s> remove       _do_upload(item)
 </s> add       try:
        _do_upload(item)
      except (requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.SSLError) as e:
        cloudlog.warning(f"athena.upload_handler.retry {e} {item}")

        if item.retry_count < MAX_RETRY_COUNT:
          item = item._replace(retry_count=item.retry_count + 1)
          upload_queue.put_nowait(item)

          for _ in range(RETRY_DELAY):
            time.sleep(1)
            if end_event.is_set():
              break
  def wait_for_upload(self):
    now = time.time()
    while time.time() - now < 5:
      if athenad.upload_queue.qsize() == 0:
        break

      athenad.Params = MockParams
      athenad.Api = MockApi
      athenad.LOCAL_PORT_WHITELIST = set([cls.SOCKET_PORT])
  
  def wait_for_upload(self):
    now = time.time()
    while time.time() - now < 5:
      if athenad.upload_queue.qsize() == 0:
        break

    def test_echo(self):
      assert dispatcher["echo"]("bob") == "bob"
  
    def test_getMessage(self):
keep keep keep add keep keep keep keep
 athenad: retry failed and timed out uploads (#21745)

* retry failed uploads

* test cleanup

* update comment

* also catch SSL error

* use defaults

* sleep in chunks 
selfdrive/athena/tests/test_athenad.py