in src/buildstream/source.py [0:0]
def _set_ref(self, new_ref, *, save):
context = self._get_context()
project = self._get_project()
toplevel = context.get_toplevel_project()
toplevel_refs = self._project_refs(toplevel)
provenance = self._get_provenance()
element_name = self.__element_name
element_idx = self.__element_index
#
# Step 1 - Obtain the node
#
node = {}
if toplevel.ref_storage == ProjectRefStorage.PROJECT_REFS:
node = toplevel_refs.lookup_ref(project.name, element_name, element_idx, write=True)
if project is toplevel and not node:
node = provenance._node
#
# Step 2 - Set the ref in memory, and determine changed state
#
clean = node.strip_node_info()
# Set the ref regardless of whether it changed, the
# TrackQueue() will want to update a specific node with
# the ref, regardless of whether the original has changed.
#
# In the following add/del/mod merge algorithm we are working with
# dictionaries, but the plugin API calls for a MappingNode.
#
modify = node.clone()
self.set_ref(new_ref, modify)
to_modify = modify.strip_node_info()
# FIXME: this will save things too often, as a ref might not have
# changed. We should optimize this to detect it differently
if not save:
return False
# Ensure the node is not from a junction
if not toplevel.ref_storage == ProjectRefStorage.PROJECT_REFS and provenance._project is not toplevel:
if provenance._project is project:
self.warn("{}: Not persisting new reference in junctioned project".format(self))
elif provenance._project is None:
assert provenance._filename == ""
assert provenance._shortname == ""
raise SourceError("{}: Error saving source reference to synthetic node.".format(self))
else:
raise SourceError(
"{}: Cannot track source in a fragment from a junction".format(provenance._shortname),
reason="tracking-junction-fragment",
)
actions = {}
for k, v in clean.items():
if k not in to_modify:
actions[k] = "del"
else:
if v != to_modify[k]:
actions[k] = "mod"
for k in to_modify.keys():
if k not in clean:
actions[k] = "add"
def walk_container(container, path):
# For each step along path, synthesise if we need to.
# If we're synthesising missing list entries, we know we're
# doing this for project.refs so synthesise empty dicts for the
# intervening entries too
lpath = path.copy()
lpath.append("") # We know the last step will be a string key
for step, next_step in zip(lpath, lpath[1:]):
if type(step) is str: # pylint: disable=unidiomatic-typecheck
# handle dict container
if step not in container:
if type(next_step) is str: # pylint: disable=unidiomatic-typecheck
container[step] = {}
else:
container[step] = []
container = container[step]
else:
# handle list container
if len(container) <= step:
while len(container) <= step:
container.append({})
container = container[step]
return container
def process_value(action, container, path, key, new_value):
container = walk_container(container, path)
if action == "del":
del container[key]
elif action == "mod":
container[key] = new_value
elif action == "add":
container[key] = new_value
else:
assert False, "BUG: Unknown action: {}".format(action)
roundtrip_cache = {}
for key, action in actions.items():
# Obtain the top level node and its file
if action == "add":
provenance = node.get_provenance()
else:
provenance = node.get_node(key).get_provenance()
toplevel_node = provenance._toplevel
# Get the path to whatever changed
if action == "add":
path = toplevel_node._find(node)
else:
full_path = toplevel_node._find(node.get_node(key))
# We want the path to the node containing the key, not to the key
path = full_path[:-1]
roundtrip_file = roundtrip_cache.get(provenance._filename)
if not roundtrip_file:
roundtrip_file = roundtrip_cache[provenance._filename] = _yaml.roundtrip_load(
provenance._filename, allow_missing=True
)
# Get the value of the round trip file that we need to change
process_value(action, roundtrip_file, path, key, to_modify.get(key))
#
# Step 3 - Apply the change in project data
#
for filename, data in roundtrip_cache.items():
# This is our roundtrip dump from the track
try:
_yaml.roundtrip_dump(data, filename)
except OSError as e:
raise SourceError(
"{}: Error saving source reference to '{}': {}".format(self, filename, e), reason="save-ref-error"
) from e
return True