Merge lp:~jason-hobbs/maas/lp-1210393 into lp:~maas-committers/maas/trunk

[Jason Hobbs]

Refactor user selection.

Don't assume a fixed number of users, or rely on any given user existing. Instead, ask the BMC what users exist, and then inspect users from that list. This eliminates the dependency on a 'User10' existing.

Pick the first user with username 'maas'. If no user with username 'maas' exists, pick the first user that has no username set. Otherwise, fail IPMI setup.

Refactor user selection.

Don't assume a fixed number of users, or rely on any given user existing. Instead, ask the BMC what users exist, and then inspect users from that list. This eliminates the dependency on a 'User10' existing.

Pick the first user with username 'maas'. If no user with username 'maas' exists, pick the first user has no username set. Otherwise, fail IPMI setup.

[Gavin Panella (allenap)]

Looks good, thanks again!

[1]

+    def test_runs_bmc_config(self):
+        """Ensure bmc-config is called with the correct args."""
+        recorder = self.patch(maas_ipmi_autodetect, 'run_command')
+
+        section = 'foo'
+        bmc_get_section(section)
+        args = recorder.call_args[0][0]
+
+        self.assertEqual('bmc-config', args[0])
+
+        for arg in ['--checkout', '--section', section]:
+            self.assertIn(arg, args)

The ordering is static so you can save some pain by using:

        recorder.assert_called_once_with(
            'bmc-config', '--checkout', '--section', section)

It'll also give a more useful error message if it fails.

I think there's also an assert_called_with() that matches any of
multiple calls, but if you need it you ought to check.

[2]

+        bgs_mock = Mock(return_value=user_text)
+        self.patch(maas_ipmi_autodetect, 'bmc_get_section', bgs_mock)
+        ps_mock = Mock(return_value=(user_number, user_record))
+        self.patch(maas_ipmi_autodetect, 'parse_section', ps_mock)

self.patch() does some of the above for you. You could also write it as:

        bgs_mock = self.patch(maas_ipmi_autodetect, 'bmc_get_section')
        bgs_mock.return_value = user_text
        ps_mock = self.patch(maas_ipmi_autodetect, 'parse_section')
        ps_mock.return_value = (user_number, user_record)

It only saves the import of Mock, but you might like it.

[3]

+        self.assertEqual(user_number, bgs_mock.call_args[0][0])
+        self.assertEqual(user_text, ps_mock.call_args[0][0])

Again, assert_called_once_with() is your friend.

If you need to match something with multiple arguments, but you can only
about one or two, there's ANY:

        from mock import ANY
        thing.assert_called_once_with("important", ANY, keyword=ANY)

There's also call(), which also works with ANY:

        from mock import call
        self.assertEqual(
            [call(1, ANY), call(2, ANY)],
            thing.call_args_list)

[4]

+def disabled_user(update=None):
+    """Record for a disabled user."""
+    base = {'Enable_User': 'No'}
+
+    if update:
+        base.update(update)
+
+    return base

Can you name this make_disabled_user()? All test factories have the
prefix "make", except where they don't, er, they have the prefix "get".

[5]

+def base_attributes(attributes_update=None):
+    """Base user records with updates in an OrderedDict."""
+
+    attributes_template = {
+        """Starting point for user records."""
+        'User1': {
+            'Enable_User': 'Yes'
+        },
+        'User2': {
+            'Username': 'admin',
+            'Enable_User': 'Yes'
+        }
+    }
+
+    base = OrderedDict(deepcopy(attributes_template))

I don't think the deepcopy() is necessary.

[6]

+    if attributes_update:
+        base.update(attributes_update)

Another habit we have is never relying on the implicit boolean-ness of
things in Python. Explicitly test for None instead:

    if attributes_update is not None:
        base.update(attributes_update)

[7]

+        ('one_disabled_not_blank_user', dict(

Fwiw, scenario names don't have to be valid identifiers; they're
free-form text I ...

review: Approve

[Jason Hobbs (jason-hobbs)]

Gavin,

On 02/19/2014 04:53 PM, Gavin Panella wrote:
> Review: Approve
>
> Looks good, thanks again!

Thanks again for the review all the pointers! I've made all the changes
you suggested and updated the branch.

Regards,
Jason

[Julian Edwards (julian-edwards)]

On Wednesday 19 Feb 2014 22:53:25 you wrote:
> Can you name this make_disabled_user()? All test factories have the
> prefix "make", except where they don't, er, they have the prefix "get".

This has been royally p!ssing me off lately... I might just pull the trigger
and change everything to make_<thing>

=== modified file 'etc/maas/templates/commissioning-user-data/snippets/maas_ipmi_autodetect.py'
--- etc/maas/templates/commissioning-user-data/snippets/maas_ipmi_autodetect.py	2014-02-18 19:26:26 +0000
+++ etc/maas/templates/commissioning-user-data/snippets/maas_ipmi_autodetect.py	2014-02-20 17:53:42 +0000
@@ -38,6 +38,10 @@
 import time
 
 
+class IPMIUserError(Exception):
+    """An error related to an IPMI user."""
+
+
 def run_command(command_args):
     """Run a command. Return output if successful or raise exception if not."""
     output = subprocess.check_output(command_args, stderr=subprocess.STDOUT)
@@ -62,18 +66,98 @@
     return '%s:%s' % (user_number, parameter)
 
 
-def bmc_user_get(user_number, parameter):
-    """Get a user parameter via bmc-config commit."""
-    key = format_user_key(user_number, parameter)
-    return bmc_get(key)
-
-
 def bmc_user_set(user_number, parameter, value):
     """Set a user parameter via bmc-config commit."""
     key = format_user_key(user_number, parameter)
     bmc_set(key, value)
 
 
+def parse_section(section):
+    """Parse the text of a section.
+
+    Returns a tuple of ('SectionName', section_attributes_dict)
+    """
+    # match lines with a word followed by space and then a non commment string
+    pattern = r'^\s*(\w+)\s+([^# \t\r\n\v\f]+).*$'
+    kv_tuples = re.findall(pattern, section, re.MULTILINE)
+    kv_dict = dict(kv_tuples)
+    section_name = kv_dict.pop('Section')
+    section_record = (section_name, kv_dict)
+
+    return section_record
+
+
+def bmc_get_section(section_name):
+    """Retrieve the text of a section from the BMC."""
+    command = ('bmc-config', '--checkout', '--section', section_name)
+    output = run_command(command)
+    return output
+
+
+def get_user_record(user_number):
+    """Return a dictionary of the user settings for a user number."""
+    section = bmc_get_section(user_number)
+    _, record = parse_section(section)
+    return record
+
+
+def bmc_list_sections():
+    """Retrieve the names of config sections from the BMC."""
+    command = ('bmc-config', '-L')
+    output = run_command(command)
+    return output
+
+
+def list_user_numbers():
+    """List the user numbers on the BMC."""
+    output = bmc_list_sections()
+    pattern = r'^(User\d+)$'
+    users = re.findall(pattern, output, re.MULTILINE)
+
+    return users
+
+
+def pick_user_number_from_list(search_username, user_numbers):
+    """Pick the best user number for a user from a list of user numbers.
+
+    If any any existing user's username matches the search username, pick
+    that user.
+
+    Otherwise, pick the first user that has no username set.
+
+    If no users match those criteria, raise an IPMIUserError.
+    """
+    first_unused = None
+
+    for user_number in user_numbers:
+        # The IPMI spec reserves User1 as anonymous.
+        if user_number == 'User1':
+            continue
+
+        user_record = get_user_record(user_number)
+
+        username = user_record.get('Username')
+
+        if username == search_username:
+            return user_number
+
+        if username is None and first_unused is None:
+                first_unused = user_number
+
+    return first_unused
+
+
+def pick_user_number(search_username):
+    """Pick the best user number for a username."""
+    user_numbers = list_user_numbers()
+    user_number = pick_user_number_from_list(search_username, user_numbers)
+
+    if not user_number:
+        raise IPMIUserError('No IPMI user slots available.')
+
+    return user_number
+
+
 def is_ipmi_dhcp():
     output = bmc_get('Lan_Conf:IP_Address_Source')
     show_re = re.compile('IP_Address_Source\s+Use_DHCP')
@@ -91,26 +175,8 @@
     return res.group()
 
 
-def get_ipmi_user_number(user):
-    for i in range(1, 17):
-        ipmi_user_number = "User%s" % i
-        # bmc-config fails if no slot for the requested user exists;
-        # instead of bailing, just keep trying remaining users.
-        try:
-            output = bmc_user_get(ipmi_user_number, 'Username')
-        except subprocess.CalledProcessError:
-            pass
-        else:
-            if user in output:
-                return ipmi_user_number
-    return None
-
-
 def commit_ipmi_user_settings(user, password):
-    ipmi_user_number = get_ipmi_user_number(user)
-    if ipmi_user_number is None:
-        bmc_user_set('User10', 'Username', user)
-        ipmi_user_number = get_ipmi_user_number(user)
+    ipmi_user_number = pick_user_number(user)
     bmc_user_set(ipmi_user_number, 'Username', user)
     bmc_user_set(ipmi_user_number, 'Password', password)
     bmc_user_set(ipmi_user_number, 'Enable_User', 'Yes')
@@ -183,7 +249,6 @@
         set_ipmi_network_source("Use_DHCP")
         # allow IPMI 120 seconds to obtain an IP address
         time.sleep(120)
-
     # create user/pass
     IPMI_MAAS_USER = "maas"
     IPMI_MAAS_PASSWORD = generate_random_password()
 
=== modified file 'etc/maas/templates/commissioning-user-data/snippets/tests/test_maas_ipmi_autodetect.py'
--- etc/maas/templates/commissioning-user-data/snippets/tests/test_maas_ipmi_autodetect.py	2014-02-18 19:26:26 +0000
+++ etc/maas/templates/commissioning-user-data/snippets/tests/test_maas_ipmi_autodetect.py	2014-02-20 17:53:42 +0000
@@ -16,14 +16,24 @@
 
 import subprocess
 
+from collections import OrderedDict
+
 from maastesting.testcase import MAASTestCase
 from maastesting.factory import factory
 
-import snippets.maas_ipmi_autodetect
+from snippets import maas_ipmi_autodetect
 
 from snippets.maas_ipmi_autodetect import (
+    bmc_get_section,
+    bmc_list_sections,
     format_user_key,
+    get_user_record,
+    list_user_numbers,
+    parse_section,
+    pick_user_number,
+    pick_user_number_from_list,
     run_command,
+    IPMIUserError
     )
 
 
@@ -63,8 +73,8 @@
         self.assertEqual(expected, user_key)
 
 
-class TestBMCMethods(MAASTestCase):
-    """Tests for the bmc_* methods."""
+class TestBMCKeyPairMethods(MAASTestCase):
+    """Tests for methods that use bmc-config --key-pair"""
 
     scenarios = [
         ('bmc_get', dict(
@@ -73,9 +83,6 @@
         ('bmc_set', dict(
             method_name='bmc_set', args=['Test:Key', 'myval'],
             key_pair_fmt='--key-pair=%s=%s', direction='--commit')),
-        ('bmc_user_get', dict(
-            method_name='bmc_user_get', args=['User10', 'Username'],
-            key_pair_fmt='--key-pair=%s:%s', direction='--checkout')),
         ('bmc_user_set', dict(
             method_name='bmc_user_set', args=['User10', 'Username', 'maas'],
             key_pair_fmt='--key-pair=%s:%s=%s', direction='--commit'))
@@ -84,24 +91,237 @@
     def test_runs_bmc_config(self):
         """Ensure bmc-config is run properly."""
 
-        recorder = self.patch(snippets.maas_ipmi_autodetect, 'run_command')
+        recorder = self.patch(maas_ipmi_autodetect, 'run_command')
 
         # Grab the method from the class module where it lives.
-        method = getattr(snippets.maas_ipmi_autodetect, self.method_name)
+        method = getattr(maas_ipmi_autodetect, self.method_name)
 
         method(*self.args)
 
-        # call_args[0] is a tuple of ordered args to run_command.
-        tokens = recorder.call_args[0][0]
-
-        # Ensure bmc-config is being called.
-        self.assertEqual(tokens[0], 'bmc-config')
-
-        # Ensure the correct --commit or --checkout is used.
-        self.assertIn(self.direction, tokens)
-
         # Note that the fmt string must use positional argument specifiers
         # if the order of appearance of args in the fmt string doesn't match
         # the order of args to the method.
         key_pair_string = self.key_pair_fmt % tuple(self.args)
-        self.assertIn(key_pair_string, tokens)
+
+        expected_args = ('bmc-config', self.direction, key_pair_string)
+        recorder.assert_called_once_with(expected_args)
+
+
+class TestBMCGetSection(MAASTestCase):
+    """Tests for bmc_get_section()."""
+
+    def test_runs_bmc_config(self):
+        """Ensure bmc-config is called with the correct args."""
+        recorder = self.patch(maas_ipmi_autodetect, 'run_command')
+
+        section = 'foo'
+        bmc_get_section(section)
+
+        recorder.assert_called_once_with(
+            ('bmc-config', '--checkout', '--section', section))
+
+
+class TestGetUserRecord(MAASTestCase):
+    """Tests for get_user_record()."""
+
+    def test_get_user_record(self):
+        """Ensure get_user_record() processes requests properly."""
+        user_number = 'User1'
+        user_text = 'some text'
+        user_record = {'bar': 'baz'}
+
+        bgs_mock = self.patch(maas_ipmi_autodetect, 'bmc_get_section')
+        bgs_mock.return_value = user_text
+        ps_mock = self.patch(maas_ipmi_autodetect, 'parse_section')
+        ps_mock.return_value = (user_number, user_record)
+
+        record = get_user_record(user_number)
+        self.assertEqual(record, user_record)
+        bgs_mock.assert_called_once_with(user_number)
+        ps_mock.assert_called_once_with(user_text)
+
+
+class TestBMCListSections(MAASTestCase):
+    """Tests for bmc_list_sections()."""
+
+    def test_bmc_list_sections(self):
+        """Ensure bmc-config is called with the correct args."""
+        recorder = self.patch(maas_ipmi_autodetect, 'run_command')
+        bmc_list_sections()
+        recorder.assert_called_once_with(('bmc-config', '-L'))
+
+
+class TestListUserNumbers(MAASTestCase):
+    """Tests for list_user_numbers()."""
+
+    section_names = (
+        "User1\n"
+        "User4\n"
+        "User3\n"
+        "User16\n"
+        "User7more\n"
+        "Userworld\n"
+        "Otherwise\n"
+        "4User5\n"
+        "4User\n"
+        "User\n"
+        "3"
+    )
+
+    def test_matching(self):
+        """Ensure only properly formatted User sections match."""
+        mock = self.patch(maas_ipmi_autodetect, 'bmc_list_sections')
+        mock.return_value = self.section_names
+        expected = ['User1', 'User4', 'User3', 'User16']
+        user_numbers = list_user_numbers()
+        self.assertEqual(expected, user_numbers)
+
+    def test_empty(self):
+        """Ensure an empty list is handled correctly."""
+        mock = self.patch(maas_ipmi_autodetect, 'bmc_list_sections')
+        mock.return_value = ''
+        results = list_user_numbers()
+        self.assertEqual([], results)
+
+
+class TestParseSection(MAASTestCase):
+    """Tests for parse_section()."""
+
+    section_template = (
+        "Section Test\n"                # Section line, no leading space.
+        "\tUsername\t\tBob\n"           # Normal line.
+        " Enabled_User\t\tNo\n"         # Leading space, not tab.
+        "\t\tPassword\t\tBobPass\n"     # Multiple leading tab.
+        "\tAnother  Value\n"            # Separating space, not tab.
+        "\tCharacters\t,./;:'\"[]{}|\\`~!@$%^&*()-_+="  # Gunk.
+        "\n"                            # Blank line.
+        "\tNotMe\n"                     # Single word.
+        "\t#Or Me\n"                    # Comment line.
+        "\tMe #Neither\n"               # Word followed by comment.
+        "\tThree\tWord\tLine\n"         # More than two words.
+        "\tFinal\tLine"                 # No trailing whitespace.
+    )
+
+    def test_matching(self):
+        """Ensure only properly formatted User sections match."""
+
+        record = parse_section(self.section_template)
+
+        expected_attributes = {
+            'Username': 'Bob',
+            'Enabled_User': 'No',
+            'Password': 'BobPass',
+            'Another': 'Value',
+            'Characters': ",./;:'\"[]{}|\\`~!@$%^&*()-_+=",
+            'Three': 'Word',
+            'Final': 'Line'
+        }
+
+        self.assertEqual('Test', record[0])
+        self.assertEqual(expected_attributes, record[1])
+
+    def test_fail_missing_section_line(self):
+        """Ensure an exception is raised if the section header is missing."""
+        no_section = self.section_template.replace('Section Test', '')
+        self.assertRaises(Exception, parse_section, no_section)
+
+
+def make_user(update=None):
+    """Make a simple user record."""
+
+    base = {'Lan_Enable_IPMI_Msgs': 'No'}
+
+    if update:
+        base.update(update)
+
+    return base
+
+
+def make_attributes(attributes_update=None):
+    """Base user records with updates in an OrderedDict."""
+
+    attributes_template = {
+        'User1': {
+            'Enable_User': 'Yes'
+        },
+        'User2': {
+            'Username': 'admin',
+            'Enable_User': 'Yes'
+        }
+    }
+
+    base = OrderedDict(attributes_template)
+
+    if attributes_update is not None:
+        base.update(attributes_update)
+
+    return base
+
+
+class TestPickUserNumberFromList(MAASTestCase):
+    """Tests for pick_user_number_from_list()."""
+
+    scenarios = [
+        ('Empty user list', dict(
+            user_attributes={},
+            expected=None)),
+        ('Existing MAAS user', dict(
+            user_attributes=make_attributes({
+                'User4': make_user(),
+                'User5': {'Username': 'maas'}}),
+            expected='User5')),
+        ('One blank user', dict(
+            user_attributes=make_attributes(
+                {'User7': make_user()}),
+            expected='User7')),
+        ('Multiple blank users', dict(
+            user_attributes=make_attributes({
+                'User7': make_user(),
+                'User8': make_user()}),
+            expected='User7')),
+        ('One not blank user', dict(
+            user_attributes=make_attributes(
+                {'User7': make_user({'Username': 'foo'})}),
+            expected=None)),
+        ('One enabled blank user', dict(
+            user_attributes=make_attributes({
+                'User7': {'Enable_User': 'Yes'}}),
+            expected='User7')),
+        ('Skip User1', dict(
+            user_attributes=make_attributes(
+                {'User1': make_user()}),
+            expected=None))
+    ]
+
+    def fetch_user_record(self, user_number):
+        """Return the mock user data for a user_number."""
+        return self.user_attributes[user_number]
+
+    def test_user_choice(self):
+        """Ensure the correct user, if any, is chosen."""
+
+        mock = self.patch(maas_ipmi_autodetect, 'get_user_record')
+        mock.side_effect = self.fetch_user_record
+        current_users = self.user_attributes.keys()
+        user = pick_user_number_from_list('maas', current_users)
+        self.assertEqual(self.expected, user)
+
+
+class TestPickUserNumber(MAASTestCase):
+    """Tests for pick_user_number()."""
+
+    def test_pick_user_number(self):
+        """Ensure proper listing and selection of a user."""
+        lun_mock = self.patch(maas_ipmi_autodetect, 'list_user_numbers')
+        lun_mock.return_value = ['User1', 'User2']
+        punfl_mock = self.patch(maas_ipmi_autodetect,
+                                'pick_user_number_from_list')
+        punfl_mock.return_value = 'User2'
+        user_number = pick_user_number('maas')
+        self.assertEqual('User2', user_number)
+
+    def test_fail_raise_exception(self):
+        """Ensure an exception is raised if no acceptable user is found."""
+        mock = self.patch(maas_ipmi_autodetect, 'list_user_numbers')
+        mock.return_value = []
+        self.assertRaises(IPMIUserError, pick_user_number, 'maas')