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 - repos: list of space separated additional repos - timeout: maximum allowed time of build, build will fail if exceeded # unused - project_owner: - project_name: - 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.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.pkg_main_version = None self.pkg_epoch = None self.pkg_release = None # TODO: validate update data, user marshmallow for key, val in task_data.items(): key = str(key) setattr(self, key, val) self.arch = self.chroot.split("-")[2] self.repos = [r for r in task_data["repos"].split(" ") if r.strip()] self.destdir = os.path.normpath(os.path.join( worker_opts.destdir, task_data["project_owner"], task_data["project_name"] )) self.results = u"/".join([ worker_opts.results_baseurl, task_data["project_owner"], task_data["project_name"] ]) self.results_repo_url = self.results self.built_packages = "" @property def chroot_repos_extended(self): repos = list(self.repos) repos.append("{}/{}".format(self.results_repo_url, self.chroot)) repos.append("{}/{}/devel".format(self.results_repo_url, self.chroot)) return repos @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): 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}, " \ u"git branch: {git_branch}, git_hash: {git_hash}, status: {status} >".format(**self.__dict__)