-
Notifications
You must be signed in to change notification settings - Fork 440
/
parts.py
356 lines (305 loc) · 12.9 KB
/
parts.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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
#
# Copyright 2022-2024 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Craft-parts lifecycle wrapper."""
import pathlib
import subprocess
import types
from typing import Any, Dict, List, Optional, Set
import craft_parts
from craft_archives import repo
from craft_cli import emit
from craft_parts import Action, ActionType, Step
from craft_parts.packages import Repository
from xdg import BaseDirectory # type: ignore
from snapcraft import errors
from snapcraft.meta import ExtractedMetadata
from snapcraft.parts.extract_metadata import extract_lifecycle_metadata
from snapcraft.services.lifecycle import get_prime_dirs_from_project
from snapcraft.utils import convert_architecture_deb_to_platform, get_host_architecture
_LIFECYCLE_STEPS = {
"pull": Step.PULL,
"overlay": Step.OVERLAY,
"build": Step.BUILD,
"stage": Step.STAGE,
"prime": Step.PRIME,
}
class PartsLifecycle:
"""Create and manage the parts lifecycle.
:param all_parts: A dictionary containing the parts defined in the project.
:param work_dir: The working directory for parts processing.
:param assets_dir: The directory containing project assets.
:param adopt_info: The name of the part containing metadata do adopt.
:param extra_build_snaps: A list of additional build snaps to install.
:param partitions: A list of partitions for the project.
:raises PartsLifecycleError: On error initializing the parts lifecycle.
"""
# pylint: disable-next=too-many-locals
def __init__( # noqa PLR0913
self,
all_parts: Dict[str, Any],
*,
work_dir: pathlib.Path,
assets_dir: pathlib.Path,
base: str, # effective base
project_base: str,
confinement: str,
package_repositories: List[Dict[str, Any]],
parallel_build_count: int,
part_names: Optional[List[str]],
adopt_info: Optional[str],
parse_info: Dict[str, List[str]],
project_name: str,
project_vars: Dict[str, str],
extra_build_snaps: Optional[List[str]] = None,
target_arch: str,
track_stage_packages: bool,
partitions: Optional[List[str]],
):
self._work_dir = work_dir
self._assets_dir = assets_dir
self._package_repositories = package_repositories
self._part_names = part_names
self._adopt_info = adopt_info
self._parse_info = parse_info
self._all_part_names = [*all_parts]
self._partitions = partitions
emit.progress("Initializing parts lifecycle")
# set the cache dir for parts package management
cache_dir = BaseDirectory.save_cache_path("snapcraft")
if target_arch == "all":
target_arch = get_host_architecture()
platform_arch = convert_architecture_deb_to_platform(target_arch)
try:
self._lcm = craft_parts.LifecycleManager(
{"parts": all_parts},
application_name="snapcraft",
work_dir=work_dir,
cache_dir=cache_dir,
arch=platform_arch,
base=base,
ignore_local_sources=["*.snap"],
extra_build_snaps=extra_build_snaps,
track_stage_packages=track_stage_packages,
parallel_build_count=parallel_build_count,
project_name=project_name,
project_vars_part_name=adopt_info,
project_vars=project_vars,
# custom arguments
project_base=project_base,
confinement=confinement,
partitions=self._partitions,
)
except craft_parts.PartsError as err:
raise errors.PartsLifecycleError(str(err)) from err
def get_prime_dir(self, component: str | None = None) -> pathlib.Path:
"""Get the prime directory path for the default prime dir or a component.
:param component: Name of the component to get the prime directory for.
:returns: The default prime directory or a component's prime directory.
:raises SnapcraftError: If the component does not exist.
"""
try:
return self.prime_dirs[component]
except KeyError as err:
raise errors.SnapcraftError(
f"Could not get prime directory for component {component!r} "
"because it does not exist."
) from err
@property
def prime_dir(self) -> pathlib.Path:
"""Return the parts prime directory path."""
return self._lcm.project_info.prime_dir
@property
def prime_dirs(self) -> dict[str | None, pathlib.Path]:
"""Return a mapping of component names to prime directories.
'None' maps to the default prime directory.
"""
return get_prime_dirs_from_project(self._lcm.project_info)
@property
def target_arch(self) -> str:
"""Return the parts project target architecture."""
return self._lcm.project_info.target_arch
@property
def target_arch_triplet(self) -> str:
"""Return the parts project target architecture."""
return self._lcm.project_info.arch_triplet
@property
def project_vars(self) -> Dict[str, str]:
"""Return the value of project variable ``version``."""
return {
"version": self._lcm.project_info.get_project_var("version"),
"grade": self._lcm.project_info.get_project_var("grade"),
}
def run(
self,
step_name: str,
*,
shell: bool = False,
shell_after: bool = False,
rerun_step: bool = False,
) -> None:
"""Run the parts lifecycle.
:param target_step: The final step to execute.
:param shell: Enter a shell instead of running step_name.
:param shell_after: Enter a shell after running step_name.
:param rerun_step: Force running step_name.
:raises PartsLifecycleError: On error during lifecycle.
:raises RuntimeError: On unexpected error.
"""
target_step = _LIFECYCLE_STEPS.get(step_name)
if not target_step:
raise RuntimeError(f"Invalid target step {step_name!r}")
if shell:
# convert shell to shell_after for the previous step
previous_steps = target_step.previous_steps()
target_step = previous_steps[-1] if previous_steps else None
shell_after = True
try:
if target_step:
actions = self._lcm.plan(target_step, part_names=self._part_names)
else:
actions = []
self._install_package_repositories()
with self._lcm.action_executor() as aex:
for action in actions:
# Workaround until canonical/craft-parts#540 is fixed
if action.step == target_step and rerun_step:
action = craft_parts.Action( # noqa PLW2901
part_name=action.part_name,
step=action.step,
action_type=ActionType.RERUN,
reason="forced rerun",
project_vars=action.project_vars,
properties=action.properties,
)
message = _get_parts_action_message(action)
emit.progress(message)
with emit.open_stream() as stream:
aex.execute(action, stdout=stream, stderr=stream)
if shell_after:
launch_shell()
except RuntimeError as err:
raise RuntimeError(f"Parts processing internal error: {err}") from err
except OSError as err:
msg = err.strerror
if err.filename:
msg = f"{err.filename}: {msg}"
raise errors.PartsLifecycleError(msg) from err
except Exception as err:
raise errors.PartsLifecycleError(str(err)) from err
def _install_package_repositories(self) -> None:
if not self._package_repositories:
return
# Install pre-requisite packages for apt-key, if not installed.
required_packages = ["gnupg", "dirmngr"]
if any(p for p in required_packages if not Repository.is_package_installed(p)):
Repository.install_packages(required_packages, refresh_package_cache=True)
emit.progress("Installing package repositories...")
refresh_required = repo.install(
self._package_repositories, key_assets=self._assets_dir / "keys"
)
if refresh_required:
emit.progress("Refreshing package repositories...")
# TODO: craft-parts API for: force_refresh=refresh_required
# pylint: disable=C0415
from craft_parts.packages import deb
deb.Ubuntu.refresh_packages_list.cache_clear()
self._lcm.refresh_packages_list()
emit.progress("Installed package repositories", permanent=True)
def clean(self, *, part_names: Optional[List[str]] = None) -> None:
"""Remove lifecycle artifacts.
:param part_names: The names of the parts to clean. If not
specified, all parts will be cleaned.
"""
if part_names:
message = "Cleaning parts: " + ", ".join(part_names)
else:
message = "Cleaning all parts"
emit.progress(message)
self._lcm.clean(part_names=part_names)
def extract_metadata(self) -> List[ExtractedMetadata]:
"""Obtain metadata information."""
return extract_lifecycle_metadata(
self._adopt_info, self._parse_info, self._work_dir, self._partitions
)
def get_primed_stage_packages(self) -> List[str]:
"""Obtain the list of primed stage packages from all parts."""
primed_stage_packages: Set[str] = set()
for name in self._all_part_names:
stage_packages = self._lcm.get_primed_stage_packages(part_name=name)
if stage_packages:
primed_stage_packages |= set(stage_packages)
package_list = list(primed_stage_packages)
package_list.sort()
return package_list
def get_part_pull_assets(self, *, part_name: str) -> Optional[Dict[str, Any]]:
"""Obtain the pull state assets."""
return self._lcm.get_pull_assets(part_name=part_name)
def launch_shell(*, cwd: Optional[pathlib.Path] = None) -> None:
"""Launch a user shell for debugging environment.
:param cwd: Working directory to start user in.
"""
emit.progress("Launching shell on build environment...", permanent=True)
with emit.pause():
subprocess.run(["bash"], check=False, cwd=cwd)
ACTION_MESSAGES = types.MappingProxyType(
{
Step.PULL: types.MappingProxyType(
{
ActionType.RUN: "Pulling",
ActionType.RERUN: "Repulling",
ActionType.SKIP: "Skipping pull for",
ActionType.UPDATE: "Updating sources for",
}
),
Step.OVERLAY: types.MappingProxyType(
{
ActionType.RUN: "Overlaying",
ActionType.RERUN: "Re-overlaying",
ActionType.SKIP: "Skipping overlay for",
ActionType.UPDATE: "Updating overlay for",
ActionType.REAPPLY: "Reapplying",
}
),
Step.BUILD: types.MappingProxyType(
{
ActionType.RUN: "Building",
ActionType.RERUN: "Rebuilding",
ActionType.SKIP: "Skipping build for",
ActionType.UPDATE: "Updating build for",
}
),
Step.STAGE: types.MappingProxyType(
{
ActionType.RUN: "Staging",
ActionType.RERUN: "Restaging",
ActionType.SKIP: "Skipping stage for",
}
),
Step.PRIME: types.MappingProxyType(
{
ActionType.RUN: "Priming",
ActionType.RERUN: "Repriming",
ActionType.SKIP: "Skipping prime for",
}
),
}
)
def _get_parts_action_message(action: Action) -> str:
"""Get a user-readable message for a particular craft-parts action."""
message = f"{ACTION_MESSAGES[action.step][action.action_type]} {action.part_name}"
if action.reason:
return message + f" ({action.reason})"
return message