Code review comment for lp:~gmb/maas/fix-bug-1184589

Tip top, looks good. There are some things that still need doing I think, but it's the right approach. But first, a Panella service announcement:

I guess there are two ways of addressing this bug:

1. Finding each place that reports CalledProcessError (explicitly or
   by catching a superclass), and update them to treat
   CalledProcessError specially, e.g. by logging e.output too.

2. Finding each place we use check_call() or check_output() and
   switching them to use wrapped_check_{call,output}(), and 'fixing'
   the exception to include the log output when it's dumped later on.

3. Push a change upstream into Python to fix CalledProcessError to be
   like #2.

That's 3 things.

Only #1 and #2 are feasible really, and you've chosen #2 (after asking
all of us, I might add). It's sad when we can't use the standard
library unadorned, but the str/unicode crap in Python 2.x kind of
forces our hand. It's easier to find uses of check_{call,output} in
the codebase than it is to find those points where exceptions leak out
into log files and suchlike, so #2 is the better approach, even though
it's less "pretty" than #1 to my mind.

This long ramble mainly for my own benefit is now over.

[1]

wrapped_check_call and wrapped_check_output are ugly names... but I
can't think of anything better. They talk about mechanism and
implementation instead of what we expect to gain from their use.

Maybe:

  check_call --> call_and_check
  check_output --> call_capture_and_check

?

[2]

I also have problems with ExternalActionError's name. Sorry :-/ This
is a hard problem. I think it's the "Action" bit; it sounds too
specific, and not Unix enough. How about "ExternalProcessError"?

[3]

+from provisioningserver.utils import (
+    read_text_file,
+    )

Try:

  utilities/format-new-and-modified-import -r submit:

and it'll unwrap that. It'll also fix imports in a few other modules
too.

[4]

+class TestCallDnsSecKeygen(MAASTestCase):
+    """Tests for omshell.call_dnssec_keygen."""
+
+    def test_runs_external_script(self):
+        check_output = self.patch(subprocess, 'check_output')
+        target_dir = self.make_dir()
+        path = os.environ.get("PATH", "").split(os.pathsep)
+        path.append("/usr/sbin")
+        env = dict(os.environ, PATH=os.pathsep.join(path))
+        call_dnssec_keygen(target_dir)
+        check_output.assert_called_with([
+            'dnssec-keygen', '-r', '/dev/urandom', '-a', 'HMAC-MD5',
+             '-b', '512', '-n', 'HOST', '-K', target_dir, '-q', 'omapi_key'
+             ], env=env)

Woah, where did that come from? :)

If it's only called once, you could do:

        check_output.assert_called_once_with([
            ...

Also, if the env isn't /that/ important here, you could do:

        from mock import ANY
        check_output.assert_called_with([
            'dnssec-keygen', '-r', '/dev/urandom', '-a', 'HMAC-MD5',
             '-b', '512', '-n', 'HOST', '-K', target_dir, '-q', 'omapi_key'
             ], env=ANY)

[5]

+        recorder = self.patch(tasks, 'wrapped_check_call', Mock())

Not your code, but you can remove the Mock() bit here; it's done
automatically by patch():

        recorder = self.patch(tasks, 'wrapped_check_call')

[6]

+    try:
+        return subprocess.check_call(command, *args, **kwargs)
+    except subprocess.CalledProcessError as error:
+        raise ExternalActionError(error.returncode, error.cmd, error.output)

You can totally cheat here ;)

    try:
        return subprocess.check_call(command, *args, **kwargs)
    except subprocess.CalledProcessError as error:
        error.__class__ = ExternalActionError
        raise

[7]

+    def __str__(self):
+        return self.__unicode__()

This ain't good; __str__ should return bytes.

I have a different implementation in mind:

{{{
class ExternalActionError(CalledProcessError):
    """Raised when there's a problem calling an external command.

    Unlike CalledProcessError, ExternalActionError.__str__() contains
    the output of the failed external process, if available.
    """

    @staticmethod
    def _to_unicode(string):
        if isinstance(string, bytes):
            return string.decode("ascii", "replace")
        else:
            return unicode(string)

    @staticmethod
    def _to_ascii(string):
        if isinstance(string, unicode):
            return string.encode("ascii", "replace")
        else:
            string = bytes(string)
            try:
                string.decode("ascii")
            except UnicodeDecodeError:
                return "<non-ASCII string>"
            else:
                return string

    def __unicode__(self):
        cmd = u" ".join(quote(self._to_unicode(part)) for part in self.cmd)
        output = self._to_unicode(self.output)
        return u"Command `%s` returned non-zero exit status %d:\n%s" % (
            cmd, self.returncode, output)

    def __str__(self):
        cmd = b" ".join(quote(self._to_ascii(part)) for part in self.cmd)
        output = self._to_ascii(self.output)
        return b"Command `%s` returned non-zero exit status %d:\n%s" % (
            cmd, self.returncode, output)
}}}

Yep, it's a big one, and it'll need a little more testing, but I
suspect it's going to be more useful overall.