-status=closed
+status=fixed
-status=open
+status=fixed
--- /dev/null
+Problem was due to
+ open-value-file
+ write-value-file
+ add/update-value-file
+which should be (and now is)
+ open-value-file
+ write-value-file
+ close-value-file
+ add/update-value-file
+since it was getting added before the changes we'd written were flushed out.
--- /dev/null
+
+
+
+Content-type=text/plain
+
+
+
+
+
+
+Date=Wed, 19 Nov 2008 01:12:37 +0000
+
+
+
+
+
+
+From=John Doe <jdoe@example.com>
+
+
+
--- /dev/null
+It looks like the mapfiles are not being 'git add'ed after changes.
+
+$ mkdir BEtest
+$ cd BEtest
+$ git init
+$ be set-root .
+$ be new 'new'
+$ git status
+# On branch master
+#
+# Initial commit
+#
+# Changes to be committed:
+# (use "git rm --cached <file>..." to unstage)
+#
+# new file: .be/bugs/8f021d79-44f5-479f-af12-c37e2caf3ce1/values
+# new file: .be/settings
+# new file: .be/version
+#
+# Changed but not updated:
+# (use "git add <file>..." to update what will be committed)
+#
+# modified: .be/bugs/8f021d79-44f5-479f-af12-c37e2caf3ce1/values
+# modified: .be/settings
+#
+
--- /dev/null
+
+
+
+Content-type=text/plain
+
+
+
+
+
+
+Date=Mon, 17 Nov 2008 15:03:58 +0000
+
+
+
+
+
+
+From=wking
+
+
+
--- /dev/null
+
+
+
+creator=wking
+
+
+
+
+
+
+severity=serious
+
+
+
+
+
+
+status=open
+
+
+
+
+
+
+summary=BE not notifying git of some changed files
+
+
+
+
+
+
+time=Mon, 17 Nov 2008 15:02:15 +0000
+
+
+
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Assign an individual or group to fix a bug"""
-from libbe import cmdutil, names
+from libbe import cmdutil
__desc__ = __doc__
def execute(args):
"""
- >>> from libbe import tests, names
+ >>> from libbe import bugdir
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> dir.get_bug("a").assigned is None
True
>>> execute(["a"])
- >>> dir.get_bug("a").assigned == names.creator()
+ >>> dir.get_bug("a").assigned == dir.rcs.get_user_id()
True
>>> execute(["a", "someone"])
>>> dir.get_bug("a").assigned
>>> execute(["a","none"])
>>> dir.get_bug("a").assigned is None
True
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
assert(len(args) in (0, 1, 2))
return
bug = cmdutil.get_bug(args[0])
if len(args) == 1:
- bug.assigned = names.creator()
+ bug.assigned = bug.rcs.get_user_id()
elif len(args) == 2:
if args[1] == "none":
bug.assigned = None
def execute(args):
"""
- >>> from libbe import tests
+ >>> from libbe import bugdir
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> dir.get_bug("a").status
u'open'
>>> execute(["a"])
>>> dir.get_bug("a").status
u'closed'
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
if len(args) !=1:
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Add a comment to a bug"""
-from libbe import cmdutil, names, utility
+from libbe import cmdutil, utility
from libbe.bug import new_comment
import os
__desc__ = __doc__
def execute(args):
"""
- >>> from libbe import tests, names
+ >>> from libbe import bugdir
>>> import os, time
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> execute(["a", "This is a comment about a"])
>>> comment = dir.get_bug("a").list_comments()[0]
>>> comment.body
u'This is a comment about a\\n'
- >>> comment.From == names.creator()
+ >>> comment.From == dir.rcs.get_user_id()
True
>>> comment.time <= int(time.time())
True
>>> execute(["b"])
>>> dir.get_bug("b").list_comments()[0].body
u'I like cheese\\n'
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
if len(args) < 1:
if title != None:
print cmdutil.underlined(title)
for bug in cur_bugs:
- print bug.string(all_bugs, shortlist=True),
+ print bug.string(all_bugs, shortlist=True)
list_bugs(bugs, no_target=False)
short = "-%c" % s[0]
long = "--%s" % s[1]
help = s[2]
- parser.add_option(short, long, action="store_true", dest=attr, help=help)
+ parser.add_option(short, long, action="store_true",
+ dest=attr, help=help)
return parser
longhelp="""
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Create a new bug"""
-from libbe import cmdutil, names, utility
+from libbe import cmdutil, names
from libbe.bug import new_bug
__desc__ = __doc__
def execute(args):
"""
>>> import os, time
- >>> from libbe import tests
- >>> dir = tests.bug_arch_dir()
+ >>> from libbe import bugdir
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
- >>> names.uuid = lambda: "a"
+ >>> names.uuid = lambda: "X"
>>> execute (["this is a test",])
- Created bug with ID a
- >>> bug = list(dir.list())[0]
+ Created bug with ID X
+ >>> bug = cmdutil.get_bug("X", dir)
>>> bug.summary
u'this is a test'
>>> bug.creator = os.environ["LOGNAME"]
u'minor'
>>> bug.target == None
True
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
if len(args) != 1:
def execute(args):
"""
- >>> from libbe import tests
+ >>> from libbe import bugdir
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> dir.get_bug("b").status
u'closed'
>>> execute(["b"])
>>> dir.get_bug("b").status
u'open'
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
if len(args) !=1:
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Remove (delete) a bug and its comments"""
-from libbe import cmdutil, names, utility
-from libbe.bug import new_bug
+from libbe import cmdutil
__desc__ = __doc__
def execute(args):
"""
>>> import os
- >>> from libbe import tests, mapfile
- >>> dir = tests.simple_bug_dir()
+ >>> from libbe import bugdir, mapfile
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> dir.get_bug("b").status
u'closed'
... except mapfile.NoSuchFile:
... print "Bug not found"
Bug not found
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
if len(args) != 1:
def execute(args):
"""
- >>> from libbe import tests
+ >>> from libbe import bugdir
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> execute(["a"])
None
>>> execute(["a", "none"])
>>> execute(["a"])
None
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
if len(args) > 2:
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Assign the root directory for bug tracking"""
+import os.path
from libbe import bugdir, cmdutil, rcs
__desc__ = __doc__
def execute(args):
"""
- >>> from libbe import tests
+ >>> from libbe import utility
>>> import os
- >>> dir = tests.Dir()
+ >>> dir = utility.Dir()
>>> try:
- ... bugdir.tree_root(dir.name)
+ ... bugdir.tree_root(dir.path)
... except bugdir.NoBugDir, e:
... True
True
- >>> execute([dir.name])
+ >>> execute([dir.path])
No revision control detected.
Directory initialized.
- >>> bd = bugdir.tree_root(dir.name)
- >>> bd.root = dir.name
- >>> dir = tests.arch_dir()
- >>> os.chdir(dir.name)
+ >>> bd = bugdir.tree_root(dir.path)
+ >>> bd.root = dir.path
+ >>> dir_rcs = rcs.installed_rcs()
+ >>> dir_rcs.init(bd.dir)
+ >>> bd.rcs_name = dir_rcs.name
+ >>> del(dir_rcs)
+ >>> os.chdir(bd.dir)
>>> execute(['.'])
Using Arch for revision control.
Directory initialized.
- >>> bd = bugdir.tree_root(dir.name+"/{arch}")
- >>> bd.root = dir.name
>>> try:
... execute(['.'])
... except cmdutil.UserError, e:
... str(e).startswith("Directory already initialized: ")
True
- >>> tests.clean_up()
>>> execute(['/highly-unlikely-to-exist'])
Traceback (most recent call last):
UserError: No such directory: /highly-unlikely-to-exist
"""
options, args = get_parser().parse_args(args)
+ basedir = args[0]
if len(args) != 1:
raise cmdutil.UsageError
- dir_rcs = rcs.detect(args[0])
+ if os.path.exists(basedir) == False:
+ raise cmdutil.UserError, "No such directory: %s" % basedir
+ dir_rcs = rcs.detect_rcs(basedir)
+ dir_rcs.root(basedir)
try:
- bugdir.create_bug_dir(args[0], dir_rcs)
+ bugdir.create_bug_dir(basedir, dir_rcs)
except bugdir.NoRootEntry:
- raise cmdutil.UserError("No such directory: %s" % args[0])
+ raise cmdutil.UserError("No such directory: %s" % basedir)
except bugdir.AlreadyInitialized:
- raise cmdutil.UserError("Directory already initialized: %s" % args[0])
+ raise cmdutil.UserError("Directory already initialized: %s" % basedir)
if dir_rcs.name is not "None":
print "Using %s for revision control." % dir_rcs.name
else:
def execute(args):
"""
- >>> from libbe import tests
+ >>> from libbe import bugdir
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> execute(["a"])
minor
>>> execute(["a", "none"])
Traceback (most recent call last):
UserError: Invalid severity level: none
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
assert(len(args) in (0, 1, 2))
def execute(args):
"""
- >>> from libbe import tests
+ >>> from libbe import bugdir
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> execute(["a"])
open
>>> execute(["a", "none"])
Traceback (most recent call last):
UserError: Invalid status: none
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
assert(len(args) in (0, 1, 2))
def execute(args):
"""
- >>> from libbe import tests
>>> import os
- >>> dir = tests.simple_bug_dir()
+ >>> dir = bugdir.simple_bug_dir()
>>> os.chdir(dir.dir)
>>> execute(["a"])
No target assigned.
>>> execute(["a", "none"])
>>> execute(["a"])
No target assigned.
- >>> tests.clean_up()
"""
options, args = get_parser().parse_args(args)
assert(len(args) in (0, 1, 2))
# 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
+
+# OUTDATED
+
"""Upgrade the bugs to the latest format"""
import os.path
import errno
# 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 config
-import errno
+import shutil
+import time
+import re
+import unittest
+import doctest
-from rcs import invoke
+import config
+from rcs import RCS, RCStestCase, CommandError
client = config.get_val("arch_client")
if client is None:
client = "tla"
config.set_val("arch_client", client)
-
-def invoke_client(*args, **kwargs):
- cl_args = [client]
- cl_args.extend(args)
- status,output,error = invoke(cl_args)
- if status not in (0,):
- raise Exception("Command failed: %s" % error)
- return output
-
-def get_user_id():
- try:
- return invoke_client('my-id')
- except Exception, e:
- if 'no arch user id set' in e.args[0]:
- return None
+def new():
+ return Arch()
+
+class Arch(RCS):
+ name = "Arch"
+ client = client
+ versioned = True
+ _archive_name = None
+ _archive_dir = None
+ _tmp_archive = False
+ _project_name = None
+ _tmp_project = False
+ _arch_paramdir = os.path.expanduser("~/.arch-params")
+ def _rcs_help(self):
+ status,output,error = self._u_invoke_client("--help")
+ return output
+ def _rcs_detect(self, path):
+ """Detect whether a directory is revision-controlled using Arch"""
+ 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:
- raise
-
-
-def set_user_id(value):
- invoke_client('my-id', value)
-
-
-def ensure_user_id():
- if get_user_id() is None:
- set_user_id('nobody <nobody@example.com>')
-
-
-def write_tree_settings(contents, path):
- file(os.path.join(path, "{arch}", "=tagging-method"), "wb").write(contents)
-
-def init_tree(path):
- invoke_client("init-tree", "-d", path)
-
-def temp_arch_tree(type="easy"):
- import tempfile
- ensure_user_id()
- path = tempfile.mkdtemp()
- init_tree(path)
- if type=="easy":
- write_tree_settings("source ^.*$\n", path)
- elif type=="tricky":
- write_tree_settings("source ^$\n", path)
- else:
- assert (type=="impossible")
- add_dir_rule("precious ^\.boo$", path, path)
- return path
-
-def list_added(root):
- assert os.path.exists(root)
- assert os.access(root, os.X_OK)
- root = os.path.realpath(root)
- inv_str = invoke_client("inventory", "--source", '--both', '--all', root)
- return [os.path.join(root, p) for p in inv_str.split('\n')]
-
-def tree_root(filename):
- assert os.path.exists(filename)
- if not os.path.isdir(filename):
- dirname = os.path.dirname(filename)
- else:
- dirname = filename
- return invoke_client("tree-root", dirname).rstrip('\n')
-
-def rel_filename(filename, root):
- filename = os.path.realpath(filename)
- root = os.path.realpath(root)
- assert(filename.startswith(root))
- return filename[len(root)+1:]
+ 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)
+ self._add_project_code(path)
+ def _create_archive(self, path):
+ # Create a new archive
+ # http://regexps.srparish.net/tutorial-tla/new-archive.html#Creating_a_New_Archive
+ assert self._archive_name == None
+ id = self.get_user_id()
+ name, email = self._u_parse_id(id)
+ if email == None:
+ email = "%s@example.com" % name
+ trailer = "%s-%s" % ("bugs-everywhere-auto",
+ time.strftime("%Y.%H.%M.%S"))
+ self._archive_name = "%s--%s" % (email, trailer)
+ self._archive_dir = "/tmp/%s" % trailer
+ self._tmp_archive = True
+ self._u_invoke_client("make-archive", self._archive_name,
+ self._archive_dir, directory=path)
+ def _invoke_client(self, *args, **kwargs):
+ """
+ Invoke the client on our archive.
+ """
+ assert self._archive_name != None
+ command = args[0]
+ if len(args) > 1:
+ tailargs = args[1:]
+ else:
+ tailargs = []
+ arglist = [command, "-A", self._archive_name]
+ arglist.extend(tailargs)
+ args = tuple(arglist)
+ return self._u_invoke_client(*args, **kwargs)
+ def _remove_archive(self):
+ assert self._tmp_archive == True
+ assert self._archive_dir != None
+ assert self._archive_name != None
+ os.remove(os.path.join(self._arch_paramdir,
+ "=locations", self._archive_name))
+ shutil.rmtree(self._archive_dir)
+ self._tmp_archive = False
+ self._archive_dir = False
+ self._archive_name = False
+ def _create_project(self, path):
+ # http://mwolson.org/projects/GettingStartedWithArch.html
+ # http://regexps.srparish.net/tutorial-tla/new-project.html#Starting_a_New_Project
+ category = "bugs-everywhere"
+ branch = "mainline"
+ version = "0.1"
+ self._project_name = "%s--%s--%s" % (category, branch, version)
+ self._invoke_client("archive-setup", self._project_name,
+ directory=path)
+ def _remove_project(self):
+ assert self._tmp_project == True
+ assert self._project_name != None
+ assert self._archive_dir != None
+ shutil.rmtree(os.path.join(self._archive_dir, self._project_name))
+ self._tmp_project = False
+ self._project_name = False
+ def _archive_project_name(self):
+ assert self._archive_name != None
+ assert self._project_name != None
+ return "%s/%s" % (self._archive_name, self._project_name)
+ 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(),
+ directory=path)
+ self._invoke_client("import", "--summary", "Began versioning",
+ directory=path)
+ def _rcs_cleanup(self):
+ if self._tmp_project == True:
+ self._remove_project()
+ if self._tmp_archive == True:
+ self._remove_archive()
+ def _rcs_get_user_id(self):
+ try:
+ status,output,error = self._u_invoke_client('my-id')
+ return output.rstrip('\n')
+ except Exception, e:
+ if 'no arch user id set' in e.args[0]:
+ return None
+ else:
+ raise
+ def _rcs_set_user_id(self, value):
+ self._u_invoke_client('my-id', value)
+ def _rcs_add(self, path):
+ self._u_invoke_client("add-id", path)
+ realpath = os.path.realpath(self._u_abspath(path))
+ pathAdded = realpath in self._list_added(self.rootdir)
+ if self.paranoid and not pathAdded:
+ self._force_source(path)
+ def _list_added(self, root):
+ assert os.path.exists(root)
+ assert os.access(root, os.X_OK)
+ root = os.path.realpath(root)
+ status,output,error = self._u_invoke_client("inventory", "--source",
+ "--both", "--all", root)
+ inv_str = output.rstrip('\n')
+ return [os.path.join(root, p) for p in inv_str.split('\n')]
+ def _add_dir_rule(self, rule, dirname, root):
+ inv_path = os.path.join(dirname, '.arch-inventory')
+ file(inv_path, "ab").write(rule)
+ if os.path.realpath(inv_path) not in self._list_added(root):
+ paranoid = self.paranoid
+ self.paranoid = False
+ self.add(inv_path)
+ self.paranoid = paranoid
+ def _force_source(self, path):
+ rule = "source %s\n" % self._u_rel_path(path)
+ self._add_dir_rule(rule, os.path.dirname(path), self.rootdir)
+ if os.path.realpath(path) not in self._list_added(self.rootdir):
+ raise CantAddFile(path)
+ def _rcs_remove(self, path):
+ if not '.arch-ids' in path:
+ self._u_invoke_client("delete-id", path)
+ def _rcs_update(self, path):
+ pass
+ def _rcs_get_file_contents(self, path, revision=None):
+ if revision == None:
+ return file(self._u_abspath(path), "rb").read()
+ else:
+ status,output,error = \
+ self._invoke_client("file-find", path, revision)
+ path = output.rstrip('\n')
+ return file(self._u_abspath(path), "rb").read()
+ def _rcs_duplicate_repo(self, directory, revision=None):
+ if revision == None:
+ RCS._rcs_duplicate_repo(self, directory, revision)
+ else:
+ status,output,error = \
+ self._u_invoke_client("get", revision,directory)
+ def _rcs_commit(self, commitfile):
+ summary,body = self._u_parse_commitfile(commitfile)
+ #status,output,error = self._invoke_client("make-log")
+ if body == None:
+ status,output,error \
+ = self._invoke_client("commit","--summary",summary)
+ else:
+ status,output,error \
+ = self._invoke_client("commit","--summary",summary,
+ "--log-message",body)
+ revision = None
+ revline = re.compile("[*] committed (.*)")
+ match = revline.search(output)
+ assert match != None, output+error
+ assert len(match.groups()) == 1
+ revpath = match.groups()[0]
+ assert not " " in revpath, revpath
+ assert revpath.startswith(self._archive_project_name()+'--')
+ revision = revpath[len(self._archive_project_name()+'--'):]
+ return revpath
class CantAddFile(Exception):
def __init__(self, file):
self.file = file
Exception.__init__(self, "Can't automatically add file %s" % file)
+class ArchTestCase(RCStestCase):
+ Class = Arch
-def add_dir_rule(rule, dirname, root):
- inv_filename = os.path.join(dirname, '.arch-inventory')
- file(inv_filename, "ab").write(rule)
- if os.path.realpath(inv_filename) not in list_added(root):
- add_id(inv_filename, paranoid=False)
-
-def force_source(filename, root):
- rule = "source %s\n" % rel_filename(filename, root)
- add_dir_rule(rule, os.path.dirname(filename), root)
- if os.path.realpath(filename) not in list_added(root):
- raise CantAddFile(filename)
-
-def add_id(filename, paranoid=False):
- invoke_client("add-id", filename)
- root = tree_root(filename)
- if paranoid and os.path.realpath(filename) not in list_added(root):
- force_source(filename, root)
-
-
-def delete_id(filename):
- invoke_client("delete-id", filename)
-
-def test_helper(type):
- t = temp_arch_tree(type)
- dirname = os.path.join(t, ".boo")
- return dirname, t
-
-def mkdir(path, paranoid=False):
- """
- >>> import shutil
- >>> dirname,t = test_helper("easy")
- >>> mkdir(dirname, paranoid=False)
- >>> assert os.path.realpath(dirname) in list_added(t)
- >>> assert not os.path.exists(os.path.join(t, ".arch-inventory"))
- >>> shutil.rmtree(t)
- >>> dirname,t = test_helper("tricky")
- >>> mkdir(dirname, paranoid=True)
- >>> assert os.path.realpath(dirname) in list_added(t)
- >>> assert os.path.exists(os.path.join(t, ".arch-inventory"))
- >>> shutil.rmtree(t)
- >>> dirname,t = test_helper("impossible")
- >>> try:
- ... mkdir(dirname, paranoid=True)
- ... except CantAddFile, e:
- ... print "Can't add file"
- Can't add file
- >>> shutil.rmtree(t)
- """
- os.mkdir(path)
- add_id(path, paranoid=paranoid)
-
-def set_file_contents(path, contents):
- add = not os.path.exists(path)
- file(path, "wb").write(contents)
- if add:
- add_id(path)
-
-
-def path_in_reference(bug_dir, spec):
- if spec is not None:
- return invoke_client("file-find", bug_dir, spec).rstrip('\n')
- return invoke_client("file-find", bug_dir).rstrip('\n')
-
-
-def unlink(path):
- try:
- os.unlink(path)
- delete_id(path)
- except OSError, e:
- if e.errno != 2:
- raise
-
-
-def detect(path):
- """Detect whether a directory is revision-controlled using Arch"""
- path = os.path.realpath(path)
- old_path = None
- while True:
- if os.path.exists(os.path.join(path, "{arch}")):
- return True
- if path == old_path:
- return False
- old_path = path
- path = os.path.join('..', path)
-
-def precommit(directory):
- pass
-
-def commit(directory, summary, body=None):
- pass
-
-def postcommit(directory):
- pass
-
-
-name = "Arch"
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(ArchTestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import os
import os.path
-import shutil
import errno
import names
import mapfile
import time
import utility
-from rcs import rcs_by_name
-
+import doctest
### Define and describe valid bug categories
# Use a tuple of (category, description) tuples since we don't have
severity = checked_property("severity", severity_values)
status = checked_property("status", status_values)
- def __init__(self, path, uuid, rcs_name, bugdir):
+ def __init__(self, path, uuid, rcs, bugdir):
self.path = path
self.uuid = uuid
if uuid is not None:
else:
dict = {}
- self.rcs_name = rcs_name
+ self.rcs = rcs
self.bugdir = bugdir
self.summary = dict.get("summary")
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 __repr__(self):
return "Bug(uuid=%r)" % self.uuid
statuschar = self.status[0]
severitychar = self.severity[0]
chars = "%c%c" % (statuschar, severitychar)
- return "%s:%s: %s\n" % (short_name, chars, self.summary)
+ return "%s:%s: %s" % (short_name, chars, self.summary)
+
def __str__(self):
return self.string(shortlist=True)
- 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 __cmp__(self, other):
+ return cmp_full(self, other)
def add_attr(self, map, name):
value = getattr(self, name)
map[name] = value
def save(self):
+ assert self.summary != None, "Can't save blank bug"
map = {}
self.add_attr(map, "assigned")
self.add_attr(map, "summary")
if self.time is not None:
map["time"] = utility.time_to_str(self.time)
path = self.get_path("values")
- mapfile.map_save(rcs_by_name(self.rcs_name), path, map)
-
+ mapfile.map_save(self.rcs, path, map)
+
def remove(self):
path = self.get_path()
- shutil.rmtree(path)
+ self.rcs.recursive_remove(path)
- def _get_rcs(self):
- return rcs_by_name(self.rcs_name)
-
- rcs = property(_get_rcs)
-
def new_comment(self):
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
return comm
def get_comment(self, uuid):
def new_bug(dir, uuid=None):
bug = dir.new_bug(uuid)
- bug.creator = names.creator()
+ bug.creator = bug.rcs.get_user_id()
bug.severity = "minor"
bug.status = "open"
bug.time = time.time()
def new_comment(bug, body=None):
comm = bug.new_comment()
- comm.From = names.creator()
+ comm.From = comm.rcs.get_user_id()
comm.time = time.time()
comm.body = body
return comm
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)
Exception.__init__(self, msg)
self.name = name
self.value = value
+
+suite = doctest.DocTestSuite()
import os.path
import cmdutil
import errno
+import unittest
+import doctest
import names
import mapfile
import time
import utility
-from rcs import rcs_by_name
+from rcs import rcs_by_name, installed_rcs
from bug import Bug
class NoBugDir(Exception):
Exception.__init__(self,
"Specified root is already initialized: %s" % path)
+def bugdir_root(versioning_root):
+ return os.path.join(versioning_root, ".be")
+
def create_bug_dir(path, rcs):
"""
- >>> import no_rcs, tests
- >>> create_bug_dir('/highly-unlikely-to-exist', no_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
- >>> test_dir = os.path.dirname(tests.bug_arch_dir().dir)
- >>> try:
- ... create_bug_dir(test_dir, no_rcs)
- ... except AlreadyInitialized, e:
- ... print "Already Initialized"
- Already Initialized
"""
root = os.path.join(path, ".be")
try:
- rcs.mkdir(root, paranoid=True)
+ rcs.mkdir(root)
except OSError, e:
if e.errno == errno.ENOENT:
raise NoRootEntry(path)
set_version(root, rcs)
mapfile.map_save(rcs,
os.path.join(root, "settings"), {"rcs_name": rcs.name})
- return BugDir(os.path.join(path, ".be"))
+ return BugDir(bugdir_root(path))
def setting_property(name, valid=None):
self.dir = dir
self.bugs_path = os.path.join(self.dir, "bugs")
try:
- self.settings = mapfile.map_load(os.path.join(self.dir, "settings"))
+ self.settings = mapfile.map_load(os.path.join(self.dir,"settings"))
except mapfile.NoSuchFile:
self.settings = {"rcs_name": "None"}
- rcs_name = setting_property("rcs_name", ("None", "bzr", "git", "Arch", "hg"))
+ rcs_name = setting_property("rcs_name",
+ ("None", "bzr", "git", "Arch", "hg"))
_rcs = None
target = setting_property("target")
mapfile.map_save(self.rcs,
os.path.join(self.dir, "settings"), self.settings)
- def get_rcs(self):
- if self._rcs is not None and self.rcs_name == self._rcs.name:
- return self._rcs
+ 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)
return self._rcs
- rcs = property(get_rcs)
+ rcs = property(_get_rcs)
+
+ def duplicate_bugdir(self, revision):
+ return BugDir(bugdir_root(self.rcs.duplicate_repo(revision)))
- def get_reference_bugdir(self, spec):
- return BugDir(self.rcs.path_in_reference(self.dir, spec))
+ def remove_duplicate_bugdir(self):
+ self.rcs.remove_duplicate_repo()
def list(self):
for uuid in self.list_uuids():
return bugs
def get_bug(self, uuid):
- return Bug(self.bugs_path, uuid, self.rcs_name, self)
+ return Bug(self.bugs_path, uuid, self.rcs, self)
def list_uuids(self):
for uuid in os.listdir(self.bugs_path):
uuid = names.uuid()
path = os.path.join(self.bugs_path, uuid)
self.rcs.mkdir(path)
- bug = Bug(self.bugs_path, None, self.rcs_name, self)
+ bug = Bug(self.bugs_path, None, self.rcs, self)
bug.uuid = uuid
return bug
Exception.__init__(self, msg)
self.name = name
self.value = value
+
+def simple_bug_dir():
+ """
+ For testing
+ >>> bugdir = simple_bug_dir()
+ >>> ls = list(bugdir.list_uuids())
+ >>> ls.sort()
+ >>> print ls
+ ['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._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_b.status = "closed"
+ bug_b.summary = "Bug B"
+ bug_b.save()
+ return bugdir
+
+
+class BugDirTestCase(unittest.TestCase):
+ def __init__(self, *args, **kwargs):
+ 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)
+ def tearDown(self):
+ del(self.rcs)
+ del(self.dir)
+ 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)
+ def testBugDirDuplicate(self):
+ self.assertRaises(AlreadyInitialized, create_bug_dir,
+ self.dir.path, self.rcs)
+
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(BugDirTestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
# 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 tempfile
-
-import config
-from rcs import invoke, CommandError
-
-def invoke_client(*args, **kwargs):
- directory = kwargs['directory']
- expect = kwargs.get('expect', (0, 1))
- cl_args = ["bzr"]
- cl_args.extend(args)
- status,output,error = invoke(cl_args, expect, cwd=directory)
- return status, output
-
-def add_id(filename, paranoid=False):
- invoke_client("add", filename, directory='.')
-
-def delete_id(filename):
- invoke_client("remove", filename, directory='.')
-
-def mkdir(path, paranoid=False):
- os.mkdir(path)
- add_id(path)
-
-def set_file_contents(path, contents):
- add = not os.path.exists(path)
- file(path, "wb").write(contents)
- if add:
- add_id(path)
-
-def lookup_revision(revno, directory):
- return invoke_client("lookup-revision", str(revno),
- directory=directory)[1].rstrip('\n')
-
-def export(revno, directory, revision_dir):
- invoke_client("export", "-r", str(revno), revision_dir, directory=directory)
-
-def find_or_make_export(revno, directory):
- revision_id = lookup_revision(revno, directory)
- home = os.path.expanduser("~")
- revision_root = os.path.join(home, ".bzrrevs")
- if not os.path.exists(revision_root):
- os.mkdir(revision_root)
- revision_dir = os.path.join(revision_root, revision_id)
- if not os.path.exists(revision_dir):
- export(revno, directory, revision_dir)
- return revision_dir
-
-def bzr_root(path):
- return invoke_client("root", path, directory=None)[1].rstrip('\r')
-
-def path_in_reference(bug_dir, spec):
- if spec is None:
- spec = int(invoke_client("revno", directory=bug_dir)[1])
- rel_bug_dir = bug_dir[len(bzr_root(bug_dir)):]
- export_root = find_or_make_export(spec, directory=bug_dir)
- return os.path.join(export_root, rel_bug_dir)
-
-
-def unlink(path):
- try:
- os.unlink(path)
- delete_id(path)
- except OSError, e:
- if e.errno != 2:
- raise
-
-
-def detect(path):
- """Detect whether a directory is revision-controlled using bzr"""
- path = os.path.realpath(path)
- old_path = None
- while True:
- if os.path.exists(os.path.join(path, ".bzr")):
+import re
+import unittest
+import doctest
+
+from rcs import RCS, RCStestCase, CommandError
+
+def new():
+ return Bzr()
+
+class Bzr(RCS):
+ name = "bzr"
+ client = "bzr"
+ versioned = True
+ def _rcs_help(self):
+ status,output,error = self._u_invoke_client("--help")
+ return output
+ def _rcs_detect(self, path):
+ if self._u_search_parent_directories(path, ".bzr") != None :
return True
- if path == old_path:
- return False
- old_path = path
- path = os.path.dirname(path)
-
-def precommit(directory):
- pass
-
-def commit(directory, summary, body=None):
- if body is not None:
- summary += '\n' + body
- descriptor, filename = tempfile.mkstemp()
- try:
- temp_file = os.fdopen(descriptor, 'wb')
- temp_file.write(summary)
- temp_file.close()
- invoke_client('commit', '--unchanged', '--file', filename,
- directory=directory)
- finally:
- os.unlink(filename)
-
-def postcommit(directory):
- try:
- invoke_client('merge', directory=directory)
- except CommandError, e:
- if ('No merge branch known or specified' in e.err_str or
- 'No merge location known or specified' in e.err_str):
- pass
+ return False
+ def _rcs_root(self, path):
+ """Find the root of the deepest repository containing path."""
+ status,output,error = self._u_invoke_client("root", path)
+ return output.rstrip('\n')
+ def _rcs_init(self, path):
+ self._u_invoke_client("init", directory=path)
+ def _rcs_get_user_id(self):
+ status,output,error = self._u_invoke_client("whoami")
+ return output.rstrip('\n')
+ def _rcs_set_user_id(self, value):
+ self._u_invoke_client("whoami", value)
+ def _rcs_add(self, path):
+ self._u_invoke_client("add", path)
+ def _rcs_remove(self, path):
+ # --force to also remove unversioned files.
+ self._u_invoke_client("remove", "--force", path)
+ def _rcs_update(self, path):
+ pass
+ def _rcs_get_file_contents(self, path, revision=None):
+ if revision == None:
+ return file(os.path.join(self.rootdir, path), "rb").read()
+ else:
+ status,output,error = \
+ self._u_invoke_client("cat","-r",revision,path)
+ return output
+ def _rcs_duplicate_repo(self, directory, revision=None):
+ if revision == None:
+ RCS._rcs_duplicate_repo(self, directory, revision)
else:
- status = invoke_client('revert', '--no-backup',
+ self._u_invoke_client("branch", "--revision", revision,
+ ".", directory)
+ def _rcs_commit(self, commitfile):
+ status,output,error = self._u_invoke_client("commit", "--unchanged",
+ "--file", commitfile)
+ revision = None
+ revline = re.compile("Committed revision (.*)[.]")
+ match = revline.search(error)
+ assert match != None, output+error
+ assert len(match.groups()) == 1
+ revision = match.groups()[0]
+ return revision
+ def postcommit(self):
+ try:
+ self._u_invoke_client('merge')
+ except CommandError, e:
+ if ('No merge branch known or specified' in e.err_str or
+ 'No merge location known or specified' in e.err_str):
+ pass
+ else:
+ self._u_invoke_client('revert', '--no-backup',
directory=directory)
- status = invoke_client('resolve', '--all', directory=directory)
- raise
- if len(invoke_client('status', directory=directory)[1]) > 0:
- commit(directory, 'Merge from upstream')
-
-name = "bzr"
+ self._u_invoke_client('resolve', '--all', directory=directory)
+ raise
+ if len(self._u_invoke_client('status', directory=directory)[1]) > 0:
+ self.commit('Merge from upstream')
+
+class BzrTestCase(RCStestCase):
+ Class = Bzr
+
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(BzrTestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
from textwrap import TextWrapper
from StringIO import StringIO
import utility
+import doctest
class UserError(Exception):
def __init__(self, msg):
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:
if __name__ == "__main__":
_test()
+
+suite = doctest.DocTestSuite()
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import ConfigParser
import os.path
+import doctest
+
def path():
"""Return the path to the per-user config file"""
return os.path.expanduser("~/.bugs_everywhere")
return config.get(section, name)
except ConfigParser.NoOptionError:
return None
+
+suite = doctest.DocTestSuite()
from libbe import cmdutil, bugdir
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()
return (removed, modified, added)
-def reference_diff(bugdir, spec=None):
- return diff(bugdir.get_reference_bugdir(spec), bugdir)
-
+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 comment_summary(comment, status):
return "%8s comment from %s on %s" % (status, comment.From,
time_to_str(comment.time))
+
+suite = doctest.DocTestSuite()
# 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 tempfile
-
-from rcs import invoke
-
-def strip_git(filename):
- # Find the base path of the GIT tree, in order to strip that leading
- # path from arguments to git -- it doesn't like absolute paths.
- if os.path.isabs(filename):
- absRepoDir = os.path.abspath(git_repo_for_path('.'))
- absRepoSlashedDir = os.path.join(absRepoDir,"")
- assert filename.startswith(absRepoSlashedDir), \
- "file %s not in git repo %s" % (filename, absRepoSlashedDir)
- filename = filename[len(absRepoSlashedDir):]
- return filename
-
-def invoke_client(*args, **kwargs):
- directory = kwargs['directory']
- expect = kwargs.get('expect', (0, 1))
- cl_args = ["git"]
- cl_args.extend(args)
- status,output,error = invoke(cl_args, expect, cwd=directory)
- return status, output
-
-def add_id(filename, paranoid=False):
- filename = strip_git(filename)
- invoke_client("add", filename, directory=git_repo_for_path('.'))
-
-def delete_id(filename):
- filename = strip_git(filename)
- invoke_client("rm", filename, directory=git_repo_for_path('.'))
-
-def mkdir(path, paranoid=False):
- os.mkdir(path)
-
-def set_file_contents(path, contents):
- add = not os.path.exists(path)
- file(path, "wb").write(contents)
- if add:
- add_id(path)
-
-def detect(path):
- """Detect whether a directory is revision-controlled using GIT"""
- path = os.path.realpath(path)
- old_path = None
- while True:
- if os.path.exists(os.path.join(path, ".git")):
+import re
+import unittest
+import doctest
+
+from rcs import RCS, RCStestCase, CommandError
+
+def new():
+ return Git()
+
+class Git(RCS):
+ name="git"
+ client="git"
+ versioned=True
+ def _rcs_help(self):
+ status,output,error = self._u_invoke_client("--help")
+ return output
+ def _rcs_detect(self, path):
+ if self._u_search_parent_directories(path, ".git") != None :
return True
- if path == old_path:
- return False
- old_path = path
- path = os.path.dirname(path)
-
-def precommit(directory):
- pass
-
-def commit(directory, summary, body=None):
- if body is not None:
- summary += '\n' + body
- descriptor, filename = tempfile.mkstemp()
- try:
- temp_file = os.fdopen(descriptor, 'wb')
- temp_file.write(summary)
- temp_file.close()
- invoke_client('commit', '-a', '-F', filename, directory=directory)
- finally:
- os.unlink(filename)
-
-def postcommit(directory):
- pass
-
-
-# In order to diff the bug database, you need a way to check out arbitrary
-# previous revisions and a mechanism for locating the bug_dir in the revision
-# you've checked out.
-#
-# Copying the Mercurial implementation, this feature is implemented by four
-# functions:
-#
-# git_dir_for_path : find '.git' for a git tree.
-#
-# export : check out a commit 'spec' from git-repo 'bug_dir' into a dir
-# 'revision_dir'
-#
-# find_or_make_export : check out a commit 'spec' from git repo 'directory' to
-# any location you please and return the path to the checkout
-#
-# path_in_reference : return a path to the bug_dir of the commit 'spec'
-
-def git_repo_for_path(path):
- """Find the root of the deepest repository containing path."""
- # Assume that nothing funny is going on; in particular, that we aren't
- # dealing with a bare repo.
- dirname = os.path.dirname(git_dir_for_path(path))
- if dirname == '' : # os.path.dirname('filename') == ''
- dirname = '.'
- return dirname
-
-def git_dir_for_path(path):
- """Find the git-dir of the deepest repo containing path."""
- return invoke_client("rev-parse", "--git-dir", directory=path)[1].rstrip()
-
-def export(spec, bug_dir, revision_dir):
- """Check out commit 'spec' from the git repo containing bug_dir into
- 'revision_dir'."""
- if not os.path.exists(revision_dir):
- os.makedirs(revision_dir)
- invoke_client("init", directory=revision_dir)
- invoke_client("pull", git_dir_for_path(bug_dir), directory=revision_dir)
- invoke_client("checkout", '-f', spec, directory=revision_dir)
-
-def find_or_make_export(spec, directory):
- """Checkout 'spec' from the repo at 'directory' by hook or by crook and
- return the path to the working copy."""
- home = os.path.expanduser("~")
- revision_root = os.path.join(home, ".be_revs")
- if not os.path.exists(revision_root):
- os.mkdir(revision_root)
- revision_dir = os.path.join(revision_root, spec)
- if not os.path.exists(revision_dir):
- export(spec, directory, revision_dir)
- return revision_dir
-
-def path_in_reference(bug_dir, spec):
- """Check out 'spec' and return the path to its bug_dir."""
- spec = spec or 'HEAD'
- spec = invoke_client('rev-parse', spec, directory=bug_dir)[1].rstrip()
- # This is a really hairy computation.
- # The theory is that we can't possibly be working out of a bare repo;
- # hence, we get the rel_bug_dir by chopping off dirname(git_dir_for_path(bug_dir))
- # + '/'.
- rel_bug_dir = strip_git(bug_dir)
- export_root = find_or_make_export(spec, directory=bug_dir)
- return os.path.join(export_root, rel_bug_dir)
-
-
-name = "git"
-
+ return False
+ def _rcs_root(self, path):
+ """Find the root of the deepest repository containing path."""
+ # Assume that nothing funny is going on; in particular, that we aren't
+ # dealing with a bare repo.
+ if os.path.isdir(path) != True:
+ path = os.path.dirname(path)
+ status,output,error = self._u_invoke_client("rev-parse", "--git-dir",
+ directory=path)
+ gitdir = os.path.join(path, output.rstrip('\n'))
+ dirname = os.path.abspath(os.path.dirname(gitdir))
+ return dirname
+ def _rcs_init(self, path):
+ self._u_invoke_client("init", directory=path)
+ def _rcs_get_user_id(self):
+ status,output,error = self._u_invoke_client("config", "user.name")
+ name = output.rstrip('\n')
+ status,output,error = self._u_invoke_client("config", "user.email")
+ email = output.rstrip('\n')
+ return self._u_create_id(name, email)
+ def _rcs_set_user_id(self, value):
+ name,email = self._u_parse_id(value)
+ if email != None:
+ self._u_invoke_client("config", "user.email", email)
+ self._u_invoke_client("config", "user.name", name)
+ def _rcs_add(self, path):
+ if os.path.isdir(path):
+ return
+ self._u_invoke_client("add", path)
+ def _rcs_remove(self, path):
+ if not os.path.isdir(self._u_abspath(path)):
+ self._u_invoke_client("rm", "-f", path)
+ def _rcs_update(self, path):
+ self._rcs_add(path)
+ def _rcs_get_file_contents(self, path, revision=None):
+ if revision == None:
+ return file(self._u_abspath(path), "rb").read()
+ else:
+ arg = "%s:%s" % (revision,path)
+ status,output,error = self._u_invoke_client("show", arg)
+ return output
+ def _rcs_duplicate_repo(self, directory, revision=None):
+ if revision==None:
+ RCS._rcs_duplicate_repo(self, directory, revision)
+ else:
+ #self._u_invoke_client("archive", revision, directory) # makes tarball
+ self._u_invoke_client("clone", "--no-checkout",".",directory)
+ self._u_invoke_client("checkout", revision, directory=directory)
+ def _rcs_commit(self, commitfile):
+ status,output,error = self._u_invoke_client('commit', '-a',
+ '-F', commitfile)
+ revision = None
+ revline = re.compile("Created (.*)commit (.*):(.*)")
+ match = revline.search(output)
+ assert match != None, output+error
+ assert len(match.groups()) == 3
+ revision = match.groups()[1]
+ return revision
+
+class GitTestCase(RCStestCase):
+ Class = Git
+
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(GitTestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
# 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 tempfile
-
-import config
-from rcs import invoke, CommandError
-
-def invoke_client(*args, **kwargs):
- directory = kwargs['directory']
- expect = kwargs.get('expect', (0, 1))
- cl_args = ["hg"]
- cl_args.extend(args)
- status,output,error = invoke(cl_args, expect, cwd=directory)
- return status, output
-
-def add_id(filename, paranoid=False):
- invoke_client("add", filename, directory='.')
-
-def delete_id(filename):
- invoke_client("rm", filename, directory='.')
-
-def mkdir(path, paranoid=False):
- os.mkdir(path)
-
-def set_file_contents(path, contents):
- add = not os.path.exists(path)
- file(path, "wb").write(contents)
- if add:
- add_id(path)
-
-def lookup_revision(revno, directory):
- return invoke_client('log', '--rev', str(revno), '--template={node}',
- directory=directory)[1].rstrip('\n')
-
-def export(revno, directory, revision_dir):
- invoke_client("archive", "--rev", str(revno), revision_dir,
- directory=directory)
-
-def find_or_make_export(revno, directory):
- revision_id = lookup_revision(revno, directory)
- home = os.path.expanduser("~")
- revision_root = os.path.join(home, ".be_revs")
- if not os.path.exists(revision_root):
- os.mkdir(revision_root)
- revision_dir = os.path.join(revision_root, revision_id)
- if not os.path.exists(revision_dir):
- export(revno, directory, revision_dir)
- return revision_dir
-
-def hg_root(path):
- return invoke_client("root", "-R", path, directory=None)[1].rstrip('\r')
-
-def path_in_reference(bug_dir, spec):
- if spec is None:
- spec = int(invoke_client('tip', '--template="{rev}"',
- directory=bug_dir)[1])
- rel_bug_dir = bug_dir[len(hg_root(bug_dir)):]
- export_root = find_or_make_export(spec, directory=bug_dir)
- return os.path.join(export_root, rel_bug_dir)
-
-
-def unlink(path):
- try:
- os.unlink(path)
- delete_id(path)
- except OSError, e:
- if e.errno != 2:
- raise
-
-
-def detect(path):
- """Detect whether a directory is revision-controlled using Mercurial"""
- path = os.path.realpath(path)
- old_path = None
- while True:
- if os.path.exists(os.path.join(path, ".hg")):
+import re
+import unittest
+import doctest
+
+from rcs import RCS, RCStestCase, CommandError, SettingIDnotSupported
+
+def new():
+ return Hg()
+
+class Hg(RCS):
+ name="hg"
+ client="hg"
+ versioned=True
+ def _rcs_help(self):
+ status,output,error = self._u_invoke_client("--help")
+ return output
+ def _rcs_detect(self, path):
+ """Detect whether a directory is revision-controlled using Mercurial"""
+ if self._u_search_parent_directories(path, ".hg") != None:
return True
- if path == old_path:
- return False
- old_path = path
- path = os.path.dirname(path)
-
-def precommit(directory):
- pass
-
-def commit(directory, summary, body=None):
- if body is not None:
- summary += '\n' + body
- descriptor, filename = tempfile.mkstemp()
- try:
- temp_file = os.fdopen(descriptor, 'wb')
- temp_file.write(summary)
- temp_file.close()
- invoke_client('commit', '--logfile', filename, directory=directory)
- finally:
- os.unlink(filename)
-
-def postcommit(directory):
- pass
-
-name = "hg"
+ return False
+ def _rcs_root(self, path):
+ status,output,error = self._u_invoke_client("root", directory=path)
+ return output.rstrip('\n')
+ def _rcs_init(self, path):
+ self._u_invoke_client("init", directory=path)
+ def _rcs_get_user_id(self):
+ status,output,error = self._u_invoke_client("showconfig","ui.username")
+ return output.rstrip('\n')
+ def _rcs_set_user_id(self, value):
+ """
+ Supported by the Config Extension, but that is not part of
+ standard Mercurial.
+ http://www.selenic.com/mercurial/wiki/index.cgi/ConfigExtension
+ """
+ raise SettingIDnotSupported
+ def _rcs_add(self, path):
+ self._u_invoke_client("add", path)
+ def _rcs_remove(self, path):
+ self._u_invoke_client("rm", path)
+ def _rcs_update(self, path):
+ pass
+ def _rcs_get_file_contents(self, path, revision=None):
+ if revision == None:
+ return file(os.path.join(self.rootdir, path), "rb").read()
+ else:
+ status,output,error = \
+ self._u_invoke_client("cat","-r",revision,path)
+ return output
+ def _rcs_duplicate_repo(self, directory, revision=None):
+ if revision == None:
+ RCS._rcs_duplicate_repo(self, directory, revision)
+ else:
+ self._u_invoke_client("archive", "--rev", revision, directory)
+ def _rcs_commit(self, commitfile):
+ self._u_invoke_client('commit', '--logfile', commitfile)
+ status,output,error = self._u_invoke_client('identify')
+ revision = None
+ revline = re.compile("(.*) tip")
+ match = revline.search(output)
+ assert match != None, output+error
+ assert len(match.groups()) == 1
+ revision = match.groups()[0]
+ return revision
+
+class HgTestCase(RCStestCase):
+ Class = Hg
+
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(HgTestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
import os.path
import errno
import utility
+import doctest
class IllegalKey(Exception):
def __init__(self, key):
add = not os.path.exists(path)
output = file(path, "wb")
generate(output, map)
+ output.close()
if add:
- rcs.add_id(path)
+ rcs.add(path)
+ else:
+ rcs.update(path)
class NoSuchFile(Exception):
def __init__(self, pathname):
if e.errno != errno.ENOENT:
raise e
raise NoSuchFile(path)
+
+suite = doctest.DocTestSuite()
import os
import sys
-
+import doctest
def uuid():
# this code borrowed from standard commands module
while (bug.uuid[:chars] == some_bug.uuid[:chars]):
chars+=1
return bug.uuid[:chars]
+
+suite = doctest.DocTestSuite()
+++ /dev/null
-# 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 config
-from os import unlink
-
-def add_id(filename, paranoid=False):
- """Compatibility function"""
- pass
-
-def delete_id(filename):
- """Compatibility function"""
- pass
-
-def mkdir(path, paranoid=False):
- os.mkdir(path)
-
-def set_file_contents(path, contents):
- add = not os.path.exists(path)
- file(path, "wb").write(contents)
- if add:
- add_id(path)
-
-def detect(path):
- """Compatibility function"""
- return True
-
-def precommit(directory):
- pass
-
-def commit(directory, summary, body=None):
- pass
-
-def postcommit(directory):
- pass
-
-name = "None"
import os
import os.path
import sys
+import doctest
+
def my_import(mod_name):
module = __import__(mod_name)
components = mod_name.split('.')
if plugin_path not in sys.path:
sys.path.append(plugin_path)
+suite = doctest.DocTestSuite()
+
def _test():
import doctest
doctest.testmod()
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
from subprocess import Popen, PIPE
+import os
+import os.path
+from socket import gethostname
+import re
import sys
+import tempfile
+import shutil
+import unittest
+import doctest
+from utility import Dir
-def rcs_by_name(rcs_name):
- """Return the module for the RCS with the given name"""
- if rcs_name == "Arch":
- import arch
- return arch
- elif rcs_name == "bzr":
- import bzr
- return bzr
- elif rcs_name == "hg":
- import hg
- return hg
- elif rcs_name == "git":
- import git
- return git
- elif rcs_name == "None":
- import no_rcs
- return no_rcs
-
-def detect(dir):
- """Return the module for the rcs being used in this directory"""
+def _get_matching_rcs(matchfn):
+ """Return the first module for which matchfn(RCS_instance) is true"""
import arch
import bzr
import hg
import git
- if arch.detect(dir):
- return arch
- elif bzr.detect(dir):
- return bzr
- elif hg.detect(dir):
- return hg
- elif git.detect(dir):
- return git
- import no_rcs
- return no_rcs
+ for module in [arch, bzr, hg, git]:
+ rcs = module.new()
+ if matchfn(rcs):
+ return rcs
+ else:
+ del(rcs)
+ return RCS()
+
+def rcs_by_name(rcs_name):
+ """Return the module for the RCS with the given name"""
+ return _get_matching_rcs(lambda rcs: rcs.name == rcs_name)
+
+def detect_rcs(dir):
+ """Return an RCS instance for the rcs being used in this directory"""
+ return _get_matching_rcs(lambda rcs: rcs.detect(dir))
+
+def installed_rcs():
+ """Return an instance of an installed RCS"""
+ return _get_matching_rcs(lambda rcs: rcs.installed())
+
class CommandError(Exception):
def __init__(self, err_str, status):
self.err_str = err_str
self.status = status
-def invoke(args, expect=(0,), cwd=None):
- try :
- if sys.platform != "win32":
- q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, cwd=cwd)
+class SettingIDnotSupported(NotImplementedError):
+ pass
+
+def new():
+ return RCS()
+
+class RCS(object):
+ """
+ Implement the 'no-rcs' interface.
+
+ Support for other RCSs can be added by subclassing this class, and
+ overriding methods _rcs_*() with code appropriate for your RCS.
+
+ The methods _u_*() are utility methods available to the _rcs_*()
+ methods.
+ """
+ name = "None"
+ client = "" # command-line tool for _u_invoke_client
+ versioned = False
+ def __init__(self, paranoid=False):
+ self.paranoid = paranoid
+ self.verboseInvoke = False
+ self.rootdir = None
+ self._duplicateBasedir = None
+ self._duplicateDirname = None
+ def __del__(self):
+ self.cleanup()
+
+ def _rcs_help(self):
+ """
+ Return the command help string.
+ (Allows a simple test to see if the client is installed.)
+ """
+ pass
+ def _rcs_detect(self, path=None):
+ """
+ Detect whether a directory is revision controlled with this RCS.
+ """
+ return True
+ def _rcs_root(self, path):
+ """
+ Get the RCS root. This is the default working directory for
+ future invocations. You would normally set this to the root
+ directory for your RCS.
+ """
+ if os.path.isdir(path)==False:
+ path = os.path.dirname(path)
+ if path == "":
+ path = os.path.abspath(".")
+ return path
+ def _rcs_init(self, path):
+ """
+ Begin versioning the tree based at path.
+ """
+ pass
+ def _rcs_cleanup(self):
+ """
+ Remove any cruft that _rcs_init() created outside of the
+ versioned tree.
+ """
+ pass
+ def _rcs_get_user_id(self):
+ """
+ Get the RCS's suggested user id (e.g. "John Doe <jdoe@example.com>").
+ If the RCS has not been configured with a username, return None.
+ """
+ return None
+ def _rcs_set_user_id(self, value):
+ """
+ Set the RCS's suggested user id (e.g "John Doe <jdoe@example.com>").
+ This is run if the RCS has not been configured with a usename, so
+ that commits will have a reasonable FROM value.
+ """
+ raise SettingIDnotSupported
+ def _rcs_add(self, path):
+ """
+ Add the already created file at path to version control.
+ """
+ pass
+ def _rcs_remove(self, path):
+ """
+ Remove the file at path from version control. Optionally
+ remove the file from the filesystem as well.
+ """
+ pass
+ def _rcs_update(self, path):
+ """
+ Notify the versioning system of changes to the versioned file
+ at path.
+ """
+ pass
+ def _rcs_get_file_contents(self, path, revision=None):
+ """
+ Get the file as it was in a given revision.
+ Revision==None specifies the current revision.
+ """
+ assert revision == None, \
+ "The %s RCS does not support revision specifiers" % self.name
+ return file(os.path.join(self.rootdir, path), "rb").read()
+ def _rcs_duplicate_repo(self, directory, revision=None):
+ """
+ Get the repository as it was in a given revision.
+ revision==None specifies the current revision.
+ dir specifies a directory to create the duplicate in.
+ """
+ shutil.copytree(self.rootdir, directory, True)
+ def _rcs_commit(self, commitfile):
+ """
+ Commit the current working directory, using the contents of
+ commitfile as the comment. Return the name of the old
+ revision.
+ """
+ return None
+ def installed(self):
+ try:
+ self._rcs_help()
+ return True
+ except OSError, e:
+ if e.errno == errno.ENOENT:
+ return False
+ raise e
+ def detect(self, path=None):
+ """
+ Detect whether a directory is revision controlled with this RCS.
+ """
+ return self._rcs_detect(path)
+ def root(self, path):
+ """
+ Set the root directory to the path's RCS root. This is the
+ default working directory for future invocations.
+ """
+ self.rootdir = self._rcs_root(path)
+ def init(self, path):
+ """
+ Begin versioning the tree based at path.
+ Also roots the rcs at path.
+ """
+ if os.path.isdir(path)==False:
+ path = os.path.dirname(path)
+ self._rcs_init(path)
+ self.root(path)
+ def cleanup(self):
+ self._rcs_cleanup()
+ def get_user_id(self):
+ """
+ Get the RCS's suggested user id (e.g. "John Doe <jdoe@example.com>").
+ If the RCS has not been configured with a username, return the user's
+ id.
+ """
+ id = self._rcs_get_user_id()
+ if id == None:
+ name = self._u_get_fallback_username()
+ email = self._u_get_fallback_email()
+ id = self._u_create_id(name, email)
+ print >> sys.stderr, "Guessing id '%s'" % id
+ try:
+ self.set_user_id(id)
+ except SettingIDnotSupported:
+ pass
+ return id
+ def set_user_id(self, value):
+ """
+ Set the RCS's suggested user id (e.g "John Doe <jdoe@example.com>").
+ This is run if the RCS has not been configured with a usename, so
+ that commits will have a reasonable FROM value.
+ """
+ self._rcs_set_user_id(value)
+ def add(self, path):
+ """
+ Add the already created file at path to version control.
+ """
+ self._rcs_add(self._u_rel_path(path))
+ def remove(self, path):
+ """
+ Remove a file from both version control and the filesystem.
+ """
+ self._rcs_remove(self._u_rel_path(path))
+ if os.path.exists(path):
+ os.remove(path)
+ def recursive_remove(self, dirname):
+ """
+ Remove a file/directory and all its decendents from both
+ version control and the filesystem.
+ """
+ for dirpath,dirnames,filenames in os.walk(dirname, topdown=False):
+ filenames.extend(dirnames)
+ for path in filenames:
+ fullpath = os.path.join(dirpath, path)
+ if os.path.exists(fullpath) == False:
+ continue
+ self._rcs_remove(self._u_rel_path(fullpath))
+ if os.path.exists(dirname):
+ shutil.rmtree(dirname)
+ def update(self, path):
+ """
+ Notify the versioning system of changes to the versioned file
+ at path.
+ """
+ self._rcs_update(self._u_rel_path(path))
+ def get_file_contents(self, path, revision=None):
+ """
+ Get the file as it was in a given revision.
+ Revision==None specifies the current revision.
+ """
+ relpath = self._u_rel_path(path)
+ return self._rcs_get_file_contents(relpath, revision)
+ 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)
+ if add:
+ self.add(path)
+ else:
+ self.update(path)
+ def mkdir(self, path):
+ """
+ Created directory at path under version control.
+ """
+ os.mkdir(path)
+ self.add(path)
+ def duplicate_repo(self, revision=None):
+ """
+ Get the repository as it was in a given revision.
+ revision==None specifies the current revision.
+ Return the path to the arbitrary directory at the base of the new repo.
+ """
+ # Dirname in Baseir to protect against simlink attacks.
+ if self._duplicateBasedir == None:
+ self._duplicateBasedir = tempfile.mkdtemp(prefix='BErcs')
+ self._duplicateDirname = \
+ os.path.join(self._duplicateBasedir, "duplicate")
+ self._rcs_duplicate_repo(directory=self._duplicateDirname,
+ revision=revision)
+ return self._duplicateDirname
+ def remove_duplicate_repo(self):
+ """
+ Clean up a duplicate repo created with duplicate_repo().
+ """
+ if self._duplicateBasedir != None:
+ shutil.rmtree(self._duplicateBasedir)
+ self._duplicateBasedir = None
+ self._duplicateDirname = None
+ def commit(self, summary, body=None):
+ """
+ Commit the current working directory, with a commit message
+ string summary and body. Return the name of the old revision
+ (or None if versioning is not supported).
+ """
+ if body is not None:
+ summary += '\n' + body
+ descriptor, filename = tempfile.mkstemp()
+ revision = None
+ try:
+ temp_file = os.fdopen(descriptor, 'wb')
+ temp_file.write(summary)
+ temp_file.flush()
+ revision = self._rcs_commit(filename)
+ temp_file.close()
+ finally:
+ os.remove(filename)
+ return revision
+ def precommit(self, directory):
+ pass
+ def postcommit(self, directory):
+ pass
+ def _u_invoke(self, args, expect=(0,), cwd=None):
+ if cwd == None:
+ cwd = self.rootdir
+ try :
+ if self.verboseInvoke == True:
+ print "%s$ %s" % (cwd, " ".join(args))
+ if sys.platform != "win32":
+ q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, cwd=cwd)
+ else:
+ # win32 don't have os.execvp() so have to run command in a shell
+ q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE,
+ shell=True, cwd=cwd)
+ except OSError, e :
+ strerror = "%s\nwhile executing %s" % (e.args[1], args)
+ raise CommandError(strerror, e.args[0])
+ output, error = q.communicate()
+ status = q.wait()
+ if status not in expect:
+ raise CommandError(error, status)
+ return status, output, error
+ def _u_invoke_client(self, *args, **kwargs):
+ directory = kwargs.get('directory',None)
+ expect = kwargs.get('expect', (0,))
+ cl_args = [self.client]
+ cl_args.extend(args)
+ return self._u_invoke(cl_args, expect, cwd=directory)
+ def _u_search_parent_directories(self, 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:
+ 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)
+ def _u_rel_path(self, path, root=None):
+ """
+ Return the relative path to path from root.
+ >>> rcs = new()
+ >>> rcs._u_rel_path("/a.b/c/.be", "/a.b/c")
+ '.be'
+ """
+ if root == None:
+ assert self.rootdir != None, "RCS not rooted"
+ root = self.rootdir
+ 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)
+ assert path != absRootSlashedDir, \
+ "file %s == root directory %s" % (path, absRootSlashedDir)
+ path = path[len(absRootSlashedDir):]
+ return path
+ def _u_abspath(self, path, root=None):
+ """
+ Return the absolute path from a path realtive to root.
+ >>> rcs = new()
+ >>> rcs._u_abspath(".be", "/a.b/c")
+ '/a.b/c/.be'
+ """
+ if root == None:
+ assert self.rootdir != None, "RCS not rooted"
+ root = self.rootdir
+ return os.path.abspath(os.path.join(root, path))
+ def _u_create_id(self, name, email=None):
+ """
+ >>> rcs = new()
+ >>> rcs._u_create_id("John Doe", "jdoe@example.com")
+ 'John Doe <jdoe@example.com>'
+ >>> rcs._u_create_id("John Doe")
+ 'John Doe'
+ """
+ assert len(name) > 0
+ if email == None or len(email) == 0:
+ return name
else:
- # win32 don't have os.execvp() so have to run command in a shell
- q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, shell=True,
- cwd=cwd)
- except OSError, e :
- strerror = "%s\nwhile executing %s" % (e.args[1], args)
- raise CommandError(strerror, e.args[0])
- output, error = q.communicate()
- status = q.wait()
- if status not in expect:
- raise CommandError(error, status)
- return status, output, error
+ return "%s <%s>" % (name, email)
+ def _u_parse_id(self, value):
+ """
+ >>> rcs = new()
+ >>> rcs._u_parse_id("John Doe <jdoe@example.com>")
+ ('John Doe', 'jdoe@example.com')
+ >>> rcs._u_parse_id("John Doe")
+ ('John Doe', None)
+ >>> try:
+ ... rcs._u_parse_id("John Doe <jdoe@example.com><what?>")
+ ... except AssertionError:
+ ... print "Invalid match"
+ Invalid match
+ """
+ emailexp = re.compile("(.*) <([^>]*)>(.*)")
+ match = emailexp.search(value)
+ if match == None:
+ email = None
+ name = value
+ else:
+ assert len(match.groups()) == 3
+ assert match.groups()[2] == "", match.groups()
+ email = match.groups()[1]
+ name = match.groups()[0]
+ assert name != None
+ assert len(name) > 0
+ return (name, email)
+ def _u_get_fallback_username(self):
+ name = None
+ for envariable in ["LOGNAME", "USERNAME"]:
+ if os.environ.has_key(envariable):
+ name = os.environ[envariable]
+ break
+ assert name != None
+ return name
+ def _u_get_fallback_email(self):
+ hostname = gethostname()
+ name = self._u_get_fallback_username()
+ return "%s@%s" % (name, hostname)
+ def _u_parse_commitfile(self, commitfile):
+ """
+ Split the commitfile created in self.commit() back into
+ summary and header lines.
+ """
+ f = file(commitfile, "rb")
+ summary = f.readline()
+ body = f.read()
+ body.lstrip('\n')
+ if len(body) == 0:
+ body = None
+ f.close
+ return (summary, body)
+
+
+class RCStestCase(unittest.TestCase):
+ Class = RCS
+ def __init__(self, *args, **kwargs):
+ unittest.TestCase.__init__(self, *args, **kwargs)
+ self.dirname = None
+ def instantiateRCS(self):
+ return self.Class()
+ def setUp(self):
+ self.dir = Dir()
+ self.dirname = self.dir.path
+ self.rcs = self.instantiateRCS()
+ def tearDown(self):
+ del(self.rcs)
+ del(self.dirname)
+ def fullPath(self, path):
+ return os.path.join(self.dirname, path)
+ def assertPathExists(self, path):
+ fullpath = self.fullPath(path)
+ self.failUnless(os.path.exists(fullpath)==True,
+ "path %s does not exist" % fullpath)
+ def uidTest(self):
+ user_id = self.rcs.get_user_id()
+ self.failUnless(user_id != None,
+ "unable to get a user id")
+ user_idB = "John Doe <jdoe@example.com>"
+ if self.rcs.name in ["None", "hg"]:
+ self.assertRaises(SettingIDnotSupported, self.rcs.set_user_id,
+ user_idB)
+ else:
+ self.rcs.set_user_id(user_idB)
+ self.failUnless(self.rcs.get_user_id() == user_idB,
+ "user id not set correctly (was %s, is %s)" \
+ % (user_id, self.rcs.get_user_id()))
+ self.failUnless(self.rcs.set_user_id(user_id) == None,
+ "unable to restore user id %s" % user_id)
+ self.failUnless(self.rcs.get_user_id() == user_id,
+ "unable to restore user id %s" % user_id)
+ def versionTest(self, path):
+ origpath = path
+ path = self.fullPath(path)
+ contentsA = "Lorem ipsum"
+ contentsB = "dolor sit amet"
+ self.rcs.set_file_contents(path,contentsA)
+ self.failUnless(self.rcs.get_file_contents(path)==contentsA,
+ "File contents not set or read correctly")
+ revision = self.rcs.commit("Commit current status")
+ self.failUnless(self.rcs.get_file_contents(path)==contentsA,
+ "Committing File contents not set or read correctly")
+ if self.rcs.versioned == True:
+ self.rcs.set_file_contents(path,contentsB)
+ self.failUnless(self.rcs.get_file_contents(path)==contentsB,
+ "File contents not set correctly after commit")
+ contentsArev = self.rcs.get_file_contents(path, revision)
+ self.failUnless(contentsArev==contentsA, \
+ "Original file contents not saved in revision %s\n%s\n%s\n" \
+ % (revision, contentsA, contentsArev))
+ dup = self.rcs.duplicate_repo(revision)
+ duppath = os.path.join(dup, origpath)
+ dupcont = file(duppath, "rb").read()
+ self.failUnless(dupcont == contentsA)
+ self.rcs.remove_duplicate_repo()
+ def testRun(self):
+ self.failUnless(self.rcs.installed() == True,
+ "%s RCS not found" % self.Class.name)
+ if self.Class.name != "None":
+ self.failUnless(self.rcs.detect(self.dirname)==False,
+ "Detected %s RCS before initializing" \
+ % self.Class.name)
+ self.rcs.init(self.dirname)
+ self.failUnless(self.rcs.detect(self.dirname)==True,
+ "Did not detect %s RCS after initializing" \
+ % self.Class.name)
+ rp = os.path.realpath(self.rcs.rootdir)
+ dp = os.path.realpath(self.dirname)
+ self.failUnless(dp == rp or rp == None,
+ "%s RCS root in wrong dir (%s %s)" \
+ % (self.Class.name, dp, rp))
+ self.uidTest()
+ self.rcs.mkdir(self.fullPath('a'))
+ self.rcs.mkdir(self.fullPath('a/b'))
+ self.rcs.mkdir(self.fullPath('c'))
+ self.assertPathExists('a')
+ self.assertPathExists('a/b')
+ self.assertPathExists('c')
+ self.versionTest('a/text')
+ self.versionTest('a/b/text')
+ self.rcs.recursive_remove(self.fullPath('a'))
+
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(RCStestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
from xml.etree import ElementTree # Python 2.5 (and greater?)
except ImportError :
from elementtree import ElementTree
-
+import doctest
def rest_xml(rest):
warnings = StringIO()
foldout += foldout_body
foldout.set_class('foldout')
return [foldout]
+
+suite = doctest.DocTestSuite()
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import tempfile
import shutil
-import os
-import os.path
-from libbe import bugdir, bug, arch
-cleanable = []
-def clean_up():
- global cleanable
- tmp = cleanable
- tmp.reverse()
- for obj in tmp:
- obj.clean_up()
- cleanable = []
+from libbe import utility, names, restconvert, mapfile, config, diff, rcs, \
+ arch, bzr, git, hg, bug, bugdir, plugin, cmdutil
+import unittest
-class Dir:
- def __init__(self):
- self.name = tempfile.mkdtemp(prefix="testdir")
- cleanable.append(self)
- def clean_up(self):
- shutil.rmtree(self.name)
+# can not use 'suite' or the base test.py file will include these suites twice.
+testsuite = unittest.TestSuite([utility.suite, names.suite, restconvert.suite,
+ mapfile.suite, config.suite, diff.suite,
+ rcs.suite, arch.suite, bzr.suite, git.suite,
+ hg.suite, bug.suite, bugdir.suite,
+ plugin.suite, cmdutil.suite])
-def arch_dir():
- arch.ensure_user_id()
- dir = Dir()
- arch.init_tree(dir.name)
- return dir
-
-def bug_arch_dir():
- dir = arch_dir()
- return bugdir.create_bug_dir(dir.name, arch)
-
-def simple_bug_dir():
- dir = bug_arch_dir()
- bug_a = bug.new_bug(dir, "a")
- bug_b = bug.new_bug(dir, "b")
- bug_b.status = "closed"
- bug_a.save()
- bug_b.save()
- return dir
+if __name__ == "__main__":
+ unittest.TextTestRunner(verbosity=2).run(testsuite)
import time
import os
import tempfile
+import shutil
+import doctest
class FileString(object):
"""Bare-bones pseudo-file class
else:
return f
+class Dir:
+ "A temporary directory for testing use"
+ def __init__(self):
+ self.path = tempfile.mkdtemp(prefix="BEtest")
+ def __del__(self):
+ shutil.rmtree(self.path)
+ def __call__(self):
+ return self.path
RFC_2822_TIME_FMT = "%a, %d %b %Y %H:%M:%S +0000"
break
out.append(line)
return ''.join(out)
+
+suite = doctest.DocTestSuite()
"""
from libbe import plugin
+import unittest
import doctest
import sys
+
+suite = unittest.TestSuite()
+
if len(sys.argv) > 1:
+ submodname = sys.argv[1]
match = False
- libbe_failures = libbe_tries = becommands_failures = becommands_tries = 0
- mod = plugin.get_plugin("libbe", sys.argv[1])
- if mod is not None:
- libbe_failures, libbe_tries = doctest.testmod(mod)
+ mod = plugin.get_plugin("libbe", submodname)
+ if mod is not None and hasattr(mod, "suite"):
+ suite.addTest(mod.suite)
match = True
- mod = plugin.get_plugin("becommands", sys.argv[1])
+ mod = plugin.get_plugin("becommands", submodname)
if mod is not None:
- becommands_failures, becommands_tries = doctest.testmod(mod)
+ suite.addTest(doctest.DocTestSuite(mod))
match = True
if not match:
- print "No modules match \"%s\"" % sys.argv[1]
+ print "No modules match \"%s\"" % submodname
sys.exit(1)
- else:
- sys.exit(libbe_failures or becommands_failures)
else:
failed = False
- for module in plugin.iter_plugins("libbe"):
- failures, tries = doctest.testmod(module[1])
- if failures:
- failed = True
- for module in plugin.iter_plugins("becommands"):
- failures, tries = doctest.testmod(module[1])
- if failures:
- failed = True
- sys.exit(failed)
+ for modname,module in plugin.iter_plugins("libbe"):
+ if not hasattr(module, "suite"):
+ continue
+ suite.addTest(module.suite)
+ for modname,module in plugin.iter_plugins("becommands"):
+ suite.addTest(doctest.DocTestSuite(module))
+
+#for s in suite._tests:
+# print s
+#exit(0)
+result = unittest.TextTestRunner(verbosity=2).run(suite)
+
+numErrors = len(result.errors)
+numFailures = len(result.failures)
+numBad = numErrors + numFailures
+if numBad > 126:
+ numBad = 1
+sys.exit(numBad)
# usage: test_usage.sh RCS
# where RCS is one of:
# bzr, git, hg, arch, none
+#
+# Note that this script uses the *installed* version of be, not the
+# one in your working tree.
set -e # exit imediately on failed command
set -o pipefail # pipes fail if any stage fails
exec 6>&2 # save stderr to file descriptor 6
exec 2>&1 # fd 2 now writes to stdout
-if [ $# -ne 1 ]
+if [ $# -gt 1 ]
then
- echo "usage: test_usage.sh RCS"
+ echo "usage: test_usage.sh [RCS]"
echo ""
echo "where RCS is one of"
for RCS in bzr git hg arch none
echo " $RCS"
done
exit 1
+elif [ $# -eq 0 ]
+then
+ for RCS in bzr git hg arch none
+ do
+ echo -e "\n\nTesting $RCS\n\n"
+ $0 "$RCS" || exit 1
+ done
+ exit 0
fi
RCS="$1"
then
ID=`id -nu`
else
- echo "Unrecognized RCS $RCS"
+ echo "Unrecognized RCS '$RCS'"
exit 1
fi
if [ -z "$ID" ]
be assign $BUG 'Joe' # assign the bug to Joe
be list -a Joe -s fixed # list the fixed bugs assigned to Joe
be assign $BUG none # assign the bug to noone
+be diff # see what has changed
be remove $BUG # decide that you don't like that bug after all
cd /
rm -rf $TESTDIR
projects / be.git / commitdiff