def _children(self, id=None, revision=None):
if revision == None:
id_to_path = self._cached_path_id.path
- path_to_id = self._cached_path_id.id
isdir = os.path.isdir
listdir = os.listdir
else:
id_to_path = lambda id : self._vcs_path(id, revision)
- path_to_id = self._cached_path_id.id
isdir = lambda path : self._vcs_isdir(path, revision)
listdir = lambda path : self._vcs_listdir(path, revision)
if id==None:
and self._vcs_is_versioned(cpath) == False:
children[i] = None
else:
- children[i] = path_to_id(cpath)
+ children[i] = self._u_path_to_id(cpath)
children[i]
return [c for c in children if c != None]
for child in self._vcs_listdir(path, revision):
stack.append((os.path.join(path, child),
'/'.join([long_id, child])))
- return None
+ raise InvalidID(id, revision=revision)
+
+ def _u_path_to_id(self, path):
+ return self._cached_path_id.id(path)
def _u_rel_path(self, path, root=None):
"""
if libbe.TESTING == True:
import doctest
- import sys
import unittest
else:
return self._u_invoke_client('cat', '-r', revision, path)
+ def _vcs_path(self, id, revision):
+ output = self._u_invoke_client('manifest', '--rev', revision)
+ be_dir = self._cached_path_id._spacer_dirs[0]
+ be_dir_sep = self._cached_path_id._spacer_dirs[0] + os.path.sep
+ files = [f for f in output.splitlines() if f.startswith(be_dir_sep)]
+ for file in files:
+ if not file.startswith(be_dir+os.path.sep):
+ continue
+ parts = file.split(os.path.sep)
+ dir = parts.pop(0) # don't add the first spacer dir
+ for part in parts[:-1]:
+ dir = os.path.join(dir, part)
+ if not dir in files:
+ files.append(dir)
+ for file in files:
+ if self._u_path_to_id(file) == id:
+ return file
+ raise base.InvalidId(id, revision=revision)
+
+ def _vcs_isdir(self, path, revision):
+ output = self._u_invoke_client('manifest', '--rev', revision)
+ files = output.splitlines()
+ if path in files:
+ return False
+ return True
+
+ def _vcs_listdir(self, path, revision):
+ output = self._u_invoke_client('manifest', '--rev', revision)
+ files = output.splitlines()
+ path = path.rstrip(os.path.sep) + os.path.sep
+ return [self._u_rel_path(f, path) for f in files if f.startswith(path)]
+
def _vcs_commit(self, commitfile, allow_empty=False):
args = ['commit', '--logfile', commitfile]
output = self._u_invoke_client(*args)