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]
183 if self.uuid == None:
184 self.uuid = libbe.util.id.uuid_gen()
186 self._setup_saved_settings()
187 if self.storage != None and self.storage.is_writeable():
190 # methods for saving/loading/accessing settings and properties.
192 def load_settings(self, settings_mapfile=None):
193 if settings_mapfile == None:
195 self.storage.get(self.id.storage('settings'), default='\n')
197 self.settings = mapfile.parse(settings_mapfile)
198 except mapfile.InvalidMapfileContents, e:
199 raise Exception('Invalid settings file for bugdir %s\n'
200 '(BE version missmatch?)' % self.id.user())
201 self._setup_saved_settings()
202 #self._setup_user_id(self.user_id)
203 self._setup_severities(self.severities)
204 self._setup_status(self.active_status, self.inactive_status)
206 def save_settings(self):
207 mf = mapfile.generate(self._get_saved_settings())
208 self.storage.set(self.id.storage('settings'), mf)
210 def load_all_bugs(self):
212 Warning: this could take a while.
215 for uuid in self.uuids():
220 Save any loaded contents to storage. Because of lazy loading
221 of bugs and comments, this is actually not too inefficient.
223 However, if self.storage.is_writeable() == True, then any
224 changes are automatically written to storage as soon as they
225 happen, so calling this method will just waste time (unless
226 something else has been messing with your stored files).
228 self.storage.add(self.id.storage(), directory=True)
229 self.storage.add(self.id.storage('settings'), parent=self.id.storage(),
235 # methods for managing bugs
237 def uuids(self, use_cached_disk_uuids=True):
238 if use_cached_disk_uuids==False or not hasattr(self, '_uuids_cache'):
239 self._uuids_cache = []
240 # list bugs that are in storage
241 if self.storage != None and self.storage.is_readable():
242 child_uuids = libbe.util.id.child_uuids(
243 self.storage.children(self.id.storage()))
244 for id in child_uuids:
245 self._uuids_cache.append(id)
246 return list(set([bug.uuid for bug in self] + self._uuids_cache))
248 def _clear_bugs(self):
251 del(self._uuids_cache)
254 def _load_bug(self, uuid):
255 bg = bug.Bug(bugdir=self, uuid=uuid, from_storage=True)
260 def new_bug(self, summary=None, _uuid=None):
261 bg = bug.Bug(bugdir=self, uuid=_uuid, summary=summary,
265 if hasattr(self, '_uuids_cache') and not bg.uuid in self._uuids_cache:
266 self._uuids_cache.append(bg.uuid)
269 def remove_bug(self, bug):
270 if hasattr(self, '_uuids_cache') and bug.uuid in self._uuids_cache:
271 self._uuids_cache.remove(bug.uuid)
273 if self.storage != None and self.storage.is_writeable():
276 def bug_from_uuid(self, uuid):
277 if not self.has_bug(uuid):
280 'No bug matches %s in %s' % (uuid, self.storage))
281 if self._bug_map[uuid] == None:
283 return self._bug_map[uuid]
285 def has_bug(self, bug_uuid):
286 if bug_uuid not in self._bug_map:
288 if bug_uuid not in self._bug_map:
292 # methods for id generation
294 def sibling_uuids(self):
297 class RevisionedBugDir (BugDir):
299 RevisionedBugDirs are read-only copies used for generating
300 diffs between revisions.
302 def __init__(self, bugdir, revision):
303 storage_version = bugdir.storage.storage_version(revision)
304 if storage_version != libbe.storage.STORAGE_VERSION:
305 raise libbe.storage.InvalidStorageVersion(storage_version)
306 s = copy.deepcopy(bugdir.storage)
308 class RevisionedStorage (object):
309 def __init__(self, storage, default_revision):
311 self.sget = self.s.get
312 self.sancestors = self.s.ancestors
313 self.schildren = self.s.children
314 self.schanged = self.s.changed
315 self.r = default_revision
316 def get(self, *args, **kwargs):
317 if not 'revision' in kwargs or kwargs['revision'] == None:
318 kwargs['revision'] = self.r
319 return self.sget(*args, **kwargs)
320 def ancestors(self, *args, **kwargs):
321 print 'getting ancestors', args, kwargs
322 if not 'revision' in kwargs or kwargs['revision'] == None:
323 kwargs['revision'] = self.r
324 ret = self.sancestors(*args, **kwargs)
325 print 'got ancestors', ret
327 def children(self, *args, **kwargs):
328 if not 'revision' in kwargs or kwargs['revision'] == None:
329 kwargs['revision'] = self.r
330 return self.schildren(*args, **kwargs)
331 def changed(self, *args, **kwargs):
332 if not 'revision' in kwargs or kwargs['revision'] == None:
333 kwargs['revision'] = self.r
334 return self.schanged(*args, **kwargs)
335 rs = RevisionedStorage(s, revision)
337 s.ancestors = rs.ancestors
338 s.children = rs.children
339 s.changed = rs.changed
340 BugDir.__init__(self, s, from_storage=True)
341 self.revision = revision
343 return self.storage.changed()
346 if libbe.TESTING == True:
347 class SimpleBugDir (BugDir):
349 For testing. Set memory=True for a memory-only bugdir.
350 >>> bugdir = SimpleBugDir()
351 >>> uuids = list(bugdir.uuids())
357 def __init__(self, memory=True, versioned=False):
362 self._dir_ref = dir # postpone cleanup since dir.cleanup() removes dir.
363 if versioned == False:
364 storage = libbe.storage.base.Storage(dir.path)
366 storage = libbe.storage.base.VersionedStorage(dir.path)
369 BugDir.__init__(self, storage=storage, uuid='abc123')
370 bug_a = self.new_bug(summary='Bug A', _uuid='a')
371 bug_a.creator = 'John Doe <jdoe@example.com>'
373 bug_b = self.new_bug(summary='Bug B', _uuid='b')
374 bug_b.creator = 'Jane Doe <jdoe@example.com>'
376 bug_b.status = 'closed'
377 if self.storage != None:
378 self.storage.disconnect() # flush to storage
379 self.storage.connect()
382 if self.storage != None:
383 self.storage.writeable = True
384 self.storage.disconnect()
385 self.storage.destroy()
386 if hasattr(self, '_dir_ref'):
387 self._dir_ref.cleanup()
389 def flush_reload(self):
390 if self.storage != None:
391 self.storage.disconnect()
392 self.storage.connect()
395 # class BugDirTestCase(unittest.TestCase):
397 # self.dir = utility.Dir()
398 # self.bugdir = BugDir(self.dir.path, sink_to_existing_root=False,
399 # allow_storage_init=True)
400 # self.storage = self.bugdir.storage
401 # def tearDown(self):
402 # self.bugdir.cleanup()
404 # def fullPath(self, path):
405 # return os.path.join(self.dir.path, path)
406 # def assertPathExists(self, path):
407 # fullpath = self.fullPath(path)
408 # self.failUnless(os.path.exists(fullpath)==True,
409 # "path %s does not exist" % fullpath)
410 # self.assertRaises(AlreadyInitialized, BugDir,
411 # self.dir.path, assertNewBugDir=True)
412 # def versionTest(self):
413 # if self.storage != None and self.storage.versioned == False:
415 # original = self.bugdir.storage.commit("Began versioning")
416 # bugA = self.bugdir.bug_from_uuid("a")
417 # bugA.status = "fixed"
419 # new = self.storage.commit("Fixed bug a")
420 # dupdir = self.bugdir.duplicate_bugdir(original)
421 # self.failUnless(dupdir.root != self.bugdir.root,
422 # "%s, %s" % (dupdir.root, self.bugdir.root))
423 # bugAorig = dupdir.bug_from_uuid("a")
424 # self.failUnless(bugA != bugAorig,
425 # "\n%s\n%s" % (bugA.string(), bugAorig.string()))
426 # bugAorig.status = "fixed"
427 # self.failUnless(bug.cmp_status(bugA, bugAorig)==0,
428 # "%s, %s" % (bugA.status, bugAorig.status))
429 # self.failUnless(bug.cmp_severity(bugA, bugAorig)==0,
430 # "%s, %s" % (bugA.severity, bugAorig.severity))
431 # self.failUnless(bug.cmp_assigned(bugA, bugAorig)==0,
432 # "%s, %s" % (bugA.assigned, bugAorig.assigned))
433 # self.failUnless(bug.cmp_time(bugA, bugAorig)==0,
434 # "%s, %s" % (bugA.time, bugAorig.time))
435 # self.failUnless(bug.cmp_creator(bugA, bugAorig)==0,
436 # "%s, %s" % (bugA.creator, bugAorig.creator))
437 # self.failUnless(bugA == bugAorig,
438 # "\n%s\n%s" % (bugA.string(), bugAorig.string()))
439 # self.bugdir.remove_duplicate_bugdir()
440 # self.failUnless(os.path.exists(dupdir.root)==False,
443 # self.bugdir.new_bug(uuid="a", summary="Ant")
444 # self.bugdir.new_bug(uuid="b", summary="Cockroach")
445 # self.bugdir.new_bug(uuid="c", summary="Praying mantis")
446 # length = len(self.bugdir)
447 # self.failUnless(length == 3, "%d != 3 bugs" % length)
448 # uuids = list(self.bugdir.uuids())
449 # self.failUnless(len(uuids) == 3, "%d != 3 uuids" % len(uuids))
450 # self.failUnless(uuids == ["a","b","c"], str(uuids))
451 # bugA = self.bugdir.bug_from_uuid("a")
452 # bugAprime = self.bugdir.bug_from_shortname("a")
453 # self.failUnless(bugA == bugAprime, "%s != %s" % (bugA, bugAprime))
456 # def testComments(self, sync_with_disk=False):
457 # if sync_with_disk == True:
458 # self.bugdir.set_sync_with_disk(True)
459 # self.bugdir.new_bug(uuid="a", summary="Ant")
460 # bug = self.bugdir.bug_from_uuid("a")
461 # comm = bug.comment_root
462 # rep = comm.new_reply("Ants are small.")
463 # rep.new_reply("And they have six legs.")
464 # if sync_with_disk == False:
466 # self.bugdir.set_sync_with_disk(True)
467 # self.bugdir._clear_bugs()
468 # bug = self.bugdir.bug_from_uuid("a")
469 # bug.load_comments()
470 # if sync_with_disk == False:
471 # self.bugdir.set_sync_with_disk(False)
472 # self.failUnless(len(bug.comment_root)==1, len(bug.comment_root))
473 # for index,comment in enumerate(bug.comments()):
475 # repLoaded = comment
476 # self.failUnless(repLoaded.uuid == rep.uuid, repLoaded.uuid)
477 # self.failUnless(comment.sync_with_disk == sync_with_disk,
478 # comment.sync_with_disk)
479 # self.failUnless(comment.content_type == "text/plain",
480 # comment.content_type)
481 # self.failUnless(repLoaded.settings["Content-type"] == \
483 # repLoaded.settings)
484 # self.failUnless(repLoaded.body == "Ants are small.",
487 # self.failUnless(comment.in_reply_to == repLoaded.uuid,
489 # self.failUnless(comment.body == "And they have six legs.",
493 # "Invalid comment: %d\n%s" % (index, comment))
494 # def testSyncedComments(self):
495 # self.testComments(sync_with_disk=True)
497 class SimpleBugDirTestCase (unittest.TestCase):
499 # create a pre-existing bugdir in a temporary directory
500 self.dir = utility.Dir()
501 self.storage = libbe.storage.base.Storage(self.dir.path)
503 self.storage.connect()
504 self.bugdir = BugDir(self.storage)
505 self.bugdir.new_bug(summary="Hopefully not imported",
507 self.storage.disconnect()
508 self.storage.connect()
510 if self.storage != None:
511 self.storage.disconnect()
512 self.storage.destroy()
514 def testOnDiskCleanLoad(self):
516 SimpleBugDir(memory==False) should not import
519 bugdir = SimpleBugDir(memory=False)
520 self.failUnless(bugdir.storage.is_readable() == True,
521 bugdir.storage.is_readable())
522 self.failUnless(bugdir.storage.is_writeable() == True,
523 bugdir.storage.is_writeable())
524 uuids = sorted([bug.uuid for bug in bugdir])
525 self.failUnless(uuids == ['a', 'b'], uuids)
526 bugdir.flush_reload()
527 uuids = sorted(bugdir.uuids())
528 self.failUnless(uuids == ['a', 'b'], uuids)
529 uuids = sorted([bug.uuid for bug in bugdir])
530 self.failUnless(uuids == [], uuids)
531 bugdir.load_all_bugs()
532 uuids = sorted([bug.uuid for bug in bugdir])
533 self.failUnless(uuids == ['a', 'b'], uuids)
535 def testInMemoryCleanLoad(self):
537 SimpleBugDir(memory==True) should not import
540 bugdir = SimpleBugDir(memory=True)
541 self.failUnless(bugdir.storage == None, bugdir.storage)
542 uuids = sorted([bug.uuid for bug in bugdir])
543 self.failUnless(uuids == ['a', 'b'], uuids)
544 uuids = sorted([bug.uuid for bug in bugdir])
545 self.failUnless(uuids == ['a', 'b'], uuids)
547 uuids = sorted(bugdir.uuids())
548 self.failUnless(uuids == [], uuids)
549 uuids = sorted([bug.uuid for bug in bugdir])
550 self.failUnless(uuids == [], uuids)
553 unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
554 suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
556 # def _get_settings(self, settings_path, for_duplicate_bugdir=False):
557 # allow_no_storage = not self.storage.path_in_root(settings_path)
558 # if allow_no_storage == True:
559 # assert for_duplicate_bugdir == True
560 # if self.sync_with_disk == False and for_duplicate_bugdir == False:
561 # # duplicates can ignore this bugdir's .sync_with_disk status
562 # raise DiskAccessRequired("_get settings")
564 # settings = mapfile.map_load(self.storage, settings_path, allow_no_storage)
565 # except storage.NoSuchFile:
566 # settings = {"storage_name": "None"}
569 # def _save_settings(self, settings_path, settings,
570 # for_duplicate_bugdir=False):
571 # allow_no_storage = not self.storage.path_in_root(settings_path)
572 # if allow_no_storage == True:
573 # assert for_duplicate_bugdir == True
574 # if self.sync_with_disk == False and for_duplicate_bugdir == False:
575 # # duplicates can ignore this bugdir's .sync_with_disk status
576 # raise DiskAccessRequired("_save settings")
577 # self.storage.mkdir(self.get_path(), allow_no_storage)
578 # mapfile.map_save(self.storage, settings_path, settings, allow_no_storage)