aboutsummaryrefslogtreecommitdiffstats
path: root/libbe
diff options
context:
space:
mode:
Diffstat (limited to 'libbe')
-rw-r--r--libbe/arch.py70
-rw-r--r--libbe/bug.py287
-rw-r--r--libbe/bugdir.py356
-rw-r--r--libbe/cmdutil.py111
-rw-r--r--libbe/diff.py33
-rw-r--r--libbe/mapfile.py6
-rw-r--r--libbe/names.py51
-rw-r--r--libbe/plugin.py2
-rw-r--r--libbe/rcs.py47
-rw-r--r--libbe/utility.py26
10 files changed, 477 insertions, 512 deletions
diff --git a/libbe/arch.py b/libbe/arch.py
index 8e7390d..b35a897 100644
--- a/libbe/arch.py
+++ b/libbe/arch.py
@@ -50,14 +50,6 @@ class Arch(RCS):
if self._u_search_parent_directories(path, "{arch}") != None :
return True
return False
- def _rcs_root(self, path):
- if not os.path.isdir(path):
- dirname = os.path.dirname(path)
- else:
- dirname = path
- status,output,error = self._u_invoke_client("tree-root", dirname)
- # get archive name...
- return output.rstrip('\n')
def _rcs_init(self, path):
self._create_archive(path)
self._create_project(path)
@@ -121,11 +113,35 @@ class Arch(RCS):
assert self._archive_name != None
assert self._project_name != None
return "%s/%s" % (self._archive_name, self._project_name)
+ def _adjust_naming_conventions(self, path):
+ """
+ By default, Arch restricts source code filenames to
+ ^[_=a-zA-Z0-9].*$
+ See
+ http://regexps.srparish.net/tutorial-tla/naming-conventions.html
+ Since our bug directory '.be' doesn't satisfy these conventions,
+ we need to adjust them.
+
+ The conventions are specified in
+ project-root/{arch}/=tagging-method
+ """
+ tagpath = os.path.join(path, "{arch}", "=tagging-method")
+ lines_out = []
+ for line in file(tagpath, "rb"):
+ line.decode("utf-8")
+ if line.startswith("source "):
+ lines_out.append("source ^[._=a-zA-X0-9].*$\n")
+ else:
+ lines_out.append(line)
+ file(tagpath, "wb").write("".join(lines_out).encode("utf-8"))
+
def _add_project_code(self, path):
# http://mwolson.org/projects/GettingStartedWithArch.html
- # http://regexps.srparish.net/tutorial-tla/importing-first.html#Importing_the_First_Revision
- self._u_invoke_client("init-tree", self._archive_project_name(),
+ # http://regexps.srparish.net/tutorial-tla/new-source.html
+ # http://regexps.srparish.net/tutorial-tla/importing-first.html
+ self._invoke_client("init-tree", self._project_name,
directory=path)
+ self._adjust_naming_conventions(path)
self._invoke_client("import", "--summary", "Began versioning",
directory=path)
def _rcs_cleanup(self):
@@ -133,6 +149,40 @@ class Arch(RCS):
self._remove_project()
if self._tmp_archive == True:
self._remove_archive()
+
+ def _rcs_root(self, path):
+ if not os.path.isdir(path):
+ dirname = os.path.dirname(path)
+ else:
+ dirname = path
+ status,output,error = self._u_invoke_client("tree-root", dirname)
+ root = output.rstrip('\n')
+
+ self._get_archive_project_name(root)
+
+ return root
+
+ def _get_archive_name(self, root):
+ status,output,error = self._u_invoke_client("archives")
+ lines = output.split('\n')
+ # e.g. output:
+ # jdoe@example.com--bugs-everywhere-auto-2008.22.24.52
+ # /tmp/BEtestXXXXXX/rootdir
+ # (+ repeats)
+ for archive,location in zip(lines[::2], lines[1::2]):
+ if os.path.realpath(location) == os.path.realpath(root):
+ self._archive_name = archive
+ assert self._archive_name != None
+
+ def _get_archive_project_name(self, root):
+ # get project names
+ status,output,error = self._u_invoke_client("tree-version", directory=root)
+ # e.g output
+ # jdoe@example.com--bugs-everywhere-auto-2008.22.24.52/be--mainline--0.1
+ archive_name,project_name = output.rstrip('\n').split('/')
+ self._archive_name = archive_name
+ self._project_name = project_name
+
def _rcs_get_user_id(self):
try:
status,output,error = self._u_invoke_client('my-id')
diff --git a/libbe/bug.py b/libbe/bug.py
index a297b1a..b1e8d26 100644
--- a/libbe/bug.py
+++ b/libbe/bug.py
@@ -17,12 +17,15 @@
import os
import os.path
import errno
-import names
-import mapfile
import time
-import utility
import doctest
+from beuuid import uuid_gen
+import mapfile
+import comment
+import utility
+
+
### Define and describe valid bug categories
# Use a tuple of (category, description) tuples since we don't have
# ordered dicts in Python yet http://www.python.org/dev/peps/pep-0372/
@@ -87,45 +90,46 @@ class Bug(object):
severity = checked_property("severity", severity_values)
status = checked_property("status", status_values)
- def __init__(self, path, uuid, rcs, bugdir):
- self.path = path
- self.uuid = uuid
- if uuid is not None:
- dict = mapfile.map_load(self.get_path("values"))
- else:
- dict = {}
-
- self.rcs = rcs
- self.bugdir = bugdir
-
- self.summary = dict.get("summary")
- self.creator = dict.get("creator")
- self.target = dict.get("target")
- self.status = dict.get("status", "open")
- self.severity = dict.get("severity", "minor")
- self.assigned = dict.get("assigned")
- self.time = dict.get("time")
- if self.time is not None:
- self.time = utility.str_to_time(self.time)
-
- def get_path(self, file=None):
- if file == None:
- return os.path.join(self.path, self.uuid)
- else:
- return os.path.join(self.path, self.uuid, file)
-
def _get_active(self):
return self.status in active_status_values
active = property(_get_active)
+ def __init__(self, bugdir=None, uuid=None, loadNow=False, summary=None):
+ self.bugdir = bugdir
+ if bugdir != None:
+ self.rcs = bugdir.rcs
+ else:
+ self.rcs = None
+ if loadNow == True:
+ self.uuid = uuid
+ self.load()
+ else:
+ # Note: defaults should match those in Bug.load()
+ if uuid != None:
+ self.uuid = uuid
+ else:
+ self.uuid = uuid_gen()
+ self.summary = summary
+ 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 = time.time()
+ self.comment_root = comment.Comment(self, uuid=comment.INVALID_UUID)
+
def __repr__(self):
return "Bug(uuid=%r)" % self.uuid
- def string(self, bugs=None, shortlist=False):
- if bugs == None:
- bugs = list(self.bugdir.list())
- short_name = names.unique_name(self, bugs)
+ def string(self, shortlist=False, show_comments=False):
+ if self.bugdir == None:
+ shortname = self.uuid
+ else:
+ shortname = self.bugdir.bug_shortname(self)
if shortlist == False:
if self.time == None:
timestring = ""
@@ -134,7 +138,7 @@ class Bug(object):
ftime = utility.time_to_str(self.time)
timestring = "%s (%s)" % (htime, ftime)
info = [("ID", self.uuid),
- ("Short name", short_name),
+ ("Short name", shortname),
("Severity", self.severity),
("Status", self.status),
("Assigned", self.assigned),
@@ -150,12 +154,20 @@ class Bug(object):
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]
- return "".join(infolines) + "%s\n" % self.summary
+ bugout = "".join(infolines) + "%s" % self.summary.rstrip('\n')
else:
statuschar = self.status[0]
severitychar = self.severity[0]
chars = "%c%c" % (statuschar, severitychar)
- return "%s:%s: %s" % (short_name, chars, self.summary)
+ bugout = "%s:%s: %s" % (shortname, chars, self.summary.rstrip('\n'))
+
+ if show_comments == True:
+ comout = self.comment_root.string_thread(auto_name_map=True,
+ bug_shortname=shortname)
+ output = bugout + '\n' + comout.rstrip('\n')
+ else :
+ output = bugout
+ return output
def __str__(self):
return self.string(shortlist=True)
@@ -163,7 +175,28 @@ class Bug(object):
def __cmp__(self, other):
return cmp_full(self, other)
- def add_attr(self, map, name):
+ def get_path(self, name=None):
+ my_dir = os.path.join(self.bugdir.get_path("bugs"), self.uuid)
+ if name is None:
+ return my_dir
+ assert name in ["values", "comments"]
+ return os.path.join(my_dir, name)
+
+ def load(self):
+ map = mapfile.map_load(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)
+
+ self.comment_root = comment.loadComments(self)
+
+ def _add_attr(self, map, name):
value = getattr(self, name)
if value is not None:
map[name] = value
@@ -171,134 +204,39 @@ class Bug(object):
def save(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")
+ 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)
+
+ self.rcs.mkdir(self.get_path())
path = self.get_path("values")
mapfile.map_save(self.rcs, path, map)
+ if len(self.comment_root) > 0:
+ self.rcs.mkdir(self.get_path("comments"))
+ comment.saveComments(self)
+
def remove(self):
+ self.comment_root.remove()
path = self.get_path()
self.rcs.recursive_remove(path)
def new_comment(self, body=None):
- if not os.path.exists(self.get_path("comments")):
- self.rcs.mkdir(self.get_path("comments"))
- comm = Comment(None, self)
- comm.uuid = names.uuid()
- comm.rcs = self.rcs
- comm.From = self.rcs.get_user_id()
- comm.time = time.time()
- comm.body = body
+ comm = comment.comment_root.new_reply(body=body)
return comm
- def get_comment(self, uuid):
- return Comment(uuid, self)
-
- def iter_comment_ids(self):
- path = self.get_path("comments")
- if not os.path.isdir(path):
- return
- try:
- for uuid in os.listdir(path):
- if (uuid.startswith('.')):
- continue
- yield uuid
- except OSError, e:
- if e.errno != errno.ENOENT:
- raise
- return
-
- def list_comments(self):
- comments = [Comment(id, self) for id in self.iter_comment_ids()]
- comments.sort(cmp_time)
- return comments
-
-def add_headers(obj, map, names):
- map_names = {}
- for name in names:
- map_names[name] = pyname_to_header(name)
- add_attrs(obj, map, names, map_names)
-
-def add_attrs(obj, map, names, map_names=None):
- if map_names is None:
- map_names = {}
- for name in names:
- map_names[name] = name
-
- for name in names:
- value = getattr(obj, name)
- if value is not None:
- map[map_names[name]] = value
-
-
-class Comment(object):
- def __init__(self, uuid, bug):
- object.__init__(self)
- self.uuid = uuid
- self.bug = bug
- if self.uuid is not None and self.bug is not None:
- map = mapfile.map_load(self.get_path("values"))
- self.time = utility.str_to_time(map["Date"])
- self.From = map["From"]
- self.in_reply_to = map.get("In-reply-to")
- self.content_type = map.get("Content-type", "text/plain")
- self.body = file(self.get_path("body")).read().decode("utf-8")
- else:
- self.time = None
- self.From = None
- self.in_reply_to = None
- self.content_type = "text/plain"
- self.body = None
+ def comment_from_shortname(self, shortname, *args, **kwargs):
+ return self.comment_root.comment_from_shortname(shortname, *args, **kwargs)
- def save(self):
- map_file = {"Date": utility.time_to_str(self.time)}
- add_headers(self, map_file, ("From", "in_reply_to", "content_type"))
- if not os.path.exists(self.get_path()):
- self.bug.rcs.mkdir(self.get_path())
- mapfile.map_save(self.bug.rcs, self.get_path("values"), map_file)
- self.bug.rcs.set_file_contents(self.get_path("body"),
- self.body.encode('utf-8'))
-
- def get_path(self, name=None):
- my_dir = os.path.join(self.bug.get_path("comments"), self.uuid)
- if name is None:
- return my_dir
- return os.path.join(my_dir, name)
-
-
-def thread_comments(comments):
- child_map = {}
- top_comments = []
- for comment in comments:
- child_map[comment.uuid] = []
- for comment in comments:
- if comment.in_reply_to is None or comment.in_reply_to not in child_map:
- top_comments.append(comment)
- continue
- child_map[comment.in_reply_to].append(comment)
-
- def recurse_children(comment):
- child_list = []
- for child in child_map[comment.uuid]:
- child_list.append(recurse_children(child))
- return (comment, child_list)
- return [recurse_children(c) for c in top_comments]
-
-def pyname_to_header(name):
- return name.capitalize().replace('_', '-')
+ def comment_from_uuid(self, uuid):
+ return self.comment_root.comment_from_uuid(uuid)
-
-class MockBug:
- def __init__(self, attr, value):
- setattr(self, attr, value)
-
# 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
@@ -307,32 +245,42 @@ class MockBug:
def cmp_severity(bug_1, bug_2):
"""
- Compare the severity levels of two bugs, with more severe bugs comparing
- as less.
-
- >>> attr="severity"
- >>> cmp_severity(MockBug(attr,"wishlist"), MockBug(attr,"wishlist")) == 0
+ Compare the severity levels of two bugs, with more severe bugs
+ comparing as less.
+ >>> bugA = Bug()
+ >>> bugB = Bug()
+ >>> bugA.severity = bugB.severity = "wishlist"
+ >>> cmp_severity(bugA, bugB) == 0
True
- >>> cmp_severity(MockBug(attr,"wishlist"), MockBug(attr,"minor")) > 0
+ >>> bugB.severity = "minor"
+ >>> cmp_severity(bugA, bugB) > 0
True
- >>> cmp_severity(MockBug(attr,"critical"), MockBug(attr,"wishlist")) < 0
+ >>> bugA.severity = "critical"
+ >>> cmp_severity(bugA, bugB) < 0
True
"""
+ if not hasattr(bug_2, "severity") :
+ return 1
return -cmp(severity_index[bug_1.severity], severity_index[bug_2.severity])
def cmp_status(bug_1, bug_2):
"""
Compare the status levels of two bugs, with more 'open' bugs
comparing as less.
-
- >>> attr="status"
- >>> cmp_status(MockBug(attr,"open"), MockBug(attr,"open")) == 0
+ >>> bugA = Bug()
+ >>> bugB = Bug()
+ >>> bugA.status = bugB.status = "open"
+ >>> cmp_status(bugA, bugB) == 0
True
- >>> cmp_status(MockBug(attr,"open"), MockBug(attr,"closed")) < 0
+ >>> bugB.status = "closed"
+ >>> cmp_status(bugA, bugB) < 0
True
- >>> cmp_status(MockBug(attr,"closed"), MockBug(attr,"open")) > 0
+ >>> bugA.status = "fixed"
+ >>> cmp_status(bugA, bugB) > 0
True
"""
+ if not hasattr(bug_2, "status") :
+ return 1
val_2 = status_index[bug_2.status]
return cmp(status_index[bug_1.status], status_index[bug_2.status])
@@ -342,13 +290,20 @@ def cmp_attr(bug_1, bug_2, attr, invert=False):
comparison rule for that attribute type. If invert == True, sort
*against* that convention.
>>> attr="severity"
- >>> cmp_attr(MockBug(attr,1), MockBug(attr,2), attr, invert=False) < 0
+ >>> bugA = Bug()
+ >>> bugB = Bug()
+ >>> bugA.severity = "critical"
+ >>> bugB.severity = "wishlist"
+ >>> cmp_attr(bugA, bugB, attr) < 0
True
- >>> cmp_attr(MockBug(attr,1), MockBug(attr,2), attr, invert=True) > 0
+ >>> cmp_attr(bugA, bugB, attr, invert=True) > 0
True
- >>> cmp_attr(MockBug(attr,1), MockBug(attr,1), attr) == 0
+ >>> bugB.severity = "critical"
+ >>> cmp_attr(bugA, bugB, attr) == 0
True
"""
+ if not hasattr(bug_2, attr) :
+ return 1
if invert == True :
return -cmp(getattr(bug_1, attr), getattr(bug_2, attr))
else :
diff --git a/libbe/bugdir.py b/libbe/bugdir.py
index 41f0fec..6152e3f 100644
--- a/libbe/bugdir.py
+++ b/libbe/bugdir.py
@@ -16,16 +16,17 @@
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import os
import os.path
-import cmdutil
import errno
+import time
import unittest
import doctest
-import names
+
+from beuuid import uuid_gen
import mapfile
-import time
+import bug
+import cmdutil
import utility
-from rcs import rcs_by_name, installed_rcs
-from bug import Bug
+from rcs import rcs_by_name, detect_rcs, installed_rcs, PathNotInRoot
class NoBugDir(Exception):
def __init__(self, path):
@@ -33,48 +34,6 @@ class NoBugDir(Exception):
Exception.__init__(self, msg)
self.path = path
-
-def iter_parent_dirs(cur_dir):
- cur_dir = os.path.realpath(cur_dir)
- old_dir = None
- while True:
- yield cur_dir
- old_dir = cur_dir
- cur_dir = os.path.normpath(os.path.join(cur_dir, '..'))
- if old_dir == cur_dir:
- break;
-
-
-def tree_root(dir, old_version=False):
- for rootdir in iter_parent_dirs(dir):
- versionfile=os.path.join(rootdir, ".be", "version")
- if os.path.exists(versionfile):
- if not old_version:
- test_version(versionfile)
- return BugDir(os.path.join(rootdir, ".be"))
- elif not os.path.exists(rootdir):
- raise NoRootEntry(rootdir)
- old_rootdir = rootdir
- rootdir=os.path.join('..', rootdir)
-
- raise NoBugDir(dir)
-
-class BadTreeVersion(Exception):
- def __init__(self, version):
- Exception.__init__(self, "Unsupported tree version: %s" % version)
- self.version = version
-
-def test_version(path):
- tree_version = file(path, "rb").read()
- if tree_version != TREE_VERSION_STRING:
- raise BadTreeVersion(tree_version)
-
-def set_version(path, rcs):
- rcs.set_file_contents(os.path.join(path, "version"), TREE_VERSION_STRING)
-
-
-TREE_VERSION_STRING = "Bugs Everywhere Tree 1 0\n"
-
class NoRootEntry(Exception):
def __init__(self, path):
self.path = path
@@ -86,32 +45,15 @@ class AlreadyInitialized(Exception):
Exception.__init__(self,
"Specified root is already initialized: %s" % path)
-def bugdir_root(versioning_root):
- return os.path.join(versioning_root, ".be")
+class InvalidValue(ValueError):
+ def __init__(self, name, value):
+ msg = "Cannot assign value %s to %s" % (value, name)
+ Exception.__init__(self, msg)
+ self.name = name
+ self.value = value
+
-def create_bug_dir(path, rcs):
- """
- >>> import tests
- >>> rcs = rcs_by_name("None")
- >>> create_bug_dir('/highly-unlikely-to-exist', rcs)
- Traceback (most recent call last):
- NoRootEntry: Specified root does not exist: /highly-unlikely-to-exist
- """
- root = os.path.join(path, ".be")
- try:
- rcs.mkdir(root)
- except OSError, e:
- if e.errno == errno.ENOENT:
- raise NoRootEntry(path)
- elif e.errno == errno.EEXIST:
- raise AlreadyInitialized(path)
- else:
- raise
- rcs.mkdir(os.path.join(root, "bugs"))
- set_version(root, rcs)
- mapfile.map_save(rcs,
- os.path.join(root, "settings"), {"rcs_name": rcs.name})
- return BugDir(bugdir_root(path))
+TREE_VERSION_STRING = "Bugs Everywhere Tree 1 0\n"
def setting_property(name, valid=None):
@@ -130,83 +72,232 @@ def setting_property(name, valid=None):
del self.settings[name]
else:
self.settings[name] = value
- self.save_settings()
+ self.save()
return property(getter, setter)
-class BugDir:
- def __init__(self, dir):
- self.dir = dir
- self.bugs_path = os.path.join(self.dir, "bugs")
+class BugDir (list):
+ def __init__(self, root=None, sink_to_existing_root=True,
+ assert_new_BugDir=False, allow_rcs_init=False,
+ loadNow=False, rcs=None):
+ list.__init__(self)
+ 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
+ if loadNow == True:
+ self.load()
+ else:
+ if assert_new_BugDir:
+ if os.path.exists(self.get_path()):
+ raise AlreadyInitialized, self.get_path()
+ if rcs == None:
+ rcs = self.guess_rcs(allow_rcs_init)
+ self.settings = {"rcs_name": self.rcs_name}
+ self.rcs_name = rcs.name
+
+ def find_root(self, path):
+ """
+ Search for an existing bug database dir and it's ancestors and
+ return a BugDir rooted there.
+ """
+ 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
+ else:
+ beroot = utility.search_parent_directories(path, ".be")
+ if beroot == None:
+ raise NoBugDir(path)
+ return beroot
+
+ def get_version(self, path=None):
+ if path == None:
+ path = self.get_path("version")
try:
- self.settings = mapfile.map_load(os.path.join(self.dir,"settings"))
- except mapfile.NoSuchFile:
- self.settings = {"rcs_name": "None"}
+ tree_version = self.rcs.get_file_contents(path)
+ except AttributeError, e:
+ # haven't initialized rcs yet
+ tree_version = file(path, "rb").read().decode("utf-8")
+ return tree_version
+
+ def set_version(self):
+ self.rcs.set_file_contents(self.get_path("version"), TREE_VERSION_STRING)
rcs_name = setting_property("rcs_name",
("None", "bzr", "git", "Arch", "hg"))
- _rcs = None
- target = setting_property("target")
-
- def save_settings(self):
- mapfile.map_save(self.rcs,
- os.path.join(self.dir, "settings"), self.settings)
+ _rcs = None
def _get_rcs(self):
if self._rcs is not None:
if self.rcs_name == self._rcs.name:
return self._rcs
self._rcs = rcs_by_name(self.rcs_name)
- self._rcs.root(self.dir)
+ self._rcs.root(self.root)
return self._rcs
rcs = property(_get_rcs)
+ target = setting_property("target")
+
+ def get_path(self, *args):
+ 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)
+
+ def guess_rcs(self, allow_rcs_init=False):
+ deepdir = self.get_path()
+ if not os.path.exists(deepdir):
+ deepdir = os.path.dirname(deepdir)
+ rcs = detect_rcs(deepdir)
+ if rcs.name == "None":
+ if allow_rcs_init == True:
+ rcs = installed_rcs()
+ rcs.init(self.root)
+ self.settings = {"rcs_name": rcs.name}
+ self.rcs_name = rcs.name
+ return rcs
+
+ def load(self):
+ version = self.get_version()
+ 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.settings = self._get_settings(self.get_path("settings"))
+ self._clear_bugs()
+ for uuid in self.list_uuids():
+ self._load_bug(uuid)
+
+ self._bug_map_gen()
+
+ def save(self):
+ self.rcs.mkdir(self.get_path())
+ self.set_version()
+ self._save_settings(self.get_path("settings"), self.settings)
+ self.rcs.mkdir(self.get_path("bugs"))
+ for bug in self:
+ bug.save()
+
+ def _get_settings(self, settings_path):
+ try:
+ settings = mapfile.map_load(settings_path)
+ except mapfile.NoSuchFile:
+ settings = {"rcs_name": "None"}
+ return settings
+
+ def _save_settings(self, settings_path, settings):
+ try:
+ mapfile.map_save(self.rcs, settings_path, settings)
+ except PathNotInRoot, e:
+ # Handling duplicate bugdir settings, special case
+ none_rcs = rcs_by_name("None")
+ none_rcs.root(settings_path)
+ mapfile.map_save(none_rcs, settings_path, settings)
+
def duplicate_bugdir(self, revision):
- return BugDir(bugdir_root(self.rcs.duplicate_repo(revision)))
+ duplicate_path = self.rcs.duplicate_repo(revision)
- def remove_duplicate_bugdir(self):
- self.rcs.remove_duplicate_repo()
+ # setup revision RCS as None, since the duplicate may not be versioned
+ duplicate_settings_path = os.path.join(duplicate_path, ".be", "settings")
+ duplicate_settings = self._get_settings(duplicate_settings_path)
+ if "rcs_name" in duplicate_settings:
+ duplicate_settings["rcs_name"] = "None"
+ self._save_settings(duplicate_settings_path, duplicate_settings)
- def list(self):
- for uuid in self.list_uuids():
- yield self.get_bug(uuid)
+ return BugDir(duplicate_path, loadNow=True)
- def bug_map(self):
- bugs = {}
- for bug in self.list():
- bugs[bug.uuid] = bug
- return bugs
+ def remove_duplicate_bugdir(self):
+ self.rcs.remove_duplicate_repo()
- def get_bug(self, uuid):
- return Bug(self.bugs_path, uuid, self.rcs, self)
+ def _bug_map_gen(self):
+ map = {}
+ for bug in self:
+ map[bug.uuid] = bug
+ self.bug_map = map
def list_uuids(self):
- for uuid in os.listdir(self.bugs_path):
+ for uuid in os.listdir(self.get_path("bugs")):
if (uuid.startswith('.')):
continue
yield uuid
- def new_bug(self, uuid=None):
- if uuid is None:
- uuid = names.uuid()
- path = os.path.join(self.bugs_path, uuid)
- self.rcs.mkdir(path)
- bug = Bug(self.bugs_path, None, self.rcs, self)
- bug.uuid = uuid
- bug.creator = self.rcs.get_user_id()
- bug.severity = "minor"
- bug.status = "open"
- bug.time = time.time()
- return bug
+ def _clear_bugs(self):
+ while len(self) > 0:
+ self.pop()
+
+ def _load_bug(self, uuid):
+ bg = bug.Bug(bugdir=self, uuid=uuid, loadNow=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)
+ self.append(bg)
+ self._bug_map_gen()
+ return bg
+
+ def remove_bug(self, bug):
+ 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()
+ >>> bug_a = bd.bug_from_shortname('a')
+ >>> print type(bug_a)
+ <class 'libbe.bug.Bug'>
+ >>> print bug_a
+ a:om: Bug A
+ """
+ matches = []
+ for bug in self:
+ if bug.uuid.startswith(shortname):
+ matches.append(bug)
+ if len(matches) > 1:
+ raise cmdutil.UserError("More than one bug matches %s. Please be more"
+ " specific." % shortname)
+ if len(matches) == 1:
+ return matches[0]
+ raise KeyError("No bug matches %s" % shortname)
+
+ def bug_from_uuid(self, uuid):
+ if uuid not in self.bug_map:
+ self._bug_map_gen()
+ if uuid not in self.bug_map:
+ raise KeyError("No bug matches %s" % uuid +str(self.bug_map)+str(self))
+ return self.bug_map[uuid]
-class InvalidValue(ValueError):
- def __init__(self, name, value):
- msg = "Cannot assign value %s to %s" % (value, name)
- Exception.__init__(self, msg)
- self.name = name
- self.value = value
def simple_bug_dir():
"""
@@ -218,18 +309,17 @@ def simple_bug_dir():
['a', 'b']
"""
dir = utility.Dir()
- rcs = installed_rcs()
- rcs.init(dir.path)
assert os.path.exists(dir.path)
- bugdir = create_bug_dir(dir.path, rcs)
+ bugdir = BugDir(dir.path, sink_to_existing_root=False, allow_rcs_init=True)
bugdir._dir_ref = dir # postpone cleanup since dir.__del__() removes dir.
- bug_a = bugdir.new_bug("a")
- bug_a.summary = "Bug A"
- bug_a.save()
- bug_b = bugdir.new_bug("b")
+ 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"
- bug_b.summary = "Bug B"
- bug_b.save()
+ bugdir.save()
return bugdir
@@ -238,9 +328,8 @@ class BugDirTestCase(unittest.TestCase):
unittest.TestCase.__init__(self, *args, **kwargs)
def setUp(self):
self.dir = utility.Dir()
- self.rcs = installed_rcs()
- self.rcs.init(self.dir.path)
- self.bugdir = create_bug_dir(self.dir.path, self.rcs)
+ self.bugdir = BugDir(self.dir.path, sink_to_existing_root=False, allow_rcs_init=True)
+ self.rcs = self.bugdir.rcs
def tearDown(self):
del(self.rcs)
del(self.dir)
@@ -250,9 +339,8 @@ class BugDirTestCase(unittest.TestCase):
fullpath = self.fullPath(path)
self.failUnless(os.path.exists(fullpath)==True,
"path %s does not exist" % fullpath)
- def testBugDirDuplicate(self):
- self.assertRaises(AlreadyInitialized, create_bug_dir,
- self.dir.path, self.rcs)
+ self.assertRaises(AlreadyInitialized, BugDir,
+ self.dir.path, assertNewBugDir=True)
unitsuite = unittest.TestLoader().loadTestsFromTestCase(BugDirTestCase)
suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
diff --git a/libbe/cmdutil.py b/libbe/cmdutil.py
index 62a0c7c..55a7a72 100644
--- a/libbe/cmdutil.py
+++ b/libbe/cmdutil.py
@@ -14,16 +14,17 @@
# 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-import bugdir
-import plugin
-import locale
-import os
import optparse
+import os
+import locale
from textwrap import TextWrapper
from StringIO import StringIO
-import utility
import doctest
+import bugdir
+import plugin
+import utility
+
class UserError(Exception):
def __init__(self, msg):
Exception.__init__(self, msg)
@@ -33,40 +34,6 @@ class UserErrorWrap(UserError):
UserError.__init__(self, str(exception))
self.exception = exception
-def get_bug(spec, bug_dir=None):
- """
- >>> bd = bugdir.simple_bug_dir()
- >>> bug_a = get_bug('a', bd)
- >>> print type(bug_a)
- <class 'libbe.bug.Bug'>
- >>> print bug_a
- a:om: Bug A
- >>> print bd.get_bug('a')
- a:om: Bug A
- >>> bug_a == bd.get_bug('a')
- True
- """
- matches = []
- try:
- if bug_dir is None:
- bug_dir = bugdir.tree_root('.')
- except bugdir.NoBugDir, e:
- raise UserErrorWrap(e)
- bugs = list(bug_dir.list())
- for bug in bugs:
- if bug.uuid.startswith(spec):
- matches.append(bug)
- if len(matches) > 1:
- raise UserError("More than one bug matches %s. Please be more"
- " specific." % spec)
- if len(matches) == 1:
- return matches[0]
-
- matches = []
- if len(matches) == 0:
- raise UserError("No bug matches %s" % spec)
- return matches[0]
-
def iter_commands():
for name, module in plugin.iter_plugins("becommands"):
yield name.replace("_", "-"), module
@@ -115,34 +82,6 @@ class UsageError(Exception):
def raise_get_help(option, opt, value, parser):
raise GetHelp
-
-def iter_comment_name(bug, unique_name):
- """Iterate through id, comment pairs, in date order.
- (This is a user-friendly id, not the comment uuid)
- """
- def key(comment):
- return comment.time
- for num, comment in enumerate(sorted(bug.list_comments(), key=key)):
- yield ("%s:%d" % (unique_name, num+1), comment)
-
-
-def comment_from_name(bug, unique_name, name):
- """Use a comment name to look up a comment"""
- for cur_name, comment in iter_comment_name(bug, unique_name):
- if name == cur_name:
- return comment
- raise KeyError(name)
-
-
-def get_bug_and_comment(identifier, bug_dir=None):
- ids = identifier.split(':')
- bug = get_bug(ids[0], bug_dir)
- if len(ids) == 2:
- comment = comment_from_name(bug, ids[0], identifier)
- else:
- comment = None
- return bug, comment
-
class CmdOptionParser(optparse.OptionParser):
def __init__(self, usage):
@@ -174,44 +113,6 @@ def underlined(instring):
return "%s\n%s" % (instring, "="*len(instring))
-def print_threaded_comments(comments, name_map, indent=""):
- """Print a threaded display of comments"""
- tw = TextWrapper(initial_indent = indent, subsequent_indent = indent,
- width=80)
- for comment, children in comments:
- s = StringIO()
- print >> s, "--------- Comment ---------"
- print >> s, "Name: %s" % name_map[comment.uuid]
- print >> s, "From: %s" % comment.From
- print >> s, "Date: %s\n" % utility.time_to_str(comment.time)
- print >> s, comment.body.rstrip('\n')
-
- s.seek(0)
- for line in s:
- print tw.fill(line).rstrip('\n')
- print_threaded_comments(children, name_map, indent=indent+" ")
-
-
-def bug_tree(dir=None):
- """Retrieve the bug tree specified by the user. If no directory is
- specified, the current working directory is used.
-
- :param dir: The directory to search for the bug tree in.
-
- >>> bug_tree() is not None
- True
- >>> bug_tree("/")
- Traceback (most recent call last):
- UserErrorWrap: The directory "/" has no bug directory.
- """
- if dir is None:
- dir = os.getcwd()
- try:
- return bugdir.tree_root(dir)
- except bugdir.NoBugDir, e:
- raise UserErrorWrap(e)
-
-
def _test():
import doctest
import sys
diff --git a/libbe/diff.py b/libbe/diff.py
index 9fa3816..95d5607 100644
--- a/libbe/diff.py
+++ b/libbe/diff.py
@@ -20,33 +20,24 @@ from libbe.utility import time_to_str
from libbe.bug import cmp_severity
import doctest
-def diff(old_tree, new_tree):
- old_bug_map = old_tree.bug_map()
- new_bug_map = new_tree.bug_map()
+def diff(old_bugdir, new_bugdir):
added = []
removed = []
modified = []
- for old_bug in old_bug_map.itervalues():
- new_bug = new_bug_map.get(old_bug.uuid)
+ for old_bug in old_bugdir:
+ new_bug = new_bugdir.bug_map.get(old_bug.uuid)
if new_bug is None :
removed.append(old_bug)
else:
if old_bug != new_bug:
modified.append((old_bug, new_bug))
- for new_bug in new_bug_map.itervalues():
- if not old_bug_map.has_key(new_bug.uuid):
+ for new_bug in new_bugdir:
+ if not old_bugdir.bug_map.has_key(new_bug.uuid):
added.append(new_bug)
return (removed, modified, added)
-
-def reference_diff(bugdir, revision=None):
- d = diff(bugdir.duplicate_bugdir(revision), bugdir)
- bugdir.remove_duplicate_bugdir()
- return d
-
def diff_report(diff_data, bug_dir):
(removed, modified, added) = diff_data
- bugs = list(bug_dir.list())
def modified_cmp(left, right):
return cmp_severity(left[1], right[1])
@@ -54,7 +45,7 @@ def diff_report(diff_data, bug_dir):
removed.sort(cmp_severity)
modified.sort(modified_cmp)
- if len(added) > 0:
+ if len(added) > 0:
print "New bug reports:"
for bug in added:
print bug.string(shortlist=True)
@@ -62,7 +53,7 @@ def diff_report(diff_data, bug_dir):
if len(modified) > 0:
printed = False
for old_bug, new_bug in modified:
- change_str = bug_changes(old_bug, new_bug, bugs)
+ change_str = bug_changes(old_bug, new_bug, bug_dir)
if change_str is None:
continue
if not printed:
@@ -73,7 +64,7 @@ def diff_report(diff_data, bug_dir):
if len(removed) > 0:
print "Removed bug reports:"
for bug in removed:
- print bug.string(bugs, shortlist=True)
+ print bug.string(shortlist=True)
def change_lines(old, new, attributes):
change_list = []
@@ -91,8 +82,8 @@ def bug_changes(old, new, bugs):
change_list = change_lines(old, new, ("time", "creator", "severity",
"target", "summary", "status", "assigned"))
- old_comment_ids = list(old.iter_comment_ids())
- new_comment_ids = list(new.iter_comment_ids())
+ old_comment_ids = [c.uuid for c in old.comment_root.traverse()]
+ new_comment_ids = [c.uuid for c in new.comment_root.traverse()]
change_strings = ["%s: %s -> %s" % f for f in change_list]
for comment_id in new_comment_ids:
if comment_id not in old_comment_ids:
@@ -105,8 +96,8 @@ def bug_changes(old, new, bugs):
if len(change_strings) == 0:
return None
- return "%s%s\n" % (new.string(bugs, shortlist=True),
- "\n".join(change_strings))
+ return "%s\n %s" % (new.string(shortlist=True),
+ " \n".join(change_strings))
def comment_summary(comment, status):
diff --git a/libbe/mapfile.py b/libbe/mapfile.py
index 8f69554..9a7fa8b 100644
--- a/libbe/mapfile.py
+++ b/libbe/mapfile.py
@@ -95,11 +95,11 @@ def parse(f):
f = utility.get_file(f)
result = {}
for line in f:
- line = line.rstrip('\n')
+ line = line.decode("utf-8").rstrip('\n')
if len(line) == 0:
continue
- name,value = [f.decode('utf-8') for f in line.split('=', 1)]
- assert not result.has_key('name')
+ name,value = [f for f in line.split('=', 1)]
+ assert not result.has_key(name)
result[name] = value
return result
diff --git a/libbe/names.py b/libbe/names.py
deleted file mode 100644
index 6e0378e..0000000
--- a/libbe/names.py
+++ /dev/null
@@ -1,51 +0,0 @@
-# Copyright (C) 2005 Aaron Bentley and Panometrics, Inc.
-# <abentley@panoramicfeedback.com>
-#
-# 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 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, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-
-import os
-import sys
-import doctest
-
-def uuid():
- # this code borrowed from standard commands module
- # but adapted to win32
- pipe = os.popen('uuidgen', 'r')
- text = pipe.read()
- sts = pipe.close()
- if sts not in (0, None):
- raise "Failed to run uuidgen"
- if text[-1:] == '\n': text = text[:-1]
- return text
-
-def unique_name(bug, bugs):
- """
- 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 some_bug in bugs:
- if bug.uuid == some_bug.uuid:
- continue
- while (bug.uuid[:chars] == some_bug.uuid[:chars]):
- chars+=1
- return bug.uuid[:chars]
-
-suite = doctest.DocTestSuite()
diff --git a/libbe/plugin.py b/libbe/plugin.py
index 05a4398..0964fba 100644
--- a/libbe/plugin.py
+++ b/libbe/plugin.py
@@ -36,6 +36,8 @@ def iter_plugins(prefix):
modfiles = os.listdir(os.path.join(plugin_path, prefix))
modfiles.sort()
for modfile in modfiles:
+ if modfile.startswith('.'):
+ continue # the occasional emacs temporary file
if modfile.endswith(".py") and modfile != "__init__.py":
yield modfile[:-3], my_import(prefix+"."+modfile[:-3])
diff --git a/libbe/rcs.py b/libbe/rcs.py
index 2993a80..abd92cb 100644
--- a/libbe/rcs.py
+++ b/libbe/rcs.py
@@ -24,7 +24,7 @@ import tempfile
import shutil
import unittest
import doctest
-from utility import Dir
+from utility import Dir, search_parent_directories
def _get_matching_rcs(matchfn):
"""Return the first module for which matchfn(RCS_instance) is true"""
@@ -32,9 +32,9 @@ def _get_matching_rcs(matchfn):
import bzr
import hg
import git
- for module in [arch, bzr, hg, git]:
+ for module in [git, arch, bzr, hg, git]:
rcs = module.new()
- if matchfn(rcs):
+ if matchfn(rcs) == True:
return rcs
else:
del(rcs)
@@ -62,6 +62,9 @@ class CommandError(Exception):
class SettingIDnotSupported(NotImplementedError):
pass
+class PathNotInRoot(Exception):
+ pass
+
def new():
return RCS()
@@ -152,7 +155,10 @@ class RCS(object):
pass
def _rcs_get_file_contents(self, path, revision=None):
"""
- Get the file as it was in a given revision.
+ Get the file contents as they were in a given revision. Don't
+ worry about decoding the contents, the RCS.get_file_contents()
+ method will handle that.
+
Revision==None specifies the current revision.
"""
assert revision == None, \
@@ -180,7 +186,7 @@ class RCS(object):
if e.errno == errno.ENOENT:
return False
raise e
- def detect(self, path=None):
+ def detect(self, path="."):
"""
Detect whether a directory is revision controlled with this RCS.
"""
@@ -264,23 +270,28 @@ class RCS(object):
Revision==None specifies the current revision.
"""
relpath = self._u_rel_path(path)
- return self._rcs_get_file_contents(relpath, revision)
+ return self._rcs_get_file_contents(relpath, revision).decode("utf-8")
def set_file_contents(self, path, contents):
"""
Set the file contents under version control.
"""
add = not os.path.exists(path)
- file(path, "wb").write(contents)
+ file(path, "wb").write(contents.encode("utf-8"))
if add:
self.add(path)
else:
self.update(path)
def mkdir(self, path):
"""
- Created directory at path under version control.
+ Create (if neccessary) a directory at path under version
+ control.
"""
- os.mkdir(path)
- self.add(path)
+ if not os.path.exists(path):
+ os.mkdir(path)
+ self.add(path)
+ else:
+ assert os.path.isdir(path)
+ self.update(path)
def duplicate_repo(self, revision=None):
"""
Get the repository as it was in a given revision.
@@ -366,16 +377,7 @@ class RCS(object):
/.be
or None if none of those files exist.
"""
- path = os.path.realpath(path)
- assert os.path.exists(path)
- old_path = None
- while True:
- if os.path.exists(os.path.join(path, filename)):
- return os.path.join(path, filename)
- if path == old_path:
- return None
- old_path = path
- path = os.path.dirname(path)
+ return search_parent_directories(path, filename)
def _u_rel_path(self, path, root=None):
"""
Return the relative path to path from root.
@@ -389,8 +391,9 @@ class RCS(object):
if os.path.isabs(path):
absRoot = os.path.abspath(root)
absRootSlashedDir = os.path.join(absRoot,"")
- assert path.startswith(absRootSlashedDir), \
- "file %s not in root %s" % (path, absRootSlashedDir)
+ if not path.startswith(absRootSlashedDir):
+ raise PathNotInRoot, \
+ "file %s not in root %s" % (path, absRootSlashedDir)
assert path != absRootSlashedDir, \
"file %s == root directory %s" % (path, absRootSlashedDir)
path = path[len(absRootSlashedDir):]
diff --git a/libbe/utility.py b/libbe/utility.py
index f595bdb..81023cd 100644
--- a/libbe/utility.py
+++ b/libbe/utility.py
@@ -71,6 +71,32 @@ def get_file(f):
else:
return f
+def search_parent_directories(path, filename):
+ """
+ Find the file (or directory) named filename in path or in any
+ of path's parents.
+
+ e.g.
+ search_parent_directories("/a/b/c", ".be")
+ will return the path to the first existing file from
+ /a/b/c/.be
+ /a/b/.be
+ /a/.be
+ /.be
+ or None if none of those files exist.
+ """
+ path = os.path.realpath(path)
+ assert os.path.exists(path)
+ old_path = None
+ while True:
+ check_path = os.path.join(path, filename)
+ if os.path.exists(check_path):
+ return check_path
+ if path == old_path:
+ return None
+ old_path = path
+ path = os.path.dirname(path)
+
class Dir:
"A temporary directory for testing use"
def __init__(self):