Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_dict_merge():
master = {'one': 1}
dataset.dict_merge(master, {'two': 2})
assert master == {'one': 1, 'two': 2}
dataset.dict_merge(master, {'two': 2})
assert master == {'one': 1, 'two': 2}
with pytest.raises(dataset.DatasetBuildError):
dataset.dict_merge(master, {'two': 3})
def test_enforce_unique_attributes():
assert dataset.enforce_unique_attributes({'key': [1]}, ['key'])
assert not dataset.enforce_unique_attributes({'key': ['undef']}, ['key'])
with pytest.raises(dataset.DatasetBuildError):
assert dataset.enforce_unique_attributes({'key': [1, 2]}, ['key'])
squeeze=squeeze,
read_keys=read_keys,
time_dims=time_dims,
)
except DatasetBuildError as ex:
# NOTE: When a variable has more than one value for an attribute we need to raise all
# the values in the file, not just the ones associated with that variable. See #54.
key = ex.args[1]
error_message = "multiple values for unique key, try re-open the file with one of:"
fbks = []
for value in index[key]:
fbk = {key: value}
fbk.update(filter_by_keys)
fbks.append(fbk)
error_message += "\n filter_by_keys=%r" % fbk
raise DatasetBuildError(error_message, key, fbks)
short_name = data_var.attributes.get('GRIB_shortName', 'paramId_%d' % param_id)
var_name = data_var.attributes.get('GRIB_cfVarName', 'unknown')
if 'parameter' in encode_cf and var_name not in ('undef', 'unknown'):
short_name = var_name
try:
dict_merge(variables, coord_vars)
dict_merge(variables, {short_name: data_var})
dict_merge(dimensions, dims)
except ValueError:
if errors == 'ignore':
pass
elif errors == 'raise':
raise
else:
log.exception("skipping variable: paramId==%r shortName=%r", param_id, short_name)
encoding = {
def dict_merge(master, update):
for key, value in update.items():
if key not in master:
master[key] = value
elif master[key] == value:
pass
else:
raise DatasetBuildError(
"key present and new value is different: "
"key=%r value=%r new_value=%r" % (key, master[key], value)
)
def enforce_unique_attributes(index, attributes_keys, filter_by_keys={}):
# type: (messages.FileIndex, T.Sequence[str], dict) -> T.Dict[str, T.Any]
attributes = collections.OrderedDict() # type: T.Dict[str, T.Any]
for key in attributes_keys:
values = index[key]
if len(values) > 1:
fbks = []
for value in values:
fbk = {key: value}
fbk.update(filter_by_keys)
fbks.append(fbk)
raise DatasetBuildError("multiple values for key %r" % key, key, fbks)
if values and values[0] not in ('undef', 'unknown'):
attributes['GRIB_' + key] = values[0]
return attributes