fix(test): drain counter queue in follower's runner

Instead of waiting for looper to stop and return final coutner
we push all updates into queue and then drain it from main loop
after thread joined or times out.

This makes timeout tests into soft timeout, because we are giving looper
time to finish processing last batch of the changes.

Author: eiri

test/unit/features/conftest.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2022, 2023.
+# © Copyright IBM Corporation 2022, 2024.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -212,12 +212,13 @@ class TimeoutError(Exception):
 
         def looper(changes, buf):
             counter = 0
+            buf.put(counter)
             try:
                 for _ in changes:
                     counter += 1
+                    buf.put(counter)
                     if stop_after > 0 and stop_after == counter:
                         follower.stop()
-                buf.put(counter)
             except Exception as e:
                 buf.put(e)
 
@@ -229,15 +230,16 @@ def main():
             buf = queue.Queue()
             thread = threading.Thread(target=looper, args=(changes, buf))
             thread.start()
+            thread.join(timeout)
+            while not buf.empty():
+                data = buf.get()
+                if isinstance(data, Exception):
+                    raise data
             while True:
-                thread.join(timeout)
                 if thread.is_alive():
                     follower.stop()
                 else:
                     break
-            data = buf.get()
-            if isinstance(data, Exception):
-                raise data
             return data
 
         return main()

test/unit/features/test_changes_follower.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2022, 2023.
+# © Copyright IBM Corporation 2022, 2024.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -458,7 +458,7 @@ def test_start(self):
         with some batches.
         """
         try:
-            self.prepare_mock_changes(batches=MAX_BATCHES)
+            self.prepare_mock_changes(batches=3)
             follower = ChangesFollower(self.client, db='db')
             count = self.runner(follower, _Mode.LISTEN, timeout=5)
         except BaseException: