# 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 os.path
import cmdutil
import errno
import unittest
import doctest
import names
import mapfile
import time
import utility
from rcs import rcs_by_name, installed_rcs
from bug import Bug
class NoBugDir(Exception):
def __init__(self, path):
msg = "The directory \"%s\" has no bug directory." % path
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
Exception.__init__(self, "Specified root does not exist: %s" % path)
class AlreadyInitialized(Exception):
def __init__(self, path):
self.path = path
Exception.__init__(self,
"Specified root is already initialized: %s" % path)
def bugdir_root(versioning_root):
return os.path.join(versioning_root, ".be")
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))
def setting_property(name, valid=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:
raise InvalidValue(name, value)
if value is None:
del self.settings[name]
else:
self.settings[name] = value
self.save_settings()
return property(getter, setter)
class BugDir:
def __init__(self, dir):
self.dir = dir
self.bugs_path = os.path.join(self.dir, "bugs")
try:
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 = None
target = setting_property("target")
def save_settings(self):
mapfile.map_save(self.rcs,
os.path.join(self.dir, "settings"), self.settings)
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)
def duplicate_bugdir(self, revision):
return BugDir(bugdir_root(self.rcs.duplicate_repo(revision)))
def remove_duplicate_bugdir(self):
self.rcs.remove_duplicate_repo()
def list(self):
for uuid in self.list_uuids():
yield self.get_bug(uuid)
def bug_map(self):
bugs = {}
for bug in self.list():
bugs[bug.uuid] = bug
return bugs
def get_bug(self, uuid):
return Bug(self.bugs_path, uuid, self.rcs, self)
def list_uuids(self):
for uuid in os.listdir(self.bugs_path):
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
return bug
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():
"""
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()])
