import os, copy
__desc__ = __doc__
+BLOCKS_TAG="BLOCKS:"
+BLOCKED_BY_TAG="BLOCKED-BY:"
+
+class BrokenLink (Exception):
+ def __init__(self, blocked_bug, blocking_bug, blocks=True):
+ if blocks == True:
+ msg = "Missing link: %s blocks %s" \
+ % (blocking_bug.uuid, blocked_bug.uuid)
+ else:
+ msg = "Missing link: %s blocked by %s" \
+ % (blocked_bug.uuid, blocking_bug.uuid)
+ Exception.__init__(self, msg)
+ self.blocked_bug = blocked_bug
+ self.blocking_bug = blocking_bug
+
+
def execute(args, manipulate_encodings=True):
"""
>>> from libbe import utility
>>> bd.save()
>>> os.chdir(bd.root)
>>> execute(["a", "b"], manipulate_encodings=False)
- Blocks on a:
+ a blocked by:
b
>>> execute(["a"], manipulate_encodings=False)
- Blocks on a:
+ a blocked by:
b
>>> execute(["--show-status", "a"], manipulate_encodings=False) # doctest: +NORMALIZE_WHITESPACE
- Blocks on a:
+ a blocked by:
+ b closed
+ >>> execute(["b", "a"], manipulate_encodings=False)
+ b blocked by:
+ a
+ b blocks:
+ a
+ >>> execute(["--show-status", "a"], manipulate_encodings=False) # doctest: +NORMALIZE_WHITESPACE
+ a blocked by:
+ b closed
+ a blocks:
b closed
+ >>> execute(["-r", "b", "a"], manipulate_encodings=False)
+ b blocks:
+ a
>>> execute(["-r", "a", "b"], manipulate_encodings=False)
>>> bd.cleanup()
"""
bugA = cmdutil.bug_from_shortname(bd, args[0])
if len(args) == 2:
bugB = cmdutil.bug_from_shortname(bd, args[1])
- estrs = bugA.extra_strings
- depend_string = "BLOCKED-BY:%s" % bugB.uuid
if options.remove == True:
- estrs.remove(depend_string)
+ remove_block(bugA, bugB)
else: # add the dependency
- estrs.append(depend_string)
- bugA.extra_strings = estrs # reassign to notice change
-
- depends = []
- for estr in bugA.extra_strings:
- if estr.startswith("BLOCKED-BY:"):
- uuid = estr[11:]
- if options.show_status == True:
- blocker = bd.bug_from_uuid(uuid)
- block_string = "%s\t%s" % (uuid, blocker.status)
- else:
- block_string = uuid
- depends.append(block_string)
- if len(depends) > 0:
- print "Blocks on %s:" % bugA.uuid
- print '\n'.join(depends)
+ add_block(bugA, bugB)
+
+ blocked_by = get_blocked_by(bd, bugA)
+ if len(blocked_by) > 0:
+ print "%s blocked by:" % bugA.uuid
+ if options.show_status == True:
+ print '\n'.join(["%s\t%s" % (bug.uuid, bug.status)
+ for bug in blocked_by])
+ else:
+ print '\n'.join([bug.uuid for bug in blocked_by])
+ blocks = get_blocks(bd, bugA)
+ if len(blocks) > 0:
+ print "%s blocks:" % bugA.uuid
+ if options.show_status == True:
+ print '\n'.join(["%s\t%s" % (bug.uuid, bug.status)
+ for bug in blocks])
+ else:
+ print '\n'.join([bug.uuid for bug in blocks])
def get_parser():
parser = cmdutil.CmdOptionParser("be depend BUG-ID [BUG-ID]")
def help():
return get_parser().help_str() + longhelp
+
+# internal helper functions
+
+def _generate_blocks_string(blocked_bug):
+ return "%s%s" % (BLOCKS_TAG, blocked_bug.uuid)
+
+def _generate_blocked_by_string(blocking_bug):
+ return "%s%s" % (BLOCKED_BY_TAG, blocking_bug.uuid)
+
+def _parse_blocks_string(string):
+ assert string.startswith(BLOCKS_TAG)
+ return string[len(BLOCKS_TAG):]
+
+def _parse_blocked_by_string(string):
+ assert string.startswith(BLOCKED_BY_TAG)
+ return string[len(BLOCKED_BY_TAG):]
+
+def _add_remove_extra_string(bug, string, add):
+ estrs = bug.extra_strings
+ if add == True:
+ estrs.append(string)
+ else: # remove the string
+ estrs.remove(string)
+ bug.extra_strings = estrs # reassign to notice change
+
+def _get_blocks(bug):
+ uuids = []
+ for line in bug.extra_strings:
+ if line.startswith(BLOCKS_TAG):
+ uuids.append(_parse_blocks_string(line))
+ return uuids
+
+def _get_blocked_by(bug):
+ uuids = []
+ for line in bug.extra_strings:
+ if line.startswith(BLOCKED_BY_TAG):
+ uuids.append(_parse_blocked_by_string(line))
+ return uuids
+
+def _repair_one_way_link(blocked_bug, blocking_bug, blocks=None):
+ pass
+
+# functions exposed to other modules
+
+def add_block(blocked_bug, blocking_bug):
+ blocked_by_string = _generate_blocked_by_string(blocking_bug)
+ _add_remove_extra_string(blocked_bug, blocked_by_string, add=True)
+ blocks_string = _generate_blocks_string(blocked_bug)
+ _add_remove_extra_string(blocking_bug, blocks_string, add=True)
+
+def remove_block(blocked_bug, blocking_bug):
+ blocked_by_string = _generate_blocked_by_string(blocking_bug)
+ _add_remove_extra_string(blocked_bug, blocked_by_string, add=False)
+ blocks_string = _generate_blocks_string(blocked_bug)
+ _add_remove_extra_string(blocking_bug, blocks_string, add=False)
+
+def get_blocks(bugdir, bug):
+ """
+ Return a list of bugs that the given bug blocks.
+ """
+ blocks = []
+ for uuid in _get_blocks(bug):
+ blocks.append(bugdir.bug_from_uuid(uuid))
+ return blocks
+
+def get_blocked_by(bugdir, bug):
+ """
+ Return a list of bugs blocking the given bug blocks.
+ """
+ blocked_by = []
+ for uuid in _get_blocked_by(bug):
+ blocked_by.append(bugdir.bug_from_uuid(uuid))
+ return blocked_by
+
+def check_dependencies(bugdir, repair_broken_links=False):
+ """
+ Check that links are bi-directional for all bugs in bugdir.
+ """
+ bugdir.load_all_bugs()
+ good_links = []
+ fixed_links = []
+ broken_links = []
+ for bug in bugdir:
+ for blocker in get_blocked_by(bugdir, bug):
+ blocks = get_blocks(bugdir, blocker)
+ if (bug, blocks) in good_links+fixed_links+broken_links:
+ continue # already checked that link
+ if bug not in blocks:
+ if repair_broken_links == True:
+ _repair_one_way_link(bug, blocker, blocks=True)
+ fixed_links.append((bug, blocker))
+ else:
+ broken_links.append((bug, blocker))
+ else:
+ good_links.append((bug, blocker))
+ for blockee in get_blocks(bugdir, bug):
+ blocked_by = get_blocked_by(bugdir, blockee)
+ if (blockee, bug) in good_links+fixed_links+broken_links:
+ continue # already checked that link
+ if bug not in blocked_by:
+ if repair_broken_links == True:
+ _repair_one_way_link(blockee, bug, blocks=False)
+ fixed_links.append((blockee, bug))
+ else:
+ broken_links.append((blockee, bug))
+ else:
+ good_links.append((blockee, bug))
+ return (good_links, fixed_links, broken_links)
projects / be.git / commitdiff