"""
Tests of the pavelib.utils.timer module.
"""
from datetime import datetime, timedelta
from mock import patch, MagicMock
from unittest import TestCase
from pavelib.utils import timer
@timer.timed
def identity(*args, **kwargs):
"""
An identity function used as a default task to test the timing of.
"""
return args, kwargs
MOCK_OPEN = MagicMock(spec=open)
@patch.dict('pavelib.utils.timer.__builtins__', open=MOCK_OPEN)
class TimedDecoratorTests(TestCase):
"""
Tests of the pavelib.utils.timer:timed decorator.
"""
def setUp(self):
super(TimedDecoratorTests, self).setUp()
patch_dumps = patch.object(timer.json, 'dump', autospec=True)
self.mock_dump = patch_dumps.start()
self.addCleanup(patch_dumps.stop)
patch_makedirs = patch.object(timer.os, 'makedirs', autospec=True)
self.mock_makedirs = patch_makedirs.start()
self.addCleanup(patch_makedirs.stop)
patch_datetime = patch.object(timer, 'datetime', autospec=True)
self.mock_datetime = patch_datetime.start()
self.addCleanup(patch_datetime.stop)
patch_exists = patch.object(timer, 'exists', autospec=True)
self.mock_exists = patch_exists.start()
self.addCleanup(patch_exists.stop)
MOCK_OPEN.reset_mock()
def get_log_messages(self, task=identity, args=None, kwargs=None, raises=None):
"""
Return all timing messages recorded during the execution of ``task``.
"""
if args is None:
args = []
if kwargs is None:
kwargs = {}
if raises is None:
task(*args, **kwargs)
else:
self.assertRaises(raises, task, *args, **kwargs)
return [
call[0][0] # log_message
for call in self.mock_dump.call_args_list
]
@patch.object(timer, 'PAVER_TIMER_LOG', '/tmp/some-log')
def test_times(self):
start = datetime(2016, 7, 20, 10, 56, 19)
end = start + timedelta(seconds=35.6)
self.mock_datetime.utcnow.side_effect = [start, end]
messages = self.get_log_messages()
self.assertEquals(len(messages), 1)
# I'm not using assertDictContainsSubset because it is
# removed in python 3.2 (because the arguments were backwards)
# and it wasn't ever replaced by anything *headdesk*
self.assertIn('duration', messages[0])
self.assertEquals(35.6, messages[0]['duration'])
self.assertIn('started_at', messages[0])
self.assertEquals(start.isoformat(' '), messages[0]['started_at'])
self.assertIn('ended_at', messages[0])
self.assertEquals(end.isoformat(' '), messages[0]['ended_at'])
@patch.object(timer, 'PAVER_TIMER_LOG', None)
def test_no_logs(self):
messages = self.get_log_messages()
self.assertEquals(len(messages), 0)
@patch.object(timer, 'PAVER_TIMER_LOG', '/tmp/some-log')
def test_arguments(self):
messages = self.get_log_messages(args=(1, 'foo'), kwargs=dict(bar='baz'))
self.assertEquals(len(messages), 1)
# I'm not using assertDictContainsSubset because it is
# removed in python 3.2 (because the arguments were backwards)
# and it wasn't ever replaced by anything *headdesk*
self.assertIn('args', messages[0])
self.assertEquals([repr(1), repr('foo')], messages[0]['args'])
self.assertIn('kwargs', messages[0])
self.assertEquals({'bar': repr('baz')}, messages[0]['kwargs'])
@patch.object(timer, 'PAVER_TIMER_LOG', '/tmp/some-log')
def test_task_name(self):
messages = self.get_log_messages()
self.assertEquals(len(messages), 1)
# I'm not using assertDictContainsSubset because it is
# removed in python 3.2 (because the arguments were backwards)
# and it wasn't ever replaced by anything *headdesk*
self.assertIn('task', messages[0])
self.assertEquals('pavelib.paver_tests.test_timer.identity', messages[0]['task'])
@patch.object(timer, 'PAVER_TIMER_LOG', '/tmp/some-log')
def test_exceptions(self):
@timer.timed
def raises():
"""
A task used for testing exception handling of the timed decorator.
"""
raise Exception('The Message!')
messages = self.get_log_messages(task=raises, raises=Exception)
self.assertEquals(len(messages), 1)
# I'm not using assertDictContainsSubset because it is
# removed in python 3.2 (because the arguments were backwards)
# and it wasn't ever replaced by anything *headdesk*
self.assertIn('exception', messages[0])
self.assertEquals("Exception: The Message!", messages[0]['exception'])
@patch.object(timer, 'PAVER_TIMER_LOG', '/tmp/some-log-%Y-%m-%d-%H-%M-%S.log')
def test_date_formatting(self):
start = datetime(2016, 7, 20, 10, 56, 19)
end = start + timedelta(seconds=35.6)
self.mock_datetime.utcnow.side_effect = [start, end]
messages = self.get_log_messages()
self.assertEquals(len(messages), 1)
MOCK_OPEN.assert_called_once_with('/tmp/some-log-2016-07-20-10-56-19.log', 'a')
@patch.object(timer, 'PAVER_TIMER_LOG', '/tmp/some-log')
def test_nested_tasks(self):
@timer.timed
def parent():
"""
A timed task that calls another task
"""
identity()
parent_start = datetime(2016, 7, 20, 10, 56, 19)
parent_end = parent_start + timedelta(seconds=60)
child_start = parent_start + timedelta(seconds=10)
child_end = parent_end - timedelta(seconds=10)
self.mock_datetime.utcnow.side_effect = [parent_start, child_start, child_end, parent_end]
messages = self.get_log_messages(task=parent)
self.assertEquals(len(messages), 2)
# Child messages first
self.assertIn('duration', messages[0])
self.assertEquals(40, messages[0]['duration'])
self.assertIn('started_at', messages[0])
self.assertEquals(child_start.isoformat(' '), messages[0]['started_at'])
self.assertIn('ended_at', messages[0])
self.assertEquals(child_end.isoformat(' '), messages[0]['ended_at'])
# Parent messages after
self.assertIn('duration', messages[1])
self.assertEquals(60, messages[1]['duration'])
self.assertIn('started_at', messages[1])
self.assertEquals(parent_start.isoformat(' '), messages[1]['started_at'])
self.assertIn('ended_at', messages[1])
self.assertEquals(parent_end.isoformat(' '), messages[1]['ended_at'])