<s> from __future__ import absolute_import , unicode_literals <EOL> from unittest import skipUnless <EOL> from prometheus_client import Counter <EOL> from prometheus_client import CollectorRegistry , generate_latest <EOL> try : <EOL> from prometheus_client . twisted import MetricsResource <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . web . server import Site <EOL> from twisted . web . resource import Resource <EOL> from twisted . internet import reactor <EOL> from twisted . web . client import Agent <EOL> from twisted . web . client import readBody <EOL> HAVE_TWISTED = True <EOL> except ImportError : <EOL> from unittest import TestCase <EOL> HAVE_TWISTED = False <EOL> class MetricsResourceTest ( TestCase ) : <EOL> @ skipUnless ( HAVE_TWISTED , "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> self . registry = CollectorRegistry ( ) <EOL> def test_reports_metrics ( self ) : <EOL> """<STR_LIT>""" <EOL> c = Counter ( '<STR_LIT>' , '<STR_LIT>' , registry = self . registry ) <EOL> c . inc ( ) <EOL> root = Resource ( ) <EOL> root . putChild ( b'<STR_LIT>' , MetricsResource ( registry = self . registry ) ) <EOL> server = reactor . listenTCP ( <NUM_LIT:0> , Site ( root ) ) <EOL> self . addCleanup ( server . stopListening ) <EOL> agent = Agent ( reactor ) <EOL> port = server . getHost ( ) . port <EOL> url = "<STR_LIT>" . format ( port = port ) <EOL> d = agent . request ( b"<STR_LIT:GET>" , url . encode ( "<STR_LIT:ascii>" ) ) <EOL> d . addCallback ( readBody ) <EOL> d . addCallback ( self . assertEqual , generate_latest ( self . registry ) ) <EOL> return d </s>
