1 # Copyright (C) 2005-2010 Aaron Bentley and Panometrics, Inc.
2 # Alexander Belchenko <bialix@ukr.net>
3 # Chris Ball <cjb@laptop.org>
4 # Gianluca Montecchi <gian@grys.it>
5 # Oleg Romanyshyn <oromanyshyn@panoramicfeedback.com>
6 # W. Trevor King <wking@drexel.edu>
8 # This program is free software; you can redistribute it and/or modify
9 # it under the terms of the GNU General Public License as published by
10 # the Free Software Foundation; either version 2 of the License, or
11 # (at your option) any later version.
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 # GNU General Public License for more details.
18 # You should have received a copy of the GNU General Public License along
19 # with this program; if not, write to the Free Software Foundation, Inc.,
20 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
23 Define the BugDir class for representing bug comments.
33 import libbe.storage as storage
34 from libbe.storage.util.properties import Property, doc_property, \
35 local_property, defaulting_property, checked_property, \
36 fn_checked_property, cached_property, primed_property, \
37 change_hook_property, settings_property
38 import libbe.storage.util.settings_object as settings_object
39 import libbe.storage.util.mapfile as mapfile
40 import libbe.bug as bug
41 import libbe.util.utility as utility
44 if libbe.TESTING == True:
49 import libbe.storage.base
52 class NoBugDir(Exception):
53 def __init__(self, path):
54 msg = "The directory \"%s\" has no bug directory." % path
55 Exception.__init__(self, msg)
58 class NoRootEntry(Exception):
59 def __init__(self, path):
61 Exception.__init__(self, "Specified root does not exist: %s" % path)
63 class AlreadyInitialized(Exception):
64 def __init__(self, path):
66 Exception.__init__(self,
67 "Specified root is already initialized: %s" % path)
69 class MultipleBugMatches(ValueError):
70 def __init__(self, shortname, matches):
71 msg = ("More than one bug matches %s. "
72 "Please be more specific.\n%s" % (shortname, matches))
73 ValueError.__init__(self, msg)
74 self.shortname = shortname
75 self.matches = matches
77 class NoBugMatches(libbe.util.id.NoIDMatches):
78 def __init__(self, *args, **kwargs):
79 libbe.util.id.NoIDMatches.__init__(self, *args, **kwargs)
82 return 'No bug matches %s' % self.id
85 class DiskAccessRequired (Exception):
86 def __init__(self, goal):
87 msg = "Cannot %s without accessing the disk" % goal
88 Exception.__init__(self, msg)
91 class BugDir (list, settings_object.SavedSettingsObject):
93 TODO: simple bugdir manipulation examples...
96 settings_properties = []
97 required_saved_properties = []
98 _prop_save_settings = settings_object.prop_save_settings
99 _prop_load_settings = settings_object.prop_load_settings
100 def _versioned_property(settings_properties=settings_properties,
101 required_saved_properties=required_saved_properties,
103 if "settings_properties" not in kwargs:
104 kwargs["settings_properties"] = settings_properties
105 if "required_saved_properties" not in kwargs:
106 kwargs["required_saved_properties"]=required_saved_properties
107 return settings_object.versioned_property(**kwargs)
109 @_versioned_property(name="target",
110 doc="The current project development target.")
111 def target(): return {}
113 def _setup_severities(self, severities):
114 if severities not in [None, settings_object.EMPTY]:
115 bug.load_severities(severities)
116 def _set_severities(self, old_severities, new_severities):
117 self._setup_severities(new_severities)
118 self._prop_save_settings(old_severities, new_severities)
119 @_versioned_property(name="severities",
120 doc="The allowed bug severities and their descriptions.",
121 change_hook=_set_severities)
122 def severities(): return {}
124 def _setup_status(self, active_status, inactive_status):
125 bug.load_status(active_status, inactive_status)
126 def _set_active_status(self, old_active_status, new_active_status):
127 self._setup_status(new_active_status, self.inactive_status)
128 self._prop_save_settings(old_active_status, new_active_status)
129 @_versioned_property(name="active_status",
130 doc="The allowed active bug states and their descriptions.",
131 change_hook=_set_active_status)
132 def active_status(): return {}
134 def _set_inactive_status(self, old_inactive_status, new_inactive_status):
135 self._setup_status(self.active_status, new_inactive_status)
136 self._prop_save_settings(old_inactive_status, new_inactive_status)
137 @_versioned_property(name="inactive_status",
138 doc="The allowed inactive bug states and their descriptions.",
139 change_hook=_set_inactive_status)
140 def inactive_status(): return {}
142 def _extra_strings_check_fn(value):
143 return utility.iterable_full_of_strings(value, \
144 alternative=settings_object.EMPTY)
145 def _extra_strings_change_hook(self, old, new):
146 self.extra_strings.sort() # to make merging easier
147 self._prop_save_settings(old, new)
148 @_versioned_property(name="extra_strings",
149 doc="Space for an array of extra strings. Useful for storing state for functionality implemented purely in becommands/<some_function>.py.",
151 check_fn=_extra_strings_check_fn,
152 change_hook=_extra_strings_change_hook,
154 def extra_strings(): return {}
156 def _bug_map_gen(self):
160 for uuid in self.uuids():
163 self._bug_map_value = map # ._bug_map_value used by @local_property
166 @primed_property(primer=_bug_map_gen)
167 @local_property("bug_map")
168 @doc_property(doc="A dict of (bug-uuid, bug-instance) pairs.")
169 def _bug_map(): return {}
171 def __init__(self, storage, uuid=None, from_storage=False):
173 settings_object.SavedSettingsObject.__init__(self)
174 self.storage = storage
175 self.id = libbe.util.id.ID(self, 'bugdir')
177 if from_storage == True:
178 if self.uuid == None:
179 self.uuid = [c for c in self.storage.children()
180 if c != 'version'][0]
182 if self.uuid == None:
183 self.uuid = libbe.util.id.uuid_gen()
184 if self.storage != None and self.storage.is_writeable():
187 # methods for saving/loading/accessing settings and properties.
189 def load_settings(self, settings_mapfile=None):
190 if settings_mapfile == None:
192 self.storage.get(self.id.storage('settings'), default='\n')
194 self.settings = mapfile.parse(settings_mapfile)
195 except mapfile.InvalidMapfileContents, e:
196 raise Exception('Invalid settings file for bugdir %s\n'
197 '(BE version missmatch?)' % self.id.user())
198 self._setup_saved_settings()
199 #self._setup_user_id(self.user_id)
200 self._setup_severities(self.severities)
201 self._setup_status(self.active_status, self.inactive_status)
203 def save_settings(self):
204 mf = mapfile.generate(self._get_saved_settings())
205 self.storage.set(self.id.storage('settings'), mf)
207 def load_all_bugs(self):
209 Warning: this could take a while.
212 for uuid in self.uuids():
217 Save any loaded contents to storage. Because of lazy loading
218 of bugs and comments, this is actually not too inefficient.
220 However, if self.storage.is_writeable() == True, then any
221 changes are automatically written to storage as soon as they
222 happen, so calling this method will just waste time (unless
223 something else has been messing with your stored files).
225 self.storage.add(self.id.storage(), directory=True)
226 self.storage.add(self.id.storage('settings'), parent=self.id.storage(),
232 # methods for managing bugs
234 def uuids(self, use_cached_disk_uuids=True):
235 if use_cached_disk_uuids==False or not hasattr(self, '_uuids_cache'):
236 self._uuids_cache = []
237 # list bugs that are in storage
238 if self.storage != None and self.storage.is_readable():
239 child_uuids = libbe.util.id.child_uuids(
240 self.storage.children(self.id.storage()))
241 for id in child_uuids:
242 self._uuids_cache.append(id)
243 return list(set([bug.uuid for bug in self] + self._uuids_cache))
245 def _clear_bugs(self):
248 if hasattr(self, '_uuids_cache'):
249 del(self._uuids_cache)
252 def _load_bug(self, uuid):
253 bg = bug.Bug(bugdir=self, uuid=uuid, from_storage=True)
258 def new_bug(self, summary=None, _uuid=None):
259 bg = bug.Bug(bugdir=self, uuid=_uuid, summary=summary,
263 if hasattr(self, '_uuids_cache') and not bg.uuid in self._uuids_cache:
264 self._uuids_cache.append(bg.uuid)
267 def remove_bug(self, bug):
268 if hasattr(self, '_uuids_cache') and bug.uuid in self._uuids_cache:
269 self._uuids_cache.remove(bug.uuid)
271 if self.storage != None and self.storage.is_writeable():
274 def bug_from_uuid(self, uuid):
275 if not self.has_bug(uuid):
278 'No bug matches %s in %s' % (uuid, self.storage))
279 if self._bug_map[uuid] == None:
281 return self._bug_map[uuid]
283 def has_bug(self, bug_uuid):
284 if bug_uuid not in self._bug_map:
286 if bug_uuid not in self._bug_map:
290 # methods for id generation
292 def sibling_uuids(self):
295 class RevisionedBugDir (BugDir):
297 RevisionedBugDirs are read-only copies used for generating
298 diffs between revisions.
300 def __init__(self, bugdir, revision):
301 storage_version = bugdir.storage.storage_version(revision)
302 if storage_version != libbe.storage.STORAGE_VERSION:
303 raise libbe.storage.InvalidStorageVersion(storage_version)
304 s = copy.deepcopy(bugdir.storage)
306 class RevisionedStorage (object):
307 def __init__(self, storage, default_revision):
309 self.sget = self.s.get
310 self.sancestors = self.s.ancestors
311 self.schildren = self.s.children
312 self.schanged = self.s.changed
313 self.r = default_revision
314 def get(self, *args, **kwargs):
315 if not 'revision' in kwargs or kwargs['revision'] == None:
316 kwargs['revision'] = self.r
317 return self.sget(*args, **kwargs)
318 def ancestors(self, *args, **kwargs):
319 print 'getting ancestors', args, kwargs
320 if not 'revision' in kwargs or kwargs['revision'] == None:
321 kwargs['revision'] = self.r
322 ret = self.sancestors(*args, **kwargs)
323 print 'got ancestors', ret
325 def children(self, *args, **kwargs):
326 if not 'revision' in kwargs or kwargs['revision'] == None:
327 kwargs['revision'] = self.r
328 return self.schildren(*args, **kwargs)
329 def changed(self, *args, **kwargs):
330 if not 'revision' in kwargs or kwargs['revision'] == None:
331 kwargs['revision'] = self.r
332 return self.schanged(*args, **kwargs)
333 rs = RevisionedStorage(s, revision)
335 s.ancestors = rs.ancestors
336 s.children = rs.children
337 s.changed = rs.changed
338 BugDir.__init__(self, s, from_storage=True)
339 self.revision = revision
341 return self.storage.changed()
344 if libbe.TESTING == True:
345 class SimpleBugDir (BugDir):
347 For testing. Set memory=True for a memory-only bugdir.
348 >>> bugdir = SimpleBugDir()
349 >>> uuids = list(bugdir.uuids())
355 def __init__(self, memory=True, versioned=False):
360 self._dir_ref = dir # postpone cleanup since dir.cleanup() removes dir.
361 if versioned == False:
362 storage = libbe.storage.base.Storage(dir.path)
364 storage = libbe.storage.base.VersionedStorage(dir.path)
367 BugDir.__init__(self, storage=storage, uuid='abc123')
368 bug_a = self.new_bug(summary='Bug A', _uuid='a')
369 bug_a.creator = 'John Doe <jdoe@example.com>'
371 bug_b = self.new_bug(summary='Bug B', _uuid='b')
372 bug_b.creator = 'Jane Doe <jdoe@example.com>'
374 bug_b.status = 'closed'
375 if self.storage != None:
376 self.storage.disconnect() # flush to storage
377 self.storage.connect()
380 if self.storage != None:
381 self.storage.writeable = True
382 self.storage.disconnect()
383 self.storage.destroy()
384 if hasattr(self, '_dir_ref'):
385 self._dir_ref.cleanup()
387 def flush_reload(self):
388 if self.storage != None:
389 self.storage.disconnect()
390 self.storage.connect()
393 # class BugDirTestCase(unittest.TestCase):
395 # self.dir = utility.Dir()
396 # self.bugdir = BugDir(self.dir.path, sink_to_existing_root=False,
397 # allow_storage_init=True)
398 # self.storage = self.bugdir.storage
399 # def tearDown(self):
400 # self.bugdir.cleanup()
402 # def fullPath(self, path):
403 # return os.path.join(self.dir.path, path)
404 # def assertPathExists(self, path):
405 # fullpath = self.fullPath(path)
406 # self.failUnless(os.path.exists(fullpath)==True,
407 # "path %s does not exist" % fullpath)
408 # self.assertRaises(AlreadyInitialized, BugDir,
409 # self.dir.path, assertNewBugDir=True)
410 # def versionTest(self):
411 # if self.storage != None and self.storage.versioned == False:
413 # original = self.bugdir.storage.commit("Began versioning")
414 # bugA = self.bugdir.bug_from_uuid("a")
415 # bugA.status = "fixed"
417 # new = self.storage.commit("Fixed bug a")
418 # dupdir = self.bugdir.duplicate_bugdir(original)
419 # self.failUnless(dupdir.root != self.bugdir.root,
420 # "%s, %s" % (dupdir.root, self.bugdir.root))
421 # bugAorig = dupdir.bug_from_uuid("a")
422 # self.failUnless(bugA != bugAorig,
423 # "\n%s\n%s" % (bugA.string(), bugAorig.string()))
424 # bugAorig.status = "fixed"
425 # self.failUnless(bug.cmp_status(bugA, bugAorig)==0,
426 # "%s, %s" % (bugA.status, bugAorig.status))
427 # self.failUnless(bug.cmp_severity(bugA, bugAorig)==0,
428 # "%s, %s" % (bugA.severity, bugAorig.severity))
429 # self.failUnless(bug.cmp_assigned(bugA, bugAorig)==0,
430 # "%s, %s" % (bugA.assigned, bugAorig.assigned))
431 # self.failUnless(bug.cmp_time(bugA, bugAorig)==0,
432 # "%s, %s" % (bugA.time, bugAorig.time))
433 # self.failUnless(bug.cmp_creator(bugA, bugAorig)==0,
434 # "%s, %s" % (bugA.creator, bugAorig.creator))
435 # self.failUnless(bugA == bugAorig,
436 # "\n%s\n%s" % (bugA.string(), bugAorig.string()))
437 # self.bugdir.remove_duplicate_bugdir()
438 # self.failUnless(os.path.exists(dupdir.root)==False,
441 # self.bugdir.new_bug(uuid="a", summary="Ant")
442 # self.bugdir.new_bug(uuid="b", summary="Cockroach")
443 # self.bugdir.new_bug(uuid="c", summary="Praying mantis")
444 # length = len(self.bugdir)
445 # self.failUnless(length == 3, "%d != 3 bugs" % length)
446 # uuids = list(self.bugdir.uuids())
447 # self.failUnless(len(uuids) == 3, "%d != 3 uuids" % len(uuids))
448 # self.failUnless(uuids == ["a","b","c"], str(uuids))
449 # bugA = self.bugdir.bug_from_uuid("a")
450 # bugAprime = self.bugdir.bug_from_shortname("a")
451 # self.failUnless(bugA == bugAprime, "%s != %s" % (bugA, bugAprime))
454 # def testComments(self, sync_with_disk=False):
455 # if sync_with_disk == True:
456 # self.bugdir.set_sync_with_disk(True)
457 # self.bugdir.new_bug(uuid="a", summary="Ant")
458 # bug = self.bugdir.bug_from_uuid("a")
459 # comm = bug.comment_root
460 # rep = comm.new_reply("Ants are small.")
461 # rep.new_reply("And they have six legs.")
462 # if sync_with_disk == False:
464 # self.bugdir.set_sync_with_disk(True)
465 # self.bugdir._clear_bugs()
466 # bug = self.bugdir.bug_from_uuid("a")
467 # bug.load_comments()
468 # if sync_with_disk == False:
469 # self.bugdir.set_sync_with_disk(False)
470 # self.failUnless(len(bug.comment_root)==1, len(bug.comment_root))
471 # for index,comment in enumerate(bug.comments()):
473 # repLoaded = comment
474 # self.failUnless(repLoaded.uuid == rep.uuid, repLoaded.uuid)
475 # self.failUnless(comment.sync_with_disk == sync_with_disk,
476 # comment.sync_with_disk)
477 # self.failUnless(comment.content_type == "text/plain",
478 # comment.content_type)
479 # self.failUnless(repLoaded.settings["Content-type"] == \
481 # repLoaded.settings)
482 # self.failUnless(repLoaded.body == "Ants are small.",
485 # self.failUnless(comment.in_reply_to == repLoaded.uuid,
487 # self.failUnless(comment.body == "And they have six legs.",
491 # "Invalid comment: %d\n%s" % (index, comment))
492 # def testSyncedComments(self):
493 # self.testComments(sync_with_disk=True)
495 class SimpleBugDirTestCase (unittest.TestCase):
497 # create a pre-existing bugdir in a temporary directory
498 self.dir = utility.Dir()
499 self.storage = libbe.storage.base.Storage(self.dir.path)
501 self.storage.connect()
502 self.bugdir = BugDir(self.storage)
503 self.bugdir.new_bug(summary="Hopefully not imported",
505 self.storage.disconnect()
506 self.storage.connect()
508 if self.storage != None:
509 self.storage.disconnect()
510 self.storage.destroy()
512 def testOnDiskCleanLoad(self):
514 SimpleBugDir(memory==False) should not import
517 bugdir = SimpleBugDir(memory=False)
518 self.failUnless(bugdir.storage.is_readable() == True,
519 bugdir.storage.is_readable())
520 self.failUnless(bugdir.storage.is_writeable() == True,
521 bugdir.storage.is_writeable())
522 uuids = sorted([bug.uuid for bug in bugdir])
523 self.failUnless(uuids == ['a', 'b'], uuids)
524 bugdir.flush_reload()
525 uuids = sorted(bugdir.uuids())
526 self.failUnless(uuids == ['a', 'b'], uuids)
527 uuids = sorted([bug.uuid for bug in bugdir])
528 self.failUnless(uuids == [], uuids)
529 bugdir.load_all_bugs()
530 uuids = sorted([bug.uuid for bug in bugdir])
531 self.failUnless(uuids == ['a', 'b'], uuids)
533 def testInMemoryCleanLoad(self):
535 SimpleBugDir(memory==True) should not import
538 bugdir = SimpleBugDir(memory=True)
539 self.failUnless(bugdir.storage == None, bugdir.storage)
540 uuids = sorted([bug.uuid for bug in bugdir])
541 self.failUnless(uuids == ['a', 'b'], uuids)
542 uuids = sorted([bug.uuid for bug in bugdir])
543 self.failUnless(uuids == ['a', 'b'], uuids)
545 uuids = sorted(bugdir.uuids())
546 self.failUnless(uuids == [], uuids)
547 uuids = sorted([bug.uuid for bug in bugdir])
548 self.failUnless(uuids == [], uuids)
551 unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
552 suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
554 # def _get_settings(self, settings_path, for_duplicate_bugdir=False):
555 # allow_no_storage = not self.storage.path_in_root(settings_path)
556 # if allow_no_storage == True:
557 # assert for_duplicate_bugdir == True
558 # if self.sync_with_disk == False and for_duplicate_bugdir == False:
559 # # duplicates can ignore this bugdir's .sync_with_disk status
560 # raise DiskAccessRequired("_get settings")
562 # settings = mapfile.map_load(self.storage, settings_path, allow_no_storage)
563 # except storage.NoSuchFile:
564 # settings = {"storage_name": "None"}
567 # def _save_settings(self, settings_path, settings,
568 # for_duplicate_bugdir=False):
569 # allow_no_storage = not self.storage.path_in_root(settings_path)
570 # if allow_no_storage == True:
571 # assert for_duplicate_bugdir == True
572 # if self.sync_with_disk == False and for_duplicate_bugdir == False:
573 # # duplicates can ignore this bugdir's .sync_with_disk status
574 # raise DiskAccessRequired("_save settings")
575 # self.storage.mkdir(self.get_path(), allow_no_storage)
576 # mapfile.map_save(self.storage, settings_path, settings, allow_no_storage)