diff options
Diffstat (limited to 'libbe/storage/vcs/base.py')
| -rw-r--r-- | libbe/storage/vcs/base.py | 74 |
1 files changed, 40 insertions, 34 deletions
diff --git a/libbe/storage/vcs/base.py b/libbe/storage/vcs/base.py index dfd007f..579935e 100644 --- a/libbe/storage/vcs/base.py +++ b/libbe/storage/vcs/base.py @@ -44,12 +44,18 @@ from libbe.util.subproc import CommandError, invoke from libbe.util.plugin import import_by_name import libbe.storage.util.upgrade as upgrade -if libbe.TESTING == True: +if libbe.TESTING: import unittest import doctest import libbe.ui.util.user + +# https://stackoverflow.com/a/56719588/164233 +def cmp(a, b): + return (int(a) > int(b)) - (int(a) < int(b)) + + VCS_ORDER = ['bzr', 'darcs', 'git', 'hg', 'monotone'] """List VCS modules in order of preference. @@ -75,7 +81,7 @@ def _get_matching_vcs(matchfn): for submodname in VCS_ORDER: module = import_by_name('libbe.storage.vcs.%s' % submodname) vcs = module.new() - if matchfn(vcs) == True: + if matchfn(vcs): return vcs return VCS() @@ -117,7 +123,7 @@ class VCSUnableToRoot (libbe.storage.base.ConnectionError): class InvalidPath (InvalidID): def __init__(self, path, root, msg=None, **kwargs): - if msg == None: + if msg is None: msg = 'Path "%s" not in root "%s"' % (path, root) InvalidID.__init__(self, msg=msg, **kwargs) self.path = path @@ -132,7 +138,7 @@ class SpacerCollision (InvalidPath): class CachedPathID (object): """Cache Storage ID <-> path policy. - + Paths generated following:: .../.be/BUGDIR/bugs/BUG/comments/COMMENT @@ -209,7 +215,7 @@ class CachedPathID (object): UUID\tPATH """ - if cache == None: + if cache is None: self._cache = {} else: self._cache = cache @@ -231,7 +237,7 @@ class CachedPathID (object): pass if self._cache != cache: self._changed = True - if cache == None: + if cache is None: self.disconnect() def destroy(self): @@ -253,7 +259,7 @@ class CachedPathID (object): f.close() def disconnect(self): - if self._changed == True: + if self._changed: f = codecs.open(self._cache_path, 'w', self.encoding) for uuid,path in list(self._cache.items()): f.write('%s\t%s\n' % (uuid, path)) @@ -271,7 +277,7 @@ class CachedPathID (object): self.init(cache=self._cache) if uuid not in self._cache: raise InvalidID(uuid) - if relpath == True: + if relpath: return os.path.join(self._cache[uuid], *extra) return os.path.join(self._root, self._cache[uuid], *extra) @@ -285,7 +291,7 @@ class CachedPathID (object): # already added path = self.path(id) else: - if parent == None: + if parent is None: parent_path = '' spacer = self._spacer_dirs[0] else: @@ -445,7 +451,7 @@ class VCS (libbe.storage.base.VersionedStorage): def _vcs_path(self, id, revision): """ Return the relative path to object id as of revision. - + Revision will not be None. """ raise NotImplementedError @@ -454,7 +460,7 @@ class VCS (libbe.storage.base.VersionedStorage): """ Return True if path (as returned by _vcs_path) was a directory as of revision, False otherwise. - + Revision will not be None. """ raise NotImplementedError @@ -463,7 +469,7 @@ class VCS (libbe.storage.base.VersionedStorage): """ Return a list of the contents of the directory path (as returned by _vcs_path) as of revision. - + Revision will not be None, and ._vcs_isdir(path, revision) will be True. """ @@ -610,7 +616,7 @@ class VCS (libbe.storage.base.VersionedStorage): """ if not hasattr(self, 'user_id'): self.user_id = self._vcs_get_user_id() - if self.user_id == None: + if self.user_id is None: # guess missing info name = libbe.ui.util.user.get_fallback_fullname() email = libbe.ui.util.user.get_fallback_email() @@ -696,13 +702,13 @@ class VCS (libbe.storage.base.VersionedStorage): self._cached_path_id.disconnect() def path(self, id, revision=None, relpath=True): - if revision == None: + if revision is None: path = self._cached_path_id.path(id) - if relpath == True: + if relpath: return self._u_rel_path(path) return path path = self._vcs_path(id, revision) - if relpath == True: + if relpath: return path return os.path.join(self.repo, path) @@ -729,7 +735,7 @@ class VCS (libbe.storage.base.VersionedStorage): self._add_path(path, **kwargs) def _exists(self, id, revision=None): - if revision == None: + if revision is None: try: path = self.path(id, revision, relpath=False) except InvalidID as e: @@ -781,11 +787,11 @@ class VCS (libbe.storage.base.VersionedStorage): id = self._u_path_to_id(path) ancestors.append(id) except (SpacerCollision, InvalidPath): - pass + pass return ancestors def _children(self, id=None, revision=None): - if revision == None: + if revision is None: isdir = os.path.isdir listdir = os.listdir else: @@ -797,7 +803,7 @@ class VCS (libbe.storage.base.VersionedStorage): path = self.be_dir else: path = self.path(id, revision, relpath=False) - if isdir(path) == False: + if isdir(path) == False: return [] children = listdir(path) for i,c in enumerate(children): @@ -809,7 +815,7 @@ class VCS (libbe.storage.base.VersionedStorage): children[i] = None children[i] = None for i,c in enumerate(children): - if c == None: continue + if c is None: continue cpath = os.path.join(path, c) children[i] = self._u_path_to_id(cpath) return [c for c in children if c != None] @@ -861,7 +867,7 @@ class VCS (libbe.storage.base.VersionedStorage): return revision def revision_id(self, index=None): - if index == None: + if index is None: return None try: if int(index) != index: @@ -869,7 +875,7 @@ class VCS (libbe.storage.base.VersionedStorage): except ValueError: raise InvalidRevision(index) revid = self._vcs_revision_id(index) - if revid == None: + if revid is None: raise libbe.storage.base.InvalidRevision(index) return revid @@ -930,7 +936,7 @@ class VCS (libbe.storage.base.VersionedStorage): def _u_find_id_from_manifest(self, id, manifest, revision=None): """Search for the relative path to id using manifest, a list of all files. - + Returns None if the id is not found. """ be_dir = self._cached_path_id._spacer_dirs[0] @@ -991,8 +997,8 @@ class VCS (libbe.storage.base.VersionedStorage): >>> vcs._u_rel_path("./a", ".") 'a' """ - if root == None: - if self.repo == None: + if root is None: + if self.repo is None: raise VCSNotRooted(self) root = self.repo path = os.path.abspath(path) @@ -1015,7 +1021,7 @@ class VCS (libbe.storage.base.VersionedStorage): >>> vcs._u_abspath(".be", "/a.b/c") '/a.b/c/.be' """ - if root == None: + if root is None: assert self.repo != None, "VCS not rooted" root = self.repo return os.path.abspath(os.path.join(root, path)) @@ -1045,11 +1051,11 @@ class VCS (libbe.storage.base.VersionedStorage): -------- libbe.storage.util.upgrade """ - if path == None: + if path is None: path = os.path.join(self.repo, '.be', 'version') if not os.path.exists(path): raise libbe.storage.InvalidStorageVersion(None) - if revision == None: # don't require connection + if revision is None: # don't require connection return libbe.util.encoding.get_file_contents( path, decode=True).rstrip() relpath = self._u_rel_path(path) @@ -1070,7 +1076,7 @@ class VCS (libbe.storage.base.VersionedStorage): self._vcs_add(self._u_rel_path(path)) -if libbe.TESTING == True: +if libbe.TESTING: class VCSTestCase (unittest.TestCase): """ Test cases for base VCS class (in addition to the Storage test @@ -1089,13 +1095,13 @@ if libbe.TESTING == True: self.dir = Dir() self.dirname = self.dir.path self.s = self.Class(repo=self.dirname) - if self.s.installed() == True: + if self.s.installed(): self.s.init() self.s.connect() def tearDown(self): super(VCSTestCase, self).tearDown() - if self.s.installed() == True: + if self.s.installed(): self.s.disconnect() self.s.destroy() self.dir.cleanup() @@ -1147,7 +1153,7 @@ if libbe.TESTING == True: """Should get the existing user ID.""" if self.s.installed(): user_id = self.s.get_user_id() - if user_id == None: + if user_id is None: return name,email = libbe.ui.util.user.parse_user_id(user_id) if email != None: @@ -1156,7 +1162,7 @@ if libbe.TESTING == True: def make_vcs_testcase_subclasses(vcs_class, namespace): c = vcs_class() if c.installed(): - if c.versioned == True: + if c.versioned: libbe.storage.base.make_versioned_storage_testcase_subclasses( vcs_class, namespace) else: |