launchpad-reviewers team mailing list archive

[Julian Edwards <julian.edwards@xxxxxxxxxxxxx>]

Julian Edwards has proposed merging lp:~julian-edwards/maas/omshell-wrapper into lp:maas.

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~julian-edwards/maas/omshell-wrapper/+merge/116414

Add a convenient wrapper around "omshell" commands to create and remove host mappings in DHCPD.

There might be some opportunity for test refactoring here and I was considering getting rid of FakeMethod and forcing a call to "echo" instead of "omshell" but I'll see what you think.  Also, I think having a dhcpd fixture here is a little crazy so I didn't bother.
-- 
https://code.launchpad.net/~julian-edwards/maas/omshell-wrapper/+merge/116414
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~julian-edwards/maas/omshell-wrapper into lp:maas.

=== added file 'src/provisioningserver/omshell.py'
--- src/provisioningserver/omshell.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/omshell.py	2012-07-24 07:44:22 +0000
@@ -0,0 +1,84 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Python wrapper around the `omshell` utility which amends objects
+inside the DHCP server.
+"""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+__metaclass__ = type
+__all__ = [
+    "Omshell",
+    ]
+
+from subprocess import (
+    CalledProcessError,
+    PIPE,
+    Popen,
+    )
+
+
+class Omshell:
+    """Wrap up the omshell utility in Python."""
+
+    def __init__(self, server_address, shared_key):
+        self.server_address = server_address
+        self.shared_key = shared_key
+        self.proc = Popen("omshell", stdin=PIPE, stdout=PIPE)
+
+    def _run(self, stdin):
+        stdout, stderr = self.proc.communicate(stdin)
+        return stdout
+
+    def create(self, ip_address, mac_address):
+        stdin = (
+            "server %(server)s\n"
+            "key omapi_key %(key)s\n"
+            "connect\n"
+            "new host\n"
+            "set ip-address = %(ip)s\n"
+            "set hardware-address = %(mac)s\n"
+            "set name = %(ip)s\n"
+            "create\n")
+        stdin = stdin % dict(
+            server=self.server_address,
+            key=self.shared_key,
+            ip=ip_address,
+            mac=mac_address)
+
+        output = self._run(stdin)
+        # If the call to omshell doesn't result in output containing the
+        # magic string 'hardware-type' then we can be reasonably sure
+        # that the 'create' command failed.  Unfortunately there's no
+        # other output like "successful" to check so this is the best we
+        # can do.
+        if "hardware-type" not in output:
+            raise CalledProcessError(self.proc.returncode, "omshell", output)
+
+    def remove(self, ip_address):
+        stdin = (
+            "server %(server)s\n"
+            "key omapi_key %(key)s\n"
+            "connect\n"
+            "new host\n"
+            "set name = %(ip)s\n"
+            "open\n"
+            "remove\n")
+        stdin = stdin % dict(
+            server=self.server_address,
+            key=self.shared_key,
+            ip=ip_address)
+
+        output = self._run(stdin)
+
+        # If the omshell worked, the last line should reference a null
+        # object.
+        lines = output.splitlines()
+        last_line = lines[-1]
+        if last_line != "obj: <null>":
+            raise CalledProcessError(self.proc.returncode, "omshell", output)

=== added file 'src/provisioningserver/tests/test_omshell.py'
--- src/provisioningserver/tests/test_omshell.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/tests/test_omshell.py	2012-07-24 07:44:22 +0000
@@ -0,0 +1,136 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the omshell.py file."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+__metaclass__ = type
+__all__ = []
+
+from subprocess import CalledProcessError
+
+from maastesting.factory import factory
+from maastesting.fakemethod import FakeMethod
+from maastesting.testcase import TestCase
+from provisioningserver.omshell import Omshell
+from testtools.matchers import MatchesStructure
+
+
+class TestOmshell(TestCase):
+
+    def test_initialisation(self):
+        server_address = factory.getRandomString()
+        shared_key = factory.getRandomString()
+        shell = Omshell(server_address, shared_key)
+        self.assertThat(
+            shell, MatchesStructure.byEquality(
+                server_address=server_address,
+                shared_key=shared_key))
+
+    def test_create_calls_omshell_correctly(self):
+        server_address = factory.getRandomString()
+        shared_key = factory.getRandomString()
+        ip_address = factory.getRandomIPAddress()
+        mac_address = factory.getRandomMACAddress()
+        shell = Omshell(server_address, shared_key)
+
+        # Instead of calling a real omshell, we'll just record the
+        # parameters passed to 'check_call'.
+        recorder = FakeMethod(result=("hardware-type", None))
+        self.patch(shell.proc, 'communicate', recorder)
+
+        shell.create(ip_address, mac_address)
+
+        expected_args = (
+            "server %(server)s\n"
+            "key omapi_key %(key)s\n"
+            "connect\n"
+            "new host\n"
+            "set ip-address = %(ip)s\n"
+            "set hardware-address = %(mac)s\n"
+            "set name = %(ip)s\n"
+            "create\n" % dict(
+                server=server_address,
+                key=shared_key,
+                ip=ip_address,
+                mac=mac_address),)
+
+        # Check that the 'stdin' arg contains the correct set of
+        # commands.
+        self.assertEqual(
+            [1, expected_args],
+            [recorder.call_count, recorder.extract_args()[0]])
+
+    def test_create_raises_when_omshell_fails(self):
+        # If the call to omshell doesn't result in output containing the
+        # magic string 'hardware-type' it means the set of commands
+        # failed.
+
+        server_address = factory.getRandomString()
+        shared_key = factory.getRandomString()
+        ip_address = factory.getRandomIPAddress()
+        mac_address = factory.getRandomMACAddress()
+        shell = Omshell(server_address, shared_key)
+
+        # Fake a call that results in a failure with random output.
+        random_output = factory.getRandomString()
+        recorder = FakeMethod(result=(random_output, None))
+        self.patch(shell.proc, 'communicate', recorder)
+
+        exc = self.assertRaises(
+            CalledProcessError, shell.create, ip_address, mac_address)
+        self.assertEqual(random_output, exc.output)
+
+    def test_remove_calls_omshell_correctly(self):
+        server_address = factory.getRandomString()
+        shared_key = factory.getRandomString()
+        ip_address = factory.getRandomIPAddress()
+        shell = Omshell(server_address, shared_key)
+
+        # Instead of calling a real omshell, we'll just record the
+        # parameters passed to 'check_call'.
+        recorder = FakeMethod(result=("thing1\nthing2\nobj: <null>", None))
+        self.patch(shell.proc, 'communicate', recorder)
+
+        shell.remove(ip_address)
+
+        expected_args = (
+            "server %(server)s\n"
+            "key omapi_key %(key)s\n"
+            "connect\n"
+            "new host\n"
+            "set name = %(ip)s\n"
+            "open\n"
+            "remove\n" % dict(
+                server=server_address,
+                key=shared_key,
+                ip=ip_address),)
+
+        # Check that the 'stdin' arg contains the correct set of
+        # commands.
+        self.assertEqual(
+            [1, expected_args],
+            [recorder.call_count, recorder.extract_args()[0]])
+
+    def test_remove_raises_when_omshell_fails(self):
+        # If the call to omshell doesn't result in output ending in the
+        # text 'obj: <null>' we can be fairly sure this operation
+        # failed.
+        server_address = factory.getRandomString()
+        shared_key = factory.getRandomString()
+        ip_address = factory.getRandomIPAddress()
+        shell = Omshell(server_address, shared_key)
+
+        # Fake a call that results in a failure with random output.
+        random_output = factory.getRandomString()
+        recorder = FakeMethod(result=(random_output, None))
+        self.patch(shell.proc, 'communicate', recorder)
+
+        exc = self.assertRaises(
+            CalledProcessError, shell.remove, ip_address)
+        self.assertEqual(random_output, exc.output)