Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Unit tests for the graphyte module."""
import re
try:
import socketserver
except ImportError:
import SocketServer as socketserver # Python 2.x compatibility
import time
import unittest
import graphyte
class TestSender(graphyte.Sender):
def __init__(self, *args, **kwargs):
graphyte.Sender.__init__(self, 'dummy_host', *args, **kwargs)
self.messages = []
def send_socket(self, message):
self.messages.append(message)
def pop_message(self):
assert self.messages, 'no messages sent'
return self.messages.pop(0)
class TestHandler(socketserver.BaseRequestHandler):
messages = []
def handle(self):
def init(*args, **kwargs):
"""Initialize default Sender instance with given args."""
global default_sender
default_sender = Sender(*args, **kwargs)
parser.add_argument('-s', '--server', default='localhost',
help='hostname of Graphite server to send to, default %(default)s')
parser.add_argument('-p', '--port', type=int, default=2003,
help='port to send message to, default %(default)d')
parser.add_argument('-u', '--udp', action='store_true',
help='send via UDP instead of TCP')
parser.add_argument('-t', '--timestamp', type=int,
help='Unix timestamp for message (defaults to current time)')
parser.add_argument('-q', '--quiet', action='store_true',
help="quiet mode (don't log send to stdout)")
args = parser.parse_args()
if not args.quiet:
logging.basicConfig(level=logging.INFO, format='%(message)s')
sender = Sender(args.server, port=args.port, log_sends=not args.quiet,
protocol='udp' if args.udp else 'tcp')
sender.send(args.metric, args.value, timestamp=args.timestamp)