conditionally use snap_http, fall back to local module

landscape/client/manager/snapmanager.py

@@ -6,10 +6,19 @@
 from landscape.client.manager.plugin import FAILED
 from landscape.client.manager.plugin import ManagerPlugin
 from landscape.client.manager.plugin import SUCCEEDED
-from landscape.client.snap.http import INCOMPLETE_STATUSES
-from landscape.client.snap.http import SnapdHttpException
-from landscape.client.snap.http import SnapHttp
-from landscape.client.snap.http import SUCCESS_STATUSES
+
+try:
+    import snap_http
+    from snap_http import INCOMPLETE_STATUSES
+    from snap_http import SnapdHttpException
+    from snap_http import SUCCESS_STATUSES
+except ImportError:
+    from landscape.client.snap.http import INCOMPLETE_STATUSES
+    from landscape.client.snap.http import SnapdHttpException
+    from landscape.client.snap.http import SnapHttp
+    from landscape.client.snap.http import SUCCESS_STATUSES
+
+    snap_http = SnapHttp()
 
 
 class SnapManager(ManagerPlugin):
@@ -23,18 +32,17 @@ class SnapManager(ManagerPlugin):
     def __init__(self):
         super().__init__()
 
-        self._snap_http = SnapHttp()
         self.SNAP_METHODS = {
-            "install-snaps": self._snap_http.install_snap,
-            "install-snaps-batch": self._snap_http.install_snaps,
-            "remove-snaps": self._snap_http.remove_snap,
-            "remove-snaps-batch": self._snap_http.remove_snaps,
-            "refresh-snaps": self._snap_http.refresh_snap,
-            "refresh-snaps-batch": self._snap_http.refresh_snaps,
-            "hold-snaps": self._snap_http.hold_snap,
-            "hold-snaps-batch": self._snap_http.hold_snaps,
-            "unhold-snaps": self._snap_http.unhold_snap,
-            "unhold-snaps-batch": self._snap_http.unhold_snaps,
+            "install-snaps": snap_http.install,
+            "install-snaps-batch": snap_http.install_all,
+            "remove-snaps": snap_http.remove,
+            "remove-snaps-batch": snap_http.remove_all,
+            "refresh-snaps": snap_http.refresh,
+            "refresh-snaps-batch": snap_http.refresh_all,
+            "hold-snaps": snap_http.hold,
+            "hold-snaps-batch": snap_http.hold_all,
+            "unhold-snaps": snap_http.unhold,
+            "unhold-snaps-batch": snap_http.unhold_all,
         }
 
     def register(self, registry):
@@ -161,7 +169,7 @@ def get_status():
             logging.info("Polling snapd for status of pending snap changes")
 
             try:
-                result = self._snap_http.check_changes().get("result", [])
+                result = snap_http.check_changes().get("result", [])
                 result_dict = {c["id"]: c for c in result}
             except SnapdHttpException as e:
                 logging.error(f"Error checking status of snap changes: {e}")
@@ -253,7 +261,7 @@ def _respond(self, snap_results, opid, errors):
 
     def _send_installed_snap_update(self):
         try:
-            installed_snaps = self._snap_http.get_snaps()
+            installed_snaps = snap_http.list()
         except SnapdHttpException as e:
             logging.error(
                 f"Unable to list installed snaps after snap change: {e}",

landscape/client/manager/tests/test_snapmanager.py

@@ -3,25 +3,25 @@
 from landscape.client.manager.manager import FAILED
 from landscape.client.manager.manager import SUCCEEDED
 from landscape.client.manager.snapmanager import SnapManager
-from landscape.client.snap.http import SnapdHttpException
-from landscape.client.snap.http import SnapHttp as OrigSnapHttp
 from landscape.client.tests.helpers import LandscapeTest
 from landscape.client.tests.helpers import ManagerHelper
 
+try:
+    from snap_http import SnapdHttpException
+except ImportError:
+    from landscape.client.snap.http import SnapdHttpException
+
 
 class SnapManagerTest(LandscapeTest):
     helpers = [ManagerHelper]
 
     def setUp(self):
         super().setUp()
 
-        self.snap_http = mock.Mock(spec_set=OrigSnapHttp)
-        self.SnapHttp = mock.patch(
-            "landscape.client.manager.snapmanager.SnapHttp",
+        self.snap_http = mock.patch(
+            "landscape.client.manager.snapmanager.snap_http",
         ).start()
 
-        self.SnapHttp.return_value = self.snap_http
-
         self.broker_service.message_store.set_accepted_types(
             ["operation-result"],
         )
@@ -49,14 +49,14 @@ def install_snap(name, revision=None, channel=None, classic=False):
 
             return mock.DEFAULT
 
-        self.snap_http.install_snap.side_effect = install_snap
+        self.snap_http.install.side_effect = install_snap
         self.snap_http.check_changes.return_value = {
             "result": [
                 {"id": "1", "status": "Done"},
                 {"id": "2", "status": "Done"},
             ],
         }
-        self.snap_http.get_snaps.return_value = {"installed": []}
+        self.snap_http.list.return_value = {"installed": []}
 
         result = self.manager.dispatch_message(
             {
@@ -90,11 +90,11 @@ def test_install_snaps_batch(self):
         When no channels or revisions are specified, snaps are installed
         via a single call to snapd.
         """
-        self.snap_http.install_snaps.return_value = {"change": "1"}
+        self.snap_http.install_all.return_value = {"change": "1"}
         self.snap_http.check_changes.return_value = {
             "result": [{"id": "1", "status": "Done"}],
         }
-        self.snap_http.get_snaps.return_value = {
+        self.snap_http.list.return_value = {
             "installed": [
                 {
                     "name": "hello",
@@ -136,10 +136,10 @@ def got_result(r):
         return result.addCallback(got_result)
 
     def test_install_snap_immediate_error(self):
-        self.snap_http.install_snaps.side_effect = SnapdHttpException(
+        self.snap_http.install_all.side_effect = SnapdHttpException(
             b'{"result": "whoops"}',
         )
-        self.snap_http.get_snaps.return_value = {"installed": []}
+        self.snap_http.list.return_value = {"installed": []}
 
         result = self.manager.dispatch_message(
             {
@@ -168,9 +168,9 @@ def got_result(r):
         return result.addCallback(got_result)
 
     def test_install_snap_no_status(self):
-        self.snap_http.install_snaps.return_value = {"change": "1"}
+        self.snap_http.install_all.return_value = {"change": "1"}
         self.snap_http.check_changes.return_value = {"result": []}
-        self.snap_http.get_snaps.return_value = {"installed": []}
+        self.snap_http.list.return_value = {"installed": []}
 
         result = self.manager.dispatch_message(
             {
@@ -197,9 +197,9 @@ def got_result(r):
         return result.addCallback(got_result)
 
     def test_install_snap_check_error(self):
-        self.snap_http.install_snaps.return_value = {"change": "1"}
+        self.snap_http.install_all.return_value = {"change": "1"}
         self.snap_http.check_changes.side_effect = SnapdHttpException("whoops")
-        self.snap_http.get_snaps.return_value = {"installed": []}
+        self.snap_http.list.return_value = {"installed": []}
 
         result = self.manager.dispatch_message(
             {
@@ -228,11 +228,11 @@ def got_result(r):
         return result.addCallback(got_result)
 
     def test_remove_snap(self):
-        self.snap_http.remove_snaps.return_value = {"change": "1"}
+        self.snap_http.remove_all.return_value = {"change": "1"}
         self.snap_http.check_changes.return_value = {
             "result": [{"id": "1", "status": "Done"}],
         }
-        self.snap_http.get_snaps.return_value = {"installed": []}
+        self.snap_http.list.return_value = {"installed": []}
 
         result = self.manager.dispatch_message(
             {

landscape/client/monitor/snapmonitor.py

@@ -1,10 +1,17 @@
 import logging
 
 from landscape.client.monitor.plugin import DataWatcher
-from landscape.client.snap.http import SnapdHttpException
-from landscape.client.snap.http import SnapHttp
 from landscape.message_schemas.server_bound import SNAPS
 
+try:
+    import snap_http
+    from snap_http import SnapdHttpException
+except ImportError:
+    from landscape.client.snap.http import SnapHttp
+    from landscape.client.snap.http import SnapdHttpException
+
+    snap_http = SnapHttp()
+
 
 class SnapMonitor(DataWatcher):
 
@@ -13,11 +20,6 @@ class SnapMonitor(DataWatcher):
     persist_name = message_type
     scope = "snaps"
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        self._snap_http = SnapHttp()
-
     def register(self, registry):
         self.config = registry.config
         # The default interval is 30 minutes.
@@ -27,7 +29,7 @@ def register(self, registry):
 
     def get_data(self):
         try:
-            snaps = self._snap_http.get_snaps()
+            snaps = snap_http.list()
         except SnapdHttpException as e:
             logging.error(f"Unable to list installed snaps: {e}")
             return

landscape/client/monitor/tests/test_snapmonitor.py

@@ -1,8 +1,13 @@
-from unittest.mock import Mock
+from unittest.mock import patch
 
 from landscape.client.monitor.snapmonitor import SnapMonitor
-from landscape.client.snap.http import SnapdHttpException, SnapHttp
-from landscape.client.tests.helpers import LandscapeTest, MonitorHelper
+from landscape.client.tests.helpers import LandscapeTest
+from landscape.client.tests.helpers import MonitorHelper
+
+try:
+    from snap_http import SnapdHttpException
+except ImportError:
+    from landscape.client.snap.http import SnapdHttpException
 
 
 class SnapMonitorTest(LandscapeTest):
@@ -30,21 +35,19 @@ def test_get_data_snapd_http_exception(self):
         """
         Tests that we return no data if there is an error getting it.
         """
-        snap_http_mock = Mock(
-            spec=SnapHttp,
-            get_snaps=Mock(side_effect=SnapdHttpException)
-        )
         plugin = SnapMonitor()
-        plugin._snap_http = snap_http_mock
         self.monitor.add(plugin)
 
-        with self.assertLogs(level="ERROR") as cm:
+        with patch(
+            "landscape.client.monitor.snapmonitor.snap_http",
+        ) as snap_http_mock, self.assertLogs(level="ERROR") as cm:
+            snap_http_mock.list.side_effect = SnapdHttpException
             plugin.exchange()
 
         messages = self.mstore.get_pending_messages()
 
         self.assertEqual(len(messages), 0)
         self.assertEqual(
             cm.output,
-            ["ERROR:root:Unable to list installed snaps: "]
+            ["ERROR:root:Unable to list installed snaps: "],
         )