Some play callback handlers. Now uses notigen

This commit is contained in:
Sandy Walsh 2014-05-16 00:31:04 +00:00
parent 0a3ead169a
commit 6753b9baba
5 changed files with 84 additions and 19 deletions

36
shoebox/handlers.py Normal file
View File

@ -0,0 +1,36 @@
# 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 os.path
import shutil
class ArchiveCallback(object):
def on_open(self, filename):
"""Called when an Archive is opened."""
pass
def on_close(self, filename):
"""Called when an Archive is closed."""
pass
class MoveFileCallback(object):
def __init__(self, destination_folder):
self.destination_folder = destination_folder
def on_close(self, filename):
"""Move this file to destination folder."""
shutil.move(filename, self.destination_folder)

View File

@ -30,16 +30,6 @@ class NoValidFile(Exception):
pass
class ArchiveCallback(object):
def on_open(self, filename):
"""Called when an Archive is opened."""
pass
def on_close(self, filename):
"""Called when an Archive is closed."""
pass
class RollManager(object):
def __init__(self, filename_template, directory=".",
archive_class=None, archive_callback=None):

View File

@ -5,14 +5,13 @@ import os
import shutil
import unittest
import notigen
from shoebox import disk_storage
from shoebox import roll_checker
from shoebox import roll_manager
from shoebox import utils
import test.integration.gen_events as egen
TEMPDIR = "test_temp"
@ -49,10 +48,6 @@ class TestSizeRolling(unittest.TestCase):
shutil.rmtree(TEMPDIR, ignore_errors=True)
os.mkdir(TEMPDIR)
def tearDown(self):
# shutil.rmtree(TEMPDIR)
pass
def test_size_rolling(self):
callback = ArchiveCallback()
@ -62,7 +57,7 @@ class TestSizeRolling(unittest.TestCase):
TEMPDIR,
archive_callback=callback)
g = egen.EventGenerator(6000)
g = notigen.EventGenerator(6000)
entries = []
now = datetime.datetime.utcnow()
while len(entries) < 10000:

43
test/test_utils.py Normal file
View File

@ -0,0 +1,43 @@
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))

View File

@ -3,9 +3,10 @@ envlist = py26,py27
[testenv]
deps =
python-dateutil
coverage
nose
mock
coverage
notigen
python-dateutil
commands = nosetests -d -v --with-coverage --cover-inclusive --cover-package shoebox []