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_types_set():
data = TestMapping({"a": TestSequence([0])})
dpath.util.set(data, '/a/0', 1)
assert(data['a'][0] == 1)
data['a'][0] = 0
dpath.util.set(data, ['a', '0'], 1)
assert(data['a'][0] == 1)
def test_set_existing_list():
dict = {
"a": [
0
]
}
dpath.util.set(dict, '/a/0', 1)
assert(dict['a'][0] == 1)
dict['a'][0] = 0
dpath.util.set(dict, ['a', '0'], 1)
assert(dict['a'][0] == 1)
"d": 31
}
}
dpath.util.set(dict, '/a/*', 31337, afilter=afilter)
assert (dict['a']['b'] == 0)
assert (dict['a']['c'] == 1)
assert (dict['a']['d'] == 31337)
dict = {
"a": {
"b": 0,
"c": 1,
"d": 31
}
}
dpath.util.set(dict, ['a', '*'], 31337, afilter=afilter)
assert (dict['a']['b'] == 0)
assert (dict['a']['c'] == 1)
assert (dict['a']['d'] == 31337)
def test_set_filter():
def afilter(x):
if int(x) == 31:
return True
return False
dict = {
"a": {
"b": 0,
"c": 1,
"d": 31
}
}
dpath.util.set(dict, '/a/*', 31337, afilter=afilter)
assert (dict['a']['b'] == 0)
assert (dict['a']['c'] == 1)
assert (dict['a']['d'] == 31337)
dict = {
"a": {
"b": 0,
"c": 1,
"d": 31
}
}
dpath.util.set(dict, ['a', '*'], 31337, afilter=afilter)
assert (dict['a']['b'] == 0)
assert (dict['a']['c'] == 1)
assert (dict['a']['d'] == 31337)
print "CANNOT CONTINUE"
exit(1)
# get the config field, and turn it into valid JSON
config_str = entity["config"]
if log_debug:
print "LOG: Raw configStr = " + config_str
# configStr = configStr.replace("\\\"", "\"") --> don't need this anymore, depends on python behaviour
config = json.loads(config_str)
if log_debug:
print "LOG: config(t) = " + json.dumps(config, indent=4)
dpath.util.set(config, param_path, value, '.')
if log_debug:
print "LOG: config(t+1) = " + json.dumps(config, indent=4)
# put the escape characters back in the config str and write back to file
config_str = json.dumps(config)
# configStr = configStr.replace("\"", "\\\"") --> don't need this anymore, depends on python behaviour
if log_debug:
print "LOG: Modified configStr = " + config_str
entity["config"] = config_str
# write back to file
with open(entity_filepath, 'w') as data_file:
data_file.write(json.dumps(data, indent=4))
def set(self, glob, value):
return dpath.util.set(self, glob, value)
def getbody(self):
"""
Getter for the Request Body to make the call
Returns:
[dict] -- [JSON constructed body]
"""
body = self.config['body']
mappings = self.config['mappings']
for mapping in mappings:
pushvalue = mapping['push']
pullvalue = mapping['pull']
#debug(f'Attempting mapping of Body:{pushvalue} with CloudEvent:{pullvalue}')
#debug(f'Replacing >>> {dpath.util.get(body, pushvalue)} with ::: {dpath.util.get(self.event, pullvalue)}')
dpath.util.set(body, pushvalue, dpath.util.get(self.event, pullvalue))
return body
def set(self, glob, value):
return dpath.util.set(self, glob, value)
def set(self, path, data):
dpath.util.set(self.data, list(path), data)