yardstick/network_services/helpers/iniparser.py

   1 # Copyright 2012 OpenStack Foundation
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #    Unless required by applicable law or agreed to in writing, software
  10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15
  16 class ParseError(Exception):
  17
  18     def __init__(self, message, line_no, line):
  19         self.msg = message
  20         self.line = line
  21         self.line_no = line_no
  22
  23     def __str__(self):
  24         return 'at line %d, %s: %r' % (self.line_no, self.msg, self.line)
  25
  26
  27 class SectionParseError(ParseError):
  28
  29     pass
  30
  31
  32 class LineParser(object):
  33
  34     PARSE_EXC = ParseError
  35
  36     @staticmethod
  37     def strip_key_value(key, value):
  38         key = key.strip()
  39         value = value.strip()
  40         if value and value[0] == value[-1] and value.startswith(('"', "'")):
  41             value = value[1:-1]
  42         return key, [value]
  43
  44     def __init__(self, line, line_no):
  45         super(LineParser, self).__init__()
  46         self.line = line
  47         self.line_no = line_no
  48         self.continuation = line != line.lstrip()
  49         semi_active, _, semi_comment = line.partition(';')
  50         pound_active, _, pound_comment = line.partition('#')
  51         if not semi_comment and not pound_comment:
  52             self.active = line.strip()
  53             self.comment = ''
  54         elif len(semi_comment) > len(pound_comment):
  55             self.active = semi_active.strip()
  56             self.comment = semi_comment.strip()
  57         else:
  58             self.active = pound_active.strip()
  59             self.comment = pound_comment.strip()
  60         self._section_name = None
  61
  62     def __repr__(self):
  63         template = "line %d: active '%s' comment '%s'\n%s"
  64         return template % (self.line_no, self.active, self.comment, self.line)
  65
  66     @property
  67     def section_name(self):
  68         if self._section_name is None:
  69             if not self.active.startswith('['):
  70                 raise self.error_no_section_start_bracket()
  71             if not self.active.endswith(']'):
  72                 raise self.error_no_section_end_bracket()
  73             self._section_name = ''
  74             if self.active:
  75                 self._section_name = self.active[1:-1]
  76             if not self._section_name:
  77                 raise self.error_no_section_name()
  78         return self._section_name
  79
  80     def is_active_line(self):
  81         return bool(self.active)
  82
  83     def is_continuation(self):
  84         return self.continuation
  85
  86     def split_key_value(self):
  87         for sep in ['=', ':']:
  88             words = self.active.split(sep, 1)
  89             try:
  90                 return self.strip_key_value(*words)
  91             except TypeError:
  92                 pass
  93
  94         return self.active.rstrip(), '@'
  95
  96     def error_invalid_assignment(self):
  97         return self.PARSE_EXC("No ':' or '=' found in assignment", self.line_no, self.line)
  98
  99     def error_empty_key(self):
 100         return self.PARSE_EXC('Key cannot be empty', self.line_no, self.line)
 101
 102     def error_unexpected_continuation(self):
 103         return self.PARSE_EXC('Unexpected continuation line', self.line_no, self.line)
 104
 105     def error_no_section_start_bracket(self):
 106         return SectionParseError('Invalid section (must start with [)', self.line_no, self.line)
 107
 108     def error_no_section_end_bracket(self):
 109         return self.PARSE_EXC('Invalid section (must end with ])', self.line_no, self.line)
 110
 111     def error_no_section_name(self):
 112         return self.PARSE_EXC('Empty section name', self.line_no, self.line)
 113
 114
 115 class BaseParser(object):
 116
 117     def parse(self, data=None):
 118         if data is not None:
 119             return self._parse(data.splitlines())
 120
 121     def _next_key_value(self, line_parser, key, value):
 122         self.comment(line_parser)
 123
 124         if not line_parser.is_active_line():
 125             # Blank line, ends multi-line values
 126             if key:
 127                 key, value = self.assignment(key, value, line_parser)
 128             return key, value
 129
 130         if line_parser.is_continuation():
 131             # Continuation of previous assignment
 132             if key is None:
 133                 raise line_parser.error_unexpected_continuation()
 134
 135             value.append(line_parser.active.lstrip())
 136             return key, value
 137
 138         if key:
 139             # Flush previous assignment, if any
 140             key, value = self.assignment(key, value, line_parser)
 141
 142         try:
 143             # Section start
 144             self.new_section(line_parser)
 145         except SectionParseError:
 146             pass
 147         else:
 148             return key, value
 149
 150         key, value = line_parser.split_key_value()
 151         if not key:
 152             raise line_parser.error_empty_key()
 153         return key, value
 154
 155     def _parse(self, line_iter):
 156         key = None
 157         value = []
 158
 159         parse_iter = (LineParser(line, line_no) for line_no, line in enumerate(line_iter))
 160         for line_parser in parse_iter:
 161             key, value = self._next_key_value(line_parser, key, value)
 162
 163         if key:
 164             # Flush previous assignment, if any
 165             self.assignment(key, value, LineParser('EOF', -1))
 166
 167     def _assignment(self, key, value, line_parser):
 168         """Called when a full assignment is parsed."""
 169         raise NotImplementedError()
 170
 171     def assignment(self, key, value, line_parser):
 172         self._assignment(key, value, line_parser)
 173         return None, []
 174
 175     def new_section(self, line_parser):
 176         """Called when a new section is started."""
 177         raise NotImplementedError()
 178
 179     def comment(self, line_parser):
 180         """Called when a comment is parsed."""
 181         raise NotImplementedError()
 182
 183
 184 class ConfigParser(BaseParser):
 185     """Parses a single config file, populating 'sections' to look like:
 186
 187         [
 188             [
 189                 'section1',
 190                 [
 191                     ['key1', 'value1\nvalue2'],
 192                     ['key2', 'value3\nvalue4'],
 193                 ],
 194             ],
 195             [
 196                 'section2',
 197                 [
 198                     ['key3', 'value5\nvalue6'],
 199                 ],
 200             ],
 201         ]
 202     """
 203
 204     def __init__(self, filename, sections=None):
 205         super(ConfigParser, self).__init__()
 206         self.filename = filename
 207         if sections is not None:
 208             self.sections = sections
 209         else:
 210             self.sections = []
 211         self.section_name = None
 212         self.section = None
 213
 214     def parse(self, data=None):
 215         if not data:
 216             data = self.filename
 217         with open(data) as f:
 218             return self._parse(f)
 219
 220     def __iter__(self):
 221         return iter(self.sections)
 222
 223     def find_section_index(self, section_name):
 224         return next((i for i, (name, value) in enumerate(self) if name == section_name), -1)
 225
 226     def find_section(self, section_name):
 227         return next((value for name, value in self.sections if name == section_name), None)
 228
 229     def new_section(self, line_parser):
 230         section_name = line_parser.section_name
 231         index = self.find_section_index(section_name)
 232         self.section_name = section_name
 233         if index == -1:
 234             self.section = [section_name, []]
 235             self.sections.append(self.section)
 236         else:
 237             self.section = self.sections[index]
 238
 239     def _assignment(self, key, value, line_parser):
 240         if not self.section_name:
 241             raise line_parser.error_no_section_name()
 242
 243         value = '\n'.join(value)
 244         entry = [key, value]
 245         self.section[1].append(entry)
 246
 247     def comment(self, line_parser):
 248         """Called when a comment is parsed."""
 249         pass