From 67f9e6fa01ea17145bf212a77984e63d7ae48d79 Mon Sep 17 00:00:00 2001 From: Noah Gift Date: Sat, 10 Sep 2011 10:10:39 -0700 Subject: [PATCH] Fixing caching problem in main, and adding unit tests --- loggly.py | 74 +++++++++++++++++++++++++++++++++--------- main.py | 3 +- test_loggly.py | 88 ++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 148 insertions(+), 17 deletions(-) create mode 100644 test_loggly.py diff --git a/loggly.py b/loggly.py index fb5d651..d066cd3 100644 --- a/loggly.py +++ b/loggly.py @@ -1,38 +1,80 @@ +"""Asynchronous Logging To Loggly From App Engine + +Example Usage:: + +#GLOBAL +EPOINT = "https://logs.loggly.com/inputs/8f4e64db-f994-43c5-b147-2327b23680d2" +HOOVER = loggly.LogglyLogger(endpoint=epoint, + level=logging.INFO) + +logging.info("This will go to loggly via https RPC call") + +""" import logging, os from logging import handlers -import xml.dom.minidom as minidom from google.appengine.api import urlfetch -# create our async rpc object - - class LogglyHTTPSHandler(handlers.MemoryHandler): - def __init__(self, capacity, flushLevel, target, endpoint): - handlers.MemoryHandler.__init__(self, capacity, flushLevel, target) + """Custom Handler That Performs Async Calls""" + + def __init__(self, capacity, flush_level, target, endpoint): + handlers.MemoryHandler.__init__(self, capacity, flush_level, target) self.appname = os.getcwd().split('/')[-2] self.version = os.getcwd().split('/')[-1] self.endpoint = endpoint def flush(self): + """Overides flush method with rpc based flush. + + Returns : + (list) An empty list which represents an empty buffer + """ + rpc = urlfetch.create_rpc() - stuff = 'source=' + self.appname + '-' + self.version + stuff = "source=%s-%s" % (self.appname, self.version) for record in self.buffer: stuff += self.format(record) - urlfetch.make_fetch_call(rpc, url=self.endpoint, payload=stuff, method=urlfetch.POST) + urlfetch.make_fetch_call(rpc, + url=self.endpoint, + payload=stuff, + method=urlfetch.POST) self.buffer = [] + return self.buffer class LogglyLogger(): + """Configures The Loggly Logger Handler""" + def __init__(self, endpoint, level): self.endpoint = endpoint self.level = level + self.log_handler = self.configure_logger() + + def configure_logger(self): + """Configures HTTPS Logger Handler. + + Returns: + (object) log_handler object + + """ + + log_handler = LogglyHTTPSHandler(100, self.level, + None, + self.endpoint) + format_str = '''%(asctime)s level=%(levelname)s, msg="%(message)s", + module=%(module)s, file="%(filename)s", lineno=%(lineno)d''' + logging.Formatter(format_str) logger = logging.getLogger() - lh = LogglyHTTPSHandler(100, level, None, endpoint) - formatterStr = '''%(asctime)s level=%(levelname)s, msg="%(message)s", module=%(module)s, file="%(filename)s", lineno=%(lineno)d''' - formatter = logging.Formatter(formatterStr) - lh.setFormatter(formatter) - logger.addHandler(lh) - logger.setLevel(level) - self.lh = lh + logger.addHandler(log_handler) + logger.setLevel(self.level) + return log_handler def flush(self): - self.lh.flush() + """Perform RPC based flush via customized MemoryHandler + + Returns: + (list) An empty list which represents an empty buffer + + """ + + return self.log_handler.flush() + diff --git a/main.py b/main.py index 557f7e9..de26f80 100755 --- a/main.py +++ b/main.py @@ -35,6 +35,8 @@ from view import MainView import loggly +HOOVER = loggly.LogglyLogger('http://logs.loggly.com/inputs/7e7a83a8-3f2b-457e-9a1a-7805c0329d6f', logging.INFO) + # do some redirect shizzle for the bizzle that is izzle class RedirectHandler(webapp.RequestHandler): def get(self, group): @@ -92,7 +94,6 @@ def main(): ('/([a-zA-Z0-9]{1,3})?(.xml|.json|.html|.txt)?', MainHandler), ('/(.*?)', RedirectHandler) ], debug=True) - hoover = loggly.LogglyLogger('http://logs.loggly.com/inputs/7e7a83a8-3f2b-457e-9a1a-7805c0329d6f', logging.INFO) wsgiref.handlers.CGIHandler().run(application) if __name__ == '__main__': diff --git a/test_loggly.py b/test_loggly.py new file mode 100644 index 0000000..067cfc6 --- /dev/null +++ b/test_loggly.py @@ -0,0 +1,88 @@ +""" + +TEST Coverage: + +To Test This: + +1. You must set path to google_appengine SDK +2. You must install nosetests, coverage +3. Then run this command: + +nosetests --with-coverage --cover-package=loggly test_loggly.py + +""" + +import unittest +import logging +import sys +try: + import sys + path = '/usr/local/google_appengine' + sys.path.insert(0,path) + from loggly import LogglyHTTPSHandler, LogglyLogger +except ImportError: + "Cannot find Google App Engine SDK at: %s" % path + sys.exit(1) + +#import Google App Engine Test Code +from google.appengine.api import apiproxy_stub_map +from google.appengine.api import urlfetch_stub + +# Create a stub map so we can build App Engine mock stubs. +apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap() + +# Register App Engine mock stubs. +apiproxy_stub_map.apiproxy.RegisterStub( + 'urlfetch', urlfetch_stub.URLFetchServiceStub()) + +class TestLogglyHTTPSHandler(unittest.TestCase): + + def setUp(self): + self.endpoint = "http://logs.loggly.com/inputs/83e527d7-fad3-4d93-89da-0c2d8c0bcd6c" + self.log_handler = LogglyHTTPSHandler(100, logging.INFO, + None, + self.endpoint) + self.buffer = [1] + + def test_constructor(self): + self.assertTrue(self.log_handler) + + def test_flush(self): + """The buffer should be empty after flush is called""" + + my_buffer = self.log_handler.flush() + self.assertFalse(my_buffer) + + def tearDown(self): + del self.endpoint + del self.log_handler + del self.buffer + +class TestLogglyLogger(unittest.TestCase): + + def setUp(self): + self.endpoint = "http://logs.loggly.com/inputs/83e527d7-fad3-4d93-89da-0c2d8c0bcd6c" + self.loggly = LogglyLogger(self.endpoint,level=logging.INFO) + self.buffer = [1] + + def test_constructor_loggly(self): + self.assertTrue(self.loggly) + + def test_configure_logger(self): + + handler = self.loggly.configure_logger() + self.assertTrue(hasattr(handler, "flush")) + + def test_flush_loggly_loggly(self): + """The buffer should be empty after flush is called""" + + my_buffer = self.loggly.flush() + self.assertFalse(my_buffer) + + def tearDown(self): + del self.endpoint + del self.loggly + del self.buffer + +if __name__ == '__main__': + unittest.main() \ No newline at end of file