Add FakeStatsdServer.find_metrics.

docs/history.rst

@@ -6,5 +6,6 @@ Release History
 `Next Release`_
 ---------------
 - Add :class:`sprockets.mixins.metrics.StatsdMixin`
+- Add :class:`sprockets.mixins.metrics.testing.FakeStatsdServer`
 
 .. _Next Release: https://github.com/sprockets/sprockets.mixins.metrics/compare/0.0.0...master

sprockets/mixins/metrics/testing.py

@@ -1,6 +1,10 @@
+import re
 import socket
 
 
+STATS_PATTERN = re.compile(r'(?P<path>[^:]*):(?P<value>[^|]*)\|(?P<type>.*)$')
+
+
 class FakeStatsdServer(object):
     """
     Implements something resembling a statsd server.
@@ -50,3 +54,30 @@ class FakeStatsdServer(object):
         if events & self._iol.READ:
             data, _ = self.socket.recvfrom(4096)
             self.datagrams.append(data)
+
+    def find_metrics(self, prefix, metric_type):
+        """
+        Yields captured datagrams that start with `prefix`.
+
+        :param str prefix: the metric prefix to search for
+        :param str metric_type: the statsd metric type (e.g., 'ms', 'c')
+        :returns: yields (path, value, metric_type) tuples for each
+            captured metric that matches
+        :raises AssertionError: if no metrics match.
+
+        """
+        pattern = re.compile(
+            '(?P<path>{}[^:]*):(?P<value>[^|]*)\\|(?P<type>{})'.format(
+                re.escape(prefix), re.escape(metric_type)))
+        matched = False
+        for datagram in self.datagrams:
+            text_msg = datagram.decode('ascii')
+            match = pattern.match(text_msg)
+            if match:
+                yield match.groups()
+                matched = True
+
+        if not matched:
+            raise AssertionError(
+                'Expected metric starting with "{}" in {!r}'.format(
+                    prefix, self.datagrams))

tests.py

@@ -37,19 +37,10 @@ class StatsdMethodTimingTests(testing.AsyncHTTPTestCase):
         response = self.fetch('/')
         self.assertEqual(response.code, 204)
 
-        expected = 'testing.SimpleHandler.GET.204:'
-        for bin_msg in self.statsd.datagrams:
-            text_msg = bin_msg.decode('ascii')
-            if text_msg.startswith(expected):
-                _, _, tail = text_msg.partition(':')
-                measurement, _, measurement_type = tail.partition('|')
-                self.assertTrue(250.0 <= float(measurement) < 500.0,
-                                '{} looks wrong'.format(measurement))
-                self.assertTrue(measurement_type, 'ms')
-                break
-        else:
-            self.fail('Expected metric starting with {} in {!r}'.format(
-                expected, self.statsd.datagrams))
+        expected = 'testing.SimpleHandler.GET.204'
+        for path, value, stat_type in self.statsd.find_metrics(expected, 'ms'):
+            self.assertTrue(250.0 <= float(value) < 500.0,
+                            '{} looks wrong'.format(value))
 
     def test_that_cached_socket_is_used(self):
         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)