1#!/usr/bin/env python
2# ***** BEGIN LICENSE BLOCK *****
3# This Source Code Form is subject to the terms of the Mozilla Public
4# License, v. 2.0. If a copy of the MPL was not distributed with this file,
5# You can obtain one at http://mozilla.org/MPL/2.0/.
6# ***** END LICENSE BLOCK *****
7"""Generic VCS support.
8"""
9
10from __future__ import absolute_import
11import os
12import sys
13from copy import deepcopy
14
15from mozharness.base.errors import VCSException
16from mozharness.base.log import FATAL
17from mozharness.base.script import BaseScript
18from mozharness.base.vcs.gittool import GittoolVCS
19from mozharness.base.vcs.mercurial import MercurialVCS
20
21sys.path.insert(1, os.path.dirname(os.path.dirname(os.path.dirname(sys.path[0]))))
22
23
24# Update this with supported VCS name : VCS object
25VCS_DICT = {
26    "hg": MercurialVCS,
27    "gittool": GittoolVCS,
28}
29
30
31# VCSMixin {{{1
32class VCSMixin(object):
33    """Basic VCS methods that are vcs-agnostic.
34    The vcs_class handles all the vcs-specific tasks.
35    """
36
37    def query_dest(self, kwargs):
38        if "dest" in kwargs:
39            return kwargs["dest"]
40        dest = os.path.basename(kwargs["repo"])
41        # Git fun
42        if dest.endswith(".git"):
43            dest = dest.replace(".git", "")
44        return dest
45
46    def _get_revision(self, vcs_obj, dest):
47        try:
48            got_revision = vcs_obj.ensure_repo_and_revision()
49            if got_revision:
50                return got_revision
51        except VCSException:
52            self.rmtree(dest)
53            raise
54
55    def _get_vcs_class(self, vcs):
56        vcs = vcs or self.config.get("default_vcs", getattr(self, "default_vcs", None))
57        vcs_class = VCS_DICT.get(vcs)
58        return vcs_class
59
60    def vcs_checkout(self, vcs=None, error_level=FATAL, **kwargs):
61        """Check out a single repo."""
62        c = self.config
63        vcs_class = self._get_vcs_class(vcs)
64        if not vcs_class:
65            self.error("Running vcs_checkout with kwargs %s" % str(kwargs))
66            raise VCSException("No VCS set!")
67        # need a better way to do this.
68        if "dest" not in kwargs:
69            kwargs["dest"] = self.query_dest(kwargs)
70        if "vcs_share_base" not in kwargs:
71            kwargs["vcs_share_base"] = c.get(
72                "%s_share_base" % vcs, c.get("vcs_share_base")
73            )
74        vcs_obj = vcs_class(
75            log_obj=self.log_obj,
76            config=self.config,
77            vcs_config=kwargs,
78            script_obj=self,
79        )
80        return self.retry(
81            self._get_revision,
82            error_level=error_level,
83            error_message="Automation Error: Can't checkout %s!" % kwargs["repo"],
84            args=(vcs_obj, kwargs["dest"]),
85        )
86
87    def vcs_checkout_repos(
88        self, repo_list, parent_dir=None, tag_override=None, **kwargs
89    ):
90        """Check out a list of repos."""
91        orig_dir = os.getcwd()
92        c = self.config
93        if not parent_dir:
94            parent_dir = os.path.join(c["base_work_dir"], c["work_dir"])
95        self.mkdir_p(parent_dir)
96        self.chdir(parent_dir)
97        revision_dict = {}
98        kwargs_orig = deepcopy(kwargs)
99        for repo_dict in repo_list:
100            kwargs = deepcopy(kwargs_orig)
101            kwargs.update(repo_dict)
102            if tag_override:
103                kwargs["branch"] = tag_override
104            dest = self.query_dest(kwargs)
105            revision_dict[dest] = {"repo": kwargs["repo"]}
106            revision_dict[dest]["revision"] = self.vcs_checkout(**kwargs)
107        self.chdir(orig_dir)
108        return revision_dict
109
110    def vcs_query_pushinfo(self, repository, revision, vcs=None):
111        """Query the pushid/pushdate of a repository/revision
112        Returns a namedtuple with "pushid" and "pushdate" elements
113        """
114        vcs_class = self._get_vcs_class(vcs)
115        if not vcs_class:
116            raise VCSException("No VCS set in vcs_query_pushinfo!")
117        vcs_obj = vcs_class(
118            log_obj=self.log_obj,
119            config=self.config,
120            script_obj=self,
121        )
122        return vcs_obj.query_pushinfo(repository, revision)
123
124
125class VCSScript(VCSMixin, BaseScript):
126    def __init__(self, **kwargs):
127        super(VCSScript, self).__init__(**kwargs)
128
129    def pull(self, repos=None, parent_dir=None):
130        repos = repos or self.config.get("repos")
131        if not repos:
132            self.info("Pull has nothing to do!")
133            return
134        dirs = self.query_abs_dirs()
135        parent_dir = parent_dir or dirs["abs_work_dir"]
136        return self.vcs_checkout_repos(repos, parent_dir=parent_dir)
137
138
139# Specific VCS stubs {{{1
140# For ease of use.
141# This is here instead of mercurial.py because importing MercurialVCS into
142# vcsbase from mercurial, and importing VCSScript into mercurial from
143# vcsbase, was giving me issues.
144class MercurialScript(VCSScript):
145    default_vcs = "hg"
146
147
148# __main__ {{{1
149if __name__ == "__main__":
150    pass
151