"""
Wait until the mail queue is empty and all emails are sent.
+ If the mail queue is still not empty after timeout is reached, a warning is logged and a
+ :py:class:`TimeoutError` is raised.
+
:param int timeout: email transfer timeout
:param vm: vm to run on if running on a guest instead of the host
:type vm: :py:class:`virttest.qemu_vm.VM` or None
+ :raises TimeoutError: if mail transfer was not complete when timeout was reached
"""
for i in range(timeout):
if i % 10 == 0:
time.sleep(1)
log.warning('Timeout reached but SMTP queue still not empty after {} s'
.format(timeout))
+ raise TimeoutError()
def wait_for_quarantine_processing(vm_session: Any = None, max_wait: int = 30) -> bool:
end_time = time.time()
self.assertGreater(end_time - start_time, 1.5)
self.assertLess(end_time - start_time, 2.5)
+
+ def test_wait_for_email_transfer_timeout(self):
+ """Test waiting for email transfer that reaches timeout."""
+ DummyCmdOutputMapping.asserted_cmds = (
+ {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0},
+ {"cmd": 'postqueue -j', "stdout": b"not empty", "returncode": 0},
+ )
+ start_time = time.time()
+ self.assertRaises(TimeoutError, arnied_wrapper.wait_for_email_transfer, 2)
+ time_diff = time.time() - start_time
+ self.assertGreater(time_diff, 1.5)
+ self.assertLess(time_diff, 2.5)
+
+ def test_wait_for_email_transfer_succeed(self):
+ """Test waiting for email transfer that succeeds at once"""
+ DummyCmdOutputMapping.asserted_cmds = (
+ {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0},
+ {"cmd": 'postqueue -j', "stdout": b"", "returncode": 0},
+ )
+ start_time = time.time()
+ arnied_wrapper.wait_for_email_transfer()
+ time_diff = time.time() - start_time
+ self.assertLess(time_diff, 1)