github/sprockets.mixins.metrics
1
0
Fork 0
mirror of https://github.com/sprockets/sprockets.mixins.metrics.git synced 2025-04-09 01:11:08 -09:00
Code Issues Projects Releases Packages Wiki Activity

set protocal with str 'protocol' as per code review

Browse source
This commit is contained in:
Dan g 2018-07-19 16:40:59 -04:00
parent ce352daf0e
commit b5d1cae249
2 changed files with 10 additions and 8 deletions
sprockets/mixins/metrics
testing.py
tests.py

10
sprockets/mixins/metrics/testing.py
View file

@ -12,7 +12,7 @@ class FakeStatsdServer(tcpserver.TCPServer):
Implements something resembling a statsd server. Implements something resembling a statsd server.
:param tornado.ioloop.IOLoop iol: the loop to attach to :param tornado.ioloop.IOLoop iol: the loop to attach to
:param bool tcp: whether the server implements TCP or UDP :param str protocol: The StatsD protocol. May be either ``udp`` or ``tcp``.
Create an instance of this class in your asynchronous test case Create an instance of this class in your asynchronous test case
attached to the IOLoop and configure your application to send attached to the IOLoop and configure your application to send
@ -32,13 +32,15 @@ class FakeStatsdServer(tcpserver.TCPServer):
PATTERN = br'(?P<path>[^:]*):(?P<value>[^|]*)\|(?P<type>.*)$' PATTERN = br'(?P<path>[^:]*):(?P<value>[^|]*)\|(?P<type>.*)$'
def __init__(self, iol, tcp=False): def __init__(self, iol, protocol='udp'):
self.datagrams = [] self.datagrams = []
if tcp is False: if protocol == 'tcp':
self.tcp_server()
elif protocol == 'udp':
self.udp_server(iol) self.udp_server(iol)
else: else:
self.tcp_server() raise ValueError('Invalid protocol: {}'.format(protocol))
def tcp_server(self): def tcp_server(self):
self.event = locks.Event() self.event = locks.Event()

8
tests.py
View file

@ -55,7 +55,7 @@ class TCPStatsdMetricCollectionTests(testing.AsyncHTTPTestCase):
def setUp(self): def setUp(self):
self.application = None self.application = None
super(TCPStatsdMetricCollectionTests, self).setUp() super(TCPStatsdMetricCollectionTests, self).setUp()
self.statsd = FakeStatsdServer(self.io_loop, tcp=True) self.statsd = FakeStatsdServer(self.io_loop, protocol='tcp')
statsd.install(self.application, **{'namespace': 'testing', statsd.install(self.application, **{'namespace': 'testing',
'host': self.statsd.sockaddr[0], 'host': self.statsd.sockaddr[0],
@ -121,7 +121,7 @@ class TCPStatsdConfigurationTests(testing.AsyncHTTPTestCase):
def setUp(self): def setUp(self):
self.application = None self.application = None
super(TCPStatsdConfigurationTests, self).setUp() super(TCPStatsdConfigurationTests, self).setUp()
self.statsd = FakeStatsdServer(self.io_loop, tcp=True) self.statsd = FakeStatsdServer(self.io_loop, protocol='tcp')
statsd.install(self.application, **{'namespace': 'testing', statsd.install(self.application, **{'namespace': 'testing',
'host': self.statsd.sockaddr[0], 'host': self.statsd.sockaddr[0],
@ -159,7 +159,7 @@ class UDPStatsdMetricCollectionTests(testing.AsyncHTTPTestCase):
def setUp(self): def setUp(self):
self.application = None self.application = None
super(UDPStatsdMetricCollectionTests, self).setUp() super(UDPStatsdMetricCollectionTests, self).setUp()
self.statsd = FakeStatsdServer(self.io_loop, tcp=False) self.statsd = FakeStatsdServer(self.io_loop, protocol='udp')
statsd.install(self.application, **{'namespace': 'testing', statsd.install(self.application, **{'namespace': 'testing',
'host': self.statsd.sockaddr[0], 'host': self.statsd.sockaddr[0],
@ -229,7 +229,7 @@ class UDPStatsdConfigurationTests(testing.AsyncHTTPTestCase):
def setUp(self): def setUp(self):
self.application = None self.application = None
super(UDPStatsdConfigurationTests, self).setUp() super(UDPStatsdConfigurationTests, self).setUp()
self.statsd = FakeStatsdServer(self.io_loop, tcp=False) self.statsd = FakeStatsdServer(self.io_loop, protocol='udp')
statsd.install(self.application, **{'namespace': 'testing', statsd.install(self.application, **{'namespace': 'testing',
'host': self.statsd.sockaddr[0], 'host': self.statsd.sockaddr[0],