Source code for backend.job

import copy
import os


[docs]class BuildJob(object): def __init__(self, task_data, worker_opts): """ Creates build job object :param dict task_dict: dictionary with the following fields (based frontend.models.Build):: - pkgs: list of space separated urls of packages to build - timeout: maximum allowed time of build, build will fail if exceeded # unused - project_owner: - project_name: - project_dirname: - submitter: :param dict worker_opts: worker options, fields:: - destdir: worker root directory to store results - results_baseurl: root url to stored results - timeout: default worker timeout """ self.timeout = worker_opts.timeout self.memory_reqs = None self.enable_net = True self.project_owner = None self.project_name = None self.project_dirname = None self.submitter = None self.ended_on = None self.started_on = None self.submitted_on = None self.status = None self.chroot = None self.arch = None # parsed from chroot self.buildroot_pkgs = None self.task_id = None self.build_id = None self.package_name = None self.package_version = None self.git_repo = None self.git_hash = None self.git_branch = None self.source_type = None self.source_json = None self.pkg_main_version = None self.pkg_epoch = None self.pkg_release = None self.srpm_url = None self.uses_devel_repo = None # TODO: validate update data, user marshmallow for key, val in task_data.items(): key = str(key) setattr(self, key, val) if self.chroot: self.arch = self.chroot.rsplit("-", 2)[2] if str(self.task_id) == str(self.build_id): self.chroot = 'srpm-builds' self.destdir = os.path.normpath(os.path.join( worker_opts.destdir, task_data["project_owner"], task_data["project_dirname"] )) self.results_repo_url = os.path.join( worker_opts.results_baseurl, task_data["project_owner"], task_data["project_dirname"], ) self.result_dir = self.target_dir_name self.built_packages = "" @property def results_dir_url(self): return os.path.join(self.results_repo_url, self.chroot, self.target_dir_name) @property def chroot_dir(self): return os.path.normpath("{}/{}".format(self.destdir, self.chroot)) @property def results_dir(self): return os.path.join(self.chroot_dir, self.target_dir_name) @property def target_dir_name(self): if not self.package_name: return "{:08d}".format(self.build_id) return "{:08d}-{}".format(self.build_id, self.package_name) @property def chroot_log_name(self): return "build-{:08d}.log".format(self.build_id) @property def chroot_log_path(self): return os.path.join(self.chroot_dir, self.chroot_log_name) @property def rsync_log_name(self): return "build-{:08d}.rsync.log".format(self.build_id)
[docs] def update(self, data_dict): """ :param dict data_dict: """ # TODO: validate update data self.__dict__.update(data_dict)
[docs] def to_dict(self): """ :return dict: dictified build job """ result = copy.deepcopy(self.__dict__) result["id"] = self.build_id result["mockchain_macros"] = self.mockchain_macros return result
@property def mockchain_macros(self): return { "copr_username": self.project_owner, "copr_projectname": self.project_name, "vendor": "Fedora Project COPR ({0}/{1})".format( self.project_owner, self.project_name) } @property def pkg_version(self): """ Canonical version presentation release and epoch "{epoch}:{version}-{release}" """ if self.pkg_main_version is None: return None if self.pkg_epoch: full_version = "{}:{}".format(self.pkg_epoch, self.pkg_main_version) else: full_version = "{}".format(self.pkg_main_version) if self.pkg_release: full_version += "-{}".format(self.pkg_release) return full_version def __str__(self): return str(self.__unicode__()) def __unicode__(self): return u"BuildJob<id: {build_id}, owner: {project_owner}, project: {project_name}, project_dir: {project_dirname}" \ u"git branch: {git_branch}, git hash: {git_hash}, status: {status} >".format(**self.__dict__)