Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for f in files:
open(os.path.join(self.temp_d, f), 'w').write("DATA %s" % i)
i += 1
local_repo.stage(files)
local_repo.do_commit('Test commit', 'fbo@localhost',
ref='refs/heads/master')
swift.SwiftRepo.init_bare(self.scon, self.conf)
tcp_client = client.TCPGitClient(self.server_address,
port=self.port)
tcp_client.send_pack(self.fakerepo,
determine_wants,
local_repo.object_store.generate_pack_data)
swift_repo = swift.SwiftRepo("fakerepo", self.conf)
commit_sha = swift_repo.refs.read_loose_ref('refs/heads/master')
otype, data = swift_repo.object_store.get_raw(commit_sha)
commit = objects.ShaFile.from_raw_string(otype, data)
otype, data = swift_repo.object_store.get_raw(commit._tree)
tree = objects.ShaFile.from_raw_string(otype, data)
objs = tree.items()
objs_ = []
for tree_entry in objs:
objs_.append(swift_repo.object_store.get_raw(tree_entry.sha))
# Blob
self.assertEqual(objs_[1][1], 'DATA 0')
self.assertEqual(objs_[2][1], 'DATA 1')
# Tree
self.assertEqual(objs_[0][0], 2)
def make_commit(repo, tree, message, author, timezone, encoding="UTF-8"):
"""build a Commit object"""
commit = Commit()
try:
commit.parents = [repo.head()]
except KeyError:
#the initial commit has no parent
pass
if isinstance(tree, dulwich.objects.Tree): tree_id = tree.id
elif isinstance(tree, str): tree_id = tree
commit.tree = tree_id
commit.message = message
commit.author = commit.committer = author
commit.commit_time = commit.author_time = int(time())
commit.commit_timezone = commit.author_timezone = parse_timezone(timezone)
commit.encoding = encoding
return commit
def tree_entry(fctx, blob_cache):
"""Compute a dulwich TreeEntry from a filectx.
A side effect is the TreeEntry is stored in the passed cache.
Returns a 2-tuple of (dulwich.objects.TreeEntry, dulwich.objects.Blob).
"""
blob_id = blob_cache.get(fctx.filenode(), None)
blob = None
if blob_id is None:
blob = dulobjs.Blob.from_string(fctx.data())
blob_id = blob.id
blob_cache[fctx.filenode()] = blob_id
flags = fctx.flags()
if "l" in flags:
mode = 0o120000
elif "x" in flags:
mode = 0o100755
else:
mode = 0o100644
return (dulobjs.TreeEntry(os.path.basename(fctx.path()), mode, blob_id), blob)
def _get_commit_from_tag(self, tag, tag_sha):
"""Return the commit referenced by the tag and when it was tagged."""
tag_obj = self[tag_sha]
if isinstance(tag_obj, objects.Tag):
# A signed tag has its own SHA, but the tag refers to
# the commit and that's the SHA we'll see when we scan
# commits on a branch.
git_obj = tag_obj
while True:
# Tags can point to other tags, in such cases follow the chain
# of tags until there are no more.
child_obj = self[git_obj.object[1]]
if isinstance(child_obj, objects.Tag):
git_obj = child_obj
else:
break
tagged_sha = git_obj.object[1]
date = tag_obj.tag_time
elif isinstance(tag_obj, objects.Commit):
if blob_id is None:
blob = dulobjs.Blob.from_string(fctx.data())
blob_id = blob.id
blob_cache[fctx.filenode()] = blob_id
flags = fctx.flags()
if "l" in flags:
mode = 0o120000
elif "x" in flags:
mode = 0o100755
else:
mode = 0o100644
return (dulobjs.TreeEntry(os.path.basename(fctx.path()), mode, blob_id), blob)
def get_tree_for_dir(tree, dirname):
for name, mode, id in tree.iteritems():
if name == dirname:
obj = self.repository._repo[id]
if isinstance(obj, objects.Tree):
return obj
else:
raise RepositoryError("Cannot create directory %s "
"at tree %s as path is occupied and is not a "
"Tree" % (dirname, tree))
return None
def extract_trees(repo, commit="a9b19e9453e516d7528aebb05a1efd66a0cd9347"):
'''get a list of all known trees from a repo based on a commit SHA'''
store = repo.object_store.generate_pack_contents([], [commit])
generator = store.iterobjects()
trees = []
for each in generator:
if isinstance(each, dulwich.objects.Tree):
trees.append(each)
return trees
continue
foo = GitTreeElement()
foo.sha = item.sha.decode("ascii")
foo.path = os.path.join(prefix, item.path.decode("utf8"))
if isinstance(obj, dulwich.objects.Blob):
foo.type = "blob"
blobs.append(foo)
elif isinstance(obj, dulwich.objects.Tree):
foo.type = "tree"
blobs.append(foo)
# Check whether to fetch more than the most upper layer.
if recursive:
_traverse(obj, repo_obj, blobs, foo.path)
tree = repo.get_object(tgt_env.tree)
if not isinstance(tree, dulwich.objects.Tree):
return []
blobs = []
if len(tree):
_traverse(tree, repo, blobs, "")
return blobs