summaryrefslogtreecommitdiff
path: root/tests/testutils/repo/git.py
blob: 1455de264f9a6e132ae2b0b5e1c9e762c5a7a6e8 (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
import os
import shutil
import subprocess

import pytest

from .repo import Repo
from .. import site


GIT_ENV = {
    'GIT_AUTHOR_DATE': '1320966000 +0200',
    'GIT_AUTHOR_NAME': 'tomjon',
    'GIT_AUTHOR_EMAIL': 'tom@jon.com',
    'GIT_COMMITTER_DATE': '1320966000 +0200',
    'GIT_COMMITTER_NAME': 'tomjon',
    'GIT_COMMITTER_EMAIL': 'tom@jon.com'
}


class Git(Repo):

    def __init__(self, directory, subdir):
        if not site.HAVE_GIT:
            pytest.skip("git is not available")

        self.submodules = {}

        super(Git, self).__init__(directory, subdir)

    def _run_git(self, *args, **kwargs):
        argv = [site.GIT]
        argv.extend(args)
        if 'env' not in kwargs:
            kwargs['env'] = dict(GIT_ENV, PWD=self.repo)
        kwargs.setdefault('cwd', self.repo)
        kwargs.setdefault('check', True)
        return subprocess.run(argv, **kwargs)

    def create(self, directory):
        self.copy_directory(directory, self.repo)
        self._run_git('init', '.')
        self._run_git('add', '.')
        self._run_git('commit', '-m', 'Initial commit')
        return self.latest_commit()

    def add_tag(self, tag):
        self._run_git('tag', tag)

    def add_annotated_tag(self, tag, message):
        self._run_git('tag', '-a', tag, '-m', message)

    def add_commit(self):
        self._run_git('commit', '--allow-empty', '-m', 'Additional commit')
        return self.latest_commit()

    def add_file(self, filename):
        shutil.copy(filename, self.repo)
        self._run_git('add', os.path.basename(filename))
        self._run_git('commit', '-m', 'Added {}'.format(os.path.basename(filename)))
        return self.latest_commit()

    def modify_file(self, new_file, path):
        shutil.copy(new_file, os.path.join(self.repo, path))
        self._run_git('commit', path, '-m', 'Modified {}'.format(os.path.basename(path)))
        return self.latest_commit()

    def add_submodule(self, subdir, url=None, checkout=None):
        submodule = {}
        if checkout is not None:
            submodule['checkout'] = checkout
        if url is not None:
            submodule['url'] = url
        self.submodules[subdir] = submodule
        self._run_git('submodule', 'add', url, subdir)
        self._run_git('commit', '-m', 'Added the submodule')
        return self.latest_commit()

    # This can also be used to a file or a submodule
    def remove_path(self, path):
        self._run_git('rm', path)
        self._run_git('commit', '-m', 'Removing {}'.format(path))
        return self.latest_commit()

    def source_config(self, ref=None, checkout_submodules=None):
        config = {
            'kind': 'git',
            'url': 'file://' + self.repo,
            'track': 'master'
        }
        if ref is not None:
            config['ref'] = ref
        if checkout_submodules is not None:
            config['checkout-submodules'] = checkout_submodules

        if self.submodules:
            config['submodules'] = dict(self.submodules)

        return config

    def latest_commit(self):
        return self._run_git(
            'rev-parse', 'HEAD',
            stdout=subprocess.PIPE,
            universal_newlines=True,
        ).stdout.strip()

    def branch(self, branch_name):
        self._run_git('checkout', '-b', branch_name)

    def delete_tag(self, tag_name):
        self._run_git('tag', '-d', tag_name)

    def checkout(self, commit):
        self._run_git('checkout', commit)

    def merge(self, commit):
        self._run_git('merge', '-m', 'Merge', commit)
        return self.latest_commit()

    def rev_parse(self, rev):
        return self._run_git(
            'rev-parse', rev,
            stdout=subprocess.PIPE,
            universal_newlines=True,
        ).stdout.strip()