Merge lp:~javier.collado/utah/preseed-parser into lp:utah

[Javier Collado]

This branch adds a new module to parse the preseed file, modify it and write it easily (the original idea was to get something similar to what the ConfigParser module from the standard library provides).

The features are:
- Quick access to preseeded questions using the question name
- Ability to prepend/append new sections (not only questions, but also comments or blank lines)
- Ability to prepend/append strings to a value in a section (useful to prepend to the late command)

The provisioning module has been updated and I tested the code with the default preseed and I got the same customized preseed as in the development branch.

Please let me know what do you think about these changes.

[Max Brustkern (nuclearbob)]

I like this a lot. Having this would have made a few things easier in cobbler desktop support, so I know it'll be helpful for future integrations. I'd also like UTAH to be able to run an automated install from a broken or empty preseed, and this will really help with that. Good stuff.

review: Approve

=== added file 'utah/preseed.py'
--- utah/preseed.py	1970-01-01 00:00:00 +0000
+++ utah/preseed.py	2012-09-06 12:53:21 +0000
@@ -0,0 +1,352 @@
+"""
+Pressed files handling
+"""
+import string
+
+
+class Preseed(object):
+    """
+    Read/Write preseed files easily
+    """
+    def __init__(self, filename):
+        # Used to access quickly to configuration sections by question name
+        self._qnames = {}
+
+        self.filename = filename
+        self.read(filename)
+
+    def __getitem__(self, key):
+        """
+        Access lines directly by their question name
+        """
+        if isinstance(key, TextPropertyValue):
+            key = key.text
+        if not isinstance(key, basestring):
+            raise TypeError
+
+        return self._qnames[key]
+
+    def __contains__(self, key):
+        """
+        Use in operator with question names
+        """
+        if isinstance(key, TextPropertyValue):
+            key = key.text
+        if not isinstance(key, basestring):
+            raise TypeError
+
+        return key in self._qnames
+
+    def read(self, filename=None):
+        """
+        Read a whole preseed file and parse its configuration lines
+        """
+        if filename is not None:
+            self.filename = filename
+        else:
+            filename = self.filename
+
+        self.sections = []
+        with open(filename) as f:
+            # One line might be made of multiple lines
+            # that are continued with '\\'
+            output_lines = []
+            for input_line in f:
+                input_line = input_line.rstrip('\n')
+                output_lines.append(input_line)
+
+                # Line is finished only when no continuation character is found
+                if not input_line.endswith('\\'):
+                    new_section = Section.new(self, output_lines)
+                    self.append(new_section)
+                    output_lines = []
+
+    def write(self, filename=None):
+        """
+        Write the modified file to the same or a new location
+        """
+        if filename is not None:
+            self.filename = filename
+        else:
+            filename = self.filename
+
+        with open(filename, 'w') as f:
+            for section in self.sections:
+                f.write('{}'.format(section))
+
+    def append(self, new_section, ref_section=None):
+        """
+        Append a new section
+        """
+        assert isinstance(new_section, Section)
+        if ref_section is None:
+            index = len(self.sections) - 1
+        else:
+            for index, section in enumerate(self.sections):
+                if section is ref_section:
+                    break
+            else:
+                raise ValueError('Reference section not found: {}'
+                                 .format(ref_section))
+
+        if (self.sections and
+            isinstance(new_section, (BlankSection, CommentSection)) and
+            type(new_section) == type(self.sections[index])):
+            self.sections[index] += new_section
+        else:
+            index += 1
+            self._insert(index, new_section)
+
+    def prepend(self, new_section, ref_section=None):
+        """
+        Prepend a new section
+        """
+        assert isinstance(new_section, Section)
+        if ref_section is None:
+            index = 0
+        else:
+            for index, section in enumerate(self.sections):
+                if section is ref_section:
+                    break
+            else:
+                raise ValueError('Reference section not found: {}'
+                                 .format(ref_section))
+
+        if (self.sections and
+            isinstance(new_section, (BlankSection, CommentSection)) and
+            type(new_section) == type(self.sections[index])):
+            self.sections[index] = new_section + self.sections[index]
+        else:
+            self._insert(index, new_section)
+
+    def _insert(self, index, new_section):
+        """
+        Insert section or join it with another one of the same type
+        """
+        self.sections.insert(index, new_section)
+
+    def section_updated(self, section, property_name, old_value, new_value):
+        """
+        Callback called any time a section property is updated
+
+        Used to maintain question names index integrity
+        """
+        if property_name == 'qname':
+            new_text = new_value.text
+            assert new_text not in self._qnames
+
+            if old_value is not None:
+                old_text = old_value.text
+                assert old_text in self._qnames
+                assert self._qnames[old_text] == section
+                del self._qnames[old_text]
+
+            self._qnames[new_text] = section
+
+
+class Section(object):
+    """
+    Any kind of line (blank, comment or configuration)
+    """
+    def __init__(self, parent):
+        self.parent = parent
+
+    def __repr__(self):
+        return '<{}: {!r}>'.format(self.__class__.__name__, str(self))
+
+    @classmethod
+    def new(self, parent, lines):
+        """
+        Create new section subclass based on the lines in the preseed
+        """
+        if isinstance(lines, basestring):
+            lines = [lines]
+        assert isinstance(lines, list)
+
+        if all(not line for line in lines):
+            return BlankSection(parent, len(lines))
+
+        if all(line.startswith('#') for line in lines):
+            return CommentSection(parent, lines)
+
+        assert all(line and not line.startswith('#')
+                   for line in lines)
+        return ConfigurationSection(parent, lines)
+
+
+class BlankSection(Section):
+    """
+    Any number of consecutive blank lines
+    """
+    def __init__(self, parent, lines_count):
+        super(BlankSection, self).__init__(parent)
+        assert isinstance(lines_count, int)
+        self.lines_count = lines_count
+
+    def __str__(self):
+        return '\n' * self.lines_count
+
+    def __add__(self, other):
+        assert isinstance(other, BlankSection)
+        return BlankSection(self.lines_count + other.lines_count)
+
+    def __iadd__(self, other):
+        assert isinstance(other, BlankSection)
+        self.lines_count += other.lines_count
+        return self
+
+
+class CommentSection(Section):
+    """
+    Any number of consecutive comment lines
+    """
+    def __init__(self, parent, lines):
+        super(CommentSection, self).__init__(parent)
+        assert isinstance(lines, list)
+        assert all(line.startswith('#') for line in lines)
+        self.lines = lines
+
+    def __str__(self):
+        return '{}\n'.format('\n'.join(self.lines))
+
+    def __add__(self, other):
+        assert isinstance(other, CommentSection)
+        return CommentSection(self.lines + other.lines)
+
+    def __iadd__(self, other):
+        assert isinstance(other, CommentSection)
+        self.lines.extend(other.lines)
+        return self
+
+
+class TextProperty(object):
+    def __init__(self, name):
+        self.name = name
+        self.obj_name = '_{}'.format(name)
+
+    def __get__(self, obj, type=None):
+        if not hasattr(obj, self.obj_name):
+            self.__set__(obj, '')
+        value = getattr(obj, self.obj_name)
+        assert isinstance(value, TextPropertyValue)
+        return value
+
+    def __set__(self, obj, new_text):
+        assert isinstance(new_text, basestring)
+
+        if hasattr(obj, self.obj_name):
+            old_value = self.__get__(obj)
+        else:
+            old_value = None
+        new_value = TextPropertyValue(self, obj, new_text)
+        setattr(obj, self.obj_name, new_value)
+        obj.property_updated(self.name, old_value, new_value)
+
+
+class TextPropertyValue(object):
+    def __init__(self, parent, obj, text=''):
+        self.parent = parent
+        self.obj = obj
+        self.text = text
+
+    def __str__(self):
+        return self.text
+
+    def __repr__(self):
+        return '<TextPropertyValue: {!r}>'.format(self.text)
+
+    def __nonzero__(self):
+        return bool(self.text)
+
+    def __eq__(self, other):
+        if isinstance(other, TextPropertyValue):
+            return self.text == other.text
+
+        if isinstance(other, basestring):
+            return self.text == other
+
+        raise ValueError
+
+    def prepend(self, other_text):
+        """
+        Prepend a string to the stored value
+        """
+        assert isinstance(other_text, basestring)
+        self.text = other_text + self.text
+        return self
+
+    def append(self, other_text):
+        """
+        Append a string to the stored value
+        """
+        assert isinstance(other_text, basestring)
+        self.text = self.text + other_text
+        return self
+
+
+class ConfigurationSection(Section):
+    """
+    A configuration statement made of one or multiple lines
+    """
+    TRAILING_CHARS = string.whitespace + '\\'
+
+    owner = TextProperty('owner')
+    qname = TextProperty('qname')
+    qtype = TextProperty('qtype')
+    value = TextProperty('value')
+
+    def __init__(self, parent, raw_lines):
+        """
+        Parse each element, so that it can be modified if needed
+        """
+        super(ConfigurationSection, self).__init__(parent)
+        lines = [raw_lines[0].rstrip(self.TRAILING_CHARS)]
+        for raw_line in raw_lines[1:]:
+            lines.append(raw_line
+                         .lstrip()
+                         .rstrip(self.TRAILING_CHARS))
+        text = ' '.join(lines)
+        splitted_text = text.split(None, 3)
+        self.owner = splitted_text[0]
+        self.qname = splitted_text[1]
+        self.qtype = splitted_text[2]
+        if len(splitted_text) == 4:
+            self.value = splitted_text[3]
+        else:
+            self.value = ''
+
+    def __str__(self):
+        """
+        Return text representation in a single line
+        """
+        if self.value:
+            line = ('{} {} {} {}\n'
+                    .format(self.owner, self.qname, self.qtype,
+                            self.value))
+        else:
+            line = ('{} {} {}\n'
+                    .format(self.owner, self.qname, self.qtype))
+        return line
+
+    def prepend(self, new_section):
+        """
+        Prepend a new section to this one
+        """
+        if isinstance(new_section, basestring):
+            new_section = Section.new(self.parent, new_section)
+
+        assert isinstance(new_section, Section)
+        self.parent.prepend(new_section, self)
+
+    def append(self, new_section):
+        """
+        Append a new section to this one
+        """
+        if isinstance(new_section, basestring):
+            new_section = Section.new(self.parent, new_section)
+
+        assert isinstance(new_section, Section)
+        self.parent.append(new_section, self)
+
+    def property_updated(self, property_name, old_value, new_value):
+        self.parent.section_updated(self, property_name, old_value, new_value)
 
=== modified file 'utah/provisioning/provisioning.py'
--- utah/provisioning/provisioning.py	2012-08-29 18:46:24 +0000
+++ utah/provisioning/provisioning.py	2012-09-06 12:53:21 +0000
@@ -19,6 +19,7 @@
 import apt.cache
 
 from utah.iso import ISO
+from utah.preseed import Preseed
 from utah.provisioning.exceptions import UTAHProvisioningException
 from utah.retry import retry
 import utah.timeout
@@ -772,57 +773,53 @@
         self.logger.info('Setting up preseed')
         if tmpdir is None:
             tmpdir = self.tmpdir
-        preseed = open(os.path.join(tmpdir, 'initrd.d', 'preseed.cfg'), 'w')
-        for line in open(self.preseed, 'r'):
-            if 'preseed/late_command string' in line:
-                if self.installtype == 'desktop':
-                    self.logger.info('Changing d-i latecommand '
-                            + 'to ubiquity success_command '
-                            + 'and prepending ubiquity lines')
-                    line = line.replace('d-i', 'ubiquity')
-                    line = line.replace('preseed/late_command string ',
-                            ('ubiquity/success_command '
-                             'string chroot /target sh -c \''  # chroot start
-                             'export LOG_FILE=/var/log/utah-install; '
-                             'apt-get install -y --force-yes openssh-server '
-                             '>>$LOG_FILE 2>&1 ; '
-                             'apt-get install -y --force-yes gdebi-core '
-                             '>>$LOG_FILE 2>&1 ; '
-                             '\'; '),                          # chroot end
-                            1)
-                    line = ("ubiquity ubiquity/summary note\n"
-                            "ubiquity ubiquity/reboot boolean true\n"
-                            + line)
-                else:
-                    self.logger.info('Prepending latecommand')
-                    line = line.replace('preseed/late_command string ',
-                            ('preseed/late_command string '
-                             'sh utah-latecommand ; '), 1)
-                line = line.rstrip(' ;')
-            if 'ubiquity/success_command string' in line:
-                self.logger.info('Prepending success_command')
-                line = line.replace('ubiquity/success_command string',
-                        ('ubiquity/success_command string '
-                         'sh utah-latecommand ; '), 1)
-                line = line.rstrip(' ;')
-            if 'pkgsel/include' in line:
-                for pkgname in ('openssh-server', 'gdebi-core'):
-                    if pkgname not in line:
-                        self.logger.info('Adding {} to preseeded packages'
-                                         .format(pkgname))
-                        line = line.replace('string ',
-                                            'string {} '.format(pkgname))
-                        line = line.rstrip() + "\n"
-            if 'netcfg/get_hostname' in line:
-                self.logger.info('Rewriting hostname to ' + self.name)
-                line = 'd-i netcfg/get_hostname string ' + self.name + "\n"
-            if 'passwd/username' in line:
-                self.logger.info('Rewriting username to ' + config.user)
-                line = 'd-i passwd/username string ' + config.user + "\n"
-            #Now that debug keeps the directory, this may not be needed
-            #self.logger.debug('Preseed line: ' + line)
-            preseed.write(line)
-            preseed.flush()
+        preseed = Preseed(self.preseed)
+        if 'preseed/late_command' in preseed:
+            if self.installtype == 'desktop':
+                self.logger.info('Changing d-i latecommand '
+                        + 'to ubiquity success_command '
+                        + 'and prepending ubiquity lines')
+                question = preseed['preseed/late_command']
+                question.owner = 'ubiquity'
+                question.qname = 'ubiquity/success_command'
+                question.value.prepend(
+                        'chroot /target sh -c \''        # chroot start
+                        'export LOG_FILE=/var/log/utah-install; '
+                        'apt-get install -y --force-yes openssh-server '
+                        '>>$LOG_FILE 2>&1 ; '
+                        'apt-get install -y --force-yes gdebi-core '
+                        '>>$LOG_FILE 2>&1 ; '
+                        '\'; '                           # chroot end
+                        )
+                question.prepend("ubiquity ubiquity/summary note")
+                question.prepend("ubiquity ubiquity/reboot boolean true")
+            else:
+                self.logger.info('Prepending latecommand')
+                question.value.prepend('sh utah-latecommand ; ')
+        if 'ubiquity/success_command' in preseed:
+            self.logger.info('Prepending success_command')
+            question = preseed['ubiquity/success_command']
+            question.value.prepend('sh utah-latecommand ; ')
+        if 'pkgsel/include' in preseed:
+            question = preseed['pkgsel/include']
+            packages = question.value.text.split()
+            for pkgname in ('openssh-server', 'gdebi-core'):
+                if pkgname not in packages:
+                    self.logger.info('Adding {} to preseeded packages'
+                                     .format(pkgname))
+                    packages.append(pkgname)
+            question.value = ' '.join(packages)
+
+        if 'netcfg/get_hostname' in preseed:
+            self.logger.info('Rewriting hostname to ' + self.name)
+            question = preseed['netcfg/get_hostname']
+            question.value = self.name
+        if 'passwd/username' in preseed:
+            self.logger.info('Rewriting username to ' + config.user)
+            question = preseed['passwd/username']
+            question.value = config.user
+
+        preseed.write(os.path.join(tmpdir, 'initrd.d', 'preseed.cfg'))
 
         if self.installtype == 'desktop':
             self.logger.info('Inserting preseed into casper')