Merge lp:~therve/landscape-client/sources-list-plugin into lp:~landscape/landscape-client/trunk

[Thomas Herve]

Here it is. I think it's a fairly simple branch. It imports the gpg keys, and manages the sources.list. Nothing too fancy here. It relies on the full key being available in the server, so it means probably connecting to keys.ubuntu.com in Landscape. I think it's better than expected the clients to be able to open that weird port.

There is also no invalidation of keys. If someone feels it's important we can try to do it in a later branch (I think it implies creating a table to keep track of the added keys, so it's a little bit more involved).

[lp:~therve/landscape-client/sources-list-plugin]

329. By Thomas Herve on 2011-04-13

Add the plugin in all plugin list

330. By Thomas Herve on 2011-04-13

Remove temporary files

[Free Ekanayaka (free.ekanayaka)]

Nice work, nothing really blocking so feel free to address what you think is worth. +1

[1]

+ deferred.addCallback(self._handle_process_error)
+ deferred.addCallback(lambda ignore, path=path: os.unlink(path))

Probably not I huge deal, but I suspect that the key file is not removed if the process fails?

[2]

+ if not line.strip() or line.startswith("#"):

I think comments can start also after some blanks, so maybe:

+ if not line.strip() or line.strip().startswith("#"):

[3]

+class SourcesList(ManagerPlugin):

and

+ "repositories", self._wrap_handle_repositories)

It would be nice for the plugin and message name to match a bit more (that's often the cases for other plugins/messages). I'd suggest "class AptSources" and "apt-sources-replace" or something. The name "reposiories" sounds a bit generic, in case we want to add other operations later. This not blocking though.

[4]

+class ProcessError(Exception):
+ """Exception raised when running a process fails."""

and

+ def run_process(self, command, args):

This seems generic enough that it could be made a function under landscape.lib.process, like:

class ProcessError(Exception):
    """Exception raised when running a process fails."""

    def __init__(self, out, err):
        super(ProcessError, self).__init__("%s\n%s" % (out, err))

def run_process(self, command, args):
    """Run the process in an asynchronous fashion."""

    def handle_process_error(result):
        """
        Turn a failed process command (code != 0) to a C{ProcessError}.
        """
        out, err, code = result
        if code:
            raise ProcessError(out, err)
        else:
            return result

    def handle_process_failure(failure):
        """
        Turn a signaled process command to a C{ProcessError}.
        """
        out, err, signal = failure.value
        raise ProcessError(out, err)

    deferred = getProcessOutputAndValue(command, args)
    deferred.addCallback(handle_process_error)
    deferred.addErrback(handle_process_failure)
    return deferred

You could test run_process separately, and have the tests for SourcesList simply shows that ProcessError failures are handled correctly:

    def test_failed_import_no_changes(self):
        """
        If the C{apt-key} command failed for some reasons, the current
        repositories aren't changed.
        """

        def run_process(command, args):
            return fail(ProcessError("nok", "some error"))

        self.sourceslist.run_process = run_process

        sources = file(self.sourceslist.SOURCES_LIST, "w")
        sources.write("oki\n\ndoki\n#comment\n")
        sources.close()

        self.manager.dispatch_message(
            {"type": "repositories", "sources": [], "gpg-keys": ["key"],
             "operation-id": 1})

        self.assertEqual(
            "oki\n\ndoki\n#comment\n",
            file(self.sourceslist.SOURCES_LIST).read())

        service = self.broker_service
        self.assertMessages(service.message_store.get_pending_messages(),
                            [{"type": "operation-result",
                              'result-te...

review: Approve

[lp:~therve/landscape-client/sources-list-plugin]

331. By Thomas Herve on 2011-04-14

Remove temporary files after failures

332. By Thomas Herve on 2011-04-14

Handle comments not starting the line

333. By Thomas Herve on 2011-04-14

Rename message type

334. By Thomas Herve on 2011-04-14

Rename plugin

[Thomas Herve (therve)]

Thanks a lot for the careful review Free. I've handled 1 to 3, but I'll postpone 4 until we have a usecase.

[lp:~therve/landscape-client/sources-list-plugin]

335. By Thomas Herve on 2011-04-15

Run the reporter at the end of the message handling

[Kevin McDermott (bigkevmcd)]

Ok, looks fine to me

+1

review: Approve

=== added file 'landscape/manager/aptsources.py'
--- landscape/manager/aptsources.py	1970-01-01 00:00:00 +0000
+++ landscape/manager/aptsources.py	2011-04-15 08:07:29 +0000
@@ -0,0 +1,155 @@
+import glob
+import os
+import tempfile
+
+from twisted.internet.defer import succeed
+from twisted.internet.utils import getProcessOutputAndValue
+
+from landscape.manager.plugin import ManagerPlugin, SUCCEEDED, FAILED
+from landscape.package.reporter import find_reporter_command
+
+
+class ProcessError(Exception):
+    """Exception raised when running a process fails."""
+
+
+class AptSources(ManagerPlugin):
+    """A plugin managing sources.list content."""
+
+    SOURCES_LIST = "/etc/apt/sources.list"
+    SOURCES_LIST_D = "/etc/apt/sources.list.d"
+
+    def register(self, registry):
+        super(AptSources, self).register(registry)
+        registry.register_message(
+            "apt-sources-replace", self._wrap_handle_repositories)
+
+    def run_process(self, command, args):
+        """
+        Run the process in an asynchronous fashion, to be overriden in tests.
+        """
+        return getProcessOutputAndValue(command, args)
+
+    def _wrap_handle_repositories(self, message):
+        """
+        Wrap C{_handle_repositories} to generate an activity result based on
+        the returned value.
+        """
+        deferred = self._handle_repositories(message)
+
+        operation_result = {"type": "operation-result",
+                            "operation-id": message["operation-id"]}
+
+        def success(ignored):
+            operation_result["status"] = SUCCEEDED
+            return operation_result
+
+        def fail(failure):
+            operation_result["status"] = FAILED
+            text = "%s: %s" % (failure.type.__name__, failure.value)
+            operation_result["result-text"] = text
+            return operation_result
+
+        deferred.addCallbacks(success, fail)
+        deferred.addBoth(lambda result:
+                         self.manager.broker.send_message(result, urgent=True))
+
+    def _handle_process_error(self, result):
+        """
+        Turn a failed process command (code != 0) to a C{ProcessError}.
+        """
+        out, err, code = result
+        if code:
+            raise ProcessError("%s\n%s" % (out, err))
+
+    def _handle_process_failure(self, failure):
+        """
+        Turn a signaled process command to a C{ProcessError}.
+        """
+        if not failure.check(ProcessError):
+            out, err, signal = failure.value
+            raise ProcessError("%s\n%s" % (out, err))
+        else:
+            return failure
+
+    def _remove_and_continue(self, passthrough, path):
+        """
+        Remove the temporary file created for the process, and forward the
+        result.
+        """
+        os.unlink(path)
+        return passthrough
+
+    def _handle_repositories(self, message):
+        """
+        Handle a list of repositories to set on the machine.
+
+        The format is the following:
+
+        {"sources": [
+          {"name": "repository-name",
+           "content":
+              "deb http://archive.ubuntu.com/ubuntu/ maverick main\n\
+              "deb-src http://archive.ubuntu.com/ubuntu/ maverick main"}
+          {"name": "repository-name-dev",
+           "content":
+              "deb http://archive.ubuntu.com/ubuntu/ maverick universe\n\
+              "deb-src http://archive.ubuntu.com/ubuntu/ maverick universe"}],
+         "gpg-keys": ["-----BEGIN PGP PUBLIC KEY BLOCK-----\n\
+                      XXXX
+                      -----END PGP PUBLIC KEY BLOCK-----",
+                      "-----BEGIN PGP PUBLIC KEY BLOCK-----\n\
+                      YYY
+                      -----END PGP PUBLIC KEY BLOCK-----"]}
+        """
+        deferred = succeed(None)
+        for key in message["gpg-keys"]:
+            fd, path = tempfile.mkstemp()
+            os.close(fd)
+            key_file = file(path, "w")
+            key_file.write(key)
+            key_file.close()
+            deferred.addCallback(
+                lambda ignore, path=path:
+                    self.run_process("/usr/bin/apt-key", ["add", path]))
+            deferred.addCallback(self._handle_process_error)
+            deferred.addBoth(self._remove_and_continue, path)
+        deferred.addErrback(self._handle_process_failure)
+        return deferred.addCallback(
+            self._handle_sources, message["sources"])
+
+    def _handle_sources(self, ignored, sources):
+        """Handle sources repositories."""
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        new_sources = file(path, "w")
+        for line in file(self.SOURCES_LIST):
+            stripped_line = line.strip()
+            if not stripped_line or stripped_line.startswith("#"):
+                new_sources.write(line)
+            else:
+                new_sources.write("#%s" % line)
+        new_sources.close()
+        os.rename(path, self.SOURCES_LIST)
+
+        for filename in glob.glob(os.path.join(self.SOURCES_LIST_D, "*.list")):
+            os.rename(filename, "%s.save" % filename)
+
+        for source in sources:
+            filename = os.path.join(self.SOURCES_LIST_D,
+                                    "landscape-%s.list" % source["name"])
+            sources_file = file(filename, "w")
+            sources_file.write(source["content"])
+            sources_file.close()
+        return self._run_reporter()
+
+    def _run_reporter(self):
+        """Once the repositories are modified, trigger a reporter run."""
+        reporter = find_reporter_command()
+
+        # Force a smart-update run, because the sources.list has changed
+        args = ["--force-smart-update"]
+
+        if self.registry.config.config is not None:
+            args.append("--config=%s" % self.registry.config.config)
+        return self.run_process(reporter, args)

=== modified file 'landscape/manager/config.py'
--- landscape/manager/config.py	2010-08-06 17:57:42 +0000
+++ landscape/manager/config.py	2011-04-15 08:07:29 +0000
@@ -5,7 +5,7 @@
 
 
 ALL_PLUGINS = ["ProcessKiller", "PackageManager", "UserManager",
-               "ShutdownManager", "Eucalyptus"]
+               "ShutdownManager", "Eucalyptus", "AptSources"]
 
 
 class ManagerConfiguration(Configuration):

=== added file 'landscape/manager/tests/test_aptsources.py'
--- landscape/manager/tests/test_aptsources.py	1970-01-01 00:00:00 +0000
+++ landscape/manager/tests/test_aptsources.py	2011-04-15 08:07:29 +0000
@@ -0,0 +1,359 @@
+import os
+
+from twisted.internet.defer import Deferred
+
+from landscape.manager.aptsources import AptSources
+from landscape.manager.plugin import SUCCEEDED, FAILED
+
+from landscape.lib.twisted_util import gather_results
+from landscape.tests.helpers import LandscapeTest, ManagerHelper
+from landscape.package.reporter import find_reporter_command
+
+
+class AptSourcesTests(LandscapeTest):
+    helpers = [ManagerHelper]
+
+    def setUp(self):
+        super(AptSourcesTests, self).setUp()
+        self.sourceslist = AptSources()
+        self.sources_path = self.makeDir()
+        self.sourceslist.SOURCES_LIST = os.path.join(self.sources_path,
+                                                     "sources.list")
+        sources_d = os.path.join(self.sources_path, "sources.list.d")
+        os.mkdir(sources_d)
+        self.sourceslist.SOURCES_LIST_D = sources_d
+        self.manager.add(self.sourceslist)
+
+        sources = file(self.sourceslist.SOURCES_LIST, "w")
+        sources.write("\n")
+        sources.close()
+
+        service = self.broker_service
+        service.message_store.set_accepted_types(["operation-result"])
+
+        self.sourceslist.run_process = lambda cmd, args: None
+
+    def test_comment_sources_list(self):
+        """
+        When getting a repository message, L{AptSources} comments the whole
+        sources.list file.
+        """
+        sources = file(self.sourceslist.SOURCES_LIST, "w")
+        sources.write("oki\n\ndoki\n#comment\n # other comment\n")
+        sources.close()
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": [],
+             "operation-id": 1})
+
+        self.assertEqual(
+            "#oki\n\n#doki\n#comment\n # other comment\n",
+            file(self.sourceslist.SOURCES_LIST).read())
+
+        service = self.broker_service
+        self.assertMessages(service.message_store.get_pending_messages(),
+                            [{"type": "operation-result",
+                              "status": SUCCEEDED, "operation-id": 1}])
+
+    def test_random_failures(self):
+        """
+        If a failure happens during the manipulation of sources, the activity
+        is reported as FAILED with the error message.
+        """
+        self.sourceslist.SOURCES_LIST = "/doesntexist"
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": [],
+             "operation-id": 1})
+
+        msg = "IOError: [Errno 2] No such file or directory: '/doesntexist'"
+        service = self.broker_service
+        self.assertMessages(service.message_store.get_pending_messages(),
+                            [{"type": "operation-result",
+                              "result-text": msg, "status": FAILED,
+                              "operation-id": 1}])
+
+    def test_rename_sources_list_d(self):
+        """
+        The sources files in sources.list.d are renamed to .save when a message
+        is received.
+        """
+        sources1 = file(
+            os.path.join(self.sourceslist.SOURCES_LIST_D, "file1.list"), "w")
+        sources1.write("ok\n")
+        sources1.close()
+
+        sources2 = file(
+            os.path.join(self.sourceslist.SOURCES_LIST_D,
+                         "file2.list.save"), "w")
+        sources2.write("ok\n")
+        sources2.close()
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": [],
+             "operation-id": 1})
+
+        self.assertFalse(
+            os.path.exists(
+                os.path.join(self.sourceslist.SOURCES_LIST_D, "file1.list")))
+
+        self.assertTrue(
+            os.path.exists(
+                os.path.join(self.sourceslist.SOURCES_LIST_D,
+                             "file1.list.save")))
+
+        self.assertTrue(
+            os.path.exists(
+                os.path.join(self.sourceslist.SOURCES_LIST_D,
+                             "file2.list.save")))
+
+    def test_create_landscape_sources(self):
+        """
+        For every sources listed in the sources field of the message,
+        C{AptSources} creates a file with the content in sources.list.d.
+        """
+        sources = [{"name": "dev", "content": "oki\n"},
+                   {"name": "lucid", "content": "doki\n"}]
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": sources, "gpg-keys": [],
+             "operation-id": 1})
+
+        dev_file = os.path.join(self.sourceslist.SOURCES_LIST_D,
+                                "landscape-dev.list")
+        self.assertTrue(os.path.exists(dev_file))
+        self.assertEqual("oki\n", file(dev_file).read())
+
+        lucid_file = os.path.join(self.sourceslist.SOURCES_LIST_D,
+                                  "landscape-lucid.list")
+        self.assertTrue(os.path.exists(lucid_file))
+        self.assertEqual("doki\n", file(lucid_file).read())
+
+    def test_import_gpg_keys(self):
+        """
+        C{AptSources} runs a process with apt-key for every keys in the
+        message.
+        """
+        deferred = Deferred()
+
+        def run_process(command, args):
+            self.assertEqual("/usr/bin/apt-key", command)
+            self.assertEqual("add", args[0])
+            filename = args[1]
+            self.assertEqual("Some key content", file(filename).read())
+            deferred.callback(("ok", "", 0))
+            return deferred
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [],
+             "gpg-keys": ["Some key content"], "operation-id": 1})
+
+        return deferred
+
+    def test_import_delete_temporary_files(self):
+        """
+        The files created to be imported by C{apt-key} are removed after the
+        import.
+        """
+        deferred = Deferred()
+        filenames = []
+
+        def run_process(command, args):
+            if not filenames:
+                filenames.append(args[1])
+                deferred.callback(("ok", "", 0))
+                return deferred
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [],
+             "gpg-keys": ["Some key content"], "operation-id": 1})
+
+        self.assertFalse(os.path.exists(filenames[0]))
+
+        return deferred
+
+    def test_failed_import_delete_temporary_files(self):
+        """
+        The files created to be imported by C{apt-key} are removed after the
+        import, even if there is a failure.
+        """
+        deferred = Deferred()
+        filenames = []
+
+        def run_process(command, args):
+            filenames.append(args[1])
+            deferred.callback(("error", "", 1))
+            return deferred
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [],
+             "gpg-keys": ["Some key content"], "operation-id": 1})
+
+        self.assertFalse(os.path.exists(filenames[0]))
+
+        return deferred
+
+    def test_failed_import_reported(self):
+        """
+        If the C{apt-key} command failed for some reasons, the output of the
+        command is reported and the activity fails.
+        """
+        deferred = Deferred()
+
+        def run_process(command, args):
+            deferred.callback(("nok", "some error", 1))
+            return deferred
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": ["key"],
+             "operation-id": 1})
+
+        service = self.broker_service
+        msg = "ProcessError: nok\nsome error"
+        self.assertMessages(service.message_store.get_pending_messages(),
+                            [{"type": "operation-result",
+                              "result-text": msg, "status": FAILED,
+                              "operation-id": 1}])
+        return deferred
+
+    def test_signaled_import_reported(self):
+        """
+        If the C{apt-key} fails with a signal, the output of the command is
+        reported and the activity fails.
+        """
+        deferred = Deferred()
+
+        def run_process(command, args):
+            deferred.errback(("nok", "some error", 1))
+            return deferred
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": ["key"],
+             "operation-id": 1})
+
+        service = self.broker_service
+        msg = "ProcessError: nok\nsome error"
+        self.assertMessages(service.message_store.get_pending_messages(),
+                            [{"type": "operation-result",
+                              "result-text": msg, "status": FAILED,
+                              "operation-id": 1}])
+        return deferred
+
+    def test_failed_import_no_changes(self):
+        """
+        If the C{apt-key} command failed for some reasons, the current
+        repositories aren't changed.
+        """
+        deferred = Deferred()
+
+        def run_process(command, args):
+            deferred.callback(("nok", "some error", 1))
+            return deferred
+
+        self.sourceslist.run_process = run_process
+
+        sources = file(self.sourceslist.SOURCES_LIST, "w")
+        sources.write("oki\n\ndoki\n#comment\n")
+        sources.close()
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": ["key"],
+             "operation-id": 1})
+
+        self.assertEqual(
+            "oki\n\ndoki\n#comment\n",
+            file(self.sourceslist.SOURCES_LIST).read())
+
+        return deferred
+
+    def test_multiple_import_sequential(self):
+        """
+        If multiple keys are specified, the imports run sequentially, not in
+        parallel.
+        """
+        deferred1 = Deferred()
+        deferred2 = Deferred()
+        deferreds = [deferred1, deferred2]
+
+        def run_process(command, args):
+            if not deferreds:
+                return None
+            return deferreds.pop(0)
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [],
+             "gpg-keys": ["key1", "key2"], "operation-id": 1})
+
+        self.assertEqual(1, len(deferreds))
+        deferred1.callback(("ok", "", 0))
+
+        self.assertEqual(0, len(deferreds))
+        deferred2.callback(("ok", "", 0))
+
+        service = self.broker_service
+        self.assertMessages(service.message_store.get_pending_messages(),
+                            [{"type": "operation-result",
+                              "status": SUCCEEDED, "operation-id": 1}])
+        return gather_results(deferreds)
+
+    def test_multiple_import_failure(self):
+        """
+        If multiple keys are specified, and that the first one fails, the error
+        is correctly reported.
+        """
+        deferred1 = Deferred()
+        deferred2 = Deferred()
+        deferreds = [deferred1, deferred2]
+
+        def run_process(command, args):
+            return deferreds.pop(0)
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [],
+             "gpg-keys": ["key1", "key2"], "operation-id": 1})
+
+        deferred1.callback(("error", "", 1))
+        deferred2.callback(("error", "", 1))
+
+        msg = "ProcessError: error\n"
+        service = self.broker_service
+        self.assertMessages(service.message_store.get_pending_messages(),
+                            [{"type": "operation-result",
+                              "result-text": msg, "status": FAILED,
+                              "operation-id": 1}])
+        return gather_results(deferreds)
+
+    def test_run_reporter(self):
+        """
+        After receiving a message, L{AptSources} triggers a reporter run to
+        have the new packages reported to the server.
+        """
+        deferred = Deferred()
+
+        def run_process(command, args):
+            self.assertEqual(find_reporter_command(), command)
+            self.assertEqual(["--force-smart-update", "--config=%s" %
+                              self.manager.config.config], args)
+            deferred.callback(("ok", "", 0))
+            return deferred
+
+        self.sourceslist.run_process = run_process
+
+        self.manager.dispatch_message(
+            {"type": "apt-sources-replace", "sources": [], "gpg-keys": [],
+             "operation-id": 1})
+
+        return deferred

=== modified file 'landscape/manager/tests/test_config.py'
--- landscape/manager/tests/test_config.py	2010-08-06 17:57:42 +0000
+++ landscape/manager/tests/test_config.py	2011-04-15 08:07:29 +0000
@@ -12,7 +12,7 @@
     def test_plugin_factories(self):
         """By default all plugins are enabled."""
         self.assertEqual(["ProcessKiller", "PackageManager", "UserManager",
-                          "ShutdownManager", "Eucalyptus"],
+                          "ShutdownManager", "Eucalyptus", "AptSources"],
                          ALL_PLUGINS)
         self.assertEqual(ALL_PLUGINS, self.config.plugin_factories)