diff options
Diffstat (limited to 'libbe')
| -rw-r--r-- | libbe/util/id.py | 294 |
1 files changed, 254 insertions, 40 deletions
diff --git a/libbe/util/id.py b/libbe/util/id.py index d57205f..d443706 100644 --- a/libbe/util/id.py +++ b/libbe/util/id.py @@ -20,10 +20,13 @@ Handle ID creation and parsing. """ import os.path +import re import libbe if libbe.TESTING == True: + import doctest + import sys import unittest try: @@ -60,6 +63,25 @@ except ImportError: return output.rstrip('\n') +HIERARCHY = ['bugdir', 'bug', 'comment'] + + +class MultipleIDMatches (ValueError): + def __init__(self, id, matches): + msg = ("More than one id matches %s. " + "Please be more specific.\n%s" % (id, matches)) + ValueError.__init__(self, msg) + self.id = id + self.matches = matches + +class NoIDMatches (KeyError): + def __init__(self, id, possible_ids): + msg = "No id matches %s.\n%s" % (id, possible_ids) + KeyError.__init__(self, msg) + self.id = id + self.possible_ids = possible_ids + + def _assemble(*args): args = list(args) for i,arg in enumerate(args): @@ -74,50 +96,242 @@ def _split(id): args[i] = None return args -def _is_a_uuid(id): - if id.startswith('uuid:'): - return True - return False - -def _uuid_to_id(id): - return 'uuid:' + id - -def _id_to_uuid(id): - return id[len('uuid:'):] - -def bugdir_id(bugdir, *args): - return _assemble(_uuid_to_id(bugdir.uuid), *args) - -def bug_id(bug, *args): - if bug.bugdir == None: - bdid = None - else: - bdid = bugdir_id(bug.bugdir) - return _assemble(bdid, _uuid_to_id(bug.uuid), *args) - -def comment_id(comment, *args): - if comment.bug == None: - bid = None - else: - bid = bug_id(comment.bug) - return _assemble(bid, _uuid_to_id(comment.uuid), *args) - -def parse_id(id): - args = _split(id) - ret = {'bugdir':_id_to_uuid(args.pop(0))} - type = 'bugdir' - for child_name in ['bug', 'comment']: - if len(args) > 0 and _is_a_uuid(args[0]): - ret[child_name] = _id_to_uuid(args.pop(0)) - type = child_name - ret['type'] = type - ret['remaining'] = os.path.join(args) +def _truncate(uuid, other_uuids, min_length=3): + chars = min_length + for id in other_uuids: + if id == uuid: + continue + while (id[:chars] == uuid[:chars]): + chars+=1 + return uuid[:chars] + +def _expand(truncated_id, other_ids): + matches = [] + for id in other_ids: + if id.startswith(truncated_id): + matches.append(id) + if len(matches) > 1: + raise MultipleIDMatches(truncated_id, matches) + if len(matches) == 0: + raise NoIDMatches(truncated_id, other_ids) + return matches[0] + + +class ID (object): + """ + IDs have several formats specialized for different uses. + + In storage, all objects are represented by their uuid alone, + because that is the simplest globally unique identifier. You can + generate ids of this sort with the .storage() method. Because an + object's storage may be distributed across several chunks, and the + chunks may not have their own uuid, we generate chunk ids by + prepending the objects uuid to the chunk name. The user id types + do not support this chunk extension feature. + + For users, the full uuids are a bit overwhelming, so we truncate + them while retaining local uniqueness (with regards to the other + objects currently in storage). We also prepend truncated parent + ids for two reasons: + (1) so that a user can locate the repository containing the + referenced object. It would be hard to find bug 'XYZ' if + that's all you knew. Much easier with 'ABC/XYZ', where ABC + is the bugdir. Each project can publish a list of bugdir-id +x - to - location mappings, e.g. + ABC...(full uuid)...DEF https://server.com/projectX/be/ + which is easier than publishing all-object-ids-to-location + mappings. + (2) because it's easier to generate and parse truncated ids if + you don't have to fetch all the ids in the storage + repository, but can restrict yourself to a specific branch. + You can generate ids of this sort with the .user() method, + although in order to preform the truncation, your object (and its + parents must define a .sibling_uuids() method. + + + While users can use the convenient short user ids in the short + term, the truncation will inevitably lead to name collision. To + avoid that, we provide a non-truncated form of the short user ids + via the .long_user() method. These long user ids should be + converted to short user ids by intelligent user interfaces. + + Related tools: + * get uuids back out of the user ids: + parse_user() + * scan text for user ids & convert to long user ids: + short_to_long_user() + * scan text for long user ids & convert to short user ids: + long_to_short_user() + + Supported types: 'bugdir', 'bug', 'comment' + """ + def __init__(self, object, type): + self._object = object + self._type = type + assert self._type in HIERARCHY, self._type + self.uuid = self._object.uuid + + def storage(self, *args): + return _assemble(self._object.uuid, *args) + + def _ancestors(self): + ret = [self._object] + index = HIERARCHY.index(self._type) + if index == 0: + return ret + o = self._object + for i in range(index, 0, -1): + parent_name = HIERARCHY[i-1] + o = getattr(o, parent_name) + ret.insert(0, o) + return ret + + def long_user(self): + return _assemble(*[o.uuid for o in self._ancestors()]) + + def user(self): + return _assemble(*[_truncate(o.uuid, o.sibling_uuids()) + for o in self._ancestors()]) + +def parse_user(id): + """ + >>> parse_user('ABC/DEF/GHI') == \\ + ... {'bugdir':'ABC', 'bug':'DEF', 'comment':'GHI', 'type':'comment'} + True + >>> parse_user('ABC/DEF') == \\ + ... {'bugdir':'ABC', 'bug':'DEF', 'type':'bug'} + True + >>> parse_user('ABC') == \\ + ... {'bugdir':'ABC', 'type':'bugdir'} + True + """ + ret = {} + args = _split(id) + assert len(args) > 0 and len(args) < 4, 'Invalid id "%s"' % id + for type,arg in zip(HIERARCHY, args): + assert len(arg) > 0, 'Invalid part "%s" of id "%s"' % (arg, id) + ret['type'] = type + ret[type] = arg return ret +REGEXP = '#([-a-f0-9]*)(/[-a-g0-9]*)?(/[-a-g0-9]*)?#' + +class IDreplacer (object): + def __init__(self, bugdirs, direction): + self.bugdirs = bugdirs + self.direction = direction + def __call__(self, match): + ids = [m.lstrip('/') for m in match.groups() if m != None] + ids = self.switch_ids(ids) + return '#' + '/'.join(ids) + '#' + def switch_id(self, id, sibling_uuids): + if id == None: + return None + if self.direction == 'long_to_short': + return _truncate(id, sibling_uuids) + return _expand(id, sibling_uuids) + def switch_ids(self, ids): + assert ids[0] != None, ids + if self.direction == 'long_to_short': + bugdir = [bd for bd in self.bugdirs if bd.uuid == ids[0]][0] + objects = [bugdir] + if len(ids) >= 2: + bug = bugdir.bug_from_uuid(ids[1]) + objects.append(bug) + if len(ids) >= 3: + comment = bug.comment_from_uuid(ids[2]) + objects.append(comment) + for i,obj in enumerate(objects): + ids[i] = self.switch_id(ids[i], obj.sibling_uuids()) + else: + ids[0] = self.switch_id(ids[0], [bd.uuid for bd in self.bugdirs]) + if len(ids) == 1: + return ids + bugdir = [bd for bd in self.bugdirs if bd.uuid == ids[0]][0] + ids[1] = self.switch_id(ids[1], bugdir.uuids()) + if len(ids) == 2: + return ids + bug = bugdir.bug_from_uuid(ids[1]) + ids[2] = self.switch_id(ids[2], bug.uuids()) + return ids + +def short_to_long_user(bugdirs, text): + return re.sub(REGEXP, IDreplacer(bugdirs, 'short_to_long'), text) +def long_to_short_user(bugdirs, text): + return re.sub(REGEXP, IDreplacer(bugdirs, 'long_to_short'), text) + if libbe.TESTING == True: class UUIDtestCase(unittest.TestCase): def testUUID_gen(self): id = uuid_gen() - self.failUnless(len(id) == 36, "invalid UUID '%s'" % id) + self.failUnless(len(id) == 36, 'invalid UUID "%s"' % id) + + class DummyObject (object): + def __init__(self, uuid, siblings=[]): + self.uuid = uuid + self._siblings = siblings + def sibling_uuids(self): + return self._siblings + + class IDtestCase(unittest.TestCase): + def setUp(self): + self.bugdir = DummyObject('1234abcd') + self.bug = DummyObject('abcdef', ['a1234', 'ab9876']) + self.bug.bugdir = self.bugdir + self.comment = DummyObject('12345678', ['1234abcd', '1234cdef']) + self.comment.bug = self.bug + self.bd_id = ID(self.bugdir, 'bugdir') + self.b_id = ID(self.bug, 'bug') + self.c_id = ID(self.comment, 'comment') + def test_storage(self): + self.failUnless(self.bd_id.storage() == self.bugdir.uuid, + self.bd_id.storage()) + self.failUnless(self.b_id.storage() == self.bug.uuid, + self.b_id.storage()) + self.failUnless(self.c_id.storage() == self.comment.uuid, + self.c_id.storage()) + self.failUnless(self.bd_id.storage('x','y','z') == \ + '1234abcd/x/y/z', self.bd_id.storage()) + def test_long_user(self): + self.failUnless(self.bd_id.long_user() == self.bugdir.uuid, + self.bd_id.long_user()) + self.failUnless(self.b_id.long_user() == \ + '/'.join([self.bugdir.uuid, self.bug.uuid]), + self.b_id.long_user()) + self.failUnless(self.c_id.long_user() == + '/'.join([self.bugdir.uuid, self.bug.uuid, + self.comment.uuid]), + self.c_id.long_user) + def test_user(self): + self.failUnless(self.bd_id.user() == '123', + self.bd_id.user()) + self.failUnless(self.b_id.user() == '123/abc', + self.b_id.user()) + self.failUnless(self.c_id.user() == '123/abc/12345', + self.c_id.user()) + + class IDtestCase(unittest.TestCase): + def setUp(self): + self.bugdir = DummyObject('1234abcd') + self.bug = DummyObject('abcdef', ['a1234', 'ab9876']) + self.bug.bugdir = self.bugdir + self.bugdir.bug_from_uuid = lambda uuid: self.bug + self.bugdir.uuids = lambda : self.bug.sibling_uuids() + [self.bug.uuid] + self.comment = DummyObject('12345678', ['1234abcd', '1234cdef']) + self.comment.bug = self.bug + self.bug.comment_from_uuid = lambda uuid: self.comment + self.bug.uuids = lambda : self.comment.sibling_uuids() + [self.comment.uuid] + self.bd_id = ID(self.bugdir, 'bugdir') + self.b_id = ID(self.bug, 'bug') + self.c_id = ID(self.comment, 'comment') + self.short = 'bla bla #123/abc# bla bla #123/abc/12345# bla bla' + self.long = 'bla bla #1234abcd/abcdef# bla bla #1234abcd/abcdef/12345678# bla bla' + def test_short_to_long(self): + self.failUnless(short_to_long_user([self.bugdir], self.short) == self.long, + '\n' + self.short + '\n' + short_to_long_user([self.bugdir], self.short) + '\n' + self.long) + def test_long_to_short(self): + self.failUnless(long_to_short_user([self.bugdir], self.long) == self.short, + '\n' + long_to_short_user([self.bugdir], self.long) + '\n' + self.short) - suite = unittest.TestLoader().loadTestsFromTestCase(UUIDtestCase) + unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__]) + suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()]) |