hg.py 7.58 KB
Newer Older
1
from collections import namedtuple
2
import json
3
import logging
4
import os
5
6
import subprocess

Georges Racinet's avatar
Georges Racinet committed
7
8
9
10
SHORT_TEMPLATE = (
    '{rev}:{node|short} {desc|firstline} ({phase})\n'
    '{if (obsfate, "  obsolete:{obsfate}\n\n")}'
)
11
HG_EXECUTABLE = 'hg'
Georges Racinet's avatar
Georges Racinet committed
12

13
14
logger = logging.getLogger(__name__)

15

16
17
def hg_call(cwd, cmd, check_return_code=True, expected_return_code=0,
            encoding=None):
18
19
20
21
22
23
    """Wrapper to run hg, allowing process error or not.


    :param check_status_code:
        if ``True``, and hg status code isn't 0, raises like an stderr variant
        of ``subprocess.check_output()``.
24
    :param encoding: if `None`, the default encoding is used
25
    """
26
    hgcmd = [HG_EXECUTABLE]
27
28
29
30
31
32
    hgcmd.extend(str(s) if not isinstance(s, bytes) else s for s in cmd)
    if encoding is None:
        env = os.environ
    else:
        env = dict(os.environ)
        env['HGENCODING'] = encoding
33
34
    proc = subprocess.Popen(hgcmd,
                            cwd=str(cwd),  # -R not enough for hg init
35
                            env=env,
36
37
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
38
    out, err = [b.decode('utf-8', 'replace') for b in proc.communicate()]
39
    retcode = proc.poll()
40
    if check_return_code and retcode != expected_return_code:
41
42
        # logging at ERROR level because if it had been expected
        # call would have used `check_return_code=False`
43
44
45
        logger.error("Failed hg_call (code %d, expected %d), "
                     "stdout=%r, stderr=%r",
                     retcode, expected_return_code, out, err)
46
47
48
49
50
51
52
53
54
55
        raise subprocess.CalledProcessError(retcode, cmd, output=err)
    return retcode, out, err


class LocalRepo(object):
    """Represent a repository on the system where the tests are run."""

    hgrc_extensions_lines = ("[extensions]",
                             "evolve =",
                             "topic = ",
56
                             "rebase =",
57
58
                             "[ui]",
                             "username = Heptapod Tests <fonct@heptapod.test>"
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
                             )

    def __init__(self, path):
        self.path = path

    @classmethod
    def init(cls, path, default_url=None):
        """Create a repository at the given path and attach it to the given URL.

        :param path: a pytest Path instance
        :param url: Optional Heptapod URL of the repo, will become the
                    default path for push/pull operations.
                    In most cases, one would want the URL to enclose
                    Basic Authentication credentials, but it's not necessary.


        """
        hg_call('.', ['init', path])
        repo = cls(path)

        hgrc_lines = list(cls.hgrc_extensions_lines)
        if default_url is not None:
            hgrc_lines.extend(("", "[paths]", "default = " + str(default_url)))
        repo.hgrc_append_lines(hgrc_lines)

        return repo

    @classmethod
    def clone(cls, url, path):
        """Clone the repository from given URL.

        :param path: a pytest Path instance
        """
92
93
        repo = cls.init(path, default_url=url)
        repo.hg('pull', '-u')
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
        return repo

    def hgrc_append_lines(self, lines):
        with self.path.join('.hg', 'hgrc').open('a') as fp:
            fp.writelines(l + '\n' for l in lines)

    def hg(self, *args, **kwargs):
        """Invoke the `hg` executable and return its stdout.

        This behaves as `subprocess.check_output` does.
        """
        return hg_call(self.path, args, **kwargs)[1]

    def hg_unchecked(self, *args, **kwargs):
        """Invoke the `hg` executable without raising if return code is nonzero.

        :returns: (process return code, stdout, stderr).
        """
        return hg_call(self.path, args, check_return_code=False, **kwargs)
Georges Racinet's avatar
Georges Racinet committed
113
114
115
116
117
118
119
120
121
122
123
124
125
126

    def graphlog(self, hidden=False):
        """Return a full log.

        This is useful to print (captured by the runner and displayed if there
        are failures), maybe to send to a file.

        :param hidden: if True, also display hidden (obsolete) changesets.

        """
        cmd = ['log', '--graph', '-T', SHORT_TEMPLATE]
        if hidden:
            cmd.append('--hidden')
        return self.hg(*cmd)
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
    def changeset_extracts(self, template_exprs,
                           collection=tuple,
                           revs=None,
                           val_sep='|:|', eol="\n"):
        """Extract information from changesets, as a collection of entries.

        This doens't represent the graph information, but the ordering given
        by `hg log`, on which this method is based, is preserved if using
        an ordered collection.

        :param template_exprs: an iterable of template keywords.
           Examples: `['rev', 'phase', 'desc']`. Later on, for more complicated
           expressions, we could allow ``template_exprs`` elements to be
           ``('key', 'expr')`` pairs, where the key would be used in the
           as key of the resulting class:`dict` and `expr` would be evaluated.
           as template expressions, and keys
        :param collection: the wished collection class. Must be instantiable
           by a generator expression, like `set` and `tuple` are.
        :return: a named tuple class whose attributes are the elements of
          ``template_exprs`` and a set of instances of that class,
          one for each changeset.

        WARNING: there is no escaping. The command will fail utterly
        if the rendering of one template expression contains ``val_sep`` or
`       ``eol``.
        """
        template = val_sep.join("{%s}" % expr for expr in template_exprs)
        cmd = ['log', '-T', template + eol]
        if revs is not None:
            cmd.extend(('-r', revs))

        Extract = namedtuple('Extract', template_exprs)
        return Extract, collection(Extract(*line.split(val_sep))
                                   for line in self.hg(*cmd).split(eol)
                                   if line)

164
165
166
    def branches(self):
        """Return `hg branches` as a set of names."""
        return set(self.hg('branches', '-T', '{branch}\n').splitlines())
167
168
169
170

    def node(self, revspec):
        """Return the node if for specified revision (can be symbolic)"""
        return self.hg('log', '-T', "{node}", '-r', revspec)
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185

    def init_gitlab_ci(self,
                       message="Init GitLab CI",
                       script=("grep foo foo",),
                       ):
        """Create an commit a .gitlab-ci.yml file.

        A convenience method to have a valid configuration without caring
        too much with what it entails.
        """
        ci_config = dict(job=dict(script=script))
        # JSON is a subset of YaML and part of Python standard library
        self.path.join('.gitlab-ci.yml').write(json.dumps(ci_config))
        self.hg('add', '.gitlab-ci.yml')
        self.hg('commit', '-m', message)
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205


def assert_matching_changesets(repo1, repo2, fields, ordered=True, revs=None):
    """Assert changesets to have same information in the two given repos

    Of course all structural information of the changesets can be checked
    with just ``node``, but there is also outer information, like for example
    ``phases``, and ``bookmarks``, whose inclusion depends on the test case.

    Also, adding some human-readable information such as `desc` or `topic`
    can make matter more obvious to people engaged in debugging.

    The ordering can be kept or not, according to the test case.
    """
    collection = tuple if ordered else set
    extract1, extract2 = [
        repo.changeset_extracts(fields, collection=collection, revs=revs)[1]
        for repo in (repo1, repo2)
    ]
    assert extract1 == extract2