Differential D6709 Diff 24408 swh/core/tests/test_statsd.py

Changeset View

Standalone View

View Options

swh/core/tests/test_statsd.py

# See the AUTHORS file at the top-level directory of this distribution

# License: GNU General Public License version 3, or any later version

# See top-level LICENSE file for more information

# Initially imported from https://github.com/DataDog/datadogpy/

# at revision 62b3a3e89988dc18d78c282fe3ff5d1813917436

Show All 18 Lines

# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;

# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND

# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

from collections import deque

from contextlib import contextmanager

import os

import socket

import time

import unittest

import pytest

from swh.core.pytest_plugin import FakeSocket

from swh.core.statsd import Statsd, TimedContextManagerDecorator

@contextmanager

class BrokenSocket(FakeSocket):

def preserve_envvars(*envvars):

def send(self, payload):

"""Context manager preserving the value of environment variables"""

raise socket.error("Socket error")

preserved = {}

to_delete = object()

for var in envvars:

preserved[var] = os.environ.get(var, to_delete)

yield

for var in envvars:

old = preserved[var]

if old is not to_delete:

os.environ[var] = old

else:

del os.environ[var]

class FakeSocket(object):

""" A fake socket for testing. """

def __init__(self):

self.payloads = deque()

class SlowSocket(FakeSocket):

def send(self, payload):

assert type(payload) == bytes

raise socket.timeout("Socket timeout")

self.payloads.append(payload)

def recv(self):

try:

return self.payloads.popleft().decode("utf-8")

except IndexError:

return None

def close(self):

def assert_almost_equal(a, b, delta):

ardumontUnsubmitted

Not Done

def assert_almost_equal(a, b, delta):

- assert 0 <= abs(a - b) <= delta, f"|{a} - {b}| not within {delta}" % (a, b, delta)

+ assert 0 <= abs(a - b) <= delta, f"|{a} - {b}| not within {delta}"

def test_set(statsd):

ardumont:

douarddaAuthorUnsubmitted

Done

oopsie

douardda: oopsie

pass

assert 0 <= abs(a - b) <= delta, f"|{a} - {b}| not within {delta}"

def __repr__(self):

return str(self.payloads)

def test_set(statsd):

statsd.set("set", 123)

assert statsd.socket.recv() == "set:123|s"

class BrokenSocket(FakeSocket):

def send(self, payload):

raise socket.error("Socket error")

def test_gauge(statsd):

statsd.gauge("gauge", 123.4)

assert statsd.socket.recv() == "gauge:123.4|g"

class SlowSocket(FakeSocket):

def send(self, payload):

raise socket.timeout("Socket timeout")

def test_counter(statsd):

statsd.increment("page.views")

assert statsd.socket.recv() == "page.views:1|c"

class TestStatsd(unittest.TestCase):

statsd.increment("page.views", 11)

def setUp(self):

assert statsd.socket.recv() == "page.views:11|c"

"""

Set up a default Statsd instance and mock the socket.

"""

self.statsd = Statsd()

self.statsd._socket = FakeSocket()

def recv(self):

statsd.decrement("page.views")

return self.statsd.socket.recv()

assert statsd.socket.recv() == "page.views:-1|c"

def test_set(self):

statsd.decrement("page.views", 12)

self.statsd.set("set", 123)

assert statsd.socket.recv() == "page.views:-12|c"

assert self.recv() == "set:123|s"

def test_gauge(self):

self.statsd.gauge("gauge", 123.4)

assert self.recv() == "gauge:123.4|g"

def test_counter(self):

def test_histogram(statsd):

self.statsd.increment("page.views")

statsd.histogram("histo", 123.4)

self.assertEqual("page.views:1|c", self.recv())

assert statsd.socket.recv() == "histo:123.4|h"

self.statsd.increment("page.views", 11)

self.assertEqual("page.views:11|c", self.recv())

self.statsd.decrement("page.views")

def test_tagged_gauge(statsd):

self.assertEqual("page.views:-1|c", self.recv())

statsd.gauge("gt", 123.4, tags={"country": "china", "age": 45})

assert statsd.socket.recv() == "gt:123.4|g|#age:45,country:china"

self.statsd.decrement("page.views", 12)

self.assertEqual("page.views:-12|c", self.recv())

def test_histogram(self):

def test_tagged_counter(statsd):

self.statsd.histogram("histo", 123.4)

statsd.increment("ct", tags={"country": "españa"})

self.assertEqual("histo:123.4|h", self.recv())

assert statsd.socket.recv() == "ct:1|c|#country:españa"

def test_tagged_gauge(self):

self.statsd.gauge("gt", 123.4, tags={"country": "china", "age": 45})

self.assertEqual("gt:123.4|g|#age:45,country:china", self.recv())

def test_tagged_counter(self):

def test_tagged_histogram(statsd):

self.statsd.increment("ct", tags={"country": "españa"})

statsd.histogram("h", 1, tags={"test_tag": "tag_value"})

self.assertEqual("ct:1|c|#country:españa", self.recv())

assert statsd.socket.recv() == "h:1|h|#test_tag:tag_value"

def test_tagged_histogram(self):

self.statsd.histogram("h", 1, tags={"test_tag": "tag_value"})

self.assertEqual("h:1|h|#test_tag:tag_value", self.recv())

def test_sample_rate(self):

def test_sample_rate(statsd):

self.statsd.increment("c", sample_rate=0)

statsd.increment("c", sample_rate=0)

assert not self.recv()

assert not statsd.socket.recv()

for i in range(10000):

self.statsd.increment("sampled_counter", sample_rate=0.3)

statsd.increment("sampled_counter", sample_rate=0.3)

self.assert_almost_equal(3000, len(self.statsd.socket.payloads), 150)

assert_almost_equal(3000, len(statsd.socket.payloads), 150)

self.assertEqual("sampled_counter:1|c|@0.3", self.recv())

assert statsd.socket.recv() == "sampled_counter:1|c|@0.3"

def test_tags_and_samples(self):

def test_tags_and_samples(statsd):

for i in range(100):

self.statsd.gauge("gst", 23, tags={"sampled": True}, sample_rate=0.9)

statsd.gauge("gst", 23, tags={"sampled": True}, sample_rate=0.9)

assert_almost_equal(90, len(statsd.socket.payloads), 10)

assert statsd.socket.recv() == "gst:23|g|@0.9|#sampled:True"

self.assert_almost_equal(90, len(self.statsd.socket.payloads), 10)

self.assertEqual("gst:23|g|@0.9|#sampled:True", self.recv())

def test_timing(self):

def test_timing(statsd):

self.statsd.timing("t", 123)

statsd.timing("t", 123)

self.assertEqual("t:123|ms", self.recv())

assert statsd.socket.recv() == "t:123|ms"

def test_metric_namespace(self):

def test_metric_namespace(statsd):

"""

Namespace prefixes all metric names.

"""

self.statsd.namespace = "foo"

statsd.namespace = "foo"

self.statsd.gauge("gauge", 123.4)

statsd.gauge("gauge", 123.4)

self.assertEqual("foo.gauge:123.4|g", self.recv())

assert statsd.socket.recv() == "foo.gauge:123.4|g"

# Test Client level constant tags

def test_gauge_constant_tags(self):

def test_gauge_constant_tags(statsd):

self.statsd.constant_tags = {

statsd.constant_tags = {

"bar": "baz",

}

self.statsd.gauge("gauge", 123.4)

statsd.gauge("gauge", 123.4)

assert self.recv() == "gauge:123.4|g|#bar:baz"

assert statsd.socket.recv() == "gauge:123.4|g|#bar:baz"

def test_counter_constant_tag_with_metric_level_tags(self):

self.statsd.constant_tags = {

def test_counter_constant_tag_with_metric_level_tags(statsd):

statsd.constant_tags = {

"bar": "baz",

"foo": True,

}

self.statsd.increment("page.views", tags={"extra": "extra"})

statsd.increment("page.views", tags={"extra": "extra"})

self.assertEqual(

assert statsd.socket.recv() == "page.views:1|c|#bar:baz,extra:extra,foo:True"

"page.views:1|c|#bar:baz,extra:extra,foo:True", self.recv(),

)

def test_gauge_constant_tags_with_metric_level_tags_twice(self):

def test_gauge_constant_tags_with_metric_level_tags_twice(statsd):

metric_level_tag = {"foo": "bar"}

self.statsd.constant_tags = {"bar": "baz"}

statsd.constant_tags = {"bar": "baz"}

self.statsd.gauge("gauge", 123.4, tags=metric_level_tag)

statsd.gauge("gauge", 123.4, tags=metric_level_tag)

assert self.recv() == "gauge:123.4|g|#bar:baz,foo:bar"

assert statsd.socket.recv() == "gauge:123.4|g|#bar:baz,foo:bar"

# sending metrics multiple times with same metric-level tags

# should not duplicate the tags being sent

self.statsd.gauge("gauge", 123.4, tags=metric_level_tag)

statsd.gauge("gauge", 123.4, tags=metric_level_tag)

assert self.recv() == "gauge:123.4|g|#bar:baz,foo:bar"

assert statsd.socket.recv() == "gauge:123.4|g|#bar:baz,foo:bar"

def assert_almost_equal(self, a, b, delta):

self.assertTrue(

def test_socket_error(statsd):

0 <= abs(a - b) <= delta, "%s - %s not within %s" % (a, b, delta)

statsd._socket = BrokenSocket()

)

statsd.gauge("no error", 1)

def test_socket_error(self):

self.statsd._socket = BrokenSocket()

self.statsd.gauge("no error", 1)

assert True, "success"

def test_socket_timeout(self):

self.statsd._socket = SlowSocket()

def test_socket_timeout(statsd):

self.statsd.gauge("no error", 1)

statsd._socket = SlowSocket()

statsd.gauge("no error", 1)

assert True, "success"

def test_timed(self):

def test_timed(statsd):

"""

Measure the distribution of a function's run time.

"""

@self.statsd.timed("timed.test")

@statsd.timed("timed.test")

def func(a, b, c=1, d=1):

"""docstring"""

time.sleep(0.5)

return (a, b, c, d)

self.assertEqual("func", func.__name__)

assert func.__name__ == "func"

self.assertEqual("docstring", func.__doc__)

assert func.__doc__ == "docstring"

result = func(1, 2, d=3)

# Assert it handles args and kwargs correctly.

self.assertEqual(result, (1, 2, 1, 3))

assert result, (1, 2, 1 == 3)

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("ms", type_)

assert type_ == "ms"

self.assertEqual("timed.test", name)

assert name == "timed.test"

self.assert_almost_equal(500, float(value), 100)

assert_almost_equal(500, float(value), 100)

def test_timed_exception(self):

def test_timed_exception(statsd):

"""

Exception bubble out of the decorator and is reported

to statsd as a dedicated counter.

"""

@self.statsd.timed("timed.test")

@statsd.timed("timed.test")

def func(a, b, c=1, d=1):

"""docstring"""

time.sleep(0.5)

return (a / b, c, d)

self.assertEqual("func", func.__name__)

assert func.__name__ == "func"

self.assertEqual("docstring", func.__doc__)

assert func.__doc__ == "docstring"

with self.assertRaises(ZeroDivisionError):

with pytest.raises(ZeroDivisionError):

func(1, 0)

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("c", type_)

assert type_ == "c"

self.assertEqual("timed.test_error_count", name)

assert name == "timed.test_error_count"

ardumontUnsubmitted

Not Done

i usually write it the other way around (and i think it's mostly in other swh modules but meh).
That is "actual_result == expected_result"

ardumont: i usually write it the other way around (and i think it's mostly in other swh modules but meh).

douarddaAuthorUnsubmitted

Done

indeed, just ran a query-replace-regex, so dumb results expected...

douardda: indeed, just ran a query-replace-regex, so dumb results expected...

self.assertEqual(int(value), 1)

assert int(value) == 1

def test_timed_no_metric(self,):

def test_timed_no_metric(statsd):

"""

Test using a decorator without providing a metric.

"""

@self.statsd.timed()

@statsd.timed()

def func(a, b, c=1, d=1):

"""docstring"""

time.sleep(0.5)

return (a, b, c, d)

self.assertEqual("func", func.__name__)

assert func.__name__ == "func"

self.assertEqual("docstring", func.__doc__)

assert func.__doc__ == "docstring"

result = func(1, 2, d=3)

# Assert it handles args and kwargs correctly.

self.assertEqual(result, (1, 2, 1, 3))

assert result, (1, 2, 1 == 3)

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("ms", type_)

assert type_ == "ms"

self.assertEqual("swh.core.tests.test_statsd.func", name)

assert name == "swh.core.tests.test_statsd.func"

self.assert_almost_equal(500, float(value), 100)

assert_almost_equal(500, float(value), 100)

def test_timed_coroutine(self):

def test_timed_coroutine(statsd):

"""

Measure the distribution of a coroutine function's run time.

Warning: Python >= 3.5 only.

"""

import asyncio

@self.statsd.timed("timed.test")

@statsd.timed("timed.test")

@asyncio.coroutine

def print_foo():

"""docstring"""

time.sleep(0.5)

print("foo")

loop = asyncio.new_event_loop()

loop.run_until_complete(print_foo())

loop.close()

# Assert

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("ms", type_)

assert type_ == "ms"

self.assertEqual("timed.test", name)

assert name == "timed.test"

self.assert_almost_equal(500, float(value), 100)

assert_almost_equal(500, float(value), 100)

def test_timed_context(self):

def test_timed_context(statsd):

"""

Measure the distribution of a context's run time.

"""

# In milliseconds

with self.statsd.timed("timed_context.test") as timer:

with statsd.timed("timed_context.test") as timer:

self.assertIsInstance(timer, TimedContextManagerDecorator)

assert isinstance(timer, TimedContextManagerDecorator)

time.sleep(0.5)

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("ms", type_)

assert type_ == "ms"

self.assertEqual("timed_context.test", name)

assert name == "timed_context.test"

self.assert_almost_equal(500, float(value), 100)

assert_almost_equal(500, float(value), 100)

self.assert_almost_equal(500, timer.elapsed, 100)

assert_almost_equal(500, timer.elapsed, 100)

def test_timed_context_exception(self):

def test_timed_context_exception(statsd):

"""

Exception bubbles out of the `timed` context manager and is

reported to statsd as a dedicated counter.

"""

class ContextException(Exception):

pass

def func(self):

def func(statsd):

with self.statsd.timed("timed_context.test"):

with statsd.timed("timed_context.test"):

time.sleep(0.5)

raise ContextException()

# Ensure the exception was raised.

self.assertRaises(ContextException, func, self)

with pytest.raises(ContextException):

func(statsd)

# Ensure the timing was recorded.

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("c", type_)

assert type_ == "c"

self.assertEqual("timed_context.test_error_count", name)

assert name == "timed_context.test_error_count"

self.assertEqual(int(value), 1)

assert int(value) == 1

def test_timed_context_no_metric_name_exception(self):

def test_timed_context_no_metric_name_exception(statsd):

"""Test that an exception occurs if using a context manager without a

metric name.

"""

def func(self):

def func(statsd):

with self.statsd.timed():

with statsd.timed():

time.sleep(0.5)

# Ensure the exception was raised.

self.assertRaises(TypeError, func, self)

with pytest.raises(TypeError):

func(statsd)

# Ensure the timing was recorded.

packet = self.recv()

packet = statsd.socket.recv()

self.assertEqual(packet, None)

assert packet is None

def test_timed_start_stop_calls(self):

def test_timed_start_stop_calls(statsd):

timer = self.statsd.timed("timed_context.test")

timer = statsd.timed("timed_context.test")

timer.start()

time.sleep(0.5)

timer.stop()

packet = self.recv()

packet = statsd.socket.recv()

name_value, type_ = packet.split("|")

name, value = name_value.split(":")

self.assertEqual("ms", type_)

assert type_ == "ms"

self.assertEqual("timed_context.test", name)

assert name == "timed_context.test"

self.assert_almost_equal(500, float(value), 100)

assert_almost_equal(500, float(value), 100)

def test_batched(self):

self.statsd.open_buffer()

def test_batched(statsd):

self.statsd.gauge("page.views", 123)

statsd.open_buffer()

self.statsd.timing("timer", 123)

statsd.gauge("page.views", 123)

self.statsd.close_buffer()

statsd.timing("timer", 123)

statsd.close_buffer()

self.assertEqual("page.views:123|g\ntimer:123|ms", self.recv())

assert statsd.socket.recv() == "page.views:123|g\ntimer:123|ms"

def test_context_manager(self):

def test_context_manager():

fake_socket = FakeSocket()

with Statsd() as statsd:

statsd._socket = fake_socket

statsd.gauge("page.views", 123)

statsd.timing("timer", 123)

self.assertEqual("page.views:123|g\ntimer:123|ms", fake_socket.recv())

assert fake_socket.recv() == "page.views:123|g\ntimer:123|ms"

def test_batched_buffer_autoflush(self):

def test_batched_buffer_autoflush():

fake_socket = FakeSocket()

with Statsd() as statsd:

statsd._socket = fake_socket

for i in range(51):

statsd.increment("mycounter")

self.assertEqual(

assert "\n".join(["mycounter:1|c" for i in range(50)]) == fake_socket.recv()

"\n".join(["mycounter:1|c" for i in range(50)]), fake_socket.recv(),

)

self.assertEqual("mycounter:1|c", fake_socket.recv())

assert fake_socket.recv() == "mycounter:1|c"

def test_module_level_instance(self):

def test_module_level_instance(statsd):

from swh.core.statsd import statsd

self.assertTrue(isinstance(statsd, Statsd))

assert isinstance(statsd, Statsd)

def test_instantiating_does_not_connect(self):

def test_instantiating_does_not_connect():

local_statsd = Statsd()

self.assertEqual(None, local_statsd._socket)

assert local_statsd._socket is None

def test_accessing_socket_opens_socket(self):

def test_accessing_socket_opens_socket():

local_statsd = Statsd()

try:

self.assertIsNotNone(local_statsd.socket)

assert local_statsd.socket is not None

finally:

local_statsd.close_socket()

def test_accessing_socket_multiple_times_returns_same_socket(self):

def test_accessing_socket_multiple_times_returns_same_socket():

local_statsd = Statsd()

fresh_socket = FakeSocket()

local_statsd._socket = fresh_socket

self.assertEqual(fresh_socket, local_statsd.socket)

assert fresh_socket == local_statsd.socket

self.assertNotEqual(FakeSocket(), local_statsd.socket)

assert FakeSocket() != local_statsd.socket

def test_tags_from_environment(self):

with preserve_envvars("STATSD_TAGS"):

os.environ["STATSD_TAGS"] = "country:china,age:45"

statsd = Statsd()

def test_tags_from_environment(monkeypatch):

monkeypatch.setenv("STATSD_TAGS", "country:china,age:45")

statsd = Statsd()

statsd._socket = FakeSocket()

statsd.gauge("gt", 123.4)

self.assertEqual("gt:123.4|g|#age:45,country:china", statsd.socket.recv())

assert statsd.socket.recv() == "gt:123.4|g|#age:45,country:china"

def test_tags_from_environment_and_constant(self):

with preserve_envvars("STATSD_TAGS"):

def test_tags_from_environment_and_constant(monkeypatch):

os.environ["STATSD_TAGS"] = "country:china,age:45"

monkeypatch.setenv("STATSD_TAGS", "country:china,age:45")

statsd = Statsd(constant_tags={"country": "canada"})

statsd._socket = FakeSocket()

statsd.gauge("gt", 123.4)

self.assertEqual("gt:123.4|g|#age:45,country:canada", statsd.socket.recv())

assert statsd.socket.recv() == "gt:123.4|g|#age:45,country:canada"

def test_tags_from_environment_warning(self):

def test_tags_from_environment_warning(monkeypatch):

with preserve_envvars("STATSD_TAGS"):

monkeypatch.setenv("STATSD_TAGS", "valid:tag,invalid_tag")

os.environ["STATSD_TAGS"] = "valid:tag,invalid_tag"

with pytest.warns(UserWarning) as record:

statsd = Statsd()

assert len(record) == 1

assert "invalid_tag" in record[0].message.args[0]

assert "valid:tag" not in record[0].message.args[0]

assert statsd.constant_tags == {"valid": "tag"}

def test_gauge_doesnt_send_none(self):

self.statsd.gauge("metric", None)

def test_gauge_doesnt_send_none(statsd):

assert self.recv() is None

statsd.gauge("metric", None)

assert statsd.socket.recv() is None

def test_increment_doesnt_send_none(self):

self.statsd.increment("metric", None)

assert self.recv() is None

def test_increment_doesnt_send_none(statsd):

statsd.increment("metric", None)

def test_decrement_doesnt_send_none(self):

assert statsd.socket.recv() is None

self.statsd.decrement("metric", None)

assert self.recv() is None

def test_decrement_doesnt_send_none(statsd):

def test_timing_doesnt_send_none(self):

statsd.decrement("metric", None)

self.statsd.timing("metric", None)

assert statsd.socket.recv() is None

assert self.recv() is None

def test_histogram_doesnt_send_none(self):

def test_timing_doesnt_send_none(statsd):

self.statsd.histogram("metric", None)

statsd.timing("metric", None)

assert self.recv() is None

assert statsd.socket.recv() is None

def test_param_host(self):

with preserve_envvars("STATSD_HOST", "STATSD_PORT"):

def test_histogram_doesnt_send_none(statsd):

os.environ["STATSD_HOST"] = "test-value"

statsd.histogram("metric", None)

os.environ["STATSD_PORT"] = ""

assert statsd.socket.recv() is None

def test_param_host(monkeypatch):

monkeypatch.setenv("STATSD_HOST", "test-value")

monkeypatch.setenv("STATSD_PORT", "")

local_statsd = Statsd(host="actual-test-value")

self.assertEqual(local_statsd.host, "actual-test-value")

assert local_statsd.host == "actual-test-value"

self.assertEqual(local_statsd.port, 8125)

assert local_statsd.port == 8125

def test_param_port(self):

def test_param_port(monkeypatch):

with preserve_envvars("STATSD_HOST", "STATSD_PORT"):

monkeypatch.setenv("STATSD_HOST", "")

os.environ["STATSD_HOST"] = ""

monkeypatch.setenv("STATSD_PORT", "12345")

os.environ["STATSD_PORT"] = "12345"

local_statsd = Statsd(port=4321)

assert local_statsd.host == "localhost"

assert local_statsd.port == 4321

self.assertEqual(local_statsd.host, "localhost")

self.assertEqual(local_statsd.port, 4321)

def test_envvar_host(self):

def test_envvar_host(monkeypatch):

with preserve_envvars("STATSD_HOST", "STATSD_PORT"):

monkeypatch.setenv("STATSD_HOST", "test-value")

os.environ["STATSD_HOST"] = "test-value"

monkeypatch.setenv("STATSD_PORT", "")

os.environ["STATSD_PORT"] = ""

local_statsd = Statsd()

assert local_statsd.host == "test-value"

assert local_statsd.port == 8125

self.assertEqual(local_statsd.host, "test-value")

self.assertEqual(local_statsd.port, 8125)

def test_envvar_port(self):

def test_envvar_port(monkeypatch):

with preserve_envvars("STATSD_HOST", "STATSD_PORT"):

monkeypatch.setenv("STATSD_HOST", "")

os.environ["STATSD_HOST"] = ""

monkeypatch.setenv("STATSD_PORT", "12345")

os.environ["STATSD_PORT"] = "12345"

local_statsd = Statsd()

self.assertEqual(local_statsd.host, "localhost")

assert local_statsd.host == "localhost"

self.assertEqual(local_statsd.port, 12345)

assert local_statsd.port == 12345

def test_namespace_added(self):

def test_namespace_added():

local_statsd = Statsd(namespace="test-namespace")

local_statsd._socket = FakeSocket()

local_statsd.gauge("gauge", 123.4)

assert local_statsd.socket.recv() == "test-namespace.gauge:123.4|g"

def test_contextmanager_empty(self):

with self.statsd:

def test_contextmanager_empty(statsd):

with statsd:

assert True, "success"

def test_contextmanager_buffering(self):

with self.statsd as s:

def test_contextmanager_buffering(statsd):

with statsd as s:

s.gauge("gauge", 123.4)

s.gauge("gauge_other", 456.78)

self.assertIsNone(s.socket.recv())

assert s.socket.recv() is None

assert statsd.socket.recv() == "gauge:123.4|g\ngauge_other:456.78|g"

self.assertEqual(self.recv(), "gauge:123.4|g\ngauge_other:456.78|g")

def test_timed_elapsed(self):

def test_timed_elapsed(statsd):

with self.statsd.timed("test_timer") as t:

with statsd.timed("test_timer") as t:

pass

self.assertGreaterEqual(t.elapsed, 0)

assert t.elapsed >= 0

self.assertEqual(self.recv(), "test_timer:%s|ms" % t.elapsed)

assert statsd.socket.recv() == "test_timer:%s|ms" % t.elapsed