Commit 64a03a90 authored by ale's avatar ale

make sure parse_* functions always use UTC; fix tests

parent dc864afa
......@@ -11,6 +11,7 @@ from lens2 import logstore, syslog_parser, utils
class FakeResultSet(object):
def __init__(self, count=5):
self._count = count
self.total = count
......@@ -18,11 +19,13 @@ class FakeResultSet(object):
self._result = []
for x in range(count):
self._result.append({'_source': 'fake %d' % x})
def __iter__(self):
return iter(self._result)
class LogStoreTest(unittest.TestCase):
class LogStoreTest(mox.MoxTestBase):
def _index_name(self, when=None):
if when is None:
when = datetime.datetime.utcnow()
......@@ -31,7 +34,7 @@ class LogStoreTest(unittest.TestCase):
def _gen_indices(self, days):
"""Generate index names for the last n days."""
now = datetime.datetime.utcnow()
for i in range(days):
for i in xrange(days):
yield self._index_name(now - datetime.timedelta(days=i))
def test_expire_indices(self):
......@@ -57,7 +60,6 @@ class LogStoreTest(unittest.TestCase):
ls.expire(since)
self.mox.VerifyAll()
def test_search_indices(self):
"""Test searching 3d ago having 10d logs."""
days_available = 10
......@@ -126,12 +128,6 @@ class LogStoreTest(unittest.TestCase):
ls.insert("Feb 23 18:49:50 void prog[6666]: message", parser)
self.mox.VerifyAll()
def setUp(self):
self.mox = mox.Mox()
def tearDown(self):
self.mox.UnsetStubs()
if __name__ == '__main__':
#logging.basicConfig(level=logging.DEBUG)
......
import time
import unittest
from datetime import datetime
from lens2 import syslog_parser
......@@ -21,10 +22,11 @@ class SyslogParserTest(unittest.TestCase):
for entry in out:
self.assertEquals("sleppa", entry["host"])
# check timestamps
self.assertEquals(utils._to_timestamp(datetime(2010, 3, 13, 7, 28, 45, 0, utils.UTC())),
out[0]["timestamp"])
self.assertEquals(utils._to_timestamp(datetime(2010, 3, 13, 7, 28, 50, 0, utils.UTC())),
out[2]["timestamp"])
cur_year = datetime.now().year
t0 = time.mktime(datetime(cur_year, 3, 13, 7, 28, 45, 0, utils.UTC()).utctimetuple())
self.assertEquals(t0, out[0]["timestamp"])
t2 = time.mktime(datetime(cur_year, 3, 13, 7, 28, 50, 0, utils.UTC()).utctimetuple())
self.assertEquals(t2, out[2]["timestamp"])
# check facility
self.assertEquals("daemon", out[0]["facility"])
self.assertEquals("kernel", out[2]["facility"])
......
......@@ -5,21 +5,26 @@ from lens2 import utils
class UtilsTest(unittest.TestCase):
def setUp(self):
# Stub out utils._now
self.now = datetime.utcnow()
def _now():
return self.now
self.old_now = utils._now
utils._now = _now
def tearDown(self):
utils._now = self.old_now
def test_parse_iso8601_date(self):
d = datetime(2010, 3, 13, 7, 28, 45, 0, utils.UTC())
self.assertEquals(d, utils.parse_iso8601_date('2010-03-13T07:28:45'))
def test_parse_timespec(self):
now = datetime.now()
def _now():
return now
old_now = utils._now
utils._now = _now
data = [
('1d', (now - timedelta(1, 0), None)),
('2d', (now - timedelta(2, 0), None)),
('2d:1d', (now - timedelta(2, 0), now - timedelta(1, 0))),
('1d', (self.now - timedelta(1, 0), None)),
('2d', (self.now - timedelta(2, 0), None)),
('2d:1d', (self.now - timedelta(2, 0), self.now - timedelta(1, 0))),
('2010/12/13', (datetime(2010, 12, 13, 0, 0, 0),
datetime(2010, 12, 14, 0, 0, 0))),
('2010/12/13:2011/01/01', (datetime(2010, 12, 13, 0, 0, 0),
......@@ -30,10 +35,9 @@ class UtilsTest(unittest.TestCase):
self.assertEquals(expected_range, result,
"parse_time_range('%s') failed: %s (expected: %s)" % (
spec, result, expected_range))
utils._now = old_now
def test_format_log(self):
log = {'timestamp': utils._to_timestamp(datetime(2010, 12, 13, 1, 2, 3, 0, utils.UTC())),
log = {'timestamp': datetime(2010, 12, 13, 1, 2, 3, 0, utils.UTC()),
'host': 'server', 'program': 'sshd', 'pid': 1234,
'facility': 'auth', 'severity': 'info',
'msg': 'Something'}
......@@ -44,3 +48,10 @@ class UtilsTest(unittest.TestCase):
expected = '2010-12-13T01:02:03 server auth.info sshd: Something\n'
self.assertEquals(expected,
utils.format_log(log))
def test_parse_time_range(self):
self.assertEquals(
(self.now - timedelta(3), None),
utils.parse_time_range('3d'))
import re
import time
from datetime import datetime, tzinfo, timedelta
......@@ -14,12 +15,7 @@ _utc = UTC()
# Make this a local function to simplify testing.
def _now():
return datetime.now()
def _to_timestamp(d):
from_epoch = (d - datetime(1970, 1,1, tzinfo=UTC())).total_seconds()
return from_epoch
return datetime.utcnow()
def parse_iso8601_date(s):
......@@ -84,7 +80,6 @@ def format_log(log):
tag = log['program']
if log.get('pid'):
tag = '%s[%s]' % (tag, log['pid'])
stamp = datetime.fromtimestamp(log['timestamp'] / 1000000.0)
return '%s %s %s.%s %s: %s\n' % (
stamp.isoformat(), log['host'],
log['timestamp'].strftime('%Y-%m-%dT%H:%M:%S'), log['host'],
log.get('facility', '<none>'), log.get('severity', '<none>'), tag, log['msg'])
[egg_info]
tag_build = .dev
tag_svn_revision = true
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment