Back to index

obnam  1.1
hooks_tests.py
Go to the documentation of this file.
00001 # Copyright (C) 2009  Lars Wirzenius
00002 #
00003 # This program is free software: you can redistribute it and/or modify
00004 # it under the terms of the GNU General Public License as published by
00005 # the Free Software Foundation, either version 3 of the License, or
00006 # (at your option) any later version.
00007 #
00008 # This program is distributed in the hope that it will be useful,
00009 # but WITHOUT ANY WARRANTY; without even the implied warranty of
00010 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
00011 # GNU General Public License for more details.
00012 #
00013 # You should have received a copy of the GNU General Public License
00014 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
00015 
00016 
00017 import unittest
00018 
00019 import obnamlib
00020 
00021 import base64
00022 
00023 class HookTests(unittest.TestCase):
00024 
00025     def setUp(self):
00026         self.hook = obnamlib.Hook()
00027         
00028     def callback(self, *args, **kwargs):
00029         self.args = args
00030         self.kwargs = kwargs
00031         
00032     def callback2(self, *args, **kwargs):
00033         self.args2 = args
00034         self.kwargs2 = kwargs
00035 
00036     def test_has_no_callbacks_by_default(self):
00037         self.assertEqual(self.hook.callbacks, [])
00038         
00039     def test_adds_callback(self):
00040         self.hook.add_callback(self.callback)
00041         self.assertEqual(self.hook.callbacks, [self.callback])
00042         
00043     def test_adds_callback_only_once(self):
00044         self.hook.add_callback(self.callback)
00045         self.hook.add_callback(self.callback)
00046         self.assertEqual(self.hook.callbacks, [self.callback])
00047         
00048     def test_adds_two_callbacks(self):
00049         id1 = self.hook.add_callback(self.callback)
00050         id2 = self.hook.add_callback(self.callback2, 
00051                                      obnamlib.Hook.DEFAULT_PRIORITY + 1)
00052         self.assertEqual(self.hook.callbacks, [self.callback, self.callback2])
00053         self.assertNotEqual(id1, id2)
00054         
00055     def test_adds_callbacks_in_reverse_order(self):
00056         id1 = self.hook.add_callback(self.callback)
00057         id2 = self.hook.add_callback(self.callback2, 
00058                                      obnamlib.Hook.DEFAULT_PRIORITY - 1)
00059         self.assertEqual(self.hook.callbacks, [self.callback2, self.callback])
00060         self.assertNotEqual(id1, id2)
00061 
00062     def test_calls_callback(self):
00063         self.hook.add_callback(self.callback)
00064         self.hook.call_callbacks('bar', kwarg='foobar')
00065         self.assertEqual(self.args, ('bar',))
00066         self.assertEqual(self.kwargs, { 'kwarg': 'foobar' })
00067 
00068     def test_removes_callback(self):
00069         cb_id = self.hook.add_callback(self.callback)
00070         self.hook.remove_callback(cb_id)
00071         self.assertEqual(self.hook.callbacks, [])
00072 
00073 class NeverAddsFilter(object):
00074 
00075     def __init__(self):
00076         self.tag = "never"
00077 
00078     def filter_read(self, data, *args, **kwargs):
00079         self.args = args
00080         self.kwargs = kwargs
00081         self.wasread = True
00082         return data
00083 
00084     def filter_write(self, data, *args, **kwargs):
00085         self.args = args
00086         self.kwargs = kwargs
00087         self.wasread = False
00088         return data
00089 
00090 class Base64Filter(object):
00091 
00092     def __init__(self):
00093         self.tag = "base64"
00094 
00095     def filter_read(self, data, *args, **kwargs):
00096         self.args = args
00097         self.kwargs = kwargs
00098         self.wasread = True
00099         return base64.b64decode(data)
00100 
00101     def filter_write(self, data, *args, **kwargs):
00102         self.args = args
00103         self.kwargs = kwargs
00104         self.wasread = False
00105         return base64.b64encode(data)
00106 
00107 class FilterHookTests(unittest.TestCase):
00108 
00109     def setUp(self):
00110         self.hook = obnamlib.FilterHook()
00111 
00112     def test_add_filter_ok(self):
00113         self.hook.add_callback(NeverAddsFilter())
00114 
00115     def test_never_filter_no_tags(self):
00116         self.hook.add_callback(NeverAddsFilter())
00117         self.assertEquals(self.hook.run_filter_write("foo"), "\0foo")
00118         
00119     def test_never_filter_clean_revert(self):
00120         self.hook.add_callback(NeverAddsFilter())
00121         self.assertEquals(self.hook.run_filter_read("\0foo"), "foo")
00122 
00123     def test_base64_filter_encode(self):
00124         self.hook.add_callback(Base64Filter())
00125         self.assertEquals(self.hook.run_filter_write("OK"), "base64\0AE9L")
00126 
00127     def test_base64_filter_decode(self):
00128         self.hook.add_callback(Base64Filter())
00129         self.assertEquals(self.hook.run_filter_read("base64\0AE9L"), "OK")
00130 
00131     def test_missing_filter_raises(self):
00132         self.assertRaises(obnamlib.MissingFilterError,
00133                           self.hook.run_filter_read, "missing\0")
00134 
00135     def test_missing_filter_gives_tag(self):
00136         try:
00137             self.hook.run_filter_read("missing\0")
00138         except obnamlib.MissingFilterError, e:
00139             self.assertEquals(e.tagname, "missing")
00140 
00141     def test_can_remove_filters(self):
00142         myfilter = NeverAddsFilter()
00143         filterid = self.hook.add_callback(myfilter)
00144         self.hook.remove_callback(filterid)
00145         self.assertEquals(self.hook.callbacks, [])
00146 
00147     def test_call_callbacks_raises(self):
00148         self.assertRaises(NotImplementedError, self.hook.call_callbacks, "")
00149 
00150 class HookManagerTests(unittest.TestCase):
00151 
00152     def setUp(self):
00153         self.hooks = obnamlib.HookManager()
00154         self.hooks.new('foo')
00155         
00156     def callback(self, *args, **kwargs):
00157         self.args = args
00158         self.kwargs = kwargs
00159 
00160     def test_has_no_tests_initially(self):
00161         hooks = obnamlib.HookManager()
00162         self.assertEqual(hooks.hooks, {})
00163         
00164     def test_adds_new_hook(self):
00165         self.assert_(self.hooks.hooks.has_key('foo'))
00166         
00167     def test_adds_new_filter_hook(self):
00168         self.hooks.new_filter('bar')
00169         self.assert_('bar' in self.hooks.filters)
00170         
00171     def test_adds_callback(self):
00172         self.hooks.add_callback('foo', self.callback)
00173         self.assertEqual(self.hooks.hooks['foo'].callbacks, [self.callback])
00174 
00175     def test_removes_callback(self):
00176         cb_id = self.hooks.add_callback('foo', self.callback)
00177         self.hooks.remove_callback('foo', cb_id)
00178         self.assertEqual(self.hooks.hooks['foo'].callbacks, [])
00179 
00180     def test_calls_callbacks(self):
00181         self.hooks.add_callback('foo', self.callback)
00182         self.hooks.call('foo', 'bar', kwarg='foobar')
00183         self.assertEqual(self.args, ('bar',))
00184         self.assertEqual(self.kwargs, { 'kwarg': 'foobar' })
00185 
00186     def test_filter_write_returns_value_of_callbacks(self):
00187         self.hooks.new_filter('bar')
00188         self.assertEquals(self.hooks.filter_write('bar', "foo"), "\0foo")
00189 
00190     def test_filter_read_returns_value_of_callbacks(self):
00191         self.hooks.new_filter('bar')
00192         self.assertEquals(self.hooks.filter_read('bar', "\0foo"), "foo")
00193 
00194     def test_add_callbacks_to_filters(self):
00195         self.hooks.new_filter('bar')
00196         filt = NeverAddsFilter()
00197         self.hooks.add_callback('bar', filt)
00198         self.assertEquals(self.hooks.filters['bar'].callbacks, [filt])
00199 
00200     def test_remove_callbacks_from_filters(self):
00201         self.hooks.new_filter('bar')
00202         filt = NeverAddsFilter()
00203         self.hooks.add_callback('bar', filt)
00204         self.hooks.remove_callback('bar', filt)
00205         self.assertEquals(self.hooks.filters['bar'].callbacks, [])