1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
# Copyright 2015 Canonical Ltd. All rights reserved.
import itertools
import os
import uuid
from pygit2 import (
init_repository,
GIT_FILEMODE_BLOB,
GIT_OBJ_COMMIT,
GIT_SORT_TIME,
IndexEntry,
Repository,
Signature,
)
def get_revlist(repo):
"""Return revlist for a given pygit2 repo object."""
return [commit.oid.hex for commit in repo.walk(repo.head.target)]
def open_repo(repo_path):
"""Return a pygit2 repo object for a given path."""
return Repository(repo_path)
class RepoFactory():
"""Builds a git repository in a user defined state."""
def __init__(self, repo_path=None, num_commits=None,
num_branches=None, num_tags=None):
self.author = Signature('Test Author', 'author@bar.com')
self.branches = []
self.committer = Signature('Test Commiter', 'committer@bar.com')
self.commits = []
self.num_branches = num_branches
self.num_commits = num_commits
self.num_tags = num_tags
self.repo_path = repo_path
self.repo = self.init_repo()
@property
def commits(self):
"""Walk repo from HEAD and returns list of commit objects."""
last = self.repo[self.repo.head.target]
return list(self.repo.walk(last.id, GIT_SORT_TIME))
def add_commit(self, blob_content, file_path, parents=[],
ref=None, author=None, committer=None):
"""Create a commit from blob_content and file_path."""
repo = self.repo
if not author:
author = self.author
if not committer:
committer = self.committer
blob_oid = repo.create_blob(blob_content)
blob_entry = IndexEntry(file_path, blob_oid, GIT_FILEMODE_BLOB)
repo.index.add(blob_entry)
tree_id = repo.index.write_tree()
oid = repo.create_commit(ref, author, committer,
blob_content, tree_id, parents)
return oid
def add_branch(self, name, oid):
commit = self.repo.get(oid)
branch = self.repo.create_branch('branch-{}'.format(name), commit)
self.branches.append(branch)
return branch
def add_tag(self, tag_name, tag_message, oid):
"""Create a tag from tag_name and oid."""
repo = self.repo
repo.create_tag(tag_name, oid, GIT_OBJ_COMMIT,
self.committer, tag_message)
def makeSignature(self, name, email, encoding='utf-8'):
"""Return an author or committer signature."""
return Signature(name, email, encoding=encoding)
def stage(self, file_path):
"""Stage a file and return a tree id."""
repo = self.repo
repo.index.add(file_path)
repo.index.write()
return repo.index.write_tree()
def generate_commits(self, num_commits, parents=[]):
"""Generate n number of commits."""
for i in xrange(num_commits):
blob_content = b'commit {} - {}'.format(i, uuid.uuid1())
test_file = 'test.txt'
with open(os.path.join(self.repo_path, test_file), 'w') as f:
f.write(blob_content)
self.stage(test_file)
commit_oid = self.add_commit(blob_content, test_file, parents)
self.commits.append(commit_oid)
parents = [commit_oid]
if i == num_commits - 1:
ref = 'refs/heads/master'
try:
self.repo.lookup_reference(ref)
except KeyError:
self.repo.create_reference(ref, commit_oid)
self.repo.set_head(commit_oid)
def generate_tags(self, num_tags):
"""Generate n number of tags."""
repo = self.repo
oid = repo.head.get_object().oid
for i in xrange(num_tags):
self.add_tag('tag{}'.format(i), 'tag message {}'.format(i), oid)
def generate_branches(self, num_branches, num_commits):
"""Generate n number of branches with n commits."""
repo = self.repo
parents = []
for i in xrange(num_branches):
self.generate_commits(num_commits, parents)
oid = repo.revparse_single('HEAD')
branch = repo.create_branch('branch-{}'.format(i), oid)
self.branches.append(branch)
parents.append(self.commits[0])
def nonexistent_oid(self):
"""Return an arbitrary OID that does not exist in this repo."""
for oid_chars in itertools.product('0123456789abcdef', repeat=40):
oid = ''.join(oid_chars)
if oid not in self.repo:
return oid
raise Exception("repo appears to contain every possible OID!")
def init_repo(self):
return init_repository(self.repo_path)
def build(self):
"""Return a repo, optionally with generated commits and tags."""
if self.num_branches:
self.generate_branches(self.num_branches, self.num_commits)
if not self.num_branches and self.num_commits:
self.generate_commits(self.num_commits)
if self.num_tags:
self.generate_tags(self.num_tags)
return self.repo
|