Making sure test coverage stays up

Author: apetenchea

arango/database.py

@@ -2725,7 +2725,7 @@ def __init__(self, connection: Connection, max_queue_time_seconds: float) -> Non
             executor=QueueBoundedApiExecutor(connection, max_queue_time_seconds),
         )
 
-    def __repr__(self) -> str:
+    def __repr__(self) -> str:  # pragma: no cover
         return f"<QueueBoundedDatabase {self.name}>"
 
     @property
@@ -2737,6 +2737,15 @@ def last_queue_time(self) -> float:
         """
         return self._executor.queue_time_seconds
 
+    @property
+    def max_queue_time(self) -> float:
+        """Return the maximum server-side queuing time in seconds.
+
+        :return: Maximum server-side queuing time in seconds.
+        :rtype: float
+        """
+        return self._executor.max_queue_time_seconds
+
     def adjust_max_queue_time(self, max_queue_time_seconds: float) -> None:
         """Adjust the maximum server-side queuing time in seconds.
 

arango/executor.py

@@ -450,7 +450,7 @@ def __init__(self, connection: Connection, max_queue_time_seconds: float) -> Non
         self._queue_time_seconds = 0.0
 
     @property
-    def context(self) -> str:
+    def context(self) -> str:  # pragma: no cover
         return "queue-bounded"
 
     @property

tests/test_queue_bounded.py

@@ -10,25 +10,27 @@ def flood_with_requests(bounded_db, async_db):
     We can only try and make it as large as possible. However, if the system
     is fast enough, it may still be 0.
     """
+    bounded_db.aql.execute("RETURN SLEEP(0.5)", count=True)
     for _ in range(3):
         for _ in range(500):
             async_db.aql.execute("RETURN SLEEP(0.5)", count=True)
-        for _ in range(3):
-            bounded_db.aql.execute("RETURN SLEEP(0.5)", count=True)
+        bounded_db.aql.execute("RETURN SLEEP(0.5)", count=True)
         if bounded_db.last_queue_time >= 0:
             break
 
 
 def test_queue_bounded(db):
     bounded_db = db.begin_queue_bounded_execution(100)
+    assert bounded_db.max_queue_time == 100
+
     async_db = db.begin_async_execution(return_result=True)
 
     flood_with_requests(bounded_db, async_db)
     assert bounded_db.last_queue_time >= 0
 
     # We can only emit a warning here. The test will still pass.
     if bounded_db.last_queue_time == 0:
-        warnings.warn("last_queue_time is 0, test may be unreliable")
+        warnings.warn(f"last_queue_time of {bounded_db} is 0, test may be unreliable")
 
     bounded_db.adjust_max_queue_time(0.0001)
     try:
@@ -39,5 +41,6 @@ def test_queue_bounded(db):
         assert e.error_code == 21004
     else:
         warnings.warn(
-            f"last queue time is {bounded_db.last_queue_time}, test may be unreliable"
+            f"last_queue_time of {bounded_db} is {bounded_db.last_queue_time},"
+            f"test may be unreliable"
         )