Add tests on the capped queue

tests/conftest.py

@@ -14,14 +14,40 @@ def example_queue_item():
     )
 
 
+@pytest.fixture
+def example_queue_items_gen():
+    """ Returns a generator with a sequence of points """
+
+    def generator():
+        id = 1
+        while True:
+            yield (
+                datetime.now(timezone.utc),
+                0.04,
+                'PUT',
+                f'/call/{id}',
+                'controllers.endpoint_name',
+            )
+            # Alternate between an endpoint or no endpoint
+            yield (
+                datetime.now(timezone.utc),
+                0.04,
+                'GET',
+                f'/call/{id}',
+                None,
+            )
+
+    yield generator()
+
+
 @pytest.fixture
 def queue() -> Queue:
     return Queue(maxsize=10)
 
 
 @pytest.fixture
-def queue_full(example_queue_item) -> Queue:
+def queue_full(example_queue_items_gen) -> Queue:
     queue = Queue(maxsize=10)
     for i in range(10):
-        queue.put_nowait(example_queue_item)
+        queue.put_nowait(next(example_queue_items_gen))
     return queue

tests/test_core.py

@@ -0,0 +1,45 @@
+from unittest.mock import patch
+from queue import Queue
+from howfast_apm import core
+
+
+def test_capped_queue(example_queue_items_gen):
+    """ CoreAPM.save_point should add items in the queue """
+    # TODO: find a better way to replace this queue object
+    core.queue = Queue(maxsize=10)
+    # Save one point
+    assert core.queue.qsize() == 0
+    core.CoreAPM.save_point(*next(example_queue_items_gen))
+    assert core.queue.qsize() == 1
+
+    # Save a second point
+    core.CoreAPM.save_point(*next(example_queue_items_gen))
+    assert core.queue.qsize() == 2
+
+    # Fill the queue
+    for i in range(8):
+        item = list(next(example_queue_items_gen))
+        item[3] = f'/call/{i}'  # update the URL to keep track of points
+        core.CoreAPM.save_point(*item)
+    assert core.queue.qsize() == 10
+    assert core.queue.full()
+
+    next_item = core.queue.get_nowait()
+    assert next_item
+
+
+def test_capped_queue_full(example_queue_item):
+    """ CoreAPM.save_point should discard old items should the queue be full """
+    # TODO: find a better way to replace this queue object
+    core.queue = Queue(maxsize=10)
+    # Fill the queue
+    for i in range(10):
+        item = list(example_queue_item)
+        item[3] = f'/call/{i}'  # update the URL to keep track of points
+        core.CoreAPM.save_point(*item)
+    assert core.queue.full()
+
+    # Add one more item to the full queue
+    core.CoreAPM.save_point(*example_queue_item)
+    assert core.queue.full(), 'queue should still be full'
+    assert core.queue.qsize() == 10

tests/test_middleware_flask.py

@@ -24,6 +24,8 @@ def HowFastFlaskMiddleware():
     """ Patch the save_point() method """
     from howfast_apm import HowFastFlaskMiddleware
     HowFastFlaskMiddleware.save_point = MagicMock()
+    # Prevent the background thread to actually start
+    HowFastFlaskMiddleware.start_background_thread = MagicMock()
     return HowFastFlaskMiddleware
 
 

tests/test_runner.py

@@ -1,4 +1,4 @@
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 from howfast_apm.queue import Runner
 
@@ -63,4 +63,37 @@ def test_queue_full(send_mocked, queue_full):
     assert queue_full.qsize() == 0
 
 
-# TODO: test what happens when with the core when the queue gets full
+@patch('requests.post')
+def test_send_batch(mocked_post, queue_full):
+    """ send_batch should serialize all batched points and send them to the API """
+    runner = Runner(queue=queue_full, app_id="test-dsn")
+    mocked_post.return_value.status_code = 200
+    runner.run_once()
+
+    assert mocked_post.called is True
+    assert mocked_post.call_count == 1
+    kwargs = mocked_post.call_args[1]
+    json_payload = kwargs.get('json')
+    assert json_payload.get('dsn') == 'test-dsn'
+    assert isinstance(json_payload.get('perf'), list)
+    points = json_payload.get('perf')
+    # queue_full has 10 elements
+    assert len(points) == 10
+    assert isinstance(points[0], tuple)
+    assert len(points[0]) == 5
+    assert len(points[1]) == 5
+    # Make sure we have the correct point
+    (method, uri, time_request_started, elapsed, endpoint) = points[0]
+    assert method == 'PUT'
+    assert uri == '/call/1'
+
+
+@patch('requests.post')
+def test_send_batch_api_issue(mocked_post, queue_full):
+    """ send_batch should be robust if the API isn't available """
+    runner = Runner(queue=queue_full, app_id="test-dsn")
+    mocked_post.return_value.status_code = None
+    # run_once should not crash
+    runner.run_once()
+
+    assert mocked_post.called is True