Source code for debusine.tasks.sbuild

# 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
[docs] def configure(self, task_data): """Handle sbuild-specific configuration.""" super().configure(task_data) # Handle default values self.data.setdefault("build_components", ["any"]) self.data.setdefault("sbuild_options", [])
[docs] def fetch_input(self, destination: Path) -> bool: """Download the source artifact.""" artifact_id = self.data["input"]["source_artifact_id"] self.fetch_artifact(artifact_id, destination) return True
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 configure_for_execution(self, download_directory: Path) -> bool: """ Configure Task: set variables needed for the build() step. Return True if configuration worked, False, if there was a problem. Note: self.find_file_by_suffix() write to a log file to be uploaded as artifact. """ self._dsc_file = self.find_file_by_suffix(download_directory, ".dsc") return self._dsc_file is not None
[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, )