Add: unit tests for removable_storage_watcher::main

providers/base/bin/removable_storage_watcher.py

@@ -1001,7 +1001,7 @@ def _summarize_changes(self, delta_records):
                     )
 
 
-def main():
+def main(argv=sys.argv[1:]):
     description = "Wait for the specified device to be inserted or removed."
     parser = argparse.ArgumentParser(description=description)
     parser.add_argument("action", choices=["insert", "remove"])
@@ -1046,7 +1046,7 @@ def main():
         help="Don't require drive being automounted",
     )
     parser.set_defaults(logging_level=logging.WARNING)
-    args = parser.parse_args()
+    args = parser.parse_args(argv)
 
     # Configure logging as requested
     # XXX: This may be incorrect as logging.basicConfig() fails after any other

providers/base/tests/test_removable_storage_watcher.py

@@ -0,0 +1,67 @@
+import unittest
+import sys
+from unittest.mock import patch, MagicMock
+
+sys.modules["dbus"] = MagicMock()
+sys.modules["dbus.exceptions"] = MagicMock()
+sys.modules["dbus.mainloop.glib"] = MagicMock()
+sys.modules["gi"] = MagicMock()
+sys.modules["gi.repository"] = MagicMock()
+
+import removable_storage_watcher
+
+
+class RemovableStorageWatcherTests(unittest.TestCase):
+
+    @patch("removable_storage_watcher.connect_to_system_bus")
+    @patch("removable_storage_watcher.is_udisks2_supported")
+    @patch("removable_storage_watcher.UDisks1StorageDeviceListener")
+    def test_main_udisk1(self, mock_udisk, mock_check, mock_connect):
+        """
+        Test the main function connects to the system bus,
+        opens a UDisks1 listener and check the action is performed
+        until timeout.
+        """
+
+        mock_connect.return_value = ("bus", "loop")
+        mock_check.return_value = False
+
+        argv = ["insert", "usb"]
+        removable_storage_watcher.main(argv)
+
+        mock_connect.assert_called_once_with()
+        mock_udisk.assert_called_once_with(
+            "bus",
+            "loop",
+            "insert",
+            ["usb"],
+            0,
+            False,
+        )
+
+        mock_udisk.return_value.check.assert_called_once_with(20)
+
+    @patch("removable_storage_watcher.connect_to_system_bus")
+    @patch("removable_storage_watcher.is_udisks2_supported")
+    @patch("removable_storage_watcher.UDisks2StorageDeviceListener")
+    def test_main_udisk2(self, mock_udisk, mock_check, mock_connect):
+        """
+        Test the main function uses a UDisks2 listener when
+        supported.
+        """
+
+        mock_connect.return_value = ("bus", "loop")
+        mock_check.return_value = True
+
+        argv = ["insert", "usb"]
+        removable_storage_watcher.main(argv)
+
+        mock_udisk.assert_called_once_with(
+            "bus",
+            "loop",
+            "insert",
+            ["usb"],
+            0,
+            False,
+            False,
+        )