Spaces:
Runtime error
Runtime error
File size: 5,499 Bytes
f10310e faa3d98 f10310e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
import sys
import unittest
import pytest
from mock import call, patch
from streamlit import config, metrics
class MetricsTest(unittest.TestCase):
"""Metrics Unittest class."""
def setUp(self):
"""Make sure Client singleton is always empty before starting tests."""
streamlit.metrics.Client._singleton = None
def tearDown(self):
"""Cleanup metrics client."""
config.set_option("global.metrics", False)
streamlit.metrics.Client._singleton = None
client = streamlit.metrics.Client.get_current()
client.toggle_metrics()
def test_constructor(self):
"""Test streamlit.metrics.Client."""
client = streamlit.metrics.Client()
self.assertEqual(streamlit.metrics.Client._singleton, client)
def test_get_current(self):
"""Test streamlit.metrics.clientget_current."""
client = streamlit.metrics.Client.get_current()
self.assertEqual(streamlit.metrics.Client._singleton, client)
def test_not_singleton(self):
"""Test streamlit.metrics.Client not singleton."""
client = streamlit.metrics.Client.get_current()
with pytest.raises(RuntimeError) as e:
streamlit.metrics.Client()
msg = "Client already initialized. Use .get_current() instead"
self.assertEqual(msg, str(e.value))
def test_enabled_metrics_no_prometheus(self):
"""Test streamlit.metrics.Client.toggle_metrics no prometheus."""
config.set_option("global.metrics", True)
client = streamlit.metrics.Client.get_current()
if sys.version_info <= (3, 0):
builtin_import = "__builtin__.__import__"
else:
builtin_import = "builtins.__import__"
with pytest.raises(ImportError) as e:
with patch(builtin_import, side_effect=ImportError):
client.toggle_metrics()
msg = "prometheus-client is not installed. pip install prometheus-client"
self.assertEqual(msg, str(e.value))
def test_enabled_metrics(self):
"""Test streamlit.metrics.toggle_metrics enabled."""
config.set_option("global.metrics", True)
client = streamlit.metrics.Client.get_current()
client._metrics = {}
# yapf: disable
client._raw_metrics = [
('Counter', 'unittest_counter', 'Unittest counter', []),
('Counter', 'unittest_counter_labels', 'Unittest counter labels', ['label']),
('Gauge', 'unittest_gauge', 'Unittest gauge', []),
]
# yapf: enable
client.toggle_metrics()
client.get("unittest_counter").inc()
client.get("unittest_counter_labels").labels("some_label")
client.get("unittest_gauge").set(42)
truth = [
"unittest_counter_total 1.0",
'unittest_counter_labels_total{label="some_label"} 0.0',
"unittest_gauge 42.0",
]
lines = client.generate_latest().splitlines()
metrics = [
x.decode("utf-8") for x in lines if x.decode("utf-8").startswith("unit")
]
metrics = [str(x) for x in metrics if "_created" not in x]
self.assertEqual(sorted(truth), sorted(metrics))
def test_disabled_metrics_check_value(self):
"""Test streamlit.metrics.Client.toggle_metrics disabled check value."""
with patch("streamlit.metrics.MockMetric", spec=True) as mock_metric:
config.set_option("global.metrics", False)
client = streamlit.metrics.Client.get_current()
client._metrics = {}
# yapf: disable
client._raw_metrics = [
('Counter', 'unittest_counter', 'Unittest counter', []),
('Counter', 'unittest_counter_labels', 'Unittest counter labels', ['label']),
('Gauge', 'unittest_gauge', 'Unittest gauge', []),
]
# yapf: enable
client.toggle_metrics()
# Test that handler in Server.py will return nothing.
self.assertEqual(client.generate_latest(), "")
client.get("unittest_counter").inc()
client.get("unittest_counter_labels").labels("some_label")
client.get("unittest_gauge").set(42)
client.get("unittest_gauge").dec()
calls = [
call(), # Constructor
call(), # unittest_counter
call(), # unittest_counter_labels
call(), # unittest_gauge
call().inc(),
call().labels("some_label"),
call().set(42),
call().dec(),
]
self.assertEqual(calls, mock_metric.mock_calls)
def test_disabled_metrics(self):
"""Test streamlit.metrics.Client.toggle_metrics disabled."""
config.set_option("global.metrics", False)
client = streamlit.metrics.Client.get_current()
client._metrics = {}
# yapf: disable
client._raw_metrics = [
('Counter', 'unittest_counter', 'Unittest counter', []),
('Counter', 'unittest_counter_labels', 'Unittest counter labels', ['label']),
('Gauge', 'unittest_gauge', 'Unittest gauge', []),
]
# yapf: enable
client.toggle_metrics()
client.get("unittest_counter").inc()
client.get("unittest_counter_labels").labels("some_label")
client.get("unittest_gauge").set(42)
client.get("unittest_gauge").dec()
|