x/stacktach-shoebox
Code Issues Proposed changes

Moved utils to notification_utils project

Browse Source
This commit is contained in:
Sandy Walsh 2014-05-21 12:11:50 +00:00
parent b77eebdabe
commit 3b65cab8c2
8 changed files with 18 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
shoebox/roll_checker.py
View File

@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import utils
import notification_utils
class RollChecker(object):
@ -36,11 +36,11 @@ class TimeRollChecker(RollChecker):
self.timedelta = timedelta
def start(self, archive):
self.start_time = utils.now()
self.start_time = notification_utils.now()
self.end_time = self.start_time + self.timedelta
def check(self, archive):
return utils.now() >= self.end_time
return notification_utils.now() >= self.end_time
class SizeRollChecker(RollChecker):

5
shoebox/roll_manager.py
View File

@ -17,9 +17,10 @@ import fnmatch
import os
import os.path
import notification_utils
import archive
import disk_storage
import utils
class NoMoreFiles(Exception):
@ -113,7 +114,7 @@ class WritingRollManager(RollManager):
self._roll_archive()
def _make_filename(self):
f = utils.now().strftime(self.filename_template)
f = notification_utils.now().strftime(self.filename_template)
f = f.replace(" ", "_")
f = f.replace("/", "_")
f = f.replace(":", "_")

65
shoebox/utils.py
View File

@ -1,65 +0,0 @@
# Copyright (c) 2014 Dark Secret Software Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import calendar
import collections
import datetime
import decimal
import json
def now():
"""Broken out for testing."""
return datetime.datetime.utcnow()
def dt_to_decimal(utc):
decimal.getcontext().prec = 30
return decimal.Decimal(str(calendar.timegm(utc.utctimetuple()))) + \
(decimal.Decimal(str(utc.microsecond)) /
decimal.Decimal("1000000.0"))
def dt_from_decimal(dec):
if dec == None:
return "n/a"
integer = int(dec)
micro = (dec - decimal.Decimal(integer)) * decimal.Decimal(1000000)
daittyme = datetime.datetime.utcfromtimestamp(integer)
return daittyme.replace(microsecond=micro)
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime.datetime):
if obj.utcoffset() is not None:
obj = obj - obj.utcoffset()
return str(dt_to_decimal(obj))
return super(DateTimeEncoder, self).default(obj)
# This is a hack for comparing structures load'ed from json
# (which are always unicode) back to strings. It's used
# for assertEqual() in the tests and is very slow and expensive.
def unicode_to_string(data):
if isinstance(data, basestring):
return str(data)
elif isinstance(data, collections.Mapping):
return dict(map(unicode_to_string, data.iteritems()))
elif isinstance(data, collections.Iterable):
return type(data)(map(unicode_to_string, data))
else:
return data

4
test/integration/test_rolling.py
View File

@ -5,12 +5,12 @@ import os
import shutil
import unittest
import notification_utils
import notigen
from shoebox import disk_storage
from shoebox import roll_checker
from shoebox import roll_manager
from shoebox import utils
TEMPDIR = "test_temp"
@ -70,7 +70,7 @@ class TestSizeRolling(unittest.TestCase):
'uuid': event['uuid'],
}
json_event = json.dumps(event,
cls=utils.DateTimeEncoder)
cls=notification_utils.DateTimeEncoder)
manager.write(metadata, json_event)
entries.append((metadata, json_event))

11
test/test_roll_checker.py
View File

@ -2,8 +2,9 @@ import datetime
import mock
import unittest
import notification_utils
from shoebox import roll_checker
from shoebox import utils
class TestRollChecker(unittest.TestCase):
@ -11,7 +12,7 @@ class TestRollChecker(unittest.TestCase):
one_hour = datetime.timedelta(hours=1)
x = roll_checker.TimeRollChecker(one_hour)
now = datetime.datetime.utcnow()
with mock.patch.object(utils, 'now') as dt:
with mock.patch.object(notification_utils, 'now') as dt:
dt.return_value = now
x.start(None)
self.assertEqual(x.start_time, now)
@ -23,15 +24,15 @@ class TestRollChecker(unittest.TestCase):
now = datetime.datetime.utcnow()
x.start_time = now
x.end_time = now + one_hour
with mock.patch.object(utils, 'now') as dt:
with mock.patch.object(notification_utils, 'now') as dt:
dt.return_value = now + one_hour
self.assertTrue(x.check(None))
with mock.patch.object(utils, 'now') as dt:
with mock.patch.object(notification_utils, 'now') as dt:
dt.return_value = now
self.assertFalse(x.check(None))
with mock.patch.object(utils, 'now') as dt:
with mock.patch.object(notification_utils, 'now') as dt:
dt.return_value = now + one_hour - datetime.timedelta(seconds = 1)
self.assertFalse(x.check(None))

5
test/test_roll_manager.py
View File

@ -3,10 +3,11 @@ import mock
import os
import unittest
import notification_utils
from shoebox import archive
from shoebox import roll_checker
from shoebox import roll_manager
from shoebox import utils
class FakeArchive(object):
@ -72,7 +73,7 @@ class TestWritingRollManager(unittest.TestCase):
hour=10, minute=11, second=12)
x = roll_manager.WritingRollManager("filename_%c.dat", None)
with mock.patch.object(utils, "now") as dt:
with mock.patch.object(notification_utils, "now") as dt:
dt.return_value = now
filename = x._make_filename()
self.assertEqual(filename,

43
test/test_utils.py
View File

@ -1,43 +0,0 @@
import datetime
import decimal
import unittest
import dateutil.tz
from shoebox import utils
class TestUtils(unittest.TestCase):
def setUp(self):
self.handler = utils.DateTimeEncoder()
def test_handle_datetime_non_datetime(self):
self.assertRaises(TypeError, self.handler.default, "text")
def test_handle_datetime(self):
now = datetime.datetime(day=1, month=2, year=2014,
hour=10, minute=11, second=12)
self.assertEqual("1391249472", self.handler.default(now))
def test_handle_datetime_offset(self):
now = datetime.datetime(day=1, month=2, year=2014,
hour=10, minute=11, second=12,
tzinfo=dateutil.tz.tzoffset(None, 4*60*60))
self.assertEqual("1391220672", self.handler.default(now))
class TestDatetimeToDecimal(unittest.TestCase):
def test_datetime_to_decimal(self):
expected_decimal = decimal.Decimal('1356093296.123')
utc_datetime = datetime.datetime.utcfromtimestamp(expected_decimal)
actual_decimal = utils.dt_to_decimal(utc_datetime)
self.assertEqual(actual_decimal, expected_decimal)
def test_decimal_to_datetime(self):
expected_decimal = decimal.Decimal('1356093296.123')
expected_datetime = datetime.datetime.utcfromtimestamp(expected_decimal)
actual_datetime = utils.dt_from_decimal(expected_decimal)
self.assertEqual(actual_datetime, expected_datetime)
def test_dt_from_decimal_none(self):
self.assertEqual("n/a", utils.dt_from_decimal(None))

1
tox.ini
View File

@ -7,6 +7,7 @@ deps =
nose
mock
notigen
notification_utils
python-dateutil
commands = nosetests -d -v --with-coverage --cover-inclusive --cover-package shoebox []