import unittest
import doctest
+import traceback
+import sys
+
import config
+from beuuid import uuid_gen
from rcs import RCS, RCStestCase, CommandError
client = config.get_val("arch_client")
self._create_archive(path)
self._create_project(path)
self._add_project_code(path)
+ #print "RCSid:", id(self), "init", self._archive_project_name()
+ #print "BEGIN_TRACE"
+ #traceback.print_stack(file=sys.stdout)
+ #print "END_TRACE"
def _create_archive(self, path):
# Create a new archive
# http://regexps.srparish.net/tutorial-tla/new-archive.html#Creating_a_New_Archive
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"))
+ trailer = "%s-%s" % ("bugs-everywhere-auto", uuid_gen()[0:8])
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)
+ self._u_invoke_client("archives")
def _invoke_client(self, *args, **kwargs):
"""
Invoke the client on our archive.
self._invoke_client("import", "--summary", "Began versioning",
directory=path)
def _rcs_cleanup(self):
+ #print "RCSid:", id(self), "cleaned", self._archive_project_name()
+ #print "BEGIN_TRACE"
+ #traceback.print_stack(file=sys.stdout)
+ #print "END_TRACE"
if self._tmp_project == True:
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)
def _rcs_get_user_id(self):
try:
+ self._u_invoke_client("archives")
status,output,error = self._u_invoke_client('my-id')
return output.rstrip('\n')
except Exception, e:
def _rcs_set_user_id(self, value):
self._u_invoke_client('my-id', value)
def _rcs_add(self, path):
+ self._u_invoke_client("archives")
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)
+ self._u_invoke_client("archives")
def _list_added(self, root):
assert os.path.exists(root)
assert os.access(root, os.X_OK)
def _rcs_commit(self, commitfile):
summary,body = self._u_parse_commitfile(commitfile)
#status,output,error = self._invoke_client("make-log")
+ self._u_invoke_client("tree-root")
+ self._u_invoke_client("tree-version")
+ self._u_invoke_client("archives")
if body == None:
status,output,error \
- = self._invoke_client("commit","--summary",summary)
+ = self._u_invoke_client("commit","--summary",summary)
else:
status,output,error \
- = self._invoke_client("commit","--summary",summary,
- "--log-message",body)
+ = self._u_invoke_client("commit","--summary",summary,
+ "--log-message",body)
revision = None
revline = re.compile("[*] committed (.*)")
match = revline.search(output)
TREE_VERSION_STRING = "Bugs Everywhere Tree 1 0\n"
-def setting_property(name, valid=None):
+def setting_property(name, valid=None, doc=None):
def getter(self):
value = self.settings.get(name)
if valid is not None:
if value not in valid:
raise InvalidValue(name, value)
return value
-
+
def setter(self, value):
if valid is not None:
if value not in valid and value is not None:
del self.settings[name]
else:
self.settings[name] = value
- self.save()
- return property(getter, setter)
+ self._save_settings(self.get_path("settings"), self.settings)
+
+ return property(getter, setter, doc=doc)
class BugDir (list):
+ """
+ File-system access:
+ When rooted in non-bugdir directory, BugDirs live completely in
+ memory until the first call to .save(). This creates a '.be'
+ sub-directory containing configurations options, bugs, comments,
+ etc. Once this sub-directory has been created (possibly by
+ another BugDir instance) any changes to the BugDir in memory will
+ be flushed to the file system automatically. However, the BugDir
+ will only load information from the file system when it loads new
+ bugs/comments that it doesn't already have in memory, or when it
+ explicitly asked to do so (e.g. .load() or __init__(loadNow=True)).
+ """
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)
+ self.settings = {}
if root == None:
root = os.getcwd()
if sink_to_existing_root == True:
if loadNow == True:
self.load()
else:
- if assert_new_BugDir:
+ if assert_new_BugDir == True:
if os.path.exists(self.get_path()):
raise AlreadyInitialized, self.get_path()
if rcs == None:
rcs = self.guess_rcs(allow_rcs_init)
- self.settings = {"rcs_name": self.rcs_name}
- self.rcs_name = rcs.name
+ self.rcs = rcs
def find_root(self, path):
"""
self.rcs.set_file_contents(self.get_path("version"), TREE_VERSION_STRING)
rcs_name = setting_property("rcs_name",
- ("None", "bzr", "git", "Arch", "hg"))
+ ("None", "bzr", "git", "Arch", "hg"),
+ doc="The name of the current RCS. Kept seperate to make saving/loading settings easy. Don't set this attribute. Set .rcs instead, and .rcs_name will be automatically adjusted.")
_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.root)
return self._rcs
- rcs = property(_get_rcs)
+ def _set_rcs(self, rcs):
+ if rcs == None:
+ rcs = rcs_by_name("None")
+ self._rcs = rcs
+ rcs.root(self.root)
+ self.rcs_name = rcs.name
- target = setting_property("target")
+ rcs = property(_get_rcs, _set_rcs, doc="A revision control system (RCS) instance")
+
+ target = setting_property("target", doc="The current project development target")
def get_path(self, *args):
my_dir = os.path.join(self.root, ".be")
if not os.path.exists(deepdir):
deepdir = os.path.dirname(deepdir)
rcs = detect_rcs(deepdir)
+ install = False
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
+ self.rcs = rcs
return rcs
def load(self):
if not os.path.exists(self.get_path()):
raise NoBugDir(self.get_path())
self.settings = self._get_settings(self.get_path("settings"))
+ self.rcs = rcs_by_name(self.rcs_name)
self._clear_bugs()
for uuid in self.list_uuids():
self._load_bug(uuid)
return settings
def _save_settings(self, settings_path, settings):
+ if not os.path.exists(self.get_path()):
+ # don't save settings until the bug directory has been
+ # initialized. this initialization happens the first time
+ # a bug directory is saved (BugDir.save()). If the user
+ # is just working with a BugDir in memory, we don't want
+ # to go cluttering up his file system with settings files.
+ return
try:
mapfile.map_save(self.rcs, settings_path, settings)
except PathNotInRoot, e:
- # Handling duplicate bugdir settings, special case
+ # Special case for configuring duplicate bugdir settings
none_rcs = rcs_by_name("None")
none_rcs.root(settings_path)
mapfile.map_save(none_rcs, settings_path, settings)
def duplicate_bugdir(self, revision):
duplicate_path = self.rcs.duplicate_repo(revision)
- # setup revision RCS as None, since the duplicate may not be versioned
+ # setup revision RCS as None, since the duplicate may not be initialized for versioning
duplicate_settings_path = os.path.join(duplicate_path, ".be", "settings")
duplicate_settings = self._get_settings(duplicate_settings_path)
if "rcs_name" in duplicate_settings:
self.bug_map = map
def list_uuids(self):
- for uuid in os.listdir(self.get_path("bugs")):
- if (uuid.startswith('.')):
- continue
- yield uuid
+ uuids = []
+ if os.path.exists(self.get_path()):
+ # list the uuids on disk
+ for uuid in os.listdir(self.get_path("bugs")):
+ if not (uuid.startswith('.')):
+ uuids.append(uuid)
+ yield uuid
+ # and the ones that are still just in memory
+ for bug in self:
+ if bug.uuid not in uuids:
+ uuids.append(bug.uuid)
+ yield bug.uuid
def _clear_bugs(self):
while len(self) > 0:
"path %s does not exist" % fullpath)
self.assertRaises(AlreadyInitialized, BugDir,
self.dir.path, assertNewBugDir=True)
+ def versionTest(self):
+ if self.rcs.versioned == False:
+ return
+ original = self.bugdir.rcs.commit("Began versioning")
+ bugA = self.bugdir.bug_from_uuid("a")
+ bugA.status = "fixed"
+ self.bugdir.save()
+ new = self.rcs.commit("Fixed bug a")
+ dupdir = self.bugdir.duplicate_bugdir(original)
+ self.failUnless(dupdir.root != self.bugdir.root, "%s, %s" % (dupdir.root, self.bugdir.root))
+ bugAorig = dupdir.bug_from_uuid("a")
+ self.failUnless(bugA != bugAorig, "\n%s\n%s" % (bugA.string(), bugAorig.string()))
+ bugAorig.status = "fixed"
+ self.failUnless(bug.cmp_status(bugA, bugAorig)==0, "%s, %s" % (bugA.status, bugAorig.status))
+ self.failUnless(bug.cmp_severity(bugA, bugAorig)==0, "%s, %s" % (bugA.severity, bugAorig.severity))
+ self.failUnless(bug.cmp_assigned(bugA, bugAorig)==0, "%s, %s" % (bugA.assigned, bugAorig.assigned))
+ self.failUnless(bug.cmp_time(bugA, bugAorig)==0, "%s, %s" % (bugA.time, bugAorig.time))
+ self.failUnless(bug.cmp_creator(bugA, bugAorig)==0, "%s, %s" % (bugA.creator, bugAorig.creator))
+ self.failUnless(bugA == bugAorig, "\n%s\n%s" % (bugA.string(), bugAorig.string()))
+ self.bugdir.remove_duplicate_bugdir()
+ self.failUnless(os.path.exists(dupdir.root)==False, str(dupdir.root))
+ def testRun(self):
+ self.bugdir.new_bug(uuid="a", summary="Ant")
+ self.bugdir.new_bug(uuid="b", summary="Cockroach")
+ self.bugdir.new_bug(uuid="c", summary="Praying mantis")
+ length = len(self.bugdir)
+ self.failUnless(length == 3, "%d != 3 bugs" % length)
+ uuids = list(self.bugdir.list_uuids())
+ self.failUnless(len(uuids) == 3, "%d != 3 uuids" % len(uuids))
+ self.failUnless(uuids == ["a","b","c"], str(uuids))
+ bugA = self.bugdir.bug_from_uuid("a")
+ bugAprime = self.bugdir.bug_from_shortname("a")
+ self.failUnless(bugA == bugAprime, "%s != %s" % (bugA, bugAprime))
+ self.bugdir.save()
+ self.versionTest()
unitsuite = unittest.TestLoader().loadTestsFromTestCase(BugDirTestCase)
suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
import shutil
import unittest
import doctest
+
+
from utility import Dir, search_parent_directories
+
def _get_matching_rcs(matchfn):
"""Return the first module for which matchfn(RCS_instance) is true"""
import arch
import bzr
import hg
import git
- for module in [git, arch, bzr, hg, git]:
+ for module in [arch, bzr, hg, git]:
rcs = module.new()
if matchfn(rcs) == True:
return rcs
class RCS(object):
"""
- Implement the 'no-rcs' interface.
+ This class implements a 'no-rcs' interface.
Support for other RCSs can be added by subclassing this class, and
overriding methods _rcs_*() with code appropriate for your RCS.
def _u_invoke(self, args, expect=(0,), cwd=None):
if cwd == None:
cwd = self.rootdir
+ if self.verboseInvoke == True:
+ print >> sys.stderr, "%s$ %s" % (cwd, " ".join(args))
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:
raise CommandError(strerror, e.args[0])
output, error = q.communicate()
status = q.wait()
+ if self.verboseInvoke == True:
+ print >> sys.stderr, "%d\n%s%s" % (status, output, error)
if status not in expect:
- raise CommandError(error, status)
+ strerror = "%s\nwhile executing %s\n%s" % (args[1], args, error)
+ raise CommandError(strerror, status)
return status, output, error
def _u_invoke_client(self, *args, **kwargs):
directory = kwargs.get('directory',None)
projects / be.git / commitdiff