unify close behavior for FTS source/destination & TransformRules

socorro/app/fetch_transform_save_app.py

@@ -374,8 +374,17 @@ def _setup_task_manager(self):
         self.config.executor_identity = self.task_manager.executor_identity
 
     #--------------------------------------------------------------------------
-    def _cleanup(self):
-        pass
+    def close(self):
+        try:
+            self.source.close()
+        except AttributeError:
+            # this source class has no close, we can ignore that & move on
+            pass
+        try:
+            self.destination.close()
+        except AttributeError:
+            # this destination class has no close, we can ignore that & move on
+            pass
 
     #--------------------------------------------------------------------------
     def main(self):
@@ -387,7 +396,7 @@ def main(self):
         self._setup_task_manager()
         self._setup_source_and_destination()
         self.task_manager.blocking_start(waiting_func=self.waiting_func)
-        self._cleanup()
+        self.close()
         self.config.logger.info('done.')
 
 
@@ -430,3 +439,13 @@ def _setup_source_and_destination(self):
             # the configuration failed to provide a "new_crash_source", fall
             # back to tying the "new_crash_source" to the "source".
             self.new_crash_source = self.source
+
+    #--------------------------------------------------------------------------
+    def close(self):
+        super(FetchTransformSaveWithSeparateNewCrashSourceApp, self).close()
+        if self.source != self.new_crash_source:
+            try:
+                self.new_crash_source.close()
+            except AttributeError:
+                # the new_crash_source has no close, move on without it
+                pass

socorro/lib/transform_rules.py

@@ -284,6 +284,9 @@ def __eq__(self, another):
         else:
             return False
 
+    #--------------------------------------------------------------------------
+    def close(self):
+        pass
 
 #==============================================================================
 class TransformRuleSystem(RequiredConfig):
@@ -471,6 +474,15 @@ def apply_until_predicate_fails(self, *args, **kwargs):
                 return False
         return None
 
+    #--------------------------------------------------------------------------
+    def close(self):
+        for a_rule in self.rules:
+            try:
+                a_rule.close()
+            except AttributeError:
+                # no close method mean no need to close
+                pass
+
 
 #------------------------------------------------------------------------------
 # Useful rule predicates and actions

socorro/processor/processor_2015.py

@@ -285,3 +285,15 @@ def _log_job_end(self, success, crash_id):
             'successful' if success else 'failed',
             crash_id
         )
+
+    #--------------------------------------------------------------------------
+    def close(self):
+        for a_rule_set_name, a_rule_set in self.rule_system.iteritems():
+            self.config.logger.debug('closing %s', a_rule_set_name)
+            try:
+                a_rule_set.close()
+            except AttributeError:
+                # guess we don't need to close that rule
+                pass
+        self.config.logger.debug('done closing rules')
+

socorro/processor/processor_app.py

@@ -171,10 +171,20 @@ def _setup_source_and_destination(self):
         )
 
     #--------------------------------------------------------------------------
-    def _cleanup(self):
+    def close(self):
         """when  the processor shutsdown, this function cleans up"""
-        if self.companion_process:
+        try:
             self.companion_process.close()
+        except AttributeError:
+            # there is either no companion or it doesn't have a close method
+            # we can skip on
+            pass
+        try:
+            self.processor.close()
+        except AttributeError:
+            # the processor implementation does not have a close method
+            # we can blithely skip on
+            pass
 
 
 if __name__ == '__main__':

socorro/unittest/app/test_generic_fetch_transform_save_app.py

@@ -21,6 +21,8 @@ def __init__(self, config):
                 self.the_list = []
 
             def _setup_source_and_destination(self):
+                self.source = Mock()
+                self.destination = Mock()
                 pass
 
             def _create_iter(self):
@@ -76,6 +78,7 @@ def __init__(self, config, quit_check_callback):
                                                        'Product': 'Fennicky',
                                                        'Version': '1.0'}),
                                      })
+                self.number_of_close_calls = 0
 
             def get_raw_crash(self, ooid):
                 return self.store[ooid]
@@ -87,16 +90,23 @@ def new_crashes(self):
                 for k in self.store.keys():
                     yield k
 
+            def close(self):
+                self.number_of_close_calls += 1
+
         class FakeStorageDestination(object):
 
             def __init__(self, config, quit_check_callback):
                 self.store = DotDict()
                 self.dumps = DotDict()
+                self.number_of_close_calls = 0
 
             def save_raw_crash(self, raw_crash, dump, crash_id):
                 self.store[crash_id] = raw_crash
                 self.dumps[crash_id] = dump
 
+            def close(self):
+                self.number_of_close_calls += 1
+
         logger = SilentFakeLogger()
         config = DotDict({
           'logger': logger,
@@ -125,6 +135,10 @@ def save_raw_crash(self, raw_crash, dump, crash_id):
         eq_(len(destination.dumps), 4)
         eq_(destination.dumps['1237'],
                          source.get_raw_dumps('1237'))
+        # ensure that each storage system had its close called
+        eq_(source.number_of_close_calls, 1)
+        eq_(destination.number_of_close_calls, 1)
+
 
     def test_source_iterator(self):
 

socorro/unittest/collector/test_crash_mover_app.py

@@ -3,6 +3,7 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from nose.tools import eq_, ok_, assert_raises
+from mock import Mock
 
 from socorro.collector.crashmover_app import CrashMoverApp
 from socorro.lib.threaded_task_manager import ThreadedTaskManager
@@ -19,6 +20,8 @@ def __init__(self, config):
                 self.the_list = []
 
             def _setup_source_and_destination(self):
+                self.source = Mock()
+                self.destination = Mock()
                 pass
 
             def _basic_iterator(self):
@@ -74,6 +77,10 @@ def __init__(self, config, quit_check_callback):
                                                        'Product': 'Fennicky',
                                                        'Version': '1.0'}),
                                      })
+                self.number_of_close_calls = 0
+
+            def close():
+                self.number_of_close_calls += 1
 
             def get_raw_crash(self, ooid):
                 return self.store[ooid]
@@ -86,16 +93,24 @@ def new_crashes(self):
                 for k in self.store.keys():
                     yield k
 
+            def close(self):
+                self.number_of_close_calls += 1
+                pass
+
         class FakeStorageDestination(object):
 
             def __init__(self, config, quit_check_callback):
                 self.store = DotDict()
                 self.dumps = DotDict()
+                self.number_of_close_calls = 0
 
             def save_raw_crash(self, raw_crash, dumps, crash_id):
                 self.store[crash_id] = raw_crash
                 self.dumps[crash_id] = dumps
 
+            def close(self):
+                self.number_of_close_calls += 1
+
         logger = SilentFakeLogger()
         config = DotDict({
           'logger': logger,
@@ -125,6 +140,10 @@ def save_raw_crash(self, raw_crash, dumps, crash_id):
         eq_(destination.dumps['1237'],
                          source.get_raw_dumps('1237'))
 
+        # ensure that each storage system had its close called
+        eq_(source.number_of_close_calls, 1)
+        eq_(destination.number_of_close_calls, 1)
+
     def test_source_iterator(self):
 
         class FakeStorageSource(object):

socorro/unittest/lib/test_transform_rules.py

@@ -35,6 +35,13 @@ class TestRuleTestLaughable(transform_rules.Rule):
     def _predicate(self, *args, **kwargs):
         return self.config.laughable != 'fred'
 
+    def close(self):
+        try:
+            self.close_counter += 1
+        except AttributeError:
+            self.close_counter = 1
+
+
 
 #==============================================================================
 class TestRuleTestDangerous(transform_rules.Rule):
@@ -44,6 +51,12 @@ class TestRuleTestDangerous(transform_rules.Rule):
     def _action(self, *args, **kwargs):
         return self.config.dangerous != 'sally'
 
+    def close(self):
+        try:
+            self.close_counter += 1
+        except AttributeError:
+            self.close_counter = 1
+
 
 #==============================================================================
 class TestTransformRules(TestCase):
@@ -578,3 +591,31 @@ def test_rules_in_config(self):
         ok_(isinstance(trs.rules[1], TestRuleTestDangerous))
         ok_(trs.rules[0].predicate(None))
         ok_(trs.rules[1].action(None))
+
+    def test_rules_close(self):
+        config = DotDict()
+        config.chatty_rules = False
+        config.chatty = False
+        config.tag = 'test.rule'
+        config.action = 'apply_all_rules'
+        config['TestRuleTestLaughable.laughable'] = 'wilma'
+        config['TestRuleTestDangerous.dangerous'] = 'dwight'
+        config.rules_list = DotDict()
+        config.rules_list.class_list = [
+            (
+                'TestRuleTestLaughable',
+                TestRuleTestLaughable,
+                'TestRuleTestLaughable'
+            ),
+            (
+                'TestRuleTestDangerous',
+                TestRuleTestDangerous,
+                'TestRuleTestDangerous'
+            )
+        ]
+        trs = transform_rules.TransformRuleSystem(config)
+
+        trs.close()
+
+        eq_(trs.rules[0].close_counter, 1)
+        eq_(trs.rules[1].close_counter, 1)