summaryrefslogtreecommitdiff
path: root/tasks.py
blob: e271c99311533145b4dfecaaedffb6e9db7c1d0b (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
# -*- coding: utf-8 -*-
import os
import re
import io
from contextlib import contextmanager
from datetime import datetime

from invoke import task, run

VERSION_FILE = 'pytest_catchlog.py'
CHANGE_LOG_FILE = 'CHANGES.rst'


def _path_abs_join(*nodes):
    return os.path.abspath(os.path.join(os.path.dirname(__file__), *nodes))


def _path_open(*nodes, **kwargs):
    return io.open(_path_abs_join(*nodes), **kwargs)


def _shell_quote(s):
    """Quote given string to be suitable as input for bash as argument."""
    if not s:
        return "''"
    if re.search(r'[^\w@%+=:,./-]', s) is None:
        return s
    return "'" + s.replace("'", "'\"'\"'") + "'"


def _git_do(*commands, **kwargs):
    """Execute arbitrary git commands."""
    kwargs.setdefault('hide', 'out')
    results = [run('git ' + command, **kwargs).stdout.strip('\n')
               for command in commands]
    return results if len(commands) > 1 else results[0]


def _git_checkout(branch_name):
    """Switches to the given branch name."""
    return _git_do('checkout ' + _shell_quote(branch_name))


@contextmanager
def _git_work_on(branch_name):
    """Work on given branch. Preserves current git branch."""
    original_branch = _git_do('rev-parse --abbrev-ref HEAD')
    try:
        if original_branch != branch_name:
            _git_checkout(branch_name)
        yield
    finally:
        if original_branch and original_branch != branch_name:
            _git_checkout(original_branch)


def _version_find_existing():
    """Returns set of existing versions in this repository.

    This information is backed by previously used version tags
    stored in the git repository.
    """
    git_tags = [y.strip() for y in _git_do('tag -l').split('\n')]

    _version_re = re.compile(r'^v?(\d+)(?:\.(\d+)(?:\.(\d+))?)?$')
    return {tuple(int(n) if n else 0 for n in m.groups())
            for m in (_version_re.match(t) for t in git_tags if t) if m}


def _version_find_latest():
    """Returns the most recent used version number.

    This information is backed by previously used version tags
    stored in the git repository.
    """
    return max(_version_find_existing())


def _version_guess_next(position='minor'):
    """Guess next version.

    A guess for the next version is determined by incrementing given
    position or minor level position in latest existing version.
    """
    try:
        latest_version = list(_version_find_latest())
    except ValueError:
        latest_version = [0, 0, 0]

    position_index = {'major': 0, 'minor': 1, 'patch': 2}[position]
    latest_version[position_index] += 1
    latest_version[position_index + 1:] = [0] * (2 - position_index)
    return tuple(latest_version)


def _version_format(version):
    """Return version in dotted string format."""
    return '.'.join(str(x) for x in version)


def _patch_file(file_path, line_callback):
    """Patch given file with result from line callback.

    Each line will be passed to the line callback.
    The return value of the given callback will determine
    the new content for the file.

    :param str file_path:
        The file to patch.
    :param callable line_callback:
        The patch function to run over each line.
    :return:
        Whenever the file has changed or not.
    :rtype:
        bool
    """
    new_file_content, file_changed = [], False
    with _path_open(file_path) as in_stream:
        for l in (x.strip('\n') for x in in_stream):
            alt_lines = line_callback(l) or [l]
            if alt_lines != [l]:
                file_changed = True
            new_file_content += (x + u'\n' for x in alt_lines)

    new_file_name = file_path + '.new'
    with _path_open(new_file_name, mode='w') as out_stream:
        out_stream.writelines(new_file_content)
        out_stream.flush()
        os.fsync(out_stream.fileno())
    os.rename(new_file_name, file_path)

    return file_changed


def _patch_version(new_version):
    """Patch given version into version file."""
    _patch_version_re = re.compile(r"""^(\s*__version__\s*=\s*(?:"|'))"""
                                   r"""(?:[^'"]*)(?:("|')\s*)$""")

    def __line_callback(line):
        match = _patch_version_re.match(line)
        if match:
            line_head, line_tail = match.groups()
            return [line_head + new_version + line_tail]
    return _patch_file(VERSION_FILE, __line_callback)


def _patch_change_log(new_version):
    """Patch given version into change log file."""
    def __line_callback(line):
        if line == u'`Unreleased`_':
            return [u'`{}`_'.format(new_version)]
        elif line == u'Yet to be released.':
            return [datetime.utcnow().strftime(u'Released on %F.')]
        elif line == u'.. %UNRELEASED_SECTION%':
            return [u'.. %UNRELEASED_SECTION%',
                    u'',
                    u'`Unreleased`_',
                    u'-------------',
                    u'',
                    u'Yet to be released.',
                    u'']
    return _patch_file(CHANGE_LOG_FILE, __line_callback)


@task()
def mkrelease(position='minor'):
    """Merge development state into Master Branch and tags a new Release."""
    next_version = _version_format(_version_guess_next(position))
    with _git_work_on('develop'):
        patched_files = []
        if _patch_version(next_version):
            patched_files.append(VERSION_FILE)

        if _patch_change_log(next_version):
            patched_files.append(CHANGE_LOG_FILE)

        if patched_files:
            patched_files = ' '.join(_shell_quote(x) for x in patched_files)
            _git_do('diff --color=always -- ' + patched_files,
                    ("commit -m 'Bump Version to {0}' -- {1}"
                     .format(next_version, patched_files)),
                    hide=None)

        with _git_work_on('master'):
            message = _shell_quote('Release {0}'.format(next_version))
            _git_do('merge --no-ff --no-edit -m {0} develop'.format(message),
                    "tag -a -m {0} {1}".format(message, next_version))