Coverage for r11k/puppetmodule/git.py: 86%
112 statements
« prev ^ index » next coverage.py v7.2.1, created at 2023-03-13 23:53 +0100
1"""Puppet modules backed by git repos."""
3from traceback_with_variables import activate_by_import # noqa: F401
5from datetime import datetime
6import json
7import logging
8import os.path
9import time
10import shutil
11import base64
13from typing import (
14 Any,
15 Optional,
16)
18# from git.repo.base import Repo
19# import git
20# import gitdb.exc
21import pygit2
22from pygit2 import (
23 Repository,
24 GIT_REPOSITORY_OPEN_BARE,
25 GIT_OBJ_BLOB,
26 GIT_OBJ_COMMIT,
27 GIT_OBJ_TREE,
28)
29from semver import VersionInfo
31from r11k.puppetmodule.base import PuppetModule
32from r11k.puppet import PuppetMetadata
33import r11k.config
34from r11k.gitutil import (
35 find_remote,
36 repo_last_fetch,
37 checkout,
38)
41logger = logging.getLogger(__name__)
44class GitPuppetModule(PuppetModule):
45 """
46 Puppet module backed by git repo.
48 Three different git repos are handled by this class:
49 - The upstream, referenced by the url self.git.
50 - The published module, which is given as the argument to publish()
51 - Our local cache of the upstream, contained in self.repo_path
52 (and self.__repo)
54 :param name: If the given module dosen't provide any metadata,
55 then this becomes the name of the published directory
56 for module (and must therefor match what the module
57 is written to be published as). If the module
58 contains metadata.json, the the name will be changed
59 to that, but a warning will be issued on mismatches.
60 :param git: URL to a git repo containing a Puppet module
61 :param version: Any valid git reference, such as a
62 tag,
63 branch,
64 commit,
65 etc...
66 """
68 def __init__(self,
69 name: str,
70 git: str,
71 *,
72 version: str,
73 config: r11k.config.Config = r11k.config.config):
74 super().__init__(name, version=version, config=config)
75 if version == 'HEAD': 75 ↛ 76line 75 didn't jump to line 76, because the condition on line 75 was never true
76 raise ValueError('HEAD is an invalid version refspec')
77 self.git = git
78 self.__repo: Optional[Repository] = None
80 @property
81 def repo_path(self) -> str:
82 """Return path of local cache of upstream."""
83 # using self._og_name would be nice, but will lead to
84 # problems if two repos have the same name (which is rather
85 # common, if different modules for the same functionallity is
86 # used in different part of the puppet deployment).
87 cache_name: str = base64.b64encode(self.git.encode('UTF-8'), b'-_') \
88 .decode('ASCII')
89 return os.path.join(self.config.clone_base, cache_name)
91 @property
92 def repo(self) -> Repository:
93 """
94 Return the coresponding git repo for this module, if available.
96 This will be the bare repo placed in the cache directory.
98 This method only works if the module have git as its origin.
99 Otherwise it will throw a value error.
100 """
101 if not self.__repo:
102 try:
103 self.__repo = pygit2.Repository(self.repo_path, GIT_REPOSITORY_OPEN_BARE)
104 if self.__repo.remotes['origin'].url != self.git:
105 logger.warning('Cached repo "%s" already exists, but points to "%s" rather than "%s". Replacing.' % (self.name, self.__repo.remotes['origin'].url, self.git)) # noqa: E501
106 shutil.rmtree(self.repo_path)
107 self.__repo = pygit2.clone_repository(self.git, self.repo_path, bare=True)
108 except pygit2.GitError:
109 self.__repo = pygit2.clone_repository(self.git, self.repo_path, bare=True)
111 if self.__repo.is_empty:
112 logger.warning('Empty repository cloned %s' % self.git)
114 # This fetches updates all remote branches, and then merges
115 # the local branches to match remote. Local branches is needed
116 # since those are used when this repo gets cloned.
117 # TODO what happens with diverged branches?
118 self.__repo.remotes['origin'].fetch(refspecs=['refs/heads/*:refs/heads/*'])
120 return self.__repo
122 def publish(self, path: str) -> None:
123 """Publish this module by cloning it to path."""
124 # This forces the download of upstream, which is needed
125 # to publish it.
126 assert self.repo
127 assert self.version
129 try:
130 repo = pygit2.Repository(path)
131 # TODO default origin name
132 if repo.remotes['origin'].url == self.repo_path:
133 repo.remotes['origin'].fetch(['+refs/heads/*:refs/heads/*'])
134 else:
135 logger.warning('Collision when publishing "%s", expected remote "%s", got %s". Replacing' % (self.name, self.repo_path, repo.remotes['origin'].url)) # noqa: E501
136 shutil.rmtree(path)
137 repo = pygit2.clone_repository(self.repo_path, path)
138 except pygit2.GitError:
139 # TODO Would replacing self.repo_path to self.repo.path
140 # allow us to remove the assert above?
141 repo = pygit2.clone_repository(self.repo_path, path)
143 # NOTE force without a warning isn't the best idea, but will
144 # have to do for now.
145 checkout(repo, self.version, force=True)
147 def versions(self) -> list[VersionInfo]:
148 """
149 Return all tags looking like release versions in the repo.
151 Note that any valid ref resolving to a commit is a valid
152 version, and can be specified in the puppetfile. (for example:
153 master, or a commit hash).
154 """
155 versions: list[VersionInfo] = []
156 pre = 'refs/tags/'
157 for tag in self.repo.references:
158 if not tag.startswith(pre):
159 continue
160 # if not tag.name:
161 # raise ValueError("Can't have empty tag")
162 # name: str = tag.name
163 name = tag[len(pre):]
164 if name[0] == 'v':
165 name = name[1:]
166 try:
167 versions.append(VersionInfo.parse(name))
168 except ValueError:
169 pass
171 return sorted(versions)
173 def _fetch_metadata(self) -> PuppetMetadata:
174 """Fetch metadata for this module."""
175 """Clone module form git url."""
176 name = self.name
177 version = self.version or self.latest()
178 git_url = self.git
180 repo = self.repo
182 remote = find_remote(repo.remotes, git_url)
183 if not remote: 183 ↛ 184line 183 didn't jump to line 184, because the condition on line 183 was never true
184 raise ValueError('Existing repo, but with different remote')
186 last_modified = repo_last_fetch(repo)
187 now = datetime.utcfromtimestamp(time.time())
189 if abs(now - last_modified).seconds > self.config.git_ttl: 189 ↛ 190line 189 didn't jump to line 190, because the condition on line 189 was never true
190 remote.fetch(['+refs/heads/*:refs/heads/*'])
192 # tree should have the type Optional[git.Tree], but the mypy
193 # version available at Arch Linux can't resolve it.
194 # arch (2022-10-03) $ mypy --version
195 # mypy 0.981 (compiled: no)
196 # venv $ mypy --version
197 # mypy 0.971 (compiled: yes)
198 tree: Optional[Any] = None
200 # NOTE this this always prepends/removes a 'v' from version,
201 # which will get weird with non-semver versions, e.g. HEAD
202 # becomes [HEAD, vHEAD].
203 for alt in tag_alternatives(version):
204 try:
205 tree = repo.revparse(alt).from_object.peel(GIT_OBJ_TREE)
206 if tree.type != GIT_OBJ_TREE: 206 ↛ 207line 206 didn't jump to line 207, because the condition on line 206 was never true
207 raise RuntimeError(f"Reference '{alt}' doesn't reference a commit in {git_url}")
208 except KeyError:
209 pass
211 # An empty tree is falsy. Only check for the non-existance here
212 if tree is None: 212 ↛ 213line 212 didn't jump to line 213, because the condition on line 212 was never true
213 raise RuntimeError(f'No tree for version {version}')
215 try:
216 # blob = tree.join('metadata.json')
217 blob = tree['metadata.json']
218 if blob.type != GIT_OBJ_BLOB: 218 ↛ 219line 218 didn't jump to line 219, because the condition on line 218 was never true
219 raise RuntimeError('metadata.json not a blob')
220 data = blob.data
221 # checksum, type, len, file = blob.data_stream
222 # data = file.read()
223 return PuppetMetadata(**json.loads(data))
224 except KeyError:
225 logger.info('No metadata for %s, generating default', name)
226 return PuppetMetadata(name=name,
227 version=version,
228 author='UNKNOWN',
229 license='UNKNOWN',
230 summary='-',
231 source=git_url,
232 dependencies=[])
234 def serialize(self) -> dict[str, Any]:
235 """
236 Serialize back into form from puppetfile.yaml.
238 The git version also includes the git url.
239 """
240 return {**super().serialize(),
241 'version': self.repo.revparse(self.version).from_object.peel(GIT_OBJ_COMMIT).hex,
242 'git': self.git}
244 def __repr__(self) -> str:
245 return f"GitPuppetModule('{self.name}', '{self.version}')"
248def tag_alternatives(name: str) -> list[str]:
249 """
250 Return the given tag name with and without leading 'v'.
252 Assumes that tag is either on the from
253 `v1.0.0` or `1.0.0`.
254 :return: `['1.0.0', 'v1.0.0']`
255 (non-prefixed version always being first).
256 """
257 if not name: 257 ↛ 258line 257 didn't jump to line 258, because the condition on line 257 was never true
258 raise ValueError("Can't have empty tag string")
259 if name[0] == 'v':
260 return [name[1:], name]
261 else:
262 return [name, f'v{name}']