zuul/zuul-jobs
Code Issues Proposed changes
0e67960f64
zuul-jobs/roles/use-buildset-registry/module_utils/pytoml.py
James E. Blair ec8a58ddb7 use-buildset-registry: Vendor pytoml and remarshal 
In order to edit the V2 registries.conf file used by podman, we
need to be able to manipulate toml from ansible.  There is no
standard library or Ansible support for that now, and we don't want
to install any python packages on the remote node.  Therefore,
vendor the remarshal and pytoml code into this role.

This is done in a standalone commit for easier review and auditing.

The originating projects are:

  https://github.com/dbohdan/remarshal
  https://github.com/avakar/pytoml

And both are MIT licensed.  Appropriate headers are added where
necessary.

Note that pytoml has been concatenated into one file in order to
adhere to Ansible's requirements for python modules.

Change-Id: I679ea5eb5cb29591be09d2f1b712400c49158abd
2019-12-03 14:10:23 -08:00

552 lines
16 KiB
Python
Raw Blame History

# No-notice MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
# Originally from:
# https://github.com/avakar/pytoml
from __future__ import unicode_literals
import datetime
import re, sys
import io, datetime, math, string, sys
try:
from pathlib import PurePath as _path_types
except ImportError:
_path_types = ()
if sys.version_info[0] == 3:
long = int
unicode = str
class TomlError(RuntimeError):
def __init__(self, message, line, col, filename):
RuntimeError.__init__(self, message, line, col, filename)
self.message = message
self.line = line
self.col = col
self.filename = filename
def __str__(self):
return '{}({}, {}): {}'.format(self.filename, self.line, self.col, self.message)
def __repr__(self):
return 'TomlError({!r}, {!r}, {!r}, {!r})'.format(self.message, self.line, self.col, self.filename)
rfc3339_re = re.compile(r'(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(\.\d+)?(?:Z|([+-]\d{2}):(\d{2}))')
def parse_rfc3339(v):
m = rfc3339_re.match(v)
if not m or m.group(0) != v:
return None
return parse_rfc3339_re(m)
def parse_rfc3339_re(m):
r = map(int, m.groups()[:6])
if m.group(7):
micro = float(m.group(7))
else:
micro = 0
if m.group(8):
g = int(m.group(8), 10) * 60 + int(m.group(9), 10)
tz = _TimeZone(datetime.timedelta(0, g * 60))
else:
tz = _TimeZone(datetime.timedelta(0, 0))
y, m, d, H, M, S = r
return datetime.datetime(y, m, d, H, M, S, int(micro * 1000000), tz)
def format_rfc3339(v):
offs = v.utcoffset()
offs = int(offs.total_seconds()) // 60 if offs is not None else 0
if offs == 0:
suffix = 'Z'
else:
if offs > 0:
suffix = '+'
else:
suffix = '-'
offs = -offs
suffix = '{0}{1:02}:{2:02}'.format(suffix, offs // 60, offs % 60)
if v.microsecond:
return v.strftime('%Y-%m-%dT%H:%M:%S.%f') + suffix
else:
return v.strftime('%Y-%m-%dT%H:%M:%S') + suffix
class _TimeZone(datetime.tzinfo):
def __init__(self, offset):
self._offset = offset
def utcoffset(self, dt):
return self._offset
def dst(self, dt):
return None
def tzname(self, dt):
m = self._offset.total_seconds() // 60
if m < 0:
res = '-'
m = -m
else:
res = '+'
h = m // 60
m = m - h * 60
return '{}{:.02}{:.02}'.format(res, h, m)
if sys.version_info[0] == 2:
_chr = unichr
else:
_chr = chr
def load(fin, translate=lambda t, x, v: v, object_pairs_hook=dict):
return loads(fin.read(), translate=translate, object_pairs_hook=object_pairs_hook, filename=getattr(fin, 'name', repr(fin)))
def loads(s, filename='<string>', translate=lambda t, x, v: v, object_pairs_hook=dict):
if isinstance(s, bytes):
s = s.decode('utf-8')
s = s.replace('\r\n', '\n')
root = object_pairs_hook()
tables = object_pairs_hook()
scope = root
src = _Source(s, filename=filename)
ast = _p_toml(src, object_pairs_hook=object_pairs_hook)
def error(msg):
raise TomlError(msg, pos[0], pos[1], filename)
def process_value(v, object_pairs_hook):
kind, text, value, pos = v
if kind == 'array':
if value and any(k != value[0][0] for k, t, v, p in value[1:]):
error('array-type-mismatch')
value = [process_value(item, object_pairs_hook=object_pairs_hook) for item in value]
elif kind == 'table':
value = object_pairs_hook([(k, process_value(value[k], object_pairs_hook=object_pairs_hook)) for k in value])
return translate(kind, text, value)
for kind, value, pos in ast:
if kind == 'kv':
k, v = value
if k in scope:
error('duplicate_keys. Key "{0}" was used more than once.'.format(k))
scope[k] = process_value(v, object_pairs_hook=object_pairs_hook)
else:
is_table_array = (kind == 'table_array')
cur = tables
for name in value[:-1]:
if isinstance(cur.get(name), list):
d, cur = cur[name][-1]
else:
d, cur = cur.setdefault(name, (None, object_pairs_hook()))
scope = object_pairs_hook()
name = value[-1]
if name not in cur:
if is_table_array:
cur[name] = [(scope, object_pairs_hook())]
else:
cur[name] = (scope, object_pairs_hook())
elif isinstance(cur[name], list):
if not is_table_array:
error('table_type_mismatch')
cur[name].append((scope, object_pairs_hook()))
else:
if is_table_array:
error('table_type_mismatch')
old_scope, next_table = cur[name]
if old_scope is not None:
error('duplicate_tables')
cur[name] = (scope, next_table)
def merge_tables(scope, tables):
if scope is None:
scope = object_pairs_hook()
for k in tables:
if k in scope:
error('key_table_conflict')
v = tables[k]
if isinstance(v, list):
scope[k] = [merge_tables(sc, tbl) for sc, tbl in v]
else:
scope[k] = merge_tables(v[0], v[1])
return scope
return merge_tables(root, tables)
class _Source:
def __init__(self, s, filename=None):
self.s = s
self._pos = (1, 1)
self._last = None
self._filename = filename
self.backtrack_stack = []
def last(self):
return self._last
def pos(self):
return self._pos
def fail(self):
return self._expect(None)
def consume_dot(self):
if self.s:
self._last = self.s[0]
self.s = self[1:]
self._advance(self._last)
return self._last
return None
def expect_dot(self):
return self._expect(self.consume_dot())
def consume_eof(self):
if not self.s:
self._last = ''
return True
return False
def expect_eof(self):
return self._expect(self.consume_eof())
def consume(self, s):
if self.s.startswith(s):
self.s = self.s[len(s):]
self._last = s
self._advance(s)
return True
return False
def expect(self, s):
return self._expect(self.consume(s))
def consume_re(self, re):
m = re.match(self.s)
if m:
self.s = self.s[len(m.group(0)):]
self._last = m
self._advance(m.group(0))
return m
return None
def expect_re(self, re):
return self._expect(self.consume_re(re))
def __enter__(self):
self.backtrack_stack.append((self.s, self._pos))
def __exit__(self, type, value, traceback):
if type is None:
self.backtrack_stack.pop()
else:
self.s, self._pos = self.backtrack_stack.pop()
return type == TomlError
def commit(self):
self.backtrack_stack[-1] = (self.s, self._pos)
def _expect(self, r):
if not r:
raise TomlError('msg', self._pos[0], self._pos[1], self._filename)
return r
def _advance(self, s):
suffix_pos = s.rfind('\n')
if suffix_pos == -1:
self._pos = (self._pos[0], self._pos[1] + len(s))
else:
self._pos = (self._pos[0] + s.count('\n'), len(s) - suffix_pos)
_ews_re = re.compile(r'(?:[ \t]|#[^\n]*\n|#[^\n]*\Z|\n)*')
def _p_ews(s):
s.expect_re(_ews_re)
_ws_re = re.compile(r'[ \t]*')
def _p_ws(s):
s.expect_re(_ws_re)
_escapes = { 'b': '\b', 'n': '\n', 'r': '\r', 't': '\t', '"': '"',
'\\': '\\', 'f': '\f' }
_basicstr_re = re.compile(r'[^"\\\000-\037]*')
_short_uni_re = re.compile(r'u([0-9a-fA-F]{4})')
_long_uni_re = re.compile(r'U([0-9a-fA-F]{8})')
_escapes_re = re.compile(r'[btnfr\"\\]')
_newline_esc_re = re.compile('\n[ \t\n]*')
def _p_basicstr_content(s, content=_basicstr_re):
res = []
while True:
res.append(s.expect_re(content).group(0))
if not s.consume('\\'):
break
if s.consume_re(_newline_esc_re):
pass
elif s.consume_re(_short_uni_re) or s.consume_re(_long_uni_re):
v = int(s.last().group(1), 16)
if 0xd800 <= v < 0xe000:
s.fail()
res.append(_chr(v))
else:
s.expect_re(_escapes_re)
res.append(_escapes[s.last().group(0)])
return ''.join(res)
_key_re = re.compile(r'[0-9a-zA-Z-_]+')
def _p_key(s):
with s:
s.expect('"')
r = _p_basicstr_content(s, _basicstr_re)
s.expect('"')
return r
if s.consume('\''):
if s.consume('\'\''):
s.consume('\n')
r = s.expect_re(_litstr_ml_re).group(0)
s.expect('\'\'\'')
else:
r = s.expect_re(_litstr_re).group(0)
s.expect('\'')
return r
return s.expect_re(_key_re).group(0)
_float_re = re.compile(r'[+-]?(?:0|[1-9](?:_?\d)*)(?:\.\d(?:_?\d)*)?(?:[eE][+-]?(?:\d(?:_?\d)*))?')
_basicstr_ml_re = re.compile(r'(?:""?(?!")|[^"\\\000-\011\013-\037])*')
_litstr_re = re.compile(r"[^'\000\010\012-\037]*")
_litstr_ml_re = re.compile(r"(?:(?:|'|'')(?:[^'\000-\010\013-\037]))*")
def _p_value(s, object_pairs_hook):
pos = s.pos()
if s.consume('true'):
return 'bool', s.last(), True, pos
if s.consume('false'):
return 'bool', s.last(), False, pos
if s.consume('"'):
if s.consume('""'):
s.consume('\n')
r = _p_basicstr_content(s, _basicstr_ml_re)
s.expect('"""')
else:
r = _p_basicstr_content(s, _basicstr_re)
s.expect('"')
return 'str', r, r, pos
if s.consume('\''):
if s.consume('\'\''):
s.consume('\n')
r = s.expect_re(_litstr_ml_re).group(0)
s.expect('\'\'\'')
else:
r = s.expect_re(_litstr_re).group(0)
s.expect('\'')
return 'str', r, r, pos
if s.consume_re(rfc3339_re):
m = s.last()
return 'datetime', m.group(0), parse_rfc3339_re(m), pos
if s.consume_re(_float_re):
m = s.last().group(0)
r = m.replace('_','')
if '.' in m or 'e' in m or 'E' in m:
return 'float', m, float(r), pos
else:
return 'int', m, int(r, 10), pos
if s.consume('['):
items = []
with s:
while True:
_p_ews(s)
items.append(_p_value(s, object_pairs_hook=object_pairs_hook))
s.commit()
_p_ews(s)
s.expect(',')
s.commit()
_p_ews(s)
s.expect(']')
return 'array', None, items, pos
if s.consume('{'):
_p_ws(s)
items = object_pairs_hook()
if not s.consume('}'):
k = _p_key(s)
_p_ws(s)
s.expect('=')
_p_ws(s)
items[k] = _p_value(s, object_pairs_hook=object_pairs_hook)
_p_ws(s)
while s.consume(','):
_p_ws(s)
k = _p_key(s)
_p_ws(s)
s.expect('=')
_p_ws(s)
items[k] = _p_value(s, object_pairs_hook=object_pairs_hook)
_p_ws(s)
s.expect('}')
return 'table', None, items, pos
s.fail()
def _p_stmt(s, object_pairs_hook):
pos = s.pos()
if s.consume( '['):
is_array = s.consume('[')
_p_ws(s)
keys = [_p_key(s)]
_p_ws(s)
while s.consume('.'):
_p_ws(s)
keys.append(_p_key(s))
_p_ws(s)
s.expect(']')
if is_array:
s.expect(']')
return 'table_array' if is_array else 'table', keys, pos
key = _p_key(s)
_p_ws(s)
s.expect('=')
_p_ws(s)
value = _p_value(s, object_pairs_hook=object_pairs_hook)
return 'kv', (key, value), pos
_stmtsep_re = re.compile(r'(?:[ \t]*(?:#[^\n]*)?\n)+[ \t]*')
def _p_toml(s, object_pairs_hook):
stmts = []
_p_ews(s)
with s:
stmts.append(_p_stmt(s, object_pairs_hook=object_pairs_hook))
while True:
s.commit()
s.expect_re(_stmtsep_re)
stmts.append(_p_stmt(s, object_pairs_hook=object_pairs_hook))
_p_ews(s)
s.expect_eof()
return stmts
def dumps(obj, sort_keys=False):
fout = io.StringIO()
dump(obj, fout, sort_keys=sort_keys)
return fout.getvalue()
_escapes = {'\n': 'n', '\r': 'r', '\\': '\\', '\t': 't', '\b': 'b', '\f': 'f', '"': '"'}
def _escape_string(s):
res = []
start = 0
def flush():
if start != i:
res.append(s[start:i])
return i + 1
i = 0
while i < len(s):
c = s[i]
if c in '"\\\n\r\t\b\f':
start = flush()
res.append('\\' + _escapes[c])
elif ord(c) < 0x20:
start = flush()
res.append('\\u%04x' % ord(c))
i += 1
flush()
return '"' + ''.join(res) + '"'
_key_chars = string.digits + string.ascii_letters + '-_'
def _escape_id(s):
if any(c not in _key_chars for c in s):
return _escape_string(s)
return s
def _format_value(v):
if isinstance(v, bool):
return 'true' if v else 'false'
if isinstance(v, int) or isinstance(v, long):
return unicode(v)
if isinstance(v, float):
if math.isnan(v) or math.isinf(v):
raise ValueError("{0} is not a valid TOML value".format(v))
else:
return repr(v)
elif isinstance(v, unicode) or isinstance(v, bytes):
return _escape_string(v)
elif isinstance(v, datetime.datetime):
return format_rfc3339(v)
elif isinstance(v, list):
return '[{0}]'.format(', '.join(_format_value(obj) for obj in v))
elif isinstance(v, dict):
return '{{{0}}}'.format(', '.join('{} = {}'.format(_escape_id(k), _format_value(obj)) for k, obj in v.items()))
elif isinstance(v, _path_types):
return _escape_string(str(v))
else:
raise RuntimeError(v)
def dump(obj, fout, sort_keys=False):
tables = [((), obj, False)]
while tables:
name, table, is_array = tables.pop()
if name:
section_name = '.'.join(_escape_id(c) for c in name)
if is_array:
fout.write('[[{0}]]\n'.format(section_name))
else:
fout.write('[{0}]\n'.format(section_name))
table_keys = sorted(table.keys()) if sort_keys else table.keys()
new_tables = []
has_kv = False
for k in table_keys:
v = table[k]
if isinstance(v, dict):
new_tables.append((name + (k,), v, False))
elif isinstance(v, list) and v and all(isinstance(o, dict) for o in v):
new_tables.extend((name + (k,), d, True) for d in v)
elif v is None:
# based on mojombo's comment: https://github.com/toml-lang/toml/issues/146#issuecomment-25019344
fout.write(
'#{} = null # To use: uncomment and replace null with value\n'.format(_escape_id(k)))
has_kv = True
else:
fout.write('{0} = {1}\n'.format(_escape_id(k), _format_value(v)))
has_kv = True
tables.extend(reversed(new_tables))
if (name or has_kv) and tables:
fout.write('\n')
View Git Blame Copy Permalink