-# Copyright (C) 2005-2009 Aaron Bentley and Panometrics, Inc.
+# Copyright (C) 2005-2012 Aaron Bentley <abentley@panoramicfeedback.com>
# Alexander Belchenko <bialix@ukr.net>
# Chris Ball <cjb@laptop.org>
+# Gianluca Montecchi <gian@grys.it>
# Oleg Romanyshyn <oromanyshyn@panoramicfeedback.com>
-# W. Trevor King <wking@drexel.edu>
+# W. Trevor King <wking@tremily.us>
#
-# 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 2 of the License, or
-# (at your option) any later version.
+# This file is part of Bugs Everywhere.
#
-# 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.
+# Bugs Everywhere 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 2 of the License, or (at your option) any
+# later version.
#
-# You should have received a copy of the GNU General Public License along
-# with this program; if not, write to the Free Software Foundation, Inc.,
-# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+# Bugs Everywhere 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
+# Bugs Everywhere. If not, see <http://www.gnu.org/licenses/>.
+
+"""Define :py:class:`BugDir` for storing a collection of bugs.
+"""
+
import copy
import errno
import os
import os.path
-import sys
import time
-import unittest
-import doctest
-
-from properties import Property, doc_property, local_property, \
- defaulting_property, checked_property, fn_checked_property, \
- cached_property, primed_property, change_hook_property, \
- settings_property
-import settings_object
-import mapfile
-import bug
-import rcs
-import encoding
-import utility
-
-
-class NoBugDir(Exception):
- def __init__(self, path):
- msg = "The directory \"%s\" has no bug directory." % path
- Exception.__init__(self, msg)
- self.path = path
-
-class NoRootEntry(Exception):
- def __init__(self, path):
- self.path = path
- Exception.__init__(self, "Specified root does not exist: %s" % path)
-
-class AlreadyInitialized(Exception):
- def __init__(self, path):
- self.path = path
- Exception.__init__(self,
- "Specified root is already initialized: %s" % path)
-
-class MultipleBugMatches(ValueError):
- def __init__(self, shortname, matches):
- msg = ("More than one bug matches %s. "
- "Please be more specific.\n%s" % (shortname, matches))
- ValueError.__init__(self, msg)
- self.shortname = shortname
- self.matches = matches
-
-class DiskAccessRequired (Exception):
- def __init__(self, goal):
- msg = "Cannot %s without accessing the disk" % goal
- Exception.__init__(self, msg)
-
-
-TREE_VERSION_STRING = "Bugs Everywhere Tree 1 0\n"
+import types
+try: # import core module, Python >= 2.5
+ from xml.etree import ElementTree
+except ImportError: # look for non-core module
+ from elementtree import ElementTree
+import xml.sax.saxutils
+
+import libbe
+import libbe.storage as storage
+from libbe.storage.util.properties import Property, doc_property, \
+ local_property, defaulting_property, checked_property, \
+ fn_checked_property, cached_property, primed_property, \
+ change_hook_property, settings_property
+import libbe.storage.util.settings_object as settings_object
+import libbe.storage.util.mapfile as mapfile
+import libbe.bug as bug
+import libbe.util.utility as utility
+import libbe.util.id
+
+if libbe.TESTING == True:
+ import doctest
+ import sys
+ import unittest
+
+ import libbe.storage.base
+
+
+class NoBugMatches(libbe.util.id.NoIDMatches):
+ def __init__(self, *args, **kwargs):
+ libbe.util.id.NoIDMatches.__init__(self, *args, **kwargs)
+ def __str__(self):
+ if self.msg == None:
+ return 'No bug matches %s' % self.id
+ return self.msg
class BugDir (list, settings_object.SavedSettingsObject):
- """
- Sink to existing root
- ======================
-
- Consider the following usage case:
- You have a bug directory rooted in
- /path/to/source
- by which I mean the '.be' directory is at
- /path/to/source/.be
- However, you're of in some subdirectory like
- /path/to/source/GUI/testing
- and you want to comment on a bug. Setting sink_to_root=True wen
- you initialize your BugDir will cause it to search for the '.be'
- file in the ancestors of the path you passed in as 'root'.
- /path/to/source/GUI/testing/.be miss
- /path/to/source/GUI/.be miss
- /path/to/source/.be hit!
- So it still roots itself appropriately without much work for you.
-
- File-system access
- ==================
-
- BugDirs live completely in memory when .sync_with_disk is False.
- This is the default configuration setup by BugDir(from_disk=False).
- If .sync_with_disk == True (e.g. BugDir(from_disk=True)), then
- any changes to the BugDir will be immediately written to disk.
-
- If you want to change .sync_with_disk, we suggest you use
- .set_sync_with_disk(), which propogates the new setting through to
- all bugs/comments/etc. that have been loaded into memory. If
- you've been living in memory and want to move to
- .sync_with_disk==True, but you're not sure if anything has been
- changed in memory, a call to save() immediately before the
- .set_sync_with_disk(True) call is a safe move.
-
- Regardless of .sync_with_disk, a call to .save() will write out
- all the contents that the BugDir instance has loaded into memory.
- If sync_with_disk has been True over the course of all interesting
- changes, this .save() call will be a waste of time.
-
- The BugDir will only load information from the file system when it
- loads new settings/bugs/comments that it doesn't already have in
- memory and .sync_with_disk == True.
-
- Allow RCS initialization
- ========================
-
- This one is for testing purposes. Setting it to True allows the
- BugDir to search for an installed RCS backend and initialize it in
- the root directory. This is a convenience option for supporting
- tests of versioning functionality (e.g. .duplicate_bugdir).
-
- Disable encoding manipulation
- =============================
-
- This one is for testing purposed. You might have non-ASCII
- Unicode in your bugs, comments, files, etc. BugDir instances try
- and support your preferred encoding scheme (e.g. "utf-8") when
- dealing with stream and file input/output. For stream output,
- this involves replacing sys.stdout and sys.stderr
- (libbe.encode.set_IO_stream_encodings). However this messes up
- doctest's output catching. In order to support doctest tests
- using BugDirs, set manipulate_encodings=False, and stick to ASCII
- in your tests.
+ """A BugDir is a container for :py:class:`~libbe.bug.Bug`\s, with some
+ additional attributes.
+
+ Parameters
+ ----------
+ storage : :py:class:`~libbe.storage.base.Storage`
+ Storage instance containing the bug directory. If
+ `from_storage` is `False`, `storage` may be `None`.
+ uuid : str, optional
+ Set the bugdir UUID (see :py:mod:`libbe.util.id`).
+ Useful if you are loading one of several bugdirs
+ stored in a single Storage instance.
+ from_storage : bool, optional
+ If `True`, attempt to load from storage. Otherwise,
+ setup in memory, saving to `storage` if it is not `None`.
+
+ See Also
+ --------
+ SimpleBugDir : bugdir manipulation exampes.
"""
settings_properties = []
doc="The current project development target.")
def target(): return {}
- def _guess_encoding(self):
- return encoding.get_encoding()
- def _check_encoding(value):
- if value != None:
- return encoding.known_encoding(value)
- def _setup_encoding(self, new_encoding):
- # change hook called before generator.
- if new_encoding not in [None, settings_object.EMPTY]:
- if self._manipulate_encodings == True:
- encoding.set_IO_stream_encodings(new_encoding)
- def _set_encoding(self, old_encoding, new_encoding):
- self._setup_encoding(new_encoding)
- self._prop_save_settings(old_encoding, new_encoding)
-
- @_versioned_property(name="encoding",
- doc="""The default input/output encoding to use (e.g. "utf-8").""",
- change_hook=_set_encoding,
- generator=_guess_encoding,
- check_fn=_check_encoding)
- def encoding(): return {}
-
- def _setup_user_id(self, user_id):
- self.rcs.user_id = user_id
- def _guess_user_id(self):
- return self.rcs.get_user_id()
- def _set_user_id(self, old_user_id, new_user_id):
- self._setup_user_id(new_user_id)
- self._prop_save_settings(old_user_id, new_user_id)
-
- @_versioned_property(name="user_id",
- doc=
-"""The user's prefered name, e.g. 'John Doe <jdoe@example.com>'. Note
-that the Arch RCS backend *enforces* ids with this format.""",
- change_hook=_set_user_id,
- generator=_guess_user_id)
- def user_id(): return {}
-
- @_versioned_property(name="default_assignee",
- doc=
-"""The default assignee for new bugs e.g. 'John Doe <jdoe@example.com>'.""")
- def default_assignee(): return {}
-
- @_versioned_property(name="rcs_name",
- doc="""The name of the current RCS. Kept seperate to make saving/loading
-settings easy. Don't set this attribute. Set .rcs instead, and
-.rcs_name will be automatically adjusted.""",
- default="None",
- allowed=["None", "Arch", "bzr", "darcs", "git", "hg"])
- def rcs_name(): return {}
-
- def _get_rcs(self, rcs_name=None):
- """Get and root a new revision control system"""
- if rcs_name == None:
- rcs_name = self.rcs_name
- new_rcs = rcs.rcs_by_name(rcs_name)
- self._change_rcs(None, new_rcs)
- return new_rcs
- def _change_rcs(self, old_rcs, new_rcs):
- new_rcs.encoding = self.encoding
- new_rcs.root(self.root)
- self.rcs_name = new_rcs.name
-
- @Property
- @change_hook_property(hook=_change_rcs)
- @cached_property(generator=_get_rcs)
- @local_property("rcs")
- @doc_property(doc="A revision control system instance.")
- def rcs(): return {}
-
- def _bug_map_gen(self):
- map = {}
- for bug in self:
- map[bug.uuid] = bug
- for uuid in self.list_uuids():
- if uuid not in map:
- map[uuid] = None
- self._bug_map_value = map # ._bug_map_value used by @local_property
-
- def _extra_strings_check_fn(value):
- return utility.iterable_full_of_strings(value, \
- alternative=settings_object.EMPTY)
- def _extra_strings_change_hook(self, old, new):
- self.extra_strings.sort() # to make merging easier
- self._prop_save_settings(old, new)
- @_versioned_property(name="extra_strings",
- doc="Space for an array of extra strings. Useful for storing state for functionality implemented purely in becommands/<some_function>.py.",
- default=[],
- check_fn=_extra_strings_check_fn,
- change_hook=_extra_strings_change_hook,
- mutable=True)
- def extra_strings(): return {}
-
- @Property
- @primed_property(primer=_bug_map_gen)
- @local_property("bug_map")
- @doc_property(doc="A dict of (bug-uuid, bug-instance) pairs.")
- def _bug_map(): return {}
-
def _setup_severities(self, severities):
if severities not in [None, settings_object.EMPTY]:
bug.load_severities(severities)
change_hook=_set_inactive_status)
def inactive_status(): return {}
+ def _extra_strings_check_fn(value):
+ return utility.iterable_full_of_strings(value, \
+ alternative=settings_object.EMPTY)
+ def _extra_strings_change_hook(self, old, new):
+ self.extra_strings.sort() # to make merging easier
+ self._prop_save_settings(old, new)
+ @_versioned_property(name="extra_strings",
+ doc="Space for an array of extra strings. Useful for storing state for functionality implemented purely in becommands/<some_function>.py.",
+ default=[],
+ check_fn=_extra_strings_check_fn,
+ change_hook=_extra_strings_change_hook,
+ mutable=True)
+ def extra_strings(): return {}
+
+ def _bug_map_gen(self):
+ map = {}
+ for bug in self:
+ map[bug.uuid] = bug
+ for uuid in self.uuids():
+ if uuid not in map:
+ map[uuid] = None
+ self._bug_map_value = map # ._bug_map_value used by @local_property
- def __init__(self, root=None, sink_to_existing_root=True,
- assert_new_BugDir=False, allow_rcs_init=False,
- manipulate_encodings=True, from_disk=False, rcs=None):
+ @Property
+ @primed_property(primer=_bug_map_gen)
+ @local_property("bug_map")
+ @doc_property(doc="A dict of (bug-uuid, bug-instance) pairs.")
+ def _bug_map(): return {}
+
+ def __init__(self, storage, uuid=None, from_storage=False):
list.__init__(self)
settings_object.SavedSettingsObject.__init__(self)
- self._manipulate_encodings = manipulate_encodings
- if root == None:
- root = os.getcwd()
- if sink_to_existing_root == True:
- self.root = self._find_root(root)
- else:
- if not os.path.exists(root):
- raise NoRootEntry(root)
- self.root = root
- # get a temporary rcs until we've loaded settings
- self.sync_with_disk = False
- self.rcs = self._guess_rcs()
-
- if from_disk == True:
- self.sync_with_disk = True
- self.load()
- else:
- self.sync_with_disk = False
- if assert_new_BugDir == True:
- if os.path.exists(self.get_path()):
- raise AlreadyInitialized, self.get_path()
- if rcs == None:
- rcs = self._guess_rcs(allow_rcs_init)
- self.rcs = rcs
- self._setup_user_id(self.user_id)
-
- # methods for getting the BugDir situated in the filesystem
-
- def _find_root(self, path):
- """
- Search for an existing bug database dir and it's ancestors and
- return a BugDir rooted there. Only called by __init__, and
- then only if sink_to_existing_root == True.
- """
- if not os.path.exists(path):
- raise NoRootEntry(path)
- versionfile=utility.search_parent_directories(path,
- os.path.join(".be", "version"))
- if versionfile != None:
- beroot = os.path.dirname(versionfile)
- root = os.path.dirname(beroot)
- return root
+ self.storage = storage
+ self.id = libbe.util.id.ID(self, 'bugdir')
+ self.uuid = uuid
+ if from_storage == True:
+ if self.uuid == None:
+ self.uuid = [c for c in self.storage.children()
+ if c != 'version'][0]
+ self.load_settings()
else:
- beroot = utility.search_parent_directories(path, ".be")
- if beroot == None:
- raise NoBugDir(path)
- return beroot
-
- def _guess_rcs(self, allow_rcs_init=False):
- """
- Only called by __init__.
- """
- deepdir = self.get_path()
- if not os.path.exists(deepdir):
- deepdir = os.path.dirname(deepdir)
- new_rcs = rcs.detect_rcs(deepdir)
- install = False
- if new_rcs.name == "None":
- if allow_rcs_init == True:
- new_rcs = rcs.installed_rcs()
- new_rcs.init(self.root)
- return new_rcs
-
- def get_path(self, *args):
- """
- Return a path relative to .root.
- """
- my_dir = os.path.join(self.root, ".be")
- if len(args) == 0:
- return my_dir
- assert args[0] in ["version", "settings", "bugs"], str(args)
- return os.path.join(my_dir, *args)
+ if self.uuid == None:
+ self.uuid = libbe.util.id.uuid_gen()
+ if self.storage != None and self.storage.is_writeable():
+ self.save()
# methods for saving/loading/accessing settings and properties.
- def _get_settings(self, settings_path, for_duplicate_bugdir=False):
- allow_no_rcs = not self.rcs.path_in_root(settings_path)
- if allow_no_rcs == True:
- assert for_duplicate_bugdir == True
- if self.sync_with_disk == False and for_duplicate_bugdir == False:
- # duplicates can ignore this bugdir's .sync_with_disk status
- raise DiskAccessRequired("_get settings")
+ def load_settings(self, settings_mapfile=None):
+ if settings_mapfile == None:
+ settings_mapfile = \
+ self.storage.get(self.id.storage('settings'), default='{}\n')
try:
- settings = mapfile.map_load(self.rcs, settings_path, allow_no_rcs)
- except rcs.NoSuchFile:
- settings = {"rcs_name": "None"}
- return settings
-
- def _save_settings(self, settings_path, settings,
- for_duplicate_bugdir=False):
- allow_no_rcs = not self.rcs.path_in_root(settings_path)
- if allow_no_rcs == True:
- assert for_duplicate_bugdir == True
- if self.sync_with_disk == False and for_duplicate_bugdir == False:
- # duplicates can ignore this bugdir's .sync_with_disk status
- raise DiskAccessRequired("_save settings")
- self.rcs.mkdir(self.get_path(), allow_no_rcs)
- mapfile.map_save(self.rcs, settings_path, settings, allow_no_rcs)
-
- def load_settings(self):
- self.settings = self._get_settings(self.get_path("settings"))
- self._setup_saved_settings()
- self._setup_user_id(self.user_id)
- self._setup_encoding(self.encoding)
+ settings = mapfile.parse(settings_mapfile)
+ except mapfile.InvalidMapfileContents, e:
+ raise Exception('Invalid settings file for bugdir %s\n'
+ '(BE version missmatch?)' % self.id.user())
+ self._setup_saved_settings(settings)
self._setup_severities(self.severities)
self._setup_status(self.active_status, self.inactive_status)
- self.rcs = rcs.rcs_by_name(self.rcs_name)
- self._setup_user_id(self.user_id)
def save_settings(self):
- settings = self._get_saved_settings()
- self._save_settings(self.get_path("settings"), settings)
-
- def get_version(self, path=None, use_none_rcs=False):
- """
- Requires disk access.
- """
- if self.sync_with_disk == False:
- raise DiskAccessRequired("get version")
- if use_none_rcs == True:
- RCS = rcs.rcs_by_name("None")
- RCS.root(self.root)
- RCS.encoding = encoding.get_encoding()
- else:
- RCS = self.rcs
-
- if path == None:
- path = self.get_path("version")
- tree_version = RCS.get_file_contents(path)
- return tree_version
-
- def set_version(self):
- """
- Requires disk access.
- """
- if self.sync_with_disk == False:
- raise DiskAccessRequired("set version")
- self.rcs.mkdir(self.get_path())
- self.rcs.set_file_contents(self.get_path("version"),
- TREE_VERSION_STRING)
-
- # methods controlling disk access
-
- def set_sync_with_disk(self, value):
- """
- Adjust .sync_with_disk for the BugDir and all it's children.
- See the BugDir docstring for a description of the role of
- .sync_with_disk.
- """
- self.sync_with_disk = value
- for bug in self:
- bug.set_sync_with_disk(value)
-
- def load(self):
- """
- Reqires disk access
- """
- version = self.get_version(use_none_rcs=True)
- if version != TREE_VERSION_STRING:
- raise NotImplementedError, \
- "BugDir cannot handle version '%s' yet." % version
- else:
- if not os.path.exists(self.get_path()):
- raise NoBugDir(self.get_path())
- self.load_settings()
+ mf = mapfile.generate(self._get_saved_settings())
+ self.storage.set(self.id.storage('settings'), mf)
def load_all_bugs(self):
"""
- Requires disk access.
Warning: this could take a while.
"""
- if self.sync_with_disk == False:
- raise DiskAccessRequired("load all bugs")
self._clear_bugs()
- for uuid in self.list_uuids():
+ for uuid in self.uuids():
self._load_bug(uuid)
def save(self):
"""
- Note that this command writes to disk _regardless_ of the
- status of .sync_with_disk.
+ Save any loaded contents to storage. Because of lazy loading
+ of bugs and comments, this is actually not too inefficient.
- Save any loaded contents to disk. Because of lazy loading of
- bugs and comments, this is actually not too inefficient.
-
- However, if .sync_with_disk = True, then any changes are
- automatically written to disk as soon as they happen, so
- calling this method will just waste time (unless something
- else has been messing with your on-disk files).
-
- Requires disk access.
+ However, if self.storage.is_writeable() == True, then any
+ changes are automatically written to storage as soon as they
+ happen, so calling this method will just waste time (unless
+ something else has been messing with your stored files).
"""
- sync_with_disk = self.sync_with_disk
- if sync_with_disk == False:
- self.set_sync_with_disk(True)
- self.set_version()
+ self.storage.add(self.id.storage(), directory=True)
+ self.storage.add(self.id.storage('settings'), parent=self.id.storage(),
+ directory=False)
self.save_settings()
for bug in self:
+ bug.bugdir = self
+ bug.storage = self.storage
bug.save()
- if sync_with_disk == False:
- self.set_sync_with_disk(sync_with_disk)
-
- # methods for managing duplicate BugDirs
-
- def duplicate_bugdir(self, revision):
- duplicate_path = self.rcs.duplicate_repo(revision)
-
- # setup revision RCS as None, since the duplicate may not be
- # initialized for versioning
- duplicate_settings_path = os.path.join(duplicate_path,
- ".be", "settings")
- duplicate_settings = self._get_settings(duplicate_settings_path,
- for_duplicate_bugdir=True)
- if "rcs_name" in duplicate_settings:
- duplicate_settings["rcs_name"] = "None"
- duplicate_settings["user_id"] = self.user_id
- if "disabled" in bug.status_values:
- # Hack to support old versions of BE bugs
- duplicate_settings["inactive_status"] = self.inactive_status
- self._save_settings(duplicate_settings_path, duplicate_settings,
- for_duplicate_bugdir=True)
-
- return BugDir(duplicate_path, from_disk=True, manipulate_encodings=self._manipulate_encodings)
-
- def remove_duplicate_bugdir(self):
- self.rcs.remove_duplicate_repo()
# methods for managing bugs
- def list_uuids(self):
- uuids = []
- if self.sync_with_disk == True and os.path.exists(self.get_path()):
- # list the uuids on disk
- for uuid in os.listdir(self.get_path("bugs")):
- if not (uuid.startswith('.')):
- uuids.append(uuid)
- yield uuid
- # and the ones that are still just in memory
- for bug in self:
- if bug.uuid not in uuids:
- uuids.append(bug.uuid)
- yield bug.uuid
+ def uuids(self, use_cached_disk_uuids=True):
+ if use_cached_disk_uuids==False or not hasattr(self, '_uuids_cache'):
+ self._refresh_uuid_cache()
+ self._uuids_cache = self._uuids_cache.union([bug.uuid for bug in self])
+ return self._uuids_cache
+
+ def _refresh_uuid_cache(self):
+ self._uuids_cache = set()
+ # list bugs that are in storage
+ if self.storage != None and self.storage.is_readable():
+ child_uuids = libbe.util.id.child_uuids(
+ self.storage.children(self.id.storage()))
+ for id in child_uuids:
+ self._uuids_cache.add(id)
def _clear_bugs(self):
while len(self) > 0:
self.pop()
+ if hasattr(self, '_uuids_cache'):
+ del(self._uuids_cache)
self._bug_map_gen()
def _load_bug(self, uuid):
- if self.sync_with_disk == False:
- raise DiskAccessRequired("_load bug")
- bg = bug.Bug(bugdir=self, uuid=uuid, from_disk=True)
+ bg = bug.Bug(bugdir=self, uuid=uuid, from_storage=True)
self.append(bg)
self._bug_map_gen()
return bg
- def new_bug(self, uuid=None, summary=None):
- bg = bug.Bug(bugdir=self, uuid=uuid, summary=summary)
- bg.set_sync_with_disk(self.sync_with_disk)
- if bg.sync_with_disk == True:
- bg.save()
- self.append(bg)
- self._bug_map_gen()
+ def new_bug(self, summary=None, _uuid=None):
+ bg = bug.Bug(bugdir=self, uuid=_uuid, summary=summary,
+ from_storage=False)
+ self.append(bg, update=True)
return bg
+ def append(self, bug, update=False):
+ super(BugDir, self).append(bug)
+ if update:
+ bug.bugdir = self
+ bug.storage = self.storage
+ self._bug_map_gen()
+ if (hasattr(self, '_uuids_cache') and
+ not bug.uuid in self._uuids_cache):
+ self._uuids_cache.add(bug.uuid)
+
def remove_bug(self, bug):
+ if hasattr(self, '_uuids_cache') and bug.uuid in self._uuids_cache:
+ self._uuids_cache.remove(bug.uuid)
self.remove(bug)
- bug.remove()
-
- def bug_shortname(self, bug):
- """
- Generate short names from uuids. Picks the minimum number of
- characters (>=3) from the beginning of the uuid such that the
- short names are unique.
-
- Obviously, as the number of bugs in the database grows, these
- short names will cease to be unique. The complete uuid should be
- used for long term reference.
- """
- chars = 3
- for uuid in self._bug_map.keys():
- if bug.uuid == uuid:
- continue
- while (bug.uuid[:chars] == uuid[:chars]):
- chars+=1
- return bug.uuid[:chars]
-
- def bug_from_shortname(self, shortname):
- """
- >>> bd = simple_bug_dir(sync_with_disk=False)
- >>> bug_a = bd.bug_from_shortname('a')
- >>> print type(bug_a)
- <class 'libbe.bug.Bug'>
- >>> print bug_a
- a:om: Bug A
- """
- matches = []
- self._bug_map_gen()
- for uuid in self._bug_map.keys():
- if uuid.startswith(shortname):
- matches.append(uuid)
- if len(matches) > 1:
- raise MultipleBugMatches(shortname, matches)
- if len(matches) == 1:
- return self.bug_from_uuid(matches[0])
- raise KeyError("No bug matches %s" % shortname)
+ if self.storage != None and self.storage.is_writeable():
+ bug.remove()
def bug_from_uuid(self, uuid):
if not self.has_bug(uuid):
- raise KeyError("No bug matches %s\n bug map: %s\n root: %s" \
- % (uuid, self._bug_map, self.root))
+ raise NoBugMatches(
+ uuid, self.uuids(),
+ 'No bug matches %s in %s' % (uuid, self.storage))
if self._bug_map[uuid] == None:
self._load_bug(uuid)
return self._bug_map[uuid]
return False
return True
+ def xml(self, indent=0, show_bugs=False, show_comments=False):
+ """
+ >>> bug.load_severities(bug.severity_def)
+ >>> bug.load_status(
+ ... active_status_def=bug.active_status_def,
+ ... inactive_status_def=bug.inactive_status_def)
+ >>> bugdirA = SimpleBugDir(memory=True)
+ >>> bugdirA.severities
+ >>> bugdirA.severities = (('minor', 'The standard bug level.'),)
+ >>> bugdirA.inactive_status = (
+ ... ('closed', 'The bug is no longer relevant.'),)
+ >>> bugA = bugdirA.bug_from_uuid('a')
+ >>> commA = bugA.comment_root.new_reply(body='comment A')
+ >>> commA.uuid = 'commA'
+ >>> commA.date = 'Thu, 01 Jan 1970 00:03:00 +0000'
+ >>> print(bugdirA.xml(show_bugs=True, show_comments=True))
+ ... # doctest: +REPORT_UDIFF
+ <bugdir>
+ <uuid>abc123</uuid>
+ <short-name>abc</short-name>
+ <severities>
+ <entry>
+ <key>minor</key>
+ <value>The standard bug level.</value>
+ </entry>
+ </severities>
+ <inactive-status>
+ <entry>
+ <key>closed</key>
+ <value>The bug is no longer relevant.</value>
+ </entry>
+ </inactive-status>
+ <bug>
+ <uuid>a</uuid>
+ <short-name>abc/a</short-name>
+ <severity>minor</severity>
+ <status>open</status>
+ <creator>John Doe <jdoe@example.com></creator>
+ <created>Thu, 01 Jan 1970 00:00:00 +0000</created>
+ <summary>Bug A</summary>
+ <comment>
+ <uuid>commA</uuid>
+ <short-name>abc/a/com</short-name>
+ <author></author>
+ <date>Thu, 01 Jan 1970 00:03:00 +0000</date>
+ <content-type>text/plain</content-type>
+ <body>comment A</body>
+ </comment>
+ </bug>
+ <bug>
+ <uuid>b</uuid>
+ <short-name>abc/b</short-name>
+ <severity>minor</severity>
+ <status>closed</status>
+ <creator>Jane Doe <jdoe@example.com></creator>
+ <created>Thu, 01 Jan 1970 00:00:00 +0000</created>
+ <summary>Bug B</summary>
+ </bug>
+ </bugdir>
+ >>> bug.load_severities(bug.severity_def)
+ >>> bug.load_status(
+ ... active_status_def=bug.active_status_def,
+ ... inactive_status_def=bug.inactive_status_def)
+ >>> bugdirA.cleanup()
+ """
+ info = [('uuid', self.uuid),
+ ('short-name', self.id.user()),
+ ('target', self.target),
+ ('severities', self.severities),
+ ('active-status', self.active_status),
+ ('inactive-status', self.inactive_status),
+ ]
+ lines = ['<bugdir>']
+ for (k,v) in info:
+ if v is not None:
+ if k in ['severities', 'active-status', 'inactive-status']:
+ lines.append(' <{}>'.format(k))
+ for vk,vv in v:
+ lines.extend([
+ ' <entry>',
+ ' <key>{}</key>'.format(
+ xml.sax.saxutils.escape(vk)),
+ ' <value>{}</value>'.format(
+ xml.sax.saxutils.escape(vv)),
+ ' </entry>',
+ ])
+ lines.append(' </{}>'.format(k))
+ else:
+ v = xml.sax.saxutils.escape(v)
+ lines.append(' <{0}>{1}</{0}>'.format(k, v))
+ for estr in self.extra_strings:
+ lines.append(' <extra-string>{}</extra-string>'.format(estr))
+ if show_bugs:
+ for bug in self:
+ bug_xml = bug.xml(indent=indent+2, show_comments=show_comments)
+ if bug_xml:
+ bug_xml = bug_xml[indent:] # strip leading indent spaces
+ lines.append(bug_xml)
+ lines.append('</bugdir>')
+ istring = ' '*indent
+ sep = '\n' + istring
+ return istring + sep.join(lines).rstrip('\n')
+
+ def from_xml(self, xml_string, preserve_uuids=False):
+ """
+ Note: If a bugdir uuid is given, set .alt_id to it's value.
+ >>> bug.load_severities(bug.severity_def)
+ >>> bug.load_status(
+ ... active_status_def=bug.active_status_def,
+ ... inactive_status_def=bug.inactive_status_def)
+ >>> bugdirA = SimpleBugDir(memory=True)
+ >>> bugdirA.severities = (('minor', 'The standard bug level.'),)
+ >>> bugdirA.inactive_status = (
+ ... ('closed', 'The bug is no longer relevant.'),)
+ >>> bugA = bugdirA.bug_from_uuid('a')
+ >>> commA = bugA.comment_root.new_reply(body='comment A')
+ >>> commA.uuid = 'commA'
+ >>> xml = bugdirA.xml(show_bugs=True, show_comments=True)
+ >>> bugdirB = BugDir(storage=None)
+ >>> bugdirB.from_xml(xml)
+ >>> bugdirB.xml(show_bugs=True, show_comments=True) == xml
+ False
+ >>> bugdirB.uuid = bugdirB.alt_id
+ >>> for bug_ in bugdirB:
+ ... bug_.uuid = bug_.alt_id
+ ... bug_.alt_id = None
+ ... for comm in bug_.comments():
+ ... comm.uuid = comm.alt_id
+ ... comm.alt_id = None
+ >>> bugdirB.xml(show_bugs=True, show_comments=True) == xml
+ True
+ >>> bugdirB.explicit_attrs # doctest: +NORMALIZE_WHITESPACE
+ ['severities', 'inactive_status']
+ >>> bugdirC = BugDir(storage=None)
+ >>> bugdirC.from_xml(xml, preserve_uuids=True)
+ >>> bugdirC.uuid == bugdirA.uuid
+ True
+ >>> bugdirC.xml(show_bugs=True, show_comments=True) == xml
+ True
+ >>> bug.load_severities(bug.severity_def)
+ >>> bug.load_status(
+ ... active_status_def=bug.active_status_def,
+ ... inactive_status_def=bug.inactive_status_def)
+ >>> bugdirA.cleanup()
+ """
+ if type(xml_string) == types.UnicodeType:
+ xml_string = xml_string.strip().encode('unicode_escape')
+ if hasattr(xml_string, 'getchildren'): # already an ElementTree Element
+ bugdir = xml_string
+ else:
+ bugdir = ElementTree.XML(xml_string)
+ if bugdir.tag != 'bugdir':
+ raise utility.InvalidXML(
+ 'bugdir', bugdir, 'root element must be <bugdir>')
+ tags = ['uuid', 'short-name', 'target', 'severities', 'active-status',
+ 'inactive-status', 'extra-string']
+ self.explicit_attrs = []
+ uuid = None
+ estrs = []
+ for child in bugdir.getchildren():
+ if child.tag == 'short-name':
+ pass
+ elif child.tag == 'bug':
+ bg = bug.Bug(bugdir=self)
+ bg.from_xml(child, preserve_uuids=preserve_uuids)
+ self.append(bg, update=True)
+ continue
+ elif child.tag in tags:
+ if child.text == None or len(child.text) == 0:
+ text = settings_object.EMPTY
+ elif child.tag in ['severities', 'active-status',
+ 'inactive-status']:
+ entries = []
+ for entry in child.getchildren():
+ if entry.tag != 'entry':
+ raise utility.InvalidXML(
+ '{} child element {} must be <entry>'.format(
+ child.tag, entry))
+ key = value = None
+ for kv in entry.getchildren():
+ if kv.tag == 'key':
+ if key is not None:
+ raise utility.InvalidXML(
+ ('duplicate keys ({} and {}) in {}'
+ ).format(key, kv.text, child.tag))
+ key = xml.sax.saxutils.unescape(kv.text)
+ elif kv.tag == 'value':
+ if value is not None:
+ raise utility.InvalidXML(
+ ('duplicate values ({} and {}) in {}'
+ ).format(
+ value, kv.text, child.tag))
+ value = xml.sax.saxutils.unescape(kv.text)
+ else:
+ raise utility.InvalidXML(
+ ('{} child element {} must be <key> or '
+ '<value>').format(child.tag, kv))
+ if key is None:
+ raise utility.InvalidXML(
+ 'no key for {}'.format(child.tag))
+ if value is None:
+ raise utility.InvalidXML(
+ 'no key for {}'.format(child.tag))
+ entries.append((key, value))
+ text = entries
+ else:
+ text = xml.sax.saxutils.unescape(child.text)
+ if not isinstance(text, unicode):
+ text = text.decode('unicode_escape')
+ text = text.strip()
+ if child.tag == 'uuid' and not preserve_uuids:
+ uuid = text
+ continue # don't set the bug's uuid tag.
+ elif child.tag == 'extra-string':
+ estrs.append(text)
+ continue # don't set the bug's extra_string yet.
+ attr_name = child.tag.replace('-','_')
+ self.explicit_attrs.append(attr_name)
+ setattr(self, attr_name, text)
+ else:
+ libbe.LOG.warning(
+ 'ignoring unknown tag {0} in {1}'.format(
+ child.tag, bugdir.tag))
+ if uuid != self.uuid:
+ if not hasattr(self, 'alt_id') or self.alt_id == None:
+ self.alt_id = uuid
+ self.extra_strings = estrs
+
+ def merge(self, other, accept_changes=True,
+ accept_extra_strings=True, accept_bugs=True,
+ accept_comments=True, change_exception=False):
+ """Merge info from other into this bugdir.
+
+ Overrides any attributes in self that are listed in
+ other.explicit_attrs.
+
+ >>> bugdirA = SimpleBugDir()
+ >>> bugdirA.extra_strings += ['TAG: favorite']
+ >>> bugdirB = SimpleBugDir()
+ >>> bugdirB.explicit_attrs = ['target']
+ >>> bugdirB.target = '1234'
+ >>> bugdirB.extra_strings += ['TAG: very helpful']
+ >>> bugdirB.extra_strings += ['TAG: useful']
+ >>> bugA = bugdirB.bug_from_uuid('a')
+ >>> commA = bugA.comment_root.new_reply(body='comment A')
+ >>> commA.uuid = 'uuid-commA'
+ >>> commA.date = 'Thu, 01 Jan 1970 00:01:00 +0000'
+ >>> bugC = bugdirB.new_bug(summary='bug C', _uuid='c')
+ >>> bugC.alt_id = 'alt-c'
+ >>> bugC.time_string = 'Thu, 01 Jan 1970 00:02:00 +0000'
+ >>> bugdirA.merge(
+ ... bugdirB, accept_changes=False, accept_extra_strings=False,
+ ... accept_bugs=False, change_exception=False)
+ >>> print(bugdirA.target)
+ None
+ >>> bugdirA.merge(
+ ... bugdirB, accept_changes=False, accept_extra_strings=False,
+ ... accept_bugs=False, change_exception=True)
+ Traceback (most recent call last):
+ ...
+ ValueError: Merge would change target "None"->"1234" for bugdir abc123
+ >>> print(bugdirA.target)
+ None
+ >>> bugdirA.merge(
+ ... bugdirB, accept_changes=True, accept_extra_strings=False,
+ ... accept_bugs=False, change_exception=True)
+ Traceback (most recent call last):
+ ...
+ ValueError: Merge would add extra string "TAG: useful" for bugdir abc123
+ >>> print(bugdirA.target)
+ 1234
+ >>> print(bugdirA.extra_strings)
+ ['TAG: favorite']
+ >>> bugdirA.merge(
+ ... bugdirB, accept_changes=True, accept_extra_strings=True,
+ ... accept_bugs=False, change_exception=True)
+ Traceback (most recent call last):
+ ...
+ ValueError: Merge would add bug c (alt: alt-c) to bugdir abc123
+ >>> print(bugdirA.extra_strings)
+ ['TAG: favorite', 'TAG: useful', 'TAG: very helpful']
+ >>> bugdirA.merge(
+ ... bugdirB, accept_changes=True, accept_extra_strings=True,
+ ... accept_bugs=True, change_exception=True)
+ >>> print(bugdirA.xml(show_bugs=True, show_comments=True))
+ ... # doctest: +ELLIPSIS, +REPORT_UDIFF
+ <bugdir>
+ <uuid>abc123</uuid>
+ <short-name>abc</short-name>
+ <target>1234</target>
+ <extra-string>TAG: favorite</extra-string>
+ <extra-string>TAG: useful</extra-string>
+ <extra-string>TAG: very helpful</extra-string>
+ <bug>
+ <uuid>a</uuid>
+ <short-name>abc/a</short-name>
+ <severity>minor</severity>
+ <status>open</status>
+ <creator>John Doe <jdoe@example.com></creator>
+ <created>Thu, 01 Jan 1970 00:00:00 +0000</created>
+ <summary>Bug A</summary>
+ <comment>
+ <uuid>uuid-commA</uuid>
+ <short-name>abc/a/uui</short-name>
+ <author></author>
+ <date>Thu, 01 Jan 1970 00:01:00 +0000</date>
+ <content-type>text/plain</content-type>
+ <body>comment A</body>
+ </comment>
+ </bug>
+ <bug>
+ <uuid>b</uuid>
+ <short-name>abc/b</short-name>
+ <severity>minor</severity>
+ <status>closed</status>
+ <creator>Jane Doe <jdoe@example.com></creator>
+ <created>Thu, 01 Jan 1970 00:00:00 +0000</created>
+ <summary>Bug B</summary>
+ </bug>
+ <bug>
+ <uuid>c</uuid>
+ <short-name>abc/c</short-name>
+ <severity>minor</severity>
+ <status>open</status>
+ <created>Thu, 01 Jan 1970 00:02:00 +0000</created>
+ <summary>bug C</summary>
+ </bug>
+ </bugdir>
+ >>> bugdirA.cleanup()
+ >>> bugdirB.cleanup()
+ """
+ if hasattr(other, 'explicit_attrs'):
+ for attr in other.explicit_attrs:
+ old = getattr(self, attr)
+ new = getattr(other, attr)
+ if old != new:
+ if accept_changes:
+ setattr(self, attr, new)
+ elif change_exception:
+ raise ValueError(
+ ('Merge would change {} "{}"->"{}" for bugdir {}'
+ ).format(attr, old, new, self.uuid))
+ for estr in other.extra_strings:
+ if not estr in self.extra_strings:
+ if accept_extra_strings:
+ self.extra_strings += [estr]
+ elif change_exception:
+ raise ValueError(
+ ('Merge would add extra string "{}" for bugdir {}'
+ ).format(estr, self.uuid))
+ for o_bug in other:
+ try:
+ s_bug = self.bug_from_uuid(o_bug.uuid)
+ except KeyError as e:
+ try:
+ s_bug = self.bug_from_uuid(o_bug.alt_id)
+ except KeyError as e:
+ s_bug = None
+ if s_bug is None:
+ if accept_bugs:
+ o_bug_copy = copy.copy(o_bug)
+ o_bug_copy.bugdir = self
+ o_bug_copy.id = libbe.util.id.ID(o_bug_copy, 'bug')
+ self.append(o_bug_copy)
+ elif change_exception:
+ raise ValueError(
+ ('Merge would add bug {} (alt: {}) to bugdir {}'
+ ).format(o_bug.uuid, o_bug.alt_id, self.uuid))
+ else:
+ s_bug.merge(o_bug, accept_changes=accept_changes,
+ accept_extra_strings=accept_extra_strings,
+ change_exception=change_exception)
+
+ # methods for id generation
-def simple_bug_dir(sync_with_disk=True):
+ def sibling_uuids(self):
+ return []
+
+class RevisionedBugDir (BugDir):
"""
- For testing. Set sync_with_disk==False for a memory-only bugdir.
- >>> bugdir = simple_bug_dir()
- >>> uuids = list(bugdir.list_uuids())
- >>> uuids.sort()
- >>> print uuids
- ['a', 'b']
+ RevisionedBugDirs are read-only copies used for generating
+ diffs between revisions.
"""
- if sync_with_disk == True:
- dir = utility.Dir()
- assert os.path.exists(dir.path)
- root = dir.path
- assert_new_BugDir = True
- rcs_init = True
- else:
- root = "/"
- assert_new_BugDir = False
- rcs_init = False
- bugdir = BugDir(root, sink_to_existing_root=False,
- assert_new_BugDir=assert_new_BugDir,
- allow_rcs_init=rcs_init,
- manipulate_encodings=False)
- if sync_with_disk == True: # postpone cleanup since dir.__del__() removes dir.
- bugdir._dir_ref = dir
- bug_a = bugdir.new_bug("a", summary="Bug A")
- bug_a.creator = "John Doe <jdoe@example.com>"
- bug_a.time = 0
- bug_b = bugdir.new_bug("b", summary="Bug B")
- bug_b.creator = "Jane Doe <jdoe@example.com>"
- bug_b.time = 0
- bug_b.status = "closed"
- if sync_with_disk == True:
- bugdir.save()
- bugdir.set_sync_with_disk(True)
- return bugdir
-
-
-class BugDirTestCase(unittest.TestCase):
- def setUp(self):
- self.dir = utility.Dir()
- self.bugdir = BugDir(self.dir.path, sink_to_existing_root=False,
- allow_rcs_init=True)
- self.rcs = self.bugdir.rcs
- def tearDown(self):
- self.rcs.cleanup()
- self.dir.cleanup()
- def fullPath(self, path):
- return os.path.join(self.dir.path, path)
- def assertPathExists(self, path):
- fullpath = self.fullPath(path)
- self.failUnless(os.path.exists(fullpath)==True,
- "path %s does not exist" % fullpath)
- self.assertRaises(AlreadyInitialized, BugDir,
- self.dir.path, assertNewBugDir=True)
- def versionTest(self):
- if self.rcs.versioned == False:
- return
- original = self.bugdir.rcs.commit("Began versioning")
- bugA = self.bugdir.bug_from_uuid("a")
- bugA.status = "fixed"
- self.bugdir.save()
- new = self.rcs.commit("Fixed bug a")
- dupdir = self.bugdir.duplicate_bugdir(original)
- self.failUnless(dupdir.root != self.bugdir.root,
- "%s, %s" % (dupdir.root, self.bugdir.root))
- bugAorig = dupdir.bug_from_uuid("a")
- self.failUnless(bugA != bugAorig,
- "\n%s\n%s" % (bugA.string(), bugAorig.string()))
- bugAorig.status = "fixed"
- self.failUnless(bug.cmp_status(bugA, bugAorig)==0,
- "%s, %s" % (bugA.status, bugAorig.status))
- self.failUnless(bug.cmp_severity(bugA, bugAorig)==0,
- "%s, %s" % (bugA.severity, bugAorig.severity))
- self.failUnless(bug.cmp_assigned(bugA, bugAorig)==0,
- "%s, %s" % (bugA.assigned, bugAorig.assigned))
- self.failUnless(bug.cmp_time(bugA, bugAorig)==0,
- "%s, %s" % (bugA.time, bugAorig.time))
- self.failUnless(bug.cmp_creator(bugA, bugAorig)==0,
- "%s, %s" % (bugA.creator, bugAorig.creator))
- self.failUnless(bugA == bugAorig,
- "\n%s\n%s" % (bugA.string(), bugAorig.string()))
- self.bugdir.remove_duplicate_bugdir()
- self.failUnless(os.path.exists(dupdir.root)==False, str(dupdir.root))
- def testRun(self):
- self.bugdir.new_bug(uuid="a", summary="Ant")
- self.bugdir.new_bug(uuid="b", summary="Cockroach")
- self.bugdir.new_bug(uuid="c", summary="Praying mantis")
- length = len(self.bugdir)
- self.failUnless(length == 3, "%d != 3 bugs" % length)
- uuids = list(self.bugdir.list_uuids())
- self.failUnless(len(uuids) == 3, "%d != 3 uuids" % len(uuids))
- self.failUnless(uuids == ["a","b","c"], str(uuids))
- bugA = self.bugdir.bug_from_uuid("a")
- bugAprime = self.bugdir.bug_from_shortname("a")
- self.failUnless(bugA == bugAprime, "%s != %s" % (bugA, bugAprime))
- self.bugdir.save()
- self.versionTest()
- def testComments(self, sync_with_disk=False):
- if sync_with_disk == True:
- self.bugdir.set_sync_with_disk(True)
- self.bugdir.new_bug(uuid="a", summary="Ant")
- bug = self.bugdir.bug_from_uuid("a")
- comm = bug.comment_root
- rep = comm.new_reply("Ants are small.")
- rep.new_reply("And they have six legs.")
- if sync_with_disk == False:
- self.bugdir.save()
- self.bugdir.set_sync_with_disk(True)
- self.bugdir._clear_bugs()
- bug = self.bugdir.bug_from_uuid("a")
- bug.load_comments()
- if sync_with_disk == False:
- self.bugdir.set_sync_with_disk(False)
- self.failUnless(len(bug.comment_root)==1, len(bug.comment_root))
- for index,comment in enumerate(bug.comments()):
- if index == 0:
- repLoaded = comment
- self.failUnless(repLoaded.uuid == rep.uuid, repLoaded.uuid)
- self.failUnless(comment.sync_with_disk == True,
- comment.sync_with_disk)
- self.failUnless(comment.content_type == "text/plain",
- comment.content_type)
- self.failUnless(repLoaded.settings["Content-type"]=="text/plain",
- repLoaded.settings)
- self.failUnless(repLoaded.body == "Ants are small.",
- repLoaded.body)
- elif index == 1:
- self.failUnless(comment.in_reply_to == repLoaded.uuid,
- repLoaded.uuid)
- self.failUnless(comment.body == "And they have six legs.",
- comment.body)
+ def __init__(self, bugdir, revision):
+ storage_version = bugdir.storage.storage_version(revision)
+ if storage_version != libbe.storage.STORAGE_VERSION:
+ raise libbe.storage.InvalidStorageVersion(storage_version)
+ s = copy.deepcopy(bugdir.storage)
+ s.writeable = False
+ class RevisionedStorage (object):
+ def __init__(self, storage, default_revision):
+ self.s = storage
+ self.sget = self.s.get
+ self.sancestors = self.s.ancestors
+ self.schildren = self.s.children
+ self.schanged = self.s.changed
+ self.r = default_revision
+ def get(self, *args, **kwargs):
+ if not 'revision' in kwargs or kwargs['revision'] == None:
+ kwargs['revision'] = self.r
+ return self.sget(*args, **kwargs)
+ def ancestors(self, *args, **kwargs):
+ print 'getting ancestors', args, kwargs
+ if not 'revision' in kwargs or kwargs['revision'] == None:
+ kwargs['revision'] = self.r
+ ret = self.sancestors(*args, **kwargs)
+ print 'got ancestors', ret
+ return ret
+ def children(self, *args, **kwargs):
+ if not 'revision' in kwargs or kwargs['revision'] == None:
+ kwargs['revision'] = self.r
+ return self.schildren(*args, **kwargs)
+ def changed(self, *args, **kwargs):
+ if not 'revision' in kwargs or kwargs['revision'] == None:
+ kwargs['revision'] = self.r
+ return self.schanged(*args, **kwargs)
+ rs = RevisionedStorage(s, revision)
+ s.get = rs.get
+ s.ancestors = rs.ancestors
+ s.children = rs.children
+ s.changed = rs.changed
+ BugDir.__init__(self, s, from_storage=True)
+ self.revision = revision
+ def changed(self):
+ return self.storage.changed()
+
+
+if libbe.TESTING == True:
+ class SimpleBugDir (BugDir):
+ """
+ For testing. Set ``memory=True`` for a memory-only bugdir.
+
+ >>> bugdir = SimpleBugDir()
+ >>> uuids = list(bugdir.uuids())
+ >>> uuids.sort()
+ >>> print uuids
+ ['a', 'b']
+ >>> bugdir.cleanup()
+ """
+ def __init__(self, memory=True, versioned=False):
+ if memory == True:
+ storage = None
else:
- self.failIf(True, "Invalid comment: %d\n%s" % (index, comment))
- def testSyncedComments(self):
- self.testComments(sync_with_disk=True)
-
-class SimpleBugDirTestCase (unittest.TestCase):
- def setUp(self):
- # create a pre-existing bugdir in a temporary directory
- self.dir = utility.Dir()
- os.chdir(self.dir.path)
- self.bugdir = BugDir(self.dir.path, sink_to_existing_root=False,
- allow_rcs_init=True)
- self.bugdir.new_bug("preexisting", summary="Hopefully not imported")
- self.bugdir.save()
- def tearDown(self):
- self.dir.cleanup()
- def testOnDiskCleanLoad(self):
- """simple_bug_dir(sync_with_disk==True) should not import preexisting bugs."""
- bugdir = simple_bug_dir(sync_with_disk=True)
- uuids = sorted([bug.uuid for bug in bugdir])
- self.failUnless(uuids == ['a', 'b'], uuids)
- bugdir._clear_bugs()
- uuids = sorted([bug.uuid for bug in bugdir])
- self.failUnless(uuids == [], uuids)
- bugdir.load_all_bugs()
- uuids = sorted([bug.uuid for bug in bugdir])
- self.failUnless(uuids == ['a', 'b'], uuids)
- def testInMemoryCleanLoad(self):
- """simple_bug_dir(sync_with_disk==False) should not import preexisting bugs."""
- bugdir = simple_bug_dir(sync_with_disk=False)
- uuids = sorted([bug.uuid for bug in bugdir])
- self.failUnless(uuids == ['a', 'b'], uuids)
- self.failUnlessRaises(DiskAccessRequired, bugdir.load_all_bugs)
- uuids = sorted([bug.uuid for bug in bugdir])
- self.failUnless(uuids == ['a', 'b'], uuids)
- bugdir._clear_bugs()
- uuids = sorted([bug.uuid for bug in bugdir])
- self.failUnless(uuids == [], uuids)
-
-
-unitsuite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
-suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
+ dir = utility.Dir()
+ self._dir_ref = dir # postpone cleanup since dir.cleanup() removes dir.
+ if versioned == False:
+ storage = libbe.storage.base.Storage(dir.path)
+ else:
+ storage = libbe.storage.base.VersionedStorage(dir.path)
+ storage.init()
+ storage.connect()
+ BugDir.__init__(self, storage=storage, uuid='abc123')
+ bug_a = self.new_bug(summary='Bug A', _uuid='a')
+ bug_a.creator = 'John Doe <jdoe@example.com>'
+ bug_a.time = 0
+ bug_b = self.new_bug(summary='Bug B', _uuid='b')
+ bug_b.creator = 'Jane Doe <jdoe@example.com>'
+ bug_b.time = 0
+ bug_b.status = 'closed'
+ if self.storage != None:
+ self.storage.disconnect() # flush to storage
+ self.storage.connect()
+
+ def cleanup(self):
+ if self.storage != None:
+ self.storage.writeable = True
+ self.storage.disconnect()
+ self.storage.destroy()
+ if hasattr(self, '_dir_ref'):
+ self._dir_ref.cleanup()
+
+ def flush_reload(self):
+ if self.storage != None:
+ self.storage.disconnect()
+ self.storage.connect()
+ self._clear_bugs()
+
+# class BugDirTestCase(unittest.TestCase):
+# def setUp(self):
+# self.dir = utility.Dir()
+# self.bugdir = BugDir(self.dir.path, sink_to_existing_root=False,
+# allow_storage_init=True)
+# self.storage = self.bugdir.storage
+# def tearDown(self):
+# self.bugdir.cleanup()
+# self.dir.cleanup()
+# def fullPath(self, path):
+# return os.path.join(self.dir.path, path)
+# def assertPathExists(self, path):
+# fullpath = self.fullPath(path)
+# self.failUnless(os.path.exists(fullpath)==True,
+# "path %s does not exist" % fullpath)
+# self.assertRaises(AlreadyInitialized, BugDir,
+# self.dir.path, assertNewBugDir=True)
+# def versionTest(self):
+# if self.storage != None and self.storage.versioned == False:
+# return
+# original = self.bugdir.storage.commit("Began versioning")
+# bugA = self.bugdir.bug_from_uuid("a")
+# bugA.status = "fixed"
+# self.bugdir.save()
+# new = self.storage.commit("Fixed bug a")
+# dupdir = self.bugdir.duplicate_bugdir(original)
+# self.failUnless(dupdir.root != self.bugdir.root,
+# "%s, %s" % (dupdir.root, self.bugdir.root))
+# bugAorig = dupdir.bug_from_uuid("a")
+# self.failUnless(bugA != bugAorig,
+# "\n%s\n%s" % (bugA.string(), bugAorig.string()))
+# bugAorig.status = "fixed"
+# self.failUnless(bug.cmp_status(bugA, bugAorig)==0,
+# "%s, %s" % (bugA.status, bugAorig.status))
+# self.failUnless(bug.cmp_severity(bugA, bugAorig)==0,
+# "%s, %s" % (bugA.severity, bugAorig.severity))
+# self.failUnless(bug.cmp_assigned(bugA, bugAorig)==0,
+# "%s, %s" % (bugA.assigned, bugAorig.assigned))
+# self.failUnless(bug.cmp_time(bugA, bugAorig)==0,
+# "%s, %s" % (bugA.time, bugAorig.time))
+# self.failUnless(bug.cmp_creator(bugA, bugAorig)==0,
+# "%s, %s" % (bugA.creator, bugAorig.creator))
+# self.failUnless(bugA == bugAorig,
+# "\n%s\n%s" % (bugA.string(), bugAorig.string()))
+# self.bugdir.remove_duplicate_bugdir()
+# self.failUnless(os.path.exists(dupdir.root)==False,
+# str(dupdir.root))
+# def testRun(self):
+# self.bugdir.new_bug(uuid="a", summary="Ant")
+# self.bugdir.new_bug(uuid="b", summary="Cockroach")
+# self.bugdir.new_bug(uuid="c", summary="Praying mantis")
+# length = len(self.bugdir)
+# self.failUnless(length == 3, "%d != 3 bugs" % length)
+# uuids = list(self.bugdir.uuids())
+# self.failUnless(len(uuids) == 3, "%d != 3 uuids" % len(uuids))
+# self.failUnless(uuids == ["a","b","c"], str(uuids))
+# bugA = self.bugdir.bug_from_uuid("a")
+# bugAprime = self.bugdir.bug_from_shortname("a")
+# self.failUnless(bugA == bugAprime, "%s != %s" % (bugA, bugAprime))
+# self.bugdir.save()
+# self.versionTest()
+# def testComments(self, sync_with_disk=False):
+# if sync_with_disk == True:
+# self.bugdir.set_sync_with_disk(True)
+# self.bugdir.new_bug(uuid="a", summary="Ant")
+# bug = self.bugdir.bug_from_uuid("a")
+# comm = bug.comment_root
+# rep = comm.new_reply("Ants are small.")
+# rep.new_reply("And they have six legs.")
+# if sync_with_disk == False:
+# self.bugdir.save()
+# self.bugdir.set_sync_with_disk(True)
+# self.bugdir._clear_bugs()
+# bug = self.bugdir.bug_from_uuid("a")
+# bug.load_comments()
+# if sync_with_disk == False:
+# self.bugdir.set_sync_with_disk(False)
+# self.failUnless(len(bug.comment_root)==1, len(bug.comment_root))
+# for index,comment in enumerate(bug.comments()):
+# if index == 0:
+# repLoaded = comment
+# self.failUnless(repLoaded.uuid == rep.uuid, repLoaded.uuid)
+# self.failUnless(comment.sync_with_disk == sync_with_disk,
+# comment.sync_with_disk)
+# self.failUnless(comment.content_type == "text/plain",
+# comment.content_type)
+# self.failUnless(repLoaded.settings["Content-type"] == \
+# "text/plain",
+# repLoaded.settings)
+# self.failUnless(repLoaded.body == "Ants are small.",
+# repLoaded.body)
+# elif index == 1:
+# self.failUnless(comment.in_reply_to == repLoaded.uuid,
+# repLoaded.uuid)
+# self.failUnless(comment.body == "And they have six legs.",
+# comment.body)
+# else:
+# self.failIf(True,
+# "Invalid comment: %d\n%s" % (index, comment))
+# def testSyncedComments(self):
+# self.testComments(sync_with_disk=True)
+
+ class SimpleBugDirTestCase (unittest.TestCase):
+ def setUp(self):
+ # create a pre-existing bugdir in a temporary directory
+ self.dir = utility.Dir()
+ self.storage = libbe.storage.base.Storage(self.dir.path)
+ self.storage.init()
+ self.storage.connect()
+ self.bugdir = BugDir(self.storage)
+ self.bugdir.new_bug(summary="Hopefully not imported",
+ _uuid="preexisting")
+ self.storage.disconnect()
+ self.storage.connect()
+ def tearDown(self):
+ if self.storage != None:
+ self.storage.disconnect()
+ self.storage.destroy()
+ self.dir.cleanup()
+ def testOnDiskCleanLoad(self):
+ """
+ SimpleBugDir(memory==False) should not import
+ preexisting bugs.
+ """
+ bugdir = SimpleBugDir(memory=False)
+ self.failUnless(bugdir.storage.is_readable() == True,
+ bugdir.storage.is_readable())
+ self.failUnless(bugdir.storage.is_writeable() == True,
+ bugdir.storage.is_writeable())
+ uuids = sorted([bug.uuid for bug in bugdir])
+ self.failUnless(uuids == ['a', 'b'], uuids)
+ bugdir.flush_reload()
+ uuids = sorted(bugdir.uuids())
+ self.failUnless(uuids == ['a', 'b'], uuids)
+ uuids = sorted([bug.uuid for bug in bugdir])
+ self.failUnless(uuids == [], uuids)
+ bugdir.load_all_bugs()
+ uuids = sorted([bug.uuid for bug in bugdir])
+ self.failUnless(uuids == ['a', 'b'], uuids)
+ bugdir.cleanup()
+ def testInMemoryCleanLoad(self):
+ """
+ SimpleBugDir(memory==True) should not import
+ preexisting bugs.
+ """
+ bugdir = SimpleBugDir(memory=True)
+ self.failUnless(bugdir.storage == None, bugdir.storage)
+ uuids = sorted([bug.uuid for bug in bugdir])
+ self.failUnless(uuids == ['a', 'b'], uuids)
+ uuids = sorted([bug.uuid for bug in bugdir])
+ self.failUnless(uuids == ['a', 'b'], uuids)
+ bugdir._clear_bugs()
+ uuids = sorted(bugdir.uuids())
+ self.failUnless(uuids == [], uuids)
+ uuids = sorted([bug.uuid for bug in bugdir])
+ self.failUnless(uuids == [], uuids)
+ bugdir.cleanup()
+
+ unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
+ suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
+
+# def _get_settings(self, settings_path, for_duplicate_bugdir=False):
+# allow_no_storage = not self.storage.path_in_root(settings_path)
+# if allow_no_storage == True:
+# assert for_duplicate_bugdir == True
+# if self.sync_with_disk == False and for_duplicate_bugdir == False:
+# # duplicates can ignore this bugdir's .sync_with_disk status
+# raise DiskAccessRequired("_get settings")
+# try:
+# settings = mapfile.map_load(self.storage, settings_path, allow_no_storage)
+# except storage.NoSuchFile:
+# settings = {"storage_name": "None"}
+# return settings
+
+# def _save_settings(self, settings_path, settings,
+# for_duplicate_bugdir=False):
+# allow_no_storage = not self.storage.path_in_root(settings_path)
+# if allow_no_storage == True:
+# assert for_duplicate_bugdir == True
+# if self.sync_with_disk == False and for_duplicate_bugdir == False:
+# # duplicates can ignore this bugdir's .sync_with_disk status
+# raise DiskAccessRequired("_save settings")
+# self.storage.mkdir(self.get_path(), allow_no_storage)
+# mapfile.map_save(self.storage, settings_path, settings, allow_no_storage)
projects / be.git / blobdiff