use processors for transactions, and test that it actually happens (#84)

CHANGELOG.asciidoc

@@ -14,6 +14,7 @@ https://github.com/elastic/apm-agent-python/compare/v1.0.0.dev2\...master[Check
 
  * added a background thread to process the transactions queue every 60 seconds (configurable) ({pull}68[#68])
  * adapted trace context for SQL traces to new API ({pull}77[#77])
+ * ensured that transaction data is also passed through processors ({pull}84[#84])
 
 [[release-v1.0.0.dev2]]
 [float]

elasticapm/base.py

@@ -441,7 +441,7 @@ def begin_transaction(self, transaction_type):
     def end_transaction(self, name, status_code=None):
         self.instrumentation_store.end_transaction(status_code, name)
         if self.instrumentation_store.should_collect():
-            self._traces_collect()
+            self._collect_transactions()
         if not self._send_timer:
             # send first batch of data after config._wait_to_first_send
             self._start_send_timer(timeout=min(self.config._wait_to_first_send, self.config.traces_send_frequency))
@@ -461,7 +461,7 @@ def set_transaction_extra_data(self, data, _key=None):
         transaction._context[_key] = data
 
     def close(self):
-        self._traces_collect()
+        self._collect_transactions()
         if self._send_timer:
             self._stop_send_timer()
         for url, transport in self._transports.items():
@@ -492,23 +492,28 @@ def handle_transport_fail(self, exception=None, **kwargs):
         )
         self.state.set_fail()
 
-    def _traces_collect(self):
+    def _collect_transactions(self):
         self._stop_send_timer()
-        transactions = self.instrumentation_store.get_all()
+        transactions = []
+        for transaction in self.instrumentation_store.get_all():
+            for processor in self.processors:
+                transaction = processor(self, transaction)
+            transactions.append(transaction)
         if not transactions:
             return
 
         data = self.build_msg({
             'transactions': transactions,
         })
+
         api_path = defaults.TRANSACTIONS_API_PATH
 
         self.send(server=self.config.server + api_path, **data)
         self._start_send_timer()
 
     def _start_send_timer(self, timeout=None):
         timeout = timeout or self.config.traces_send_frequency
-        self._send_timer = threading.Timer(timeout, self._traces_collect)
+        self._send_timer = threading.Timer(timeout, self._collect_transactions)
         self._send_timer.start()
 
     def _stop_send_timer(self):

elasticapm/contrib/asyncio/client.py

@@ -26,7 +26,7 @@ def _send_remote(self, url, data, headers=None):
 
     def _start_send_timer(self, timeout=None):
         timeout = timeout or self.config.traces_send_frequency
-        self._send_timer = AsyncTimer(timeout, self._traces_collect)
+        self._send_timer = AsyncTimer(timeout, self._collect_transactions)
 
     def _stop_send_timer(self):
         if self._send_timer:

tests/client/client_tests.py

@@ -290,7 +290,7 @@ def test_send_with_auth_header(time, send_remote):
 
 @mock.patch('elasticapm.transport.http_urllib3.Urllib3Transport.send')
 @mock.patch('elasticapm.transport.http_urllib3.Urllib3Transport.close')
-@mock.patch('elasticapm.base.Client._traces_collect')
+@mock.patch('elasticapm.base.Client._collect_transactions')
 def test_client_shutdown_sync(mock_traces_collect, mock_close, mock_send):
     client = Client(
         server='http://example.com',
@@ -307,7 +307,7 @@ def test_client_shutdown_sync(mock_traces_collect, mock_close, mock_send):
 
 
 @mock.patch('elasticapm.transport.http_urllib3.Urllib3Transport.send')
-@mock.patch('elasticapm.base.Client._traces_collect')
+@mock.patch('elasticapm.base.Client._collect_transactions')
 def test_client_shutdown_async(mock_traces_collect, mock_send):
     client = Client(
         server='http://example.com',

tests/processors/tests.py

@@ -1,10 +1,12 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
 
+import mock
 import pytest
 
 from elasticapm import Client, processors
 from elasticapm.utils import compat
+from tests.fixtures import elasticapm_client
 
 
 @pytest.fixture()
@@ -222,3 +224,33 @@ def test_mark_in_app_frames():
     assert not frames[3]['in_app']
     assert frames[4]['in_app']
     assert not frames[5]['in_app']
+
+
+def dummy_processor(client, data):
+    data['processed'] = True
+    return data
+
+
+@pytest.mark.parametrize('elasticapm_client', [{'processors': 'tests.processors.tests.dummy_processor'}], indirect=True)
+def test_transactions_processing(elasticapm_client):
+    for i in range(5):
+        elasticapm_client.begin_transaction('dummy')
+        elasticapm_client.end_transaction('dummy_transaction', 'success')
+    elasticapm_client._collect_transactions()
+    for transaction in elasticapm_client.events[0]['transactions']:
+        assert transaction['processed'] is True
+
+
+@pytest.mark.parametrize('elasticapm_client', [{'processors': 'tests.processors.tests.dummy_processor'}], indirect=True)
+def test_exception_processing(elasticapm_client):
+    try:
+        1 / 0
+    except ZeroDivisionError:
+        elasticapm_client.capture_exception()
+    assert elasticapm_client.events[0]['errors'][0]['processed'] is True
+
+
+@pytest.mark.parametrize('elasticapm_client', [{'processors': 'tests.processors.tests.dummy_processor'}], indirect=True)
+def test_message_processing(elasticapm_client):
+    elasticapm_client.capture_message('foo')
+    assert elasticapm_client.events[0]['errors'][0]['processed'] is True