diff options
Diffstat (limited to 'libbe/storage/vcs')
-rw-r--r-- | libbe/storage/vcs/base.py | 74 | ||||
-rw-r--r-- | libbe/storage/vcs/darcs.py | 28 | ||||
-rw-r--r-- | libbe/storage/vcs/monotone.py | 13 |
3 files changed, 66 insertions, 49 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: diff --git a/libbe/storage/vcs/darcs.py b/libbe/storage/vcs/darcs.py index 41262f5..5ec9a26 100644 --- a/libbe/storage/vcs/darcs.py +++ b/libbe/storage/vcs/darcs.py @@ -39,11 +39,17 @@ import libbe from ...util.subproc import CommandError from . import base -if libbe.TESTING == True: +if libbe.TESTING: import doctest import unittest +# https://stackoverflow.com/a/56719588/164233 +def cmp(a, b): + return (int(a) > int(b)) - (int(a) < int(b)) + + + def new(): return Darcs() @@ -98,7 +104,7 @@ class Darcs(base.VCS): NotImplementedError: Cannot parse non-integer portion "0pre2" of Darcs version "2.0.0pre2" """ if not hasattr(self, '_parsed_version') \ - or self._parsed_version == None: + or self._parsed_version is None: num_part = self.version().split(' ')[0] self._parsed_version = [] for num in num_part.split('.'): @@ -119,12 +125,12 @@ class Darcs(base.VCS): def _vcs_get_user_id(self): # following http://darcs.net/manual/node4.html#SECTION00410030000000000000 # as of June 22th, 2010 - if self.repo == None: + if self.repo is None: return None for pref_file in ['author', 'email']: for prefs_dir in [os.path.join(self.repo, '_darcs', 'prefs'), os.path.expanduser(os.path.join('~', '.darcs'))]: - if prefs_dir == None: + if prefs_dir is None: continue pref_path = os.path.join(prefs_dir, pref_file) if os.path.exists(pref_path): @@ -135,7 +141,7 @@ class Darcs(base.VCS): return None def _vcs_detect(self, path): - if self._u_search_parent_directories(path, "_darcs") != None : + if self._u_search_parent_directories(path, "_darcs") is not None : return True return False @@ -146,7 +152,7 @@ class Darcs(base.VCS): if os.path.isdir(path) != True: path = os.path.dirname(path) darcs_dir = self._u_search_parent_directories(path, '_darcs') - if darcs_dir == None: + if darcs_dir is None: return None return os.path.dirname(darcs_dir) @@ -175,7 +181,7 @@ class Darcs(base.VCS): pass # darcs notices changes def _vcs_get_file_contents(self, path, revision=None): - if revision == None: + if revision is None: return base.VCS._vcs_get_file_contents(self, path, revision) if self.version_cmp(2, 0, 0) == 1: status,output,error = self._u_invoke_client( \ @@ -266,7 +272,7 @@ class Darcs(base.VCS): else: revline = re.compile("Finished recording patch '(.*)'") match = revline.search(output) - assert match != None, output+error + assert match is not None, output+error assert len(match.groups()) == 1 revision = match.groups()[0] return revision @@ -308,7 +314,7 @@ class Darcs(base.VCS): if i+1 < len(revisions): next_rev = revisions[i+1] args.extend(['--from-patch', next_rev]) - if path != None: + if path is not None: args.append(path) kwargs = {'unicode_output':unicode_output} status,output,error = self._u_invoke_client( @@ -356,7 +362,7 @@ class Darcs(base.VCS): @@ -1 +0,0 @@ -this entry will be deleted \ No newline at end of file - + """ new = [] modified = [] @@ -404,7 +410,7 @@ class Darcs(base.VCS): return self._parse_diff(self._diff(revision)) -if libbe.TESTING == True: +if libbe.TESTING: base.make_vcs_testcase_subclasses(Darcs, sys.modules[__name__]) unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__]) diff --git a/libbe/storage/vcs/monotone.py b/libbe/storage/vcs/monotone.py index abd3142..add7be9 100644 --- a/libbe/storage/vcs/monotone.py +++ b/libbe/storage/vcs/monotone.py @@ -38,6 +38,11 @@ if libbe.TESTING == True: import sys +# https://stackoverflow.com/a/56719588/164233 +def cmp(a, b): + return (int(a) > int(b)) - (int(a) < int(b)) + + def new(): return Monotone() @@ -315,15 +320,15 @@ class Monotone (base.VCS): # old_revision [1ce9ac2cfe3166b8ad23a60555f8a70f37686c25] # # delete ".be/dir/bugs/moved" - # + # # delete ".be/dir/bugs/removed" - # + # # add_file ".be/dir/bugs/moved2" # content [33e4510df9abef16dad7c65c0775e74602cc5005] - # + # # add_file ".be/dir/bugs/new" # content [45c45b5630f7446f83b0e14ee1525e449a06131c] - # + # # patch ".be/dir/bugs/modified" # from [809bf3b80423c361849386008a0ce01199d30929] # to [f13d3ec08972e2b41afecd9a90d4bc71cdcea338] |