cc-4105: Added more tests.

python_apps/media-monitor2/tests/notifier.py

@@ -1,8 +1,12 @@
 # -*- coding: utf-8 -*-
 import unittest
+import json
 
 from media.monitor.airtime import AirtimeNotifier, AirtimeMessageReceiver
-from mock import patch
+from mock import patch, Mock
+from media.monitor.config import MMConfig
+
+def filter_ev(d): return { i : j for i,j in d.iteritems() if i != 'event_type' }
 
 class TestReceiver(unittest.TestCase):
     def setUp(self):
@@ -10,14 +14,50 @@ class TestReceiver(unittest.TestCase):
         cfg = {}
         self.amr = AirtimeMessageReceiver(cfg)
 
-    def test_message(self):
+    def test_supported_messages(self):
+        self.assertTrue( len(self.amr.supported_messages()) > 0 )
+
+    def test_supported(self):
+        # Every supported message should fire something
         for event_type in self.amr.supported_messages():
             msg = { 'event_type' : event_type, 'extra_param' : 123 }
-            filtered = { i : j for i,j in msg.iteritems() if i != 'event_type' }
+            filtered = filter_ev(msg)
             with patch.object(self.amr, 'execute_message') as mock_method:
                 mock_method.side_effect = None
                 ret = self.amr.message(msg)
                 self.assertTrue(ret)
                 mock_method.assert_called_with(event_type, filtered)
 
+    def test_no_mod_message(self):
+        ev = { 'event_type' : 'new_watch', 'directory' : 'something here' }
+        filtered = filter_ev(ev)
+        with patch.object(self.amr, 'execute_message') as mock_method:
+            mock_method.return_value = "tested"
+            ret = self.amr.message(ev)
+            self.assertTrue( ret ) # message passing worked
+            mock_method.assert_called_with(ev['event_type'], filtered)
+            # test that our copy of the message does not get modified
+            self.assertTrue( 'event_type' in ev )
+
+class TestAirtimeNotifier(unittest.TestCase):
+    def test_handle_message(self):
+        #from configobj import ConfigObj
+        test_cfg = MMConfig('./test_config.cfg')
+        ran = [False]
+        class MockReceiver(object):
+            def message(me,m):
+                self.assertTrue( 'event_type' in m )
+                self.assertEqual( m['path'], '/bs/path' )
+                ran[0] = True
+        airtime = AirtimeNotifier(cfg=test_cfg, message_receiver=MockReceiver())
+        m1 = Mock()
+        m1.ack = "ack'd message"
+        m2 = Mock()
+        m2.body = json.dumps({ 'event_type' : 'file_delete', 'path' : '/bs/path' })
+        airtime.handle_message(body=m1,message=m2)
+        self.assertTrue( ran[0] )
+
+
+
+
 if __name__ == '__main__': unittest.main()

python_apps/media-monitor2/tests/test_config.cfg

@@ -1,4 +1,4 @@
-api_client = "airtime"
+api_client = 'airtime'
 
 # where the binary files live
 bin_dir = '/usr/lib/airtime/media-monitor'
@@ -18,7 +18,7 @@ rabbitmq_vhost = '/'
 ############################################
 # Media-Monitor preferences                         #
 ############################################
-check_filesystem_events = 5 #how long to queue up events performed on the files themselves.
-check_airtime_events = 30 #how long to queue metadata input from airtime.
+check_filesystem_events = '5' 
+check_airtime_events = '30' 
 
 list_value_testing = 'val1', 'val2', 'val3'