Changeset - c7dff65ad06b
[Not reviewed]
0 1 0
Lance Edgar - 8 years ago 2016-08-11 20:21:25
ledgar@sacfoodcoop.com
Fix broken test
1 file changed with 4 insertions and 2 deletions:
0 comments (0 inline, 0 general)
rattail/tests/filemon/test_actions.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
from __future__ import unicode_literals
 

	
 
import os
 
import time
 
import Queue
 
from unittest import TestCase
 

	
 
from mock import Mock, patch, call
 
from fixture import TempIO
 

	
 
from rattail.config import make_config
 
from rattail.config import make_config, RattailConfig
 
from rattail.filemon import actions
 
from rattail.filemon.config import Profile, ProfileAction
 

	
 

	
 
class TestAction(TestCase):
 

	
 
    def test_callable_must_be_implemented_in_subclass(self):
 
        config = make_config([])
 
        action = actions.Action(config)
 
        self.assertRaises(NotImplementedError, action)
 

	
 

	
 
@@ -92,24 +92,25 @@ class TestPerformActions(TestCase):
 
                call(self.tmp.join(u'file1')),
 
                call(self.tmp.join(u'file2')),
 
                ])
 
        self.assertFalse(os.path.exists(self.tmp.join(u'file1')))
 
        self.assertTrue(os.path.exists(self.tmp.join(u'file2')))
 
        self.assertTrue(os.path.exists(self.tmp.join(u'file3')))
 

	
 

	
 
class TestInvokeAction(TestCase):
 

	
 
    def setUp(self):
 
        self.action = ProfileAction()
 
        self.action.config = RattailConfig()
 
        self.action.action = Mock(return_value=None)
 
        self.action.retry_attempts = 6
 
        self.tmp = TempIO()
 
        self.file = self.tmp.putfile(u'file', u'')
 

	
 
    def test_action_which_succeeds_is_only_called_once(self):
 
        actions.invoke_action(self.action, self.file)
 
        self.assertEqual(self.action.action.call_count, 1)
 

	
 
    def test_action_with_no_delay_does_not_pause_between_attempts(self):
 
        self.action.retry_attempts = 3
 
        self.action.action.side_effect = [RuntimeError, RuntimeError, None]
 
@@ -121,25 +122,26 @@ class TestInvokeAction(TestCase):
 
    def test_action_with_delay_pauses_between_attempts(self):
 
        self.action.retry_attempts = 3
 
        self.action.retry_delay = 1
 
        self.action.action.side_effect = [RuntimeError, RuntimeError, None]
 
        start = time.time()
 
        actions.invoke_action(self.action, self.file)
 
        self.assertEqual(self.action.action.call_count, 3)
 
        self.assertTrue(time.time() - start >= 2.0)
 

	
 
    def test_action_which_fails_is_only_attempted_the_specified_number_of_times(self):
 
        self.action.action.side_effect = RuntimeError
 
        # Last attempt will not handle the exception; assert that as well.
 
        self.assertRaises(RuntimeError, actions.invoke_action, self.action, self.file)
 
        with patch('rattail.filemon.actions.send_email') as send_email:
 
            self.assertRaises(RuntimeError, actions.invoke_action, self.action, self.file)
 
        self.assertEqual(self.action.action.call_count, 6)
 

	
 
    def test_action_which_fails_then_succeeds_stops_retrying(self):
 
        # First 2 attempts fail, third succeeds.
 
        self.action.action.side_effect = [RuntimeError, RuntimeError, None]
 
        actions.invoke_action(self.action, self.file)
 
        self.assertEqual(self.action.action.call_count, 3)
 

	
 
    def test_action_which_fails_with_different_errors_stops_retrying(self):
 
        self.action.action.side_effect = [ValueError, TypeError, None]
 
        # Second attempt will not handle the exception; assert that as well.
 
        self.assertRaises(TypeError, actions.invoke_action, self.action, self.file)
0 comments (0 inline, 0 general)