/usr/lib/python2.7/dist-packages/jenkinsapi/artifact.py is in python-jenkinsapi 0.2.16-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """
Artifacts can be used to represent data created as a side-effect of running a Jenkins build.
Artifacts are files which are associated with a single build. A build can have any number of
artifacts associated with it.
This module provides a class called Artifact which allows you to download objects from the server
and also access them as a stream.
"""
import os
import logging
import hashlib
from jenkinsapi.fingerprint import Fingerprint
from jenkinsapi.custom_exceptions import ArtifactBroken
log = logging.getLogger(__name__)
class Artifact(object):
"""
Represents a single Jenkins artifact, usually some kind of file
generated as a by-product of executing a Jenkins build.
"""
def __init__(self, filename, url, build):
self.filename = filename
self.url = url
self.build = build
def save(self, fspath):
"""
Save the artifact to an explicit path. The containing directory must exist.
Returns a reference to the file which has just been writen to.
:param fspath: full pathname including the filename, str
:return: filepath
"""
log.info(msg="Saving artifact @ %s to %s" % (self.url, fspath))
if not fspath.endswith(self.filename):
log.warn(msg="Attempt to change the filename of artifact %s on save." % self.filename)
if os.path.exists(fspath):
if self.build:
try:
if self._verify_download(fspath):
log.info(msg="Local copy of %s is already up to date." % self.filename)
return fspath
except ArtifactBroken:
log.info("Jenkins artifact could not be identified.")
else:
log.info("This file did not originate from Jenkins, so cannot check.")
else:
log.info("Local file is missing, downloading new.")
filepath = self._do_download(fspath)
try:
self._verify_download(filepath)
except ArtifactBroken:
log.warning("fingerprint of the downloaded artifact could not be verified")
return fspath
def get_jenkins_obj(self):
return self.build.get_jenkins_obj()
def get_data(self):
"""
Grab the text of the artifact
"""
response = self.get_jenkins_obj().requester.get_and_confirm_status(self.url)
return response.content
def _do_download(self, fspath):
"""
Download the the artifact to a path.
"""
with open(fspath, "wb") as out:
out.write(self.get_data())
return fspath
def _verify_download(self, fspath):
"""
Verify that a downloaded object has a valid fingerprint.
"""
local_md5 = self._md5sum(fspath)
fp = Fingerprint(self.build.job.jenkins.baseurl, local_md5, self.build.job.jenkins)
return fp.validate_for_build(os.path.basename(fspath), self.build.job.name, self.build.buildno)
def _md5sum(self, fspath, chunksize=2 ** 20):
"""
A MD5 hashing function intended to produce the same results as that used by
Jenkins.
"""
md5 = hashlib.md5()
try:
with open(fspath, 'rb') as f:
for chunk in iter(lambda: f.read(chunksize), ''):
md5.update(chunk)
except:
raise
return md5.hexdigest()
def save_to_dir(self, dirpath):
"""
Save the artifact to a folder. The containing directory must be exist, but use the artifact's
default filename.
"""
assert os.path.exists(dirpath)
assert os.path.isdir(dirpath)
outputfilepath = os.path.join(dirpath, self.filename)
return self.save(outputfilepath)
def __repr__(self):
"""
Produce a handy repr-string.
"""
return """<%s.%s %s>""" % (self.__class__.__module__, self.__class__.__name__, self.url)
|