/usr/lib/python2.7/dist-packages/jenkinsapi/artifact.py is in python-jenkinsapi 0.2.30-1.
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """
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, strict_validation=False):
"""
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, strict_validation):
log.info(
msg="Local copy of %s is already up to date." %
self.filename)
return fspath
except ArtifactBroken:
log.warning("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)
self._verify_download(filepath, strict_validation)
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, strict_validation):
"""
Verify that a downloaded object has a valid fingerprint.
"""
local_md5 = self._md5sum(fspath)
baseurl = self.build.job.jenkins.baseurl
fp = Fingerprint(
baseurl,
local_md5,
self.build.job.jenkins)
valid = fp.validate_for_build(
os.path.basename(fspath), self.build.job.name, self.build.buildno)
if not valid or (fp.unknown and strict_validation): # strict = 404 as invalid
raise ArtifactBroken("Artifact %s seems to be broken, check %s" % (local_md5, baseurl))
return True
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), ''):
if chunk:
md5.update(chunk)
else:
break
except:
raise
return md5.hexdigest()
def save_to_dir(self, dirpath, strict_validation=False):
"""
Save the artifact to a folder. The containing directory must 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, strict_validation)
def __repr__(self):
"""
Produce a handy repr-string.
"""
return """<%s.%s %s>""" % (self.__class__.__module__,
self.__class__.__name__, self.url)
|