Updated tests for GIT 1.6
[0release.git] / support.py
blob51d935a6572af6b129be410b9d34a4b964556146
1 # Copyright (C) 2007, Thomas Leonard
2 # See the README file for details, or visit http://0install.net.
4 import os, subprocess, shutil, tarfile
5 import urlparse, ftplib, httplib
6 from zeroinstall import SafeException
7 from zeroinstall.injector import model, qdom
8 from logging import info
10 release_status_file = os.path.abspath('release-status')
12 def check_call(*args, **kwargs):
13 exitstatus = subprocess.call(*args, **kwargs)
14 if exitstatus != 0:
15 if type(args[0]) in (str, unicode):
16 cmd = args[0]
17 else:
18 cmd = ' '.join(args[0])
19 raise SafeException("Command failed with exit code %d:\n%s" % (exitstatus, cmd))
21 def show_and_run(cmd, args):
22 print "Executing: %s %s" % (cmd, ' '.join("[%s]" % x for x in args))
23 check_call(['sh', '-c', cmd, '-'] + args)
25 def suggest_release_version(snapshot_version):
26 """Given a snapshot version, suggest a suitable release version.
27 >>> suggest_release_version('1.0-pre')
28 '1.0'
29 >>> suggest_release_version('0.9-post')
30 '0.10'
31 >>> suggest_release_version('3')
32 Traceback (most recent call last):
33 ...
34 SafeException: Version '3' is not a snapshot version (should end in -pre or -post)
35 """
36 version = model.parse_version(snapshot_version)
37 mod = version[-1]
38 if mod == 0:
39 raise SafeException("Version '%s' is not a snapshot version (should end in -pre or -post)" % snapshot_version)
40 if mod > 0:
41 # -post, so increment the number
42 version[-2][-1] += 1
43 version[-1] = 0 # Remove the modifier
44 return model.format_version(version)
46 def publish(iface, **kwargs):
47 args = [os.environ['0PUBLISH']]
48 for k in kwargs:
49 value = kwargs[k]
50 if value is True:
51 args += ['--' + k.replace('_', '-')]
52 elif value is not None:
53 args += ['--' + k.replace('_', '-'), value]
54 args.append(iface)
55 info("Executing %s", args)
56 check_call(args)
58 def get_singleton_impl(iface):
59 impls = iface.implementations
60 if len(impls) != 1:
61 raise SafeException("Local feed '%s' contains %d versions! I need exactly one!" % (iface.uri, len(impls)))
62 return impls.values()[0]
64 def backup_if_exists(name):
65 if not os.path.exists(name):
66 return
67 backup = name + '~'
68 if os.path.exists(backup):
69 print "(deleting old backup %s)" % backup
70 if os.path.isdir(backup):
71 shutil.rmtree(backup)
72 else:
73 os.unlink(backup)
74 os.rename(name, backup)
75 print "(renamed old %s as %s; will delete on next run)" % (name, backup)
77 def get_choice(options):
78 while True:
79 choice = raw_input('/'.join(options) + ': ').lower()
80 if not choice: continue
81 for o in options:
82 if o.lower().startswith(choice):
83 return o
85 def make_archive_name(feed_name, version):
86 return feed_name.lower().replace(' ', '-') + '-' + version
88 def in_PATH(prog):
89 for x in os.environ['PATH'].split(':'):
90 if os.path.isfile(os.path.join(x, prog)):
91 return True
92 return False
94 def show_diff(from_dir, to_dir):
95 for cmd in [['meld'], ['xxdiff'], ['diff', '-ur']]:
96 if in_PATH(cmd[0]):
97 code = os.spawnvp(os.P_WAIT, cmd[0], cmd + [from_dir, to_dir])
98 if code:
99 print "WARNING: command %s failed with exit code %d" % (cmd, code)
100 return
102 class Status(object):
103 __slots__ = ['old_snapshot_version', 'release_version', 'head_before_release', 'new_snapshot_version',
104 'head_at_release', 'created_archive', 'tagged', 'uploaded_archive', 'updated_master_feed']
105 def __init__(self):
106 for name in self.__slots__:
107 setattr(self, name, None)
109 if os.path.isfile(release_status_file):
110 for line in file(release_status_file):
111 assert line.endswith('\n')
112 line = line[:-1]
113 name, value = line.split('=')
114 setattr(self, name, value)
115 info("Loaded status %s=%s", name, value)
117 def save(self):
118 tmp_name = release_status_file + '.new'
119 tmp = file(tmp_name, 'w')
120 try:
121 lines = ["%s=%s\n" % (name, getattr(self, name)) for name in self.__slots__ if getattr(self, name)]
122 tmp.write(''.join(lines))
123 tmp.close()
124 os.rename(tmp_name, release_status_file)
125 info("Wrote status to %s", release_status_file)
126 except:
127 os.unlink(tmp_name)
128 raise
130 def host(address):
131 if hasattr(address, 'hostname'):
132 return address.hostname
133 else:
134 return address[1].split(':', 1)[0]
136 def port(address):
137 if hasattr(address, 'port'):
138 return address.port
139 else:
140 port = address[1].split(':', 1)[1:]
141 if port:
142 return int(port[0])
143 else:
144 return None
146 def get_http_size(url, ttl = 1):
147 assert url.lower().startswith('http://')
149 address = urlparse.urlparse(url)
150 http = httplib.HTTPConnection(host(address), port(address) or 80)
152 parts = url.split('/', 3)
153 if len(parts) == 4:
154 path = parts[3]
155 else:
156 path = ''
158 http.request('HEAD', '/' + path, headers = {'Host': host(address)})
159 response = http.getresponse()
160 try:
161 if response.status == 200:
162 return response.getheader('Content-Length')
163 elif response.status in (301, 302):
164 new_url_rel = response.getheader('Location') or response.getheader('URI')
165 new_url = urlparse.urljoin(url, new_url_rel)
166 else:
167 raise SafeException("HTTP error: got status code %s" % response.status)
168 finally:
169 response.close()
171 if ttl:
172 info("Resource moved! Checking new URL %s" % new_url)
173 assert new_url
174 return get_http_size(new_url, ttl - 1)
175 else:
176 raise SafeException('Too many redirections.')
178 def get_ftp_size(url):
179 address = urlparse.urlparse(url)
180 ftp = ftplib.FTP(host(address))
181 try:
182 ftp.login()
183 return ftp.size(url.split('/', 3)[3])
184 finally:
185 ftp.close()
187 def get_size(url):
188 scheme = urlparse.urlparse(url)[0].lower()
189 if scheme.startswith('http'):
190 return get_http_size(url)
191 elif scheme.startswith('ftp'):
192 return get_ftp_size(url)
193 else:
194 raise SafeException("Unknown scheme '%s' in '%s'" % (scheme, url))
196 def unpack_tarball(archive_file):
197 tar = tarfile.open(archive_file, 'r:bz2')
198 members = [m for m in tar.getmembers() if m.name != 'pax_global_header']
199 tar.extractall('.', members = members)
201 def load_feed(path):
202 stream = open(path)
203 try:
204 return model.ZeroInstallFeed(qdom.parse(stream), local_path = path)
205 finally:
206 stream.close()
208 def get_archive_basename(impl):
209 return os.path.basename(urlparse.urlparse(impl.download_sources[0].url).path)
211 def relative_path(ancestor, dst):
212 stem = os.path.abspath(os.path.dirname(ancestor))
213 dst = os.path.abspath(dst)
214 if stem != '/':
215 stem += '/'
216 assert dst.startswith(stem)
217 return dst[len(stem):]
219 assert relative_path('/foo', '/foo') == 'foo'
220 assert relative_path('/foo', '/foo/bar') == 'foo/bar'