-
Notifications
You must be signed in to change notification settings - Fork 216
/
scm.py
192 lines (164 loc) · 7.3 KB
/
scm.py
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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
# -*- coding: utf-8 -*-
# vim:expandtab:autoindent:tabstop=4:shiftwidth=4:filetype=python:textwidth=0:
# License: GPL2 or later see COPYING
# Written by Marko Myllynen
# Copyright (C) 2010 Marko Myllynen <myllynen@redhat.com>
import os
import shlex
import shutil
import subprocess
import sys
import tempfile
import six
from . import util
from .trace_decorator import traceLog
class scmWorker(object):
"""Build RPMs from SCM"""
@traceLog()
def __init__(self, log, opts, macros):
self.log = log
self.log.debug("Initializing SCM integration...")
self.name = self.version = None
self.wrk_dir = self.src_dir = None
self.sources = []
self.macros = macros
self.method = opts['method']
if self.method == "cvs":
self.get = opts['cvs_get']
elif self.method == "svn":
self.get = opts['svn_get']
elif self.method == "git":
self.get = opts['git_get']
else:
self.log.error("Unsupported SCM method: %s", self.method)
sys.exit(5)
self.branch = None
self.postget = []
if 'branch' in opts:
self.branch = opts['branch']
if self.branch:
if self.method == "cvs":
self.get = self.get.replace("SCM_BRN", "-r " + self.branch)
elif self.method == "git":
self.postget = ["git checkout " + self.branch]
if "--recursive" in self.get or "--recurse-submodules" in self.get:
self.postget.append("git submodule update --init --recursive")
elif self.method == "svn":
self.get = self.get.replace("SCM_BRN", self.branch)
else:
self.log.error("Unsupported SCM method: %s", self.method)
sys.exit(5)
elif self.method == "svn":
self.get = self.get.replace("SCM_BRN", "trunk")
self.get = self.get.replace("SCM_BRN", "")
if 'package' in opts:
self.pkg = opts['package']
else:
self.log.error("Trying to use SCM, package not defined")
sys.exit(5)
self.get = self.get.replace("SCM_PKG", self.pkg)
self.spec = opts['spec']
self.spec = self.spec.replace("SCM_PKG", self.pkg)
self.ext_src_dir = opts['ext_src_dir']
self.write_tar = opts['write_tar']
self.exclude_vcs = opts['exclude_vcs']
self.git_timestamps = opts['git_timestamps']
self.log.debug("SCM checkout command: %s", self.get)
for command in self.postget:
self.log.debug("SCM checkout post command: %s", command)
@traceLog()
def get_sources(self):
self.wrk_dir = tempfile.mkdtemp(".mock-scm." + self.pkg)
self.src_dir = self.wrk_dir + "/" + self.pkg
self.log.debug("SCM checkout directory: %s", self.wrk_dir)
util.do(shlex.split(self.get), shell=False, cwd=self.wrk_dir, env=os.environ)
for command in self.postget:
util.do(shlex.split(command), shell=False, cwd=self.src_dir, env=os.environ)
self.log.debug("Fetched sources from SCM")
@traceLog()
def adjust_git_timestamps(self):
cwd_dir = util.pretty_getcwd()
self.log.debug("Adjusting timestamps in %s", self.src_dir)
os.chdir(self.src_dir)
proc = subprocess.Popen(['git', 'ls-files', '-z'], shell=False, stdout=subprocess.PIPE)
for f in proc.communicate()[0].split('\0')[:-1]:
rev = subprocess.Popen(['git', 'rev-list', 'HEAD', f], shell=False,
stdout=subprocess.PIPE
).stdout.readlines()[0].rstrip('\n')
ts = subprocess.Popen(['git', 'show', '--pretty=format:%ai', '--abbrev-commit', rev, f],
shell=False,
stdout=subprocess.PIPE
).stdout.readlines()[0].rstrip('\n')
subprocess.Popen(['touch', '-d', ts, f], shell=False)
os.chdir(cwd_dir)
@traceLog()
def prepare_sources(self):
# import rpm after setarch
import rpm
self.log.debug("Preparing SCM sources")
# Check some helper files
if os.path.exists(self.src_dir + "/.write_tar"):
self.log.debug(".write_tar detected, will write tarball on the fly")
self.write_tar = True
# Figure out the spec file
sf = self.src_dir + "/" + self.spec
if not os.path.exists(sf):
sf = self.src_dir + "/" + self.spec.lower()
if not os.path.exists(sf):
self.log.error("Can't find spec file %s/%s", self.src_dir, self.spec)
self.clean()
sys.exit(5)
self.spec = sf
# Add passed RPM macros before parsing spec file
for macro, expression in list(self.macros.items()):
rpm.addMacro(macro.lstrip('%'), expression)
# Dig out some basic information from the spec file
self.sources = []
ts = rpm.ts()
rpm_spec = ts.parseSpec(self.spec)
self.name = rpm.expandMacro("%{name}")
self.version = rpm.expandMacro("%{version}")
tarball = None
for (filename, num, flags) in rpm_spec.sources:
self.sources.append(filename.split("/")[-1])
if num == 0 and flags == 1:
tarball = filename.split("/")[-1]
self.log.debug("Sources: %s", self.sources)
# Adjust timestamps for Git checkouts
if self.method == "git" and self.git_timestamps:
self.adjust_git_timestamps()
# Generate a tarball from the checked out sources if needed
if str(self.write_tar).lower() == "true":
tardir = self.name + "-" + self.version
if tarball is None:
tarball = tardir + ".tar.gz"
taropts = ""
# Always exclude vcs data from tarball unless told not to
if str(self.exclude_vcs).lower() == "true":
proc = subprocess.Popen(['tar', '--help'], shell=False, stdout=subprocess.PIPE)
proc_result = proc.communicate()[0]
if six.PY3:
proc_result = proc_result.decode()
if "--exclude-vcs" in proc_result:
taropts = "--exclude-vcs"
self.log.debug("Writing %s/%s...", self.src_dir, tarball)
cwd_dir = os.getcwd()
os.chdir(self.wrk_dir)
os.rename(self.name, tardir)
cmd = "tar caf " + tarball + " " + taropts + " " + tardir
util.do(shlex.split(cmd), shell=False, cwd=self.wrk_dir, env=os.environ)
os.rename(tarball, tardir + "/" + tarball)
os.rename(tardir, self.name)
os.chdir(cwd_dir)
# Get possible external sources from an external sources directory
for f in self.sources:
if not os.path.exists(self.src_dir + "/" + f) and \
os.path.exists(self.ext_src_dir + "/" + f):
self.log.debug("Copying %s/%s to %s/%s", self.ext_src_dir, f, self.src_dir, f)
shutil.copy2(self.ext_src_dir + "/" + f, self.src_dir + "/" + f)
self.log.debug("Prepared sources for building src.rpm")
return (self.src_dir, self.spec)
@traceLog()
def clean(self):
self.log.debug("Clean SCM checkout directory")
util.rmtree(self.wrk_dir)