diff options
| -rw-r--r-- | libbe/bug.py | 273 | ||||
| -rw-r--r-- | libbe/properties.py | 418 |
2 files changed, 586 insertions, 105 deletions
diff --git a/libbe/bug.py b/libbe/bug.py index f47bcba..e04d4ee 100644 --- a/libbe/bug.py +++ b/libbe/bug.py @@ -21,6 +21,9 @@ import time import doctest from beuuid import uuid_gen +from properties import Property, doc_property, local_property, \ + defaulting_property, checked_property, cached_property, \ + primed_property, change_hook_property, settings_property import mapfile import comment import utility @@ -69,76 +72,155 @@ for i in range(len(status_values)): status_index[status_values[i]] = i -def checked_property(name, valid): +# Define an invalid value for our properties, distinct from None, +# which shows that a property has been initialized but has no value. +EMPTY = -1 + + +class Bug(object): """ - Provide access to an attribute name, testing for valid values. + >>> b = Bug() + >>> print b.status + open + >>> print b.severity + minor + + There are two formats for time, int and string. Setting either + one will adjust the other appropriately. The string form is the + one stored in the bug's settings file on disk. + >>> print type(b.time) + <type 'int'> + >>> print type(b.time_string) + <type 'str'> + >>> b.time = 0 + >>> print b.time_string + Thu, 01 Jan 1970 00:00:00 +0000 + >>> b.time_string="Thu, 01 Jan 1970 00:01:00 +0000" + >>> b.time + 60 + >>> print b.settings["time"] + Thu, 01 Jan 1970 00:01:00 +0000 """ - def getter(self): - value = getattr(self, "_"+name) - if value not in valid: - raise InvalidValue(name, value) - return value + def _save_settings(self, old, new): + if self.sync_with_disk==True: + self.save_settings() + def _load_settings(self): + if self.sync_with_disk==True and self._settings_loaded==False: + self.load_settings() + else: + for property in self.settings_properties: + if property not in self.settings: + self.settings[property] = EMPTY + + settings_properties = [] + required_saved_properties = ['status','severity'] # to protect against future changes in default values + + def _versioned_property(name, doc, default=None, save=_save_settings, load=_load_settings, setprops=settings_properties, allowed=None): + "Combine the common decorators in a single function" + setprops.append(name) + def decorator(funcs): + if allowed != None: + checked = checked_property(allowed=allowed) + defaulting = defaulting_property(default=default, null=EMPTY) + change_hook = change_hook_property(hook=save) + primed = primed_property(primer=load) + settings = settings_property(name=name) + docp = doc_property(doc=doc) + deco = defaulting(change_hook(primed(settings(docp(funcs))))) + if allowed != None: + deco = checked(deco) + return Property(deco) + return decorator + + @_versioned_property(name="severity", + doc="A measure of the bug's importance", + default="minor", + allowed=severity_values) + def severity(): return {} + + @_versioned_property(name="status", + doc="The bug's current status", + default="open", + allowed=status_values) + def status(): return {} + + @property + def active(self): + return self.status in active_status_values - def setter(self, value): - if value not in valid: - raise InvalidValue(name, value) - return setattr(self, "_"+name, value) - return property(getter, setter) + @_versioned_property(name="target", + doc="The deadline for fixing this bug") + def target(): return {} + @_versioned_property(name="creator", + doc="The user who entered the bug into the system") + def creator(): return {} -class Bug(object): - severity = checked_property("severity", severity_values) - status = checked_property("status", status_values) + @_versioned_property(name="reporter", + doc="The user who reported the bug") + def reporter(): return {} - def _get_active(self): - return self.status in active_status_values + @_versioned_property(name="assigned", + doc="The developer in charge of the bug") + def assigned(): return {} + + @_versioned_property(name="time", + doc="An RFC 2822 timestamp for bug creation") + def time_string(): return {} - active = property(_get_active) + def _get_time(self): + if self.time_string == None: + return None + return utility.str_to_time(self.time_string) + def _set_time(self, value): + self.time_string = utility.time_to_str(value) + time = property(fget=_get_time, + fset=_set_time, + doc="An integere version of .time_string") + + @_versioned_property(name="summary", + doc="A one-line bug description") + def summary(): return {} def _get_comment_root(self): - if self._comment_root == None: - if self._comments_loaded == True: - self._comment_root = comment.loadComments(self) - else: - self._comment_root = comment.Comment(self, - uuid=comment.INVALID_UUID) - return self._comment_root + if self.sync_with_disk: + return comment.loadComments(self) + else: + return comment.Comment(self, uuid=comment.INVALID_UUID) + + @Property + @cached_property(generator=_get_comment_root) + @local_property("comment_root") + @doc_property(doc="The trunk of the comment tree") + def comment_root(): return {} - def _set_comment_root(self, comment_root): - self._comment_root = comment_root + def _get_rcs(self): + if hasattr(self.bugdir, "rcs"): + return self.bugdir.rcs - _comment_root = None - comment_root = property(_get_comment_root, _set_comment_root, - doc="The trunk of the comment tree") + @Property + @cached_property(generator=_get_rcs) + @local_property("rcs") + @doc_property(doc="A revision control system instance.") + def rcs(): return {} def __init__(self, bugdir=None, uuid=None, from_disk=False, load_comments=False, summary=None): self.bugdir = bugdir - if bugdir != None: - self.rcs = bugdir.rcs - else: - self.rcs = None + self.uuid = uuid + self._settings_loaded = False + self.settings = {} if from_disk == True: - self._comments_loaded = False - self.uuid = uuid - self.load(load_comments=load_comments) + self.sync_with_disk = True + #self.load(load_comments=load_comments) else: - # Note: defaults should match those in Bug.load() - self._comments_loaded = True - if uuid != None: - self.uuid = uuid - else: + self.sync_with_disk = False + if uuid == None: self.uuid = uuid_gen() - self.summary = summary + self.time = int(time.time()) # only save to second precision if self.rcs != None: self.creator = self.rcs.get_user_id() - else: - self.creator = None - self.target = None - self.status = "open" - self.severity = "minor" - self.assigned = None - self.time = int(time.time()) # only save to second precision + self.summary = summary def __repr__(self): return "Bug(uuid=%r)" % self.uuid @@ -149,27 +231,19 @@ class Bug(object): else: shortname = self.bugdir.bug_shortname(self) if shortlist == False: - if self.time == None: - timestring = "" + if self.time_string == "": + timestring = self.time_string else: htime = utility.handy_time(self.time) - ftime = utility.time_to_str(self.time) - timestring = "%s (%s)" % (htime, ftime) + timestring = "%s (%s)" % (htime, self.time_string) info = [("ID", self.uuid), ("Short name", shortname), ("Severity", self.severity), ("Status", self.status), - ("Assigned", self.assigned), - ("Target", self.target), - ("Creator", self.creator), + ("Assigned", self.assigned or ""), + ("Target", self.target or ""), + ("Creator", self.creator or ""), ("Created", timestring)] - newinfo = [] - for k,v in info: - if v == None: - newinfo.append((k,"")) - else: - newinfo.append((k,v)) - info = newinfo longest_key_len = max([len(k) for k,v in info]) infolines = [" %*s : %s\n" %(longest_key_len,k,v) for k,v in info] bugout = "".join(infolines) + "%s" % self.summary.rstrip('\n') @@ -180,8 +254,6 @@ class Bug(object): bugout = "%s:%s: %s" % (shortname,chars,self.summary.rstrip('\n')) if show_comments == True: - if self._comments_loaded == False: - self.load_comments() # take advantage of the string_thread(auto_name_map=True) # SIDE-EFFECT of sorting by comment time. comout = self.comment_root.string_thread(flatten=False, @@ -205,52 +277,38 @@ class Bug(object): assert name in ["values", "comments"] return os.path.join(my_dir, name) - def load(self, load_comments=False): - map = mapfile.map_load(self.rcs, self.get_path("values")) - self.summary = map.get("summary") - self.creator = map.get("creator") - self.target = map.get("target") - self.status = map.get("status", "open") - self.severity = map.get("severity", "minor") - self.assigned = map.get("assigned") - self.time = map.get("time") - if self.time is not None: - self.time = utility.str_to_time(self.time) - - if load_comments == True: - self.load_comments() + def load_settings(self): + self.settings = mapfile.map_load(self.rcs, self.get_path("values")) + for property in self.settings_properties: + if property not in self.settings: + self.settings[property] = EMPTY + elif self.settings[property] == None: + self.settings[property] = EMPTY + self._settings_loaded = True def load_comments(self): - # clear _comment_root, so _get_comment_root returns a fresh version - self._comment_root = None - self._comments_loaded = True - - def comments(self): - if self._comments_loaded == False: - self.load_comments() - for comment in self.comment_root.traverse(): - yield comment - - def _add_attr(self, map, name): - value = getattr(self, name) - if value is not None: - map[name] = value - - def save(self): + # Clear _comment_root, so _get_comment_root returns a fresh + # version. Turn of syncing temporarily so we don't write our + # blank comment tree to disk. + self.sync_with_disk = False + self._comment_root = None + self.sync_with_disk = True + + def save_settings(self): assert self.summary != None, "Can't save blank bug" map = {} - self._add_attr(map, "assigned") - self._add_attr(map, "summary") - self._add_attr(map, "creator") - self._add_attr(map, "target") - self._add_attr(map, "status") - self._add_attr(map, "severity") - if self.time is not None: - map["time"] = utility.time_to_str(self.time) + for k,v in self.settings.items(): + if (v != None and v != EMPTY): + map[k] = v + for k in self.required_saved_properties: + map[k] = getattr(self, k) self.rcs.mkdir(self.get_path()) path = self.get_path("values") mapfile.map_save(self.rcs, path, map) + + def save(self): + self.save_settings() if len(self.comment_root) > 0: self.rcs.mkdir(self.get_path("comments")) @@ -261,6 +319,10 @@ class Bug(object): path = self.get_path() self.rcs.recursive_remove(path) + def comments(self): + for comment in self.comment_root.traverse(): + yield comment + def new_comment(self, body=None): comm = self.comment_root.new_reply(body=body) return comm @@ -280,7 +342,8 @@ class Bug(object): for id, comment in self.comment_root.comment_shortnames(shortname): yield (id, comment) -# the general rule for bug sorting is that "more important" bugs are + +# The general rule for bug sorting is that "more important" bugs are # less than "less important" bugs. This way sorting a list of bugs # will put the most important bugs first in the list. When relative # importance is unclear, the sorting follows some arbitrary convention diff --git a/libbe/properties.py b/libbe/properties.py new file mode 100644 index 0000000..f55dc0e --- /dev/null +++ b/libbe/properties.py @@ -0,0 +1,418 @@ +# Bugs Everywhere - a distributed bugtracker +# Copyright (C) 2008 W. Trevor King <wking@drexel.edu> +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +""" +This module provides a series of useful decorators for defining +various types of properties. For example usage, consider the +unittests at the end of the module. + +See + http://www.python.org/dev/peps/pep-0318/ +and + http://www.phyast.pitt.edu/~micheles/python/documentation.html +for more information on decorators. +""" + +import unittest + +class ValueCheckError (ValueError): + def __init__(self, name, value, allowed): + msg = "%s not in %s for %s" % (value, allowed, name) + ValueError.__init__(self, msg) + self.name = name + self.value = value + self.allowed = allowed + +def Property(funcs): + """ + End a chain of property decorators, returning a property. + """ + args = {} + args["fget"] = funcs.get("fget", None) + args["fset"] = funcs.get("fset", None) + args["fdel"] = funcs.get("fdel", None) + args["doc"] = funcs.get("doc", None) + + #print "Creating a property with" + #for key, val in args.items(): print key, value + return property(**args) + +def doc_property(doc=None): + """ + Add a docstring to a chain of property decorators. + """ + def decorator(funcs=None): + """ + Takes either a dict of funcs {"fget":fnX, "fset":fnY, ...} + or a function fn() returning such a dict. + """ + if hasattr(funcs, "__call__"): + funcs = funcs() # convert from function-arg to dict + funcs["doc"] = doc + return funcs + return decorator + +def local_property(name): + """ + Define get/set access to per-parent-instance local storage. Uses + ._<name>_value to store the value for a particular owner instance. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget", None) + fset = funcs.get("fset", None) + def _fget(self): + if fget is not None: + fget(self) + value = getattr(self, "_%s_value" % name, None) + return value + def _fset(self, value): + setattr(self, "_%s_value" % name, value) + if fset is not None: + fset(self, value) + funcs["fget"] = _fget + funcs["fset"] = _fset + funcs["name"] = name + return funcs + return decorator + +def settings_property(name): + """ + Similar to local_property, except where local_property stores the + value in instance._<name>_value, settings_property stores the + value in instance.settings[name]. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget", None) + fset = funcs.get("fset", None) + def _fget(self): + if fget is not None: + fget(self) + value = self.settings.get(name, None) + return value + def _fset(self, value): + self.settings[name] = value + if fset is not None: + fset(self, value) + funcs["fget"] = _fget + funcs["fset"] = _fset + funcs["name"] = name + return funcs + return decorator + +def defaulting_property(default=None, null=None): + """ + Define a default value for get access to a property. + If the stored value is null, then default is returned. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget") + def _fget(self): + value = fget(self) + if value == null: + return default + return value + funcs["fget"] = _fget + return funcs + return decorator + +def checked_property(allowed=[]): + """ + Define allowed values for get/set access to a property. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget") + fset = funcs.get("fset") + name = funcs.get("name", "<unknown>") + def _fget(self): + value = fget(self) + if value not in allowed: + raise ValueCheckError(name, value, allowed) + return value + def _fset(self, value): + if value not in allowed: + raise ValueCheckError(name, value, allowed) + fset(self, value) + funcs["fget"] = _fget + funcs["fset"] = _fset + return funcs + return decorator + +def cached_property(generator, initVal=None): + """ + Allow caching of values generated by generator(instance), where + instance is the instance to which this property belongs. Uses + ._<name>_cache to store a cache flag for a particular owner + instance. When the cache flag is True (or missing), the normal + value is returned. Otherwise the generator is called (and it's + output stored) for every get. The cache flag is missing on + initialization. Particular instances may override by setting + their own flag. + + If caching is True, but the stored value == initVal, the parameter + is considered 'uninitialized', and the generator is called anyway. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget") + fset = funcs.get("fset") + name = funcs.get("name", "<unknown>") + def _fget(self): + cache = getattr(self, "_%s_cache" % name, True) + if cache == True: + value = fget(self) + if cache == False or (cache == True and value == initVal): + value = generator(self) + fset(self, value) + return value + funcs["fget"] = _fget + return funcs + return decorator + +def primed_property(primer, initVal=None): + """ + Just like a generator_property, except that instead of returning a + new value and running fset to cache it, the primer performs some + background manipulation (e.g. loads data into instance.settings) + such that a _second_ pass through fget succeeds. + + The 'cache' flag becomes a 'prime' flag, with priming taking place + whenever ._<name>_prime is True, or is False or missing and + value == initVal. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget") + name = funcs.get("name", "<unknown>") + def _fget(self): + prime = getattr(self, "_%s_prime" % name, False) + if prime == False: + value = fget(self) + if prime == True or (prime == False and value == initVal): + primer(self) + value = fget(self) + return value + funcs["fget"] = _fget + return funcs + return decorator + +def change_hook_property(hook): + """ + Call the function hook(instance, old_value, new_value) whenever a + value different from the current value is set (instance is a a + reference to the class instance to which this property belongs). + This is useful for saving changes to disk, etc. + """ + def decorator(funcs): + if hasattr(funcs, "__call__"): + funcs = funcs() + fget = funcs.get("fget") + fset = funcs.get("fset") + name = funcs.get("name", "<unknown>") + def _fset(self, value): + old_value = fget(self) + if value != old_value: + hook(self, old_value, value) + fset(self, value) + funcs["fset"] = _fset + return funcs + return decorator + + +class DecoratorTests(unittest.TestCase): + def testLocalDoc(self): + class Test(object): + @Property + @doc_property("A fancy property") + def x(): + return {} + self.failUnless(Test.x.__doc__ == "A fancy property", + Test.x.__doc__) + def testLocalProperty(self): + class Test(object): + @Property + @local_property(name="LOCAL") + def x(): + return {} + t = Test() + self.failUnless(t.x == None, str(t.x)) + t.x = 'z' # the first set initializes ._LOCAL_value + self.failUnless(t.x == 'z', str(t.x)) + self.failUnless("_LOCAL_value" in dir(t), dir(t)) + self.failUnless(t._LOCAL_value == 'z', t._LOCAL_value) + def testSettingsProperty(self): + class Test(object): + @Property + @settings_property(name="attr") + def x(): + return {} + def __init__(self): + self.settings = {} + t = Test() + self.failUnless(t.x == None, str(t.x)) + t.x = 'z' # the first set initializes ._LOCAL_value + self.failUnless(t.x == 'z', str(t.x)) + self.failUnless("attr" in t.settings, t.settings) + self.failUnless(t.settings["attr"] == 'z', t.settings["attr"]) + def testDefaultingLocalProperty(self): + class Test(object): + @Property + @defaulting_property(default='y', null='x') + @local_property(name="DEFAULT") + def x(): return {} + t = Test() + self.failUnless(t.x == None, str(t.x)) + t.x = 'x' + self.failUnless(t.x == 'y', str(t.x)) + t.x = 'y' + self.failUnless(t.x == 'y', str(t.x)) + t.x = 'z' + self.failUnless(t.x == 'z', str(t.x)) + def testCheckedLocalProperty(self): + class Test(object): + @Property + @checked_property(allowed=['x', 'y', 'z']) + @local_property(name="CHECKED") + def x(): return {} + def __init__(self): + self._CHECKED_value = 'x' + t = Test() + self.failUnless(t.x == 'x', str(t.x)) + try: + t.x = None + e = None + except ValueCheckError, e: + pass + self.failUnless(type(e) == ValueCheckError, type(e)) + def testTwoCheckedLocalProperties(self): + class Test(object): + @Property + @checked_property(allowed=['x', 'y', 'z']) + @local_property(name="X") + def x(): return {} + + @Property + @checked_property(allowed=['a', 'b', 'c']) + @local_property(name="A") + def a(): return {} + def __init__(self): + self._A_value = 'a' + self._X_value = 'x' + t = Test() + try: + t.x = 'a' + e = None + except ValueCheckError, e: + pass + self.failUnless(type(e) == ValueCheckError, type(e)) + t.x = 'x' + t.x = 'y' + t.x = 'z' + try: + t.a = 'x' + e = None + except ValueCheckError, e: + pass + self.failUnless(type(e) == ValueCheckError, type(e)) + t.a = 'a' + t.a = 'b' + t.a = 'c' + def testCachedLocalProperty(self): + class Gen(object): + def __init__(self): + self.i = 0 + def __call__(self, owner): + self.i += 1 + return self.i + class Test(object): + @Property + @cached_property(generator=Gen(), initVal=None) + @local_property(name="CACHED") + def x(): return {} + t = Test() + self.failIf("_CACHED_cache" in dir(t), getattr(t, "_CACHED_cache", None)) + self.failUnless(t.x == 1, t.x) + self.failUnless(t.x == 1, t.x) + self.failUnless(t.x == 1, t.x) + t.x = 8 + self.failUnless(t.x == 8, t.x) + self.failUnless(t.x == 8, t.x) + t._CACHED_cache = False + val = t.x + self.failUnless(val == 2, val) + val = t.x + self.failUnless(val == 3, val) + val = t.x + self.failUnless(val == 4, val) + t._CACHED_cache = True + self.failUnless(t.x == 4, str(t.x)) + self.failUnless(t.x == 4, str(t.x)) + self.failUnless(t.x == 4, str(t.x)) + def testPrimedLocalProperty(self): + class Test(object): + def prime(self): + self.settings["PRIMED"] = "initialized" + @Property + @primed_property(primer=prime, initVal=None) + @settings_property(name="PRIMED") + def x(): return {} + def __init__(self): + self.settings={} + t = Test() + self.failIf("_PRIMED_prime" in dir(t), getattr(t, "_PRIMED_prime", None)) + self.failUnless(t.x == "initialized", t.x) + t.x = 1 + self.failUnless(t.x == 1, t.x) + t.x = None + self.failUnless(t.x == "initialized", t.x) + t._PRIMED_prime = True + t.x = 3 + self.failUnless(t.x == "initialized", t.x) + t._PRIMED_prime = False + t.x = 3 + self.failUnless(t.x == 3, t.x) + def testChangeHookLocalProperty(self): + class Test(object): + def _hook(self, old, new): + self.old = old + self.new = new + + @Property + @change_hook_property(_hook) + @local_property(name="HOOKED") + def x(): return {} + t = Test() + t.x = 1 + self.failUnless(t.old == None, t.old) + self.failUnless(t.new == 1, t.new) + t.x = 1 + self.failUnless(t.old == None, t.old) + self.failUnless(t.new == 1, t.new) + t.x = 2 + self.failUnless(t.old == 1, t.old) + self.failUnless(t.new == 2, t.new) + +suite = unittest.TestLoader().loadTestsFromTestCase(DecoratorTests) + |