# Copyright 2021 The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.
"""
Task to build Debian packages with sbuild.
This task implements the PackageBuild generic task for its task_data:
https://freexian-team.pages.debian.net/debusine/reference/tasks.html#task-packagebuild
"""
import subprocess
from pathlib import Path
from typing import Optional
import debian.deb822 as deb822
import debusine.utils
from debusine.artifacts import (
BinaryPackages,
PackageBuildLog,
Upload,
)
from debusine.client.models import RemoteArtifact
from debusine.tasks import Task, TaskConfigError
from debusine.tasks._task_mixins import (
FetchExecUploadMixin,
TaskRunCommandMixin,
)
from debusine.tasks.sbuild_validator_mixin import SbuildValidatorMixin
from debusine.utils import read_dsc
[docs]class Sbuild(
SbuildValidatorMixin, TaskRunCommandMixin, FetchExecUploadMixin, Task
):
"""Task implementing a Debian package build with sbuild."""
TASK_VERSION = 1
TASK_DATA_SCHEMA = {
"type": "object",
"properties": {
"input": {
"type": "object",
"properties": {
"source_artifact_id": {
"type": "integer",
},
},
"required": ["source_artifact_id"],
"additionalProperties": False,
},
"distribution": {
"type": "string",
},
"host_architecture": {
"type": "string",
},
"build_components": {
"type": "array",
"items": {
"enum": ["any", "all", "source"],
},
"uniqueItems": True,
},
"sbuild_options": {
"type": "array",
"items": {"type": "string"},
},
},
"required": ["input", "distribution", "host_architecture"],
"additionalProperties": False,
}
[docs] def __init__(self) -> None:
"""Initialize the sbuild task."""
super().__init__()
self.chroots = None
self.builder = "sbuild"
# dsc_file Path. Set by self.configure_for_execution()
self._dsc_file: Optional[Path] = None
@property
def chroot_name(self) -> str:
"""Build name of required chroot."""
return "%s-%s" % (
self.data["distribution"],
self.data["host_architecture"],
)
@staticmethod
def _call_dpkg_architecture(): # pragma: no cover
return (
subprocess.check_output(["dpkg", "--print-architecture"])
.decode("utf-8")
.strip()
)
[docs] def analyze_worker(self):
"""Report metadata for this task on this worker."""
metadata = super().analyze_worker()
available_key = self.prefix_with_task_name("available")
metadata[available_key] = debusine.utils.is_command_available("sbuild")
if debusine.utils.is_command_available("schroot"):
self._update_chroots_list()
chroots_key = self.prefix_with_task_name("chroots")
metadata[chroots_key] = self.chroots.copy()
host_arch_key = self.prefix_with_task_name("host_architecture")
metadata[host_arch_key] = self._call_dpkg_architecture()
return metadata
[docs] def can_run_on(self, worker_metadata: dict) -> bool:
"""Check the specified worker can run the requested task."""
if not super().can_run_on(worker_metadata):
return False
available_key = self.prefix_with_task_name("available")
if not worker_metadata.get(available_key, False):
return False
chroot_key = self.prefix_with_task_name("chroots")
if self.chroot_name not in worker_metadata.get(chroot_key, []):
return False
return True
@staticmethod
def _call_schroot_list(): # pragma: no cover
return (
subprocess.check_output(["schroot", "--list"])
.decode("utf-8")
.strip()
)
def _update_chroots_list(self):
"""
Provide support for finding available chroots.
Ensure that aliases are supported as the DSC may explicitly refer to
<codename>-security (or -backports) etc.
Populates the self.chroots list, if the list is empty.
No return value, this is a find, not a get.
"""
if self.chroots is not None:
return
self.chroots = []
output = self._call_schroot_list()
for line in output.split("\n"):
if line.startswith("chroot:") and line.endswith("-sbuild"):
self.chroots.append(line[7:-7])
def _verify_distribution(self):
"""Verify a suitable schroot exists."""
self._update_chroots_list()
if not self.chroots:
self.logger.error("No sbuild chroots found")
return False
if self.chroot_name in self.chroots:
return True
self.logger.error("No suitable chroot found for %s", self.chroot_name)
return False
def _cmdline(self) -> list[str]:
"""
Build the sbuild command line.
Use self.data and self._dsc_file.
"""
cmd = [
self.builder,
"--no-clean",
]
if "any" in self.data["build_components"]:
cmd.append("--arch-any")
else:
cmd.append("--no-arch-any")
if "all" in self.data["build_components"]:
cmd.append("--arch-all")
else:
cmd.append("--no-arch-all")
if "source" in self.data["build_components"]:
cmd.append("--source")
else:
cmd.append("--no-source")
cmd.append("--dist=" + self.data["distribution"])
cmd.append("--arch=" + self.data["host_architecture"])
cmd.extend(self.data["sbuild_options"])
cmd.append(str(self._dsc_file))
return cmd
[docs] def execute(self) -> bool:
"""
Verify task can be executed and super().execute().
:raises: TaskConfigError.
""" # noqa: D402
if not self._verify_distribution():
raise TaskConfigError(
"No suitable schroot for %s-%s"
% (self.data["distribution"], self.data["host_architecture"])
)
return super().execute()
def _upload_package_build_log(
self, build_directory: Path, source: str, version: str
) -> RemoteArtifact:
if not self.debusine:
raise AssertionError("self.debusine not set")
package_build_log = PackageBuildLog.create(
source=source,
version=version,
file=debusine.utils.find_file_suffixes(build_directory, [".build"]),
)
return self.debusine.upload_artifact(
package_build_log,
workspace=self.workspace,
work_request=self.work_request,
)
def _upload_binary_upload(self, build_directory: Path) -> RemoteArtifact:
if not self.debusine:
raise AssertionError("self.debusine not set")
changes_path = debusine.utils.find_file_suffixes(
build_directory, [".changes"]
)
artifact_binary_upload = Upload.create(
changes_file=changes_path,
)
return self.debusine.upload_artifact(
artifact_binary_upload,
workspace=self.workspace,
work_request=self.work_request,
)
def _create_binary_package_local_artifact(
self, build_directory: Path, dsc: deb822.Dsc, suffixes: list[str]
) -> BinaryPackages:
deb_paths = debusine.utils.find_files_suffixes(
build_directory, suffixes
)
return BinaryPackages.create(
srcpkg_name=dsc["source"],
srcpkg_version=dsc["version"],
version=dsc["version"],
architecture=dsc["architecture"],
files=deb_paths,
packages=[],
)
def _upload_binary_packages(
self, build_directory: Path, dsc: deb822.Dsc
) -> list[RemoteArtifact]:
r"""Upload \*.deb and \*.udeb files."""
if not self.debusine:
raise AssertionError("self.debusine not set")
host_arch = self.data["host_architecture"]
packages = []
if "any" in self.data["build_components"]:
prefix = "_" + host_arch
packages.append(
self._create_binary_package_local_artifact(
build_directory, dsc, [prefix + ".deb", prefix + ".udeb"]
)
)
if "all" in self.data["build_components"]:
prefix = "_all"
packages.append(
self._create_binary_package_local_artifact(
build_directory, dsc, [prefix + ".deb", prefix + ".udeb"]
)
)
remote_artifacts: list[RemoteArtifact] = []
for package in packages:
if package.files:
remote_artifacts.append(
self.debusine.upload_artifact(
package,
workspace=self.workspace,
work_request=self.work_request,
)
)
return remote_artifacts
def _create_remote_binary_packages_relations(
self,
remote_build_log: RemoteArtifact,
remote_binary_upload: RemoteArtifact,
remote_binary_packages: list[RemoteArtifact],
):
if not self.debusine:
raise AssertionError("self.debusine not set")
for remote_binary_package in remote_binary_packages:
for source_artifact_id in self._source_artifacts_ids:
self.debusine.relation_create(
remote_binary_package.id,
source_artifact_id,
"built-using",
)
self.debusine.relation_create(
remote_build_log.id, remote_binary_package.id, "relates-to"
)
self.debusine.relation_create(
remote_binary_upload.id,
remote_binary_package.id,
"extends",
)
self.debusine.relation_create(
remote_binary_upload.id,
remote_binary_package.id,
"relates-to",
)
[docs] def upload_artifacts(self, directory: Path, *, execution_success: bool):
"""
Upload the artifacts from directory.
:param directory: directory containing the files that
will be uploaded.
:param execution_success: if False skip uploading .changes and
*.deb/*.udeb
"""
if not self.debusine:
raise AssertionError("self.debusine not set")
dsc = read_dsc(self._dsc_file)
if dsc is not None:
# Upload the .build file (PackageBuildLog)
remote_build_log = self._upload_package_build_log(
directory, dsc["source"], dsc["version"]
)
for source_artifact_id in self._source_artifacts_ids:
self.debusine.relation_create(
remote_build_log.id,
source_artifact_id,
"relates-to",
)
if execution_success:
# Upload the *.deb/*.udeb files (BinaryPackages)
remote_binary_packages = self._upload_binary_packages(
directory, dsc
)
# Upload the .changes and the rest of the files
remote_binary_changes = self._upload_binary_upload(directory)
# Create the relations
self._create_remote_binary_packages_relations(
remote_build_log,
remote_binary_changes,
remote_binary_packages,
)