Coverage for /usr/lib/python3/dist-packages/yaml/constructor.py: 54%

100

101

102

103

__all__ = ['BaseConstructor', 'SafeConstructor', 'Constructor',

'ConstructorError']

from .error import *

from .nodes import *

import collections.abc, datetime, base64, binascii, re, sys, types

class ConstructorError(MarkedYAMLError):

pass

class BaseConstructor:

yaml_constructors = {}

yaml_multi_constructors = {}

def __init__(self):

self.constructed_objects = {}

self.recursive_objects = {}

self.state_generators = []

self.deep_construct = False

def check_data(self):

# If there are more documents available?

return self.check_node()

def get_data(self):

# Construct and return the next document.

if self.check_node():

return self.construct_document(self.get_node())

def get_single_data(self):

# Ensure that the stream contains a single document and construct it.

node = self.get_single_node()

if node is not None:

return self.construct_document(node)

return None

def construct_document(self, node):

data = self.construct_object(node)

while self.state_generators:

state_generators = self.state_generators

self.state_generators = []

for generator in state_generators:

for dummy in generator:

pass

self.constructed_objects = {}

self.recursive_objects = {}

self.deep_construct = False

return data

def construct_object(self, node, deep=False):

if node in self.constructed_objects:

return self.constructed_objects[node]

if deep:

old_deep = self.deep_construct

self.deep_construct = True

if node in self.recursive_objects:

raise ConstructorError(None, None,

"found unconstructable recursive node", node.start_mark)

self.recursive_objects[node] = None

constructor = None

tag_suffix = None

if node.tag in self.yaml_constructors:

constructor = self.yaml_constructors[node.tag]

else:

for tag_prefix in self.yaml_multi_constructors:

if node.tag.startswith(tag_prefix):

tag_suffix = node.tag[len(tag_prefix):]

constructor = self.yaml_multi_constructors[tag_prefix]

break

else:

if None in self.yaml_multi_constructors:

tag_suffix = node.tag

constructor = self.yaml_multi_constructors[None]

elif None in self.yaml_constructors:

constructor = self.yaml_constructors[None]

elif isinstance(node, ScalarNode):

constructor = self.__class__.construct_scalar

elif isinstance(node, SequenceNode):

constructor = self.__class__.construct_sequence

elif isinstance(node, MappingNode):

constructor = self.__class__.construct_mapping

if tag_suffix is None:

data = constructor(self, node)

else:

data = constructor(self, tag_suffix, node)

if isinstance(data, types.GeneratorType):

generator = data

data = next(generator)

if self.deep_construct:

for dummy in generator:

pass

else:

self.state_generators.append(generator)

self.constructed_objects[node] = data

del self.recursive_objects[node]

if deep:

self.deep_construct = old_deep

return data

def construct_scalar(self, node):

if not isinstance(node, ScalarNode):

raise ConstructorError(None, None,

"expected a scalar node, but found %s" % node.id,

node.start_mark)

return node.value

def construct_sequence(self, node, deep=False):

if not isinstance(node, SequenceNode):

raise ConstructorError(None, None,

"expected a sequence node, but found %s" % node.id,

node.start_mark)

return [self.construct_object(child, deep=deep)

for child in node.value]

def construct_mapping(self, node, deep=False):

if not isinstance(node, MappingNode):

raise ConstructorError(None, None,

"expected a mapping node, but found %s" % node.id,

node.start_mark)

mapping = {}

for key_node, value_node in node.value:

key = self.construct_object(key_node, deep=deep)

if not isinstance(key, collections.abc.Hashable):

raise ConstructorError("while constructing a mapping", node.start_mark,

"found unhashable key", key_node.start_mark)

value = self.construct_object(value_node, deep=deep)

mapping[key] = value

return mapping

def construct_pairs(self, node, deep=False):

if not isinstance(node, MappingNode):

raise ConstructorError(None, None,

"expected a mapping node, but found %s" % node.id,

node.start_mark)

pairs = []

for key_node, value_node in node.value:

key = self.construct_object(key_node, deep=deep)

value = self.construct_object(value_node, deep=deep)

pairs.append((key, value))

return pairs

@classmethod

def add_constructor(cls, tag, constructor):

if not 'yaml_constructors' in cls.__dict__:

cls.yaml_constructors = cls.yaml_constructors.copy()

cls.yaml_constructors[tag] = constructor

@classmethod

def add_multi_constructor(cls, tag_prefix, multi_constructor):

if not 'yaml_multi_constructors' in cls.__dict__:

cls.yaml_multi_constructors = cls.yaml_multi_constructors.copy()

cls.yaml_multi_constructors[tag_prefix] = multi_constructor

class SafeConstructor(BaseConstructor):

def construct_scalar(self, node):

if isinstance(node, MappingNode):

for key_node, value_node in node.value:

if key_node.tag == 'tag:yaml.org,2002:value':

return self.construct_scalar(value_node)

return super().construct_scalar(node)

def flatten_mapping(self, node):

merge = []

index = 0

while index < len(node.value):

key_node, value_node = node.value[index]

if key_node.tag == 'tag:yaml.org,2002:merge':

del node.value[index]

if isinstance(value_node, MappingNode):

self.flatten_mapping(value_node)

merge.extend(value_node.value)

elif isinstance(value_node, SequenceNode):

submerge = []

for subnode in value_node.value:

if not isinstance(subnode, MappingNode):

raise ConstructorError("while constructing a mapping",

node.start_mark,

"expected a mapping for merging, but found %s"

% subnode.id, subnode.start_mark)

self.flatten_mapping(subnode)

submerge.append(subnode.value)

submerge.reverse()

for value in submerge:

merge.extend(value)

else:

raise ConstructorError("while constructing a mapping", node.start_mark,

"expected a mapping or list of mappings for merging, but found %s"

% value_node.id, value_node.start_mark)

elif key_node.tag == 'tag:yaml.org,2002:value':

key_node.tag = 'tag:yaml.org,2002:str'

index += 1

else:

index += 1

if merge:

node.value = merge + node.value

def construct_mapping(self, node, deep=False):

if isinstance(node, MappingNode):

self.flatten_mapping(node)

return super().construct_mapping(node, deep=deep)

def construct_yaml_null(self, node):

self.construct_scalar(node)

return None

bool_values = {

'yes': True,

'no': False,

'true': True,

'false': False,

'on': True,

'off': False,

}

def construct_yaml_bool(self, node):

value = self.construct_scalar(node)

return self.bool_values[value.lower()]

def construct_yaml_int(self, node):

value = self.construct_scalar(node)

value = value.replace('_', '')

sign = +1

if value[0] == '-':

sign = -1

if value[0] in '+-':

value = value[1:]

if value == '0':

return 0

elif value.startswith('0b'):

return sign*int(value[2:], 2)

elif value.startswith('0x'):

return sign*int(value[2:], 16)

elif value[0] == '0':

return sign*int(value, 8)

elif ':' in value:

digits = [int(part) for part in value.split(':')]

digits.reverse()

base = 1

value = 0

for digit in digits:

value += digit*base

base *= 60

return sign*value

else:

return sign*int(value)

inf_value = 1e300

while inf_value != inf_value*inf_value:

inf_value *= inf_value

nan_value = -inf_value/inf_value # Trying to make a quiet NaN (like C99).

def construct_yaml_float(self, node):

value = self.construct_scalar(node)

value = value.replace('_', '').lower()

sign = +1

if value[0] == '-':

sign = -1

if value[0] in '+-':

value = value[1:]

if value == '.inf':

return sign*self.inf_value

elif value == '.nan':

return self.nan_value

elif ':' in value:

digits = [float(part) for part in value.split(':')]

digits.reverse()

base = 1

value = 0.0

for digit in digits:

value += digit*base

base *= 60

return sign*value

else:

return sign*float(value)

def construct_yaml_binary(self, node):

try:

value = self.construct_scalar(node).encode('ascii')

except UnicodeEncodeError as exc:

raise ConstructorError(None, None,

"failed to convert base64 data into ascii: %s" % exc,

node.start_mark)

try:

if hasattr(base64, 'decodebytes'):

return base64.decodebytes(value)

else:

return base64.decodestring(value)

except binascii.Error as exc:

raise ConstructorError(None, None,

"failed to decode base64 data: %s" % exc, node.start_mark)

timestamp_regexp = re.compile(

r'''^(?P<year>[0-9][0-9][0-9][0-9])

-(?P<month>[0-9][0-9]?)

-(?P<day>[0-9][0-9]?)

(?:(?:[Tt]|[ \t]+)

(?P<hour>[0-9][0-9]?)

:(?P<minute>[0-9][0-9])

:(?P<second>[0-9][0-9])

(?:\.(?P<fraction>[0-9]*))?

(?:[ \t]*(?P<tz>Z|(?P<tz_sign>[-+])(?P<tz_hour>[0-9][0-9]?)

(?::(?P<tz_minute>[0-9][0-9]))?))?)?$''', re.X)

def construct_yaml_timestamp(self, node):

value = self.construct_scalar(node)

match = self.timestamp_regexp.match(node.value)

values = match.groupdict()

year = int(values['year'])

month = int(values['month'])

day = int(values['day'])

if not values['hour']:

return datetime.date(year, month, day)

hour = int(values['hour'])

minute = int(values['minute'])

second = int(values['second'])

fraction = 0

if values['fraction']:

fraction = values['fraction'][:6]

while len(fraction) < 6:

fraction += '0'

fraction = int(fraction)

delta = None

if values['tz_sign']:

tz_hour = int(values['tz_hour'])

tz_minute = int(values['tz_minute'] or 0)

delta = datetime.timedelta(hours=tz_hour, minutes=tz_minute)

if values['tz_sign'] == '-':

delta = -delta

data = datetime.datetime(year, month, day, hour, minute, second, fraction)

if delta:

data -= delta

return data

def construct_yaml_omap(self, node):

# Note: we do not check for duplicate keys, because it's too

# CPU-expensive.

omap = []

yield omap

if not isinstance(node, SequenceNode):

raise ConstructorError("while constructing an ordered map", node.start_mark,

"expected a sequence, but found %s" % node.id, node.start_mark)

for subnode in node.value:

if not isinstance(subnode, MappingNode):

raise ConstructorError("while constructing an ordered map", node.start_mark,

"expected a mapping of length 1, but found %s" % subnode.id,

subnode.start_mark)

if len(subnode.value) != 1:

raise ConstructorError("while constructing an ordered map", node.start_mark,

"expected a single mapping item, but found %d items" % len(subnode.value),

subnode.start_mark)

key_node, value_node = subnode.value[0]

key = self.construct_object(key_node)

value = self.construct_object(value_node)

omap.append((key, value))

def construct_yaml_pairs(self, node):

# Note: the same code as `construct_yaml_omap`.

pairs = []

yield pairs

if not isinstance(node, SequenceNode):

raise ConstructorError("while constructing pairs", node.start_mark,

"expected a sequence, but found %s" % node.id, node.start_mark)

for subnode in node.value:

if not isinstance(subnode, MappingNode):

raise ConstructorError("while constructing pairs", node.start_mark,

"expected a mapping of length 1, but found %s" % subnode.id,

subnode.start_mark)

if len(subnode.value) != 1:

raise ConstructorError("while constructing pairs", node.start_mark,

"expected a single mapping item, but found %d items" % len(subnode.value),

subnode.start_mark)

key_node, value_node = subnode.value[0]

key = self.construct_object(key_node)

value = self.construct_object(value_node)

pairs.append((key, value))

def construct_yaml_set(self, node):

data = set()

yield data

value = self.construct_mapping(node)

data.update(value)

def construct_yaml_str(self, node):

return self.construct_scalar(node)

def construct_yaml_seq(self, node):

data = []

yield data

data.extend(self.construct_sequence(node))

def construct_yaml_map(self, node):

data = {}

yield data

value = self.construct_mapping(node)

data.update(value)

def construct_yaml_object(self, node, cls):

data = cls.__new__(cls)

yield data

if hasattr(data, '__setstate__'):

state = self.construct_mapping(node, deep=True)

data.__setstate__(state)

else:

state = self.construct_mapping(node)

data.__dict__.update(state)

def construct_undefined(self, node):

raise ConstructorError(None, None,

"could not determine a constructor for the tag %r" % node.tag,

node.start_mark)