# Copyright 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 use reprotest in debusine."""
from pathlib import Path
from typing import Any
from debusine import utils
from debusine.artifacts.local_artifact import ReprotestArtifact
from debusine.artifacts.models import (
ArtifactCategory,
CollectionCategory,
DebianSourcePackage,
DebianUpload,
WorkRequestResults,
get_source_package_name,
get_source_package_version,
)
from debusine.client.models import RelationType
from debusine.tasks import BaseTaskWithExecutor, RunCommandTask
from debusine.tasks.models import ReprotestData, ReprotestDynamicData
from debusine.tasks.server import TaskDatabaseInterface
class Reprotest(
RunCommandTask[ReprotestData, ReprotestDynamicData],
BaseTaskWithExecutor[ReprotestData, ReprotestDynamicData],
):
"""Task to use reprotest in debusine."""
TASK_VERSION = 1
CAPTURE_OUTPUT_FILENAME = "reprotest.log"
def __init__(
self,
task_data: dict[str, Any],
dynamic_task_data: dict[str, Any] None = None,
) -> None:
"""Initialize object."""
super().__init__(task_data, dynamic_task_data)
self._reprotest_target: Path None = None
def build_dynamic_data(
self, task_database: TaskDatabaseInterface
) -> ReprotestDynamicData:
"""Compute and return ReprotestDynamicData."""
input_source_artifact = task_database.lookup_single_artifact(
self.data.source_artifact
)
assert input_source_artifact is not None
self.ensure_artifact_categories(
configuration_key="input.source_artifact",
category=input_source_artifact.category,
expected=(
ArtifactCategory.SOURCE_PACKAGE,
ArtifactCategory.UPLOAD,
),
)
assert isinstance(
input_source_artifact.data, (DebianSourcePackage, DebianUpload)
)
subject = get_source_package_name(input_source_artifact.data)
version = get_source_package_version(input_source_artifact.data)
assert self.data.environment is not None
environment = self.get_environment(
task_database,
self.data.environment,
default_category=CollectionCategory.ENVIRONMENTS,
)
return ReprotestDynamicData(
source_artifact_id=input_source_artifact.id,
subject=subject,
parameter_summary=f" subject _ version ",
environment_id=environment.id,
)
def get_input_artifacts_ids(self) -> list[int]:
"""Return the list of input artifact IDs used by this task."""
if not self.dynamic_data:
return []
return [
self.dynamic_data.source_artifact_id,
self.dynamic_data.environment_id,
]
def fetch_input(self, destination: Path) -> bool:
"""Download the required artifacts."""
assert self.dynamic_data
artifact_id = self.dynamic_data.source_artifact_id
assert artifact_id is not None
self.fetch_artifact(artifact_id, destination)
return True
def configure_for_execution(self, download_directory: Path) -> bool:
"""
Find a .dsc in download_directory.
Install reprotest and other utilities used in _cmdline.
Set self._reprotest_target to it.
:param download_directory: where to search the files
:return: True if valid files were found
"""
self._prepare_executor_instance()
if self.executor_instance is None:
raise AssertionError("self.executor_instance cannot be None")
self.run_executor_command(
["apt-get", "update"],
log_filename="install.log",
run_as_root=True,
check=True,
)
self.run_executor_command(
[
"apt-get",
"--yes",
"--no-install-recommends",
"install",
"reprotest",
"dpkg-dev",
"devscripts",
"equivs",
"sudo",
],
log_filename="install.log",
run_as_root=True,
)
self._reprotest_target = utils.find_file_suffixes(
download_directory, [".dsc"]
)
return True
def _cmdline(self) -> list[str]:
"""
Build the reprotest command line.
Use configuration of self.data and self._reprotest_target.
"""
target = self._reprotest_target
assert target is not None
cmd = [
"bash",
"-c",
f"TMPDIR=/tmp ; cd /tmp ; dpkg-source -x target package/; "
"cd package/ ; mk-build-deps ; apt-get install --yes ./*.deb ; "
"rm *.deb ; "
"reprotest --vary=-time,-user_group,-fileordering,-domain_host .",
]
return cmd
@staticmethod
def _cmdline_as_root() -> bool:
r"""apt-get install --yes ./\*.deb must be run as root."""
return True
def task_result(
self,
returncode: int None,
execute_directory: Path, # noqa: U100
) -> WorkRequestResults:
"""
Evaluate task output and return success.
For a successful run of reprotest:
-must have the output file
-exit code is 0
:return: WorkRequestResults.SUCCESS or WorkRequestResults.FAILURE.
"""
reprotest_file = execute_directory / self.CAPTURE_OUTPUT_FILENAME
if reprotest_file.exists() and returncode == 0:
return WorkRequestResults.SUCCESS
return WorkRequestResults.FAILURE
def upload_artifacts(
self, exec_directory: Path, *, execution_result: WorkRequestResults
) -> None:
"""Upload the ReprotestArtifact with the files and relationships."""
if not self.debusine:
raise AssertionError("self.debusine not set")
assert self.dynamic_data is not None
assert self.dynamic_data.parameter_summary is not None
reprotest_artifact = ReprotestArtifact.create(
reprotest_output=exec_directory / self.CAPTURE_OUTPUT_FILENAME,
reproducible=execution_result == WorkRequestResults.SUCCESS,
package=self.dynamic_data.parameter_summary,
)
uploaded = self.debusine.upload_artifact(
reprotest_artifact,
workspace=self.workspace_name,
work_request=self.work_request_id,
)
assert self.dynamic_data is not None
assert self.dynamic_data.source_artifact_id is not None
self.debusine.relation_create(
uploaded.id,
self.dynamic_data.source_artifact_id,
RelationType.RELATES_TO,
)