aboutsummaryrefslogtreecommitdiffhomepage
path: root/gbp/pkg/upstreamsource.py
blob: 384513e52ae195c1e017bbc510f44cbda069ce87 (plain)
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
193
194
195
196
197
198
199
200
201
202
203
204
# vim: set fileencoding=utf-8 :
#
# (C) 2017 Guido Günther <agx@sigxcpu.org>
#    This program is free software; you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation; either version 2 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with this program; if not, please see
#    <http://www.gnu.org/licenses/>

import glob
import os

import gbp.command_wrappers as gbpc

from gbp.pkg.compressor import Compressor
from gbp.pkg.pkgpolicy import PkgPolicy

from gbp.errors import GbpError


class UpstreamSource(object):
    """
    Upstream source. Can be either an unpacked dir, a tarball or another type
    of archive

    @cvar _orig: are the upstream sources already suitable as an upstream
                 tarball
    @type _orig: boolean
    @cvar _path: path to the upstream sources
    @type _path: string
    @cvar _unpacked: path to the unpacked source tree
    @type _unpacked: string
    """
    def __init__(self, name, unpacked=None, pkg_policy=PkgPolicy, sig=None):
        self._orig = False
        self._pkg_policy = pkg_policy
        self._path = name
        self.unpacked = unpacked
        self._sig = sig

        self._check_orig()
        if self.is_dir():
            self.unpacked = self.path

    def __repr__(self):
        path = "'%s'" % self.path if self.path is not None else None
        sig = "'%s'" % self.signaturefile if self.signaturefile is not None else None
        return "<%s path=%s signaturefile=%s>" % (self.__class__.__name__,
                                                  path, sig)

    def _check_orig(self):
        """
        Check if upstream source format can be used as orig tarball.
        This doesn't imply that the tarball is correctly named.

        @return: C{True} if upstream source format is suitable
            as upstream tarball, C{False} otherwise.
        @rtype: C{bool}
        """
        if self.is_dir():
            self._orig = False
            return

        parts = self._path.split('.')
        try:
            if parts[-1] == 'tgz':
                self._orig = True
            elif parts[-2] == 'tar':
                if (parts[-1] in Compressor.Opts or
                        parts[-1] in Compressor.Aliases):
                    self._orig = True
        except IndexError:
            self._orig = False

    def is_orig(self):
        """
        @return: C{True} if sources are suitable as orig tarball,
            C{False} otherwise
        @rtype: C{bool}
        """
        return self._orig

    def is_dir(self):
        """
        @return: C{True} if if upstream sources are an unpacked directory,
            C{False} otherwise
        @rtype: C{bool}
        """
        return True if os.path.isdir(self._path) else False

    @property
    def path(self):
        return self._path.rstrip('/')

    @property
    def signaturefile(self):
        return self._sig

    def unpack(self, dir, filters=None):
        """
        Unpack packed upstream sources into a given directory
        (filtering out files specified by filters) and determine the
        toplevel of the source tree.
        """
        if self.is_dir():
            raise GbpError("Cannot unpack directory %s" % self.path)

        if not filters:
            filters = []

        if not isinstance(filters, list):
            raise GbpError("Filters must be a list")

        self._unpack_archive(dir, filters)
        self.unpacked = self._unpacked_toplevel(dir)

    def _unpack_archive(self, dir, filters):
        """
        Unpack packed upstream sources into a given directory
        allowing to filter out files in case of tar archives.
        """
        ext = os.path.splitext(self.path)[1]
        if ext in [".zip", ".xpi"]:
            if filters:
                raise GbpError("Can only filter tar archives: %s", (ext, self.path))
            self._unpack_zip(dir)
        else:
            self._unpack_tar(dir, filters)

    def _unpack_zip(self, dir):
        try:
            gbpc.UnpackZipArchive(self.path, dir)()
        except gbpc.CommandExecFailed:
            raise GbpError("Unpacking of %s failed" % self.path)

    def _unpacked_toplevel(self, dir):
        """unpacked archives can contain a leading directory or not"""
        unpacked = glob.glob('%s/*' % dir)
        unpacked.extend(glob.glob("%s/.*" % dir))  # include hidden files and folders
        # Check that dir contains nothing but a single folder:
        if len(unpacked) == 1 and os.path.isdir(unpacked[0]):
            return unpacked[0]
        else:
            return dir

    def _unpack_tar(self, dir, filters):
        """
        Unpack a tarball to I{dir} applying a list of I{filters}. Leave the
        cleanup to the caller in case of an error.
        """
        try:
            unpackArchive = gbpc.UnpackTarArchive(self.path, dir, filters)
            unpackArchive()
        except gbpc.CommandExecFailed:
            # unpackArchive already printed an error message
            raise GbpError

    def pack(self, newarchive, filters=None):
        """
        Recreate a new archive from the current one

        @param newarchive: the name of the new archive
        @type newarchive: string
        @param filters: tar filters to apply
        @type filters: array of strings
        @return: the new upstream source
        @rtype: UpstreamSource
        """
        if not self.unpacked:
            raise GbpError("Need an unpacked source tree to pack")

        if not filters:
            filters = []

        if not isinstance(filters, list):
            raise GbpError("Filters must be a list")

        try:
            unpacked = self.unpacked.rstrip('/')
            repackArchive = gbpc.PackTarArchive(newarchive,
                                                os.path.dirname(unpacked),
                                                os.path.basename(unpacked),
                                                filters)
            repackArchive()
        except gbpc.CommandExecFailed:
            # repackArchive already printed an error
            raise GbpError
        return type(self)(newarchive)

    @staticmethod
    def known_compressions():
        return Compressor.Exts.values()

    def guess_version(self, extra_regex=r''):
        return self._pkg_policy.guess_upstream_src_version(self.path,
                                                           extra_regex)