-
-
Notifications
You must be signed in to change notification settings - Fork 3.6k
/
base.py
166 lines (128 loc) · 4.95 KB
/
base.py
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
"""Base classes for VCS backends."""
import os
import structlog
from readthedocs.core.utils.filesystem import safe_rmtree
from readthedocs.doc_builder.exceptions import BuildCancelled, BuildUserError
from readthedocs.projects.exceptions import RepositoryError
log = structlog.get_logger(__name__)
class VCSVersion:
"""
Represents a Version (tag or branch) in a VCS.
This class should only be instantiated in BaseVCS subclasses.
It can act as a context manager to temporarily switch to this tag (eg to
build docs for this tag).
"""
def __init__(self, repository, identifier, verbose_name):
self.repository = repository
self.identifier = identifier
self.verbose_name = verbose_name
def __repr__(self):
return '<VCSVersion: {}:{}'.format(
self.repository.repo_url,
self.verbose_name,
)
class BaseVCS:
"""
Base for VCS Classes.
VCS commands are executed inside a ``BaseBuildEnvironment`` subclass.
"""
supports_tags = False # Whether this VCS supports tags or not.
supports_branches = False # Whether this VCS supports branches or not.
supports_submodules = False
# Whether this VCS supports listing remotes (branches, tags) without cloning
supports_lsremote = False
# =========================================================================
# General methods
# =========================================================================
# Defining a base API, so we'll have unused args
# pylint: disable=unused-argument
def __init__(
self,
project,
version_slug,
environment,
verbose_name=None,
version_type=None,
**kwargs
):
self.default_branch = project.default_branch
self.project = project
self.name = project.name
self.repo_url = project.clean_repo
self.working_dir = project.checkout_path(version_slug)
# required for External versions
self.verbose_name = verbose_name
self.version_type = version_type
self.environment = environment
def check_working_dir(self):
if not os.path.exists(self.working_dir):
os.makedirs(self.working_dir)
def make_clean_working_dir(self):
"""Ensures that the working dir exists and is empty."""
safe_rmtree(self.working_dir, ignore_errors=True)
self.check_working_dir()
def update(self, identifier=None):
"""
Update a local copy of the repository in self.working_dir.
If self.working_dir is already a valid local copy of the repository,
update the repository, else create a new local copy of the repository.
"""
self.check_working_dir()
def run(self, *cmd, **kwargs):
kwargs.update({
'cwd': self.working_dir,
'shell': False,
})
try:
build_cmd = self.environment.run(*cmd, **kwargs)
except BuildCancelled:
# Catch ``BuildCancelled`` here and re raise it. Otherwise, if we
# raise a ``RepositoryError`` then the ``on_failure`` method from
# Celery won't treat this problem as a ``BuildCancelled`` issue.
raise BuildCancelled
except BuildUserError as e:
# Re raise as RepositoryError to handle it properly from outside
if hasattr(e, "message"):
raise RepositoryError(e.message)
raise RepositoryError
# Return a tuple to keep compatibility
return (build_cmd.exit_code, build_cmd.output, build_cmd.error)
# =========================================================================
# Tag / Branch related methods
# These methods only apply if supports_tags = True and/or
# support_branches = True
# =========================================================================
@property
def tags(self):
"""
Returns a list of VCSVersion objects.
See VCSVersion for more information.
"""
raise NotImplementedError
@property
def branches(self):
"""
Returns a list of VCSVersion objects.
See VCSVersion for more information.
"""
raise NotImplementedError
@property
def commit(self):
"""Returns a string representing the current commit."""
raise NotImplementedError
def checkout(self, identifier=None):
"""
Set the state to the given identifier.
If identifier is None, checkout to the latest revision.
The type and format of identifier may change from VCS to VCS, so each
backend is responsible to understand it's identifiers.
"""
self.check_working_dir()
def update_submodules(self, config):
"""
Update the submodules of the current checkout.
:type config: readthedocs.config.BuildConfigBase
"""
raise NotImplementedError
def repo_exists(self):
raise NotImplementedError