sts-sponsors team mailing list archive

[Jack Lloyd-Walters <mp+434995@xxxxxxxxxxxxxxxxxx>]

Review: Needs Fixing

Inline nit/questions, looks good elsewhere though.

Diff comments:

> diff --git a/src/provisioningserver/drivers/power/tests/test_cps.py b/src/provisioningserver/drivers/power/tests/test_cps.py
> new file mode 100644
> index 0000000..0663c80
> --- /dev/null
> +++ b/src/provisioningserver/drivers/power/tests/test_cps.py
> @@ -0,0 +1,154 @@
> +# Copyright 2015-2016 Canonical Ltd.  This software is licensed under the
> +# GNU Affero General Public License version 3 (see the file LICENSE).
> +
> +"""Tests for `provisioningserver.drivers.power.cps`."""
> +
> +import random
> +
> +from testtools.matchers import Equals

We're trying to migrate away from testtools, could you remove this?

> +
> +from maastesting.factory import factory
> +from maastesting.matchers import MockCalledOnceWith
> +from maastesting.testcase import MAASTestCase
> +from provisioningserver.drivers.power import cps as cps_module
> +from provisioningserver.drivers.power import PowerActionError
> +from provisioningserver.utils.shell import has_command_available, ProcessResult
> +
> +COMMON_ARGS_STATUS = "-c private -v1 {} 1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4.{}"
> +COMMON_ARGS_CONTROL = "-c private -v1 {} 1.3.6.1.4.1.3808.1.1.3.3.3.1.1.4.{}"
> +COMMON_OUTPUT = "iso.3.6.1.4.1.3808.1.1.3.3.5.1.1.4.%s = INTEGER: 1\n"
> +
> +
> +class TestCPSPowerDriver(MAASTestCase):
> +    def make_context(self, outlet=None):
> +        context = {
> +            "power_address": factory.make_name("power_address"),
> +            "node_outlet": (
> +                outlet if outlet is not None else random.randrange(1, 8)
> +            ),
> +        }
> +        return context
> +
> +    def test_missing_packages(self):
> +        mock = self.patch(has_command_available)
> +        mock.return_value = False
> +        driver = cps_module.CPSPowerDriver()
> +        missing = driver.detect_missing_packages()
> +        self.assertEqual(["snmp"], missing)
> +
> +    def test_no_missing_packages(self):
> +        mock = self.patch(has_command_available)
> +        mock.return_value = True
> +        driver = cps_module.CPSPowerDriver()
> +        missing = driver.detect_missing_packages()
> +        self.assertEqual([], missing)
> +
> +    def patch_run_command(self, stdout="", stderr="", returncode=0):
> +        mock_run_command = self.patch(cps_module.shell, "run_command")
> +        mock_run_command.return_value = ProcessResult(
> +            stdout=stdout, stderr=stderr, returncode=returncode
> +        )
> +        return mock_run_command
> +
> +    def test_run_process_calls_command_and_returns_output(self):
> +        driver = cps_module.CPSPowerDriver()
> +        context = self.make_context()
> +        command = ["snmpget"] + COMMON_ARGS_STATUS.format(
> +            context["power_address"], context["node_outlet"]
> +        ).split()
> +        mock_run_command = self.patch_run_command(
> +            stdout=COMMON_OUTPUT % context["node_outlet"],
> +            stderr="error_output",
> +        )
> +        output = driver.run_process(*command)
> +        mock_run_command.assert_called_once_with(*command)
> +        self.expectThat(

could you change this to something using unittests assertions (see the table below https://docs.python.org/3/library/unittest.html#unittest.TestCase.debug)

As this test is at the end of the function call, and we don't need to execute code further, we can safely convert this to a straight assertEqual, ie:
`self.assertEqual(output, cps_module.CPS_OUTLET_STATE_COMMANDS["immediateOn"])`

> +            output, Equals(cps_module.CPS_OUTLET_STATE_COMMANDS["immediateOn"])
> +        )
> +
> +    def test_run_process_crashes_on_external_process_error(self):
> +        driver = cps_module.CPSPowerDriver()
> +        self.patch_run_command(returncode=1)
> +        self.assertRaises(
> +            PowerActionError, driver.run_process, factory.make_name("command")
> +        )
> +
> +    def test_run_process_crashes_on_no_power_state_match_found(self):
> +        driver = cps_module.CPSPowerDriver()
> +        self.patch_run_command(stdout="Error")
> +        self.assertRaises(
> +            PowerActionError, driver.run_process, factory.make_name("command")
> +        )
> +
> +    def test_power_on_calls_run_process(self):
> +        driver = cps_module.CPSPowerDriver()
> +        system_id = factory.make_name("system_id")
> +        context = self.make_context()
> +        mock_power_query = self.patch(driver, "power_query")
> +        mock_power_query.return_value = "on"
> +        self.patch(driver, "power_off")
> +        mock_run_process = self.patch(driver, "run_process")
> +        driver.power_on(system_id, context)
> +
> +        self.expectThat(

Swap out this one too.

If you still want the function to execute after an error, as `expectThat()` would imply, I would recommend `mock_power_query.assert_called_once_with(system_id, context)`, rather than using the testtools matcher.

If not, that just `self.assertEqual()` would do of course.

> +            mock_power_query, MockCalledOnceWith(system_id, context)
> +        )
> +        command = (
> +            ["snmpset"]
> +            + COMMON_ARGS_CONTROL.format(
> +                context["power_address"], context["node_outlet"]
> +            ).split()
> +            + ["i", cps_module.CPS_OUTLET_STATE_COMMANDS["immediateOn"]]
> +        )
> +        mock_run_process.assert_called_once_with(*command)
> +
> +    def test_power_off_calls_run_process(self):
> +        driver = cps_module.CPSPowerDriver()
> +        system_id = factory.make_name("system_id")
> +        context = self.make_context()
> +        mock_run_process = self.patch(driver, "run_process")
> +        driver.power_off(system_id, context)
> +        command = (
> +            ["snmpset"]
> +            + COMMON_ARGS_CONTROL.format(
> +                context["power_address"], context["node_outlet"]
> +            ).split()
> +            + ["i", cps_module.CPS_OUTLET_STATE_COMMANDS["immediateOff"]]
> +        )
> +        mock_run_process.assert_called_once_with(*command)
> +
> +    def test_power_query_returns_power_state_on(self):
> +        driver = cps_module.CPSPowerDriver()
> +        system_id = factory.make_name("system_id")
> +        context = self.make_context()
> +        mock_run_process = self.patch(driver, "run_process")
> +        mock_run_process.return_value = "1"
> +        result = driver.power_query(system_id, context)
> +        command = ["snmpget"] + COMMON_ARGS_STATUS.format(
> +            context["power_address"], context["node_outlet"]
> +        ).split()
> +        mock_run_process.assert_called_once_with(*command)
> +        self.expectThat(result, Equals("on"))

Swap out here too.

> +
> +    def test_power_query_returns_power_state_off(self):
> +        driver = cps_module.CPSPowerDriver()
> +        system_id = factory.make_name("system_id")
> +        context = self.make_context()
> +        mock_run_process = self.patch(driver, "run_process")
> +        mock_run_process.return_value = "2"
> +        result = driver.power_query(system_id, context)
> +        command = ["snmpget"] + COMMON_ARGS_STATUS.format(
> +            context["power_address"], context["node_outlet"]
> +        ).split()
> +        mock_run_process.assert_called_once_with(*command)
> +        self.expectThat(result, Equals("off"))

Same here

> +
> +    def test_power_query_crashes_for_uknown_power_state(self):
> +        driver = cps_module.CPSPowerDriver()
> +        system_id = factory.make_name("system_id")
> +        context = self.make_context()
> +        mock_run_process = self.patch(driver, "run_process")
> +        mock_run_process.return_value = "Error"
> +        self.assertRaises(
> +            PowerActionError, driver.power_query, system_id, context
> +        )


-- 
https://code.launchpad.net/~syntroniks/maas/+git/maas/+merge/434995
Your team MAAS Committers is subscribed to branch maas:master.