Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def fetch(index):
if func(index):
raise BuildInfoNotFound("")
return index
def test_get_valid_build_got_exception(self):
data = self.create_inbound_build_data('c0', 'c3')
data.info_fetcher.find_build_info = \
Mock(side_effect=errors.BuildInfoNotFound)
data.mid_point()
self.assertEqual(len(data), 0)
return {
"status": {
"runs": [
{"state": "completed", "runId": 0, "resolved": "2015-06-01T22:13:02.115Z"}
]
}
}
def list_artifacts(taskid, run_id):
return {"artifacts": []}
Index.findTask = find_task
Queue.status = status
Queue.listArtifacts = list_artifacts
with self.assertRaises(errors.BuildInfoNotFound):
self.info_fetcher.find_build_info(create_push("123456789", 1))
def test_access(range_creator):
build_range = range_creator.create(list(range(5)))
assert build_range[0] == 0
build_range.build_info_fetcher.find_build_info.side_effect = BuildInfoNotFound
assert build_range[1] is False
# even if one build is invalid, item access do not modify the range
assert len(build_range) == 5
# but the following will remove the invalid entry
build_range.filter_invalid_builds()
assert len(build_range) == 4
def _get_valid_build_for_date(self, date):
try:
return self.info_fetcher.find_build_info(date)
except errors.BuildInfoNotFound:
return False
if run_id is None:
raise BuildInfoNotFound("Unable to find completed runs for task %s" % task_id)
artifacts = tc_queue.listArtifacts(task_id, run_id)["artifacts"]
# look over the artifacts of that run
build_url = None
for a in artifacts:
name = os.path.basename(a["name"])
if self.build_regex.search(name):
meth = tc_queue.buildUrl
if self.fetch_config.tk_needs_auth():
meth = tc_queue.buildSignedUrl
build_url = meth("getArtifact", task_id, run_id, a["name"])
break
if build_url is None:
raise BuildInfoNotFound(
"unable to find a build url for the" " changeset %r" % changeset
)
return IntegrationBuildInfo(
self.fetch_config,
build_url=build_url,
build_date=build_date,
changeset=changeset,
repo_url=self.jpushes.repo_url,
task_id=task_id,
)
def build_info(self):
if self._build_info is None:
try:
self._build_info = self._fetch()
except BuildInfoNotFound as exc:
LOG.warning("Skipping build %s: %s" % (self.data, exc))
self._build_info = False
return self._build_info
data = {}
if not url.endswith("/"):
url += "/"
links = url_links(url)
if not self.fetch_config.has_build_info:
links += url_links(self.fetch_config.get_nightly_info_url(url))
for link in links:
if "build_url" not in data and self.build_regex.search(link):
data["build_url"] = link
elif "build_txt_url" not in data and self.build_info_regex.search(link):
data["build_txt_url"] = link
if data:
# Check that we found all required data. The URL in build_url is
# required. build_txt_url is optional.
if "build_url" not in data:
raise BuildInfoNotFound(
"Failed to find a build file in directory {} that "
"matches regex '{}'".format(url, self.build_regex.pattern)
)
with self._fetch_lock:
lst.append((index, data))