diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 57ba449c..14e7ee03 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -5,9 +5,17 @@ name: CI_TEST on: push: + paths: + - 'requirements.txt' + - '**.py' + - '**.yml' pull_request: branches: [ "master" ] +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: build: @@ -15,11 +23,34 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.8", "3.9", "3.10"] + python-version: ["3.8", "3.9", "3.10", "3.11"] token: ["stable", "latest"] steps: - - uses: actions/checkout@v3 + - name: Harden Runner + uses: step-security/harden-runner@1f99358870fe1c846a3ccba386cc2b2246836776 # v2.2.1 + with: + egress-policy: block + allowed-endpoints: > + azure.archive.ubuntu.com:80 + esm.ubuntu.com:443 + files.pythonhosted.org:443 + ftp-chi.osuosl.org:443 + ftp-nyc.osuosl.org:443 + get.jenkins.io:443 + github.com:443 + mirror.xmission.com:443 + motd.ubuntu.com:443 + packages.microsoft.com:443 + ppa.launchpadcontent.net:443 + pypi.org:443 + updates.jenkins-ci.org:80 + updates.jenkins.io:443 + updates.jenkins.io:80 + + - name: Checkout + uses: actions/checkout@v3 + - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v3 with: @@ -51,9 +82,8 @@ jobs: - name: Lint with flake8 run: | - # stop the build if there are Python syntax errors or undefined names - flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics - flake8 . --count --exit-zero --max-complexity=10 --max-line-length=79 --statistics + flake8 jenkinsapi/ --count --select=E9,F63,F7,F82 --ignore F821,W503,W504 --show-source --statistics + flake8 jenkinsapi/ --count --exit-zero --max-complexity=10 --max-line-length=79 --statistics - name: Test with pytest env: diff --git a/.gitignore b/.gitignore index 649c5973..0033879a 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,4 @@ coverage.xml *.war venv/ tags +.pytype/ diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 67b26723..be18a7fa 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -32,7 +32,3 @@ repos: - id: debug-statements - id: check-yaml files: .*\.(yaml|yml)$ - - repo: https://github.com/PyCQA/flake8 - rev: 5.0.4 - hooks: - - id: flake8 diff --git a/LICENSE b/LICENSE new file mode 100644 index 00000000..a5e18342 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2023 PyContribs + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.rst b/README.rst index 2ca660ce..b6386ce8 100644 --- a/README.rst +++ b/README.rst @@ -46,11 +46,8 @@ Python versions The project has been tested against Python versions: -* 2.7 -* 3.4 -* 3.5 -* 3.6 -* 3.7 +* 2.7 - last version compatible with Python 2.7 is tagged Py2 in repository and available on PyPi as version 0.3.13 +* 3.8 - 3.11 Jenkins versions ---------------- @@ -146,25 +143,26 @@ missing test dependencies: virtualenv source .venv/bin/active - (venv) python setup.py test + (.venv) pip install -r requirements.txt + (.venv) python setup.py test Development ----------- -* Make sure that you have Java_ installed. +* Make sure that you have Java_ installed. Jenkins will be automatically + downloaded and started during tests. * Create virtual environment for development * Install package in development mode .. code-block:: bash - (venv) pip install -e . - (venv) pip install -r test-requirements.txt + (.venv) pip install -r test-requirements.txt * Make your changes, write tests and check your code .. code-block:: bash - (venv) tox + (.venv) pytest Project Contributors diff --git a/doc/source/rules_for_contributors.rst b/doc/CONTRIBUTING.md similarity index 67% rename from doc/source/rules_for_contributors.rst rename to doc/CONTRIBUTING.md index 33edcb64..19304239 100644 --- a/doc/source/rules_for_contributors.rst +++ b/doc/CONTRIBUTING.md @@ -6,14 +6,17 @@ The JenkinsAPI project welcomes contributions via GitHub. Please bear in mind th Python compatibility -------------------- -The project currently targets Python 2.6 and Python 2.7. Support for Python 3.x will be introduced soon. Please do not add any features which -will break our supported Python 2.x versions or make it harder for us to migrate to Python 3.x +The project currently targets Python 3.8+. Last version compatible with Python 2.7 is tagged as Py2. + +Code formatting +--------------- + +The project follows strict PEP8 guidelines. Please use a tool like black to format your code before submitting a pull request. Tell black to use 79 characters per line (black -l 79). Test Driven Development ----------------------- -Please do not submit pull requests without tests. That's really important. Our project is all about test-driven development. It would be -embarrasing if our project failed because of a lack of tests! +Please do not submit pull requests without tests. That's really important. Our project is all about test-driven development. It would be embarrasing if our project failed because of a lack of tests! You might want to follow a typical test driven development cycle: http://en.wikipedia.org/wiki/Test-driven_development @@ -24,7 +27,7 @@ Features implemented without tests will be removed. Unmaintained features (which Check the CI status before comitting ------------------------------------ -We have a Travis CI account - please verify that your branch works before making a pull request. +Project uses Github Actions, please verify that your branch passes all tests before making a pull request. Any problems? ------------- diff --git a/jenkinsapi/api.py b/jenkinsapi/api.py index a4334c73..a920cc1f 100644 --- a/jenkinsapi/api.py +++ b/jenkinsapi/api.py @@ -5,74 +5,90 @@ hence they have simple string arguments. """ import os +import re import time import logging -import six +from typing import List, Dict -import six.moves.urllib.parse as urlparse +from urllib.parse import urlparse from jenkinsapi import constants -from jenkinsapi.jenkins import Jenkins from jenkinsapi.artifact import Artifact +from jenkinsapi.jenkins import Jenkins +from jenkinsapi.view import View +from jenkinsapi.job import Job +from jenkinsapi.build import Build from jenkinsapi.custom_exceptions import ArtifactsMissing, TimeOut, BadURL +from jenkinsapi.result_set import ResultSet -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) def get_latest_test_results( - jenkinsurl, jobname, username=None, password=None, ssl_verify=True -): + jenkinsurl: str, + jobname: str, + username: str = "", + password: str = "", + ssl_verify: bool = True, +) -> ResultSet: """ A convenience function to fetch down the very latest test results from a jenkins job. """ - latestbuild = get_latest_build( + latestbuild: Build = get_latest_build( jenkinsurl, jobname, username=username, password=password, ssl_verify=ssl_verify, ) - res = latestbuild.get_resultset() - return res + return latestbuild.get_resultset() def get_latest_build( - jenkinsurl, jobname, username=None, password=None, ssl_verify=True -): + jenkinsurl: str, + jobname: str, + username: str = "", + password: str = "", + ssl_verify: bool = True, +) -> Build: """ A convenience function to fetch down the very latest test results from a jenkins job. """ - jenkinsci = Jenkins( + jenkinsci: Jenkins = Jenkins( jenkinsurl, username=username, password=password, ssl_verify=ssl_verify ) - job = jenkinsci[jobname] + job: Job = jenkinsci[jobname] return job.get_last_build() def get_latest_complete_build( - jenkinsurl, jobname, username=None, password=None, ssl_verify=True -): + jenkinsurl: str, + jobname: str, + username: str = "", + password: str = "", + ssl_verify: bool = True, +) -> Build: """ A convenience function to fetch down the very latest test results from a jenkins job. """ - jenkinsci = Jenkins( + jenkinsci: Jenkins = Jenkins( jenkinsurl, username=username, password=password, ssl_verify=ssl_verify ) - job = jenkinsci[jobname] + job: Job = jenkinsci[jobname] return job.get_last_completed_build() def get_build( - jenkinsurl, - jobname, - build_no, - username=None, - password=None, - ssl_verify=True, -): + jenkinsurl: str, + jobname: str, + build_no: int, + username: str = "", + password: str = "", + ssl_verify: bool = True, +) -> Build: """ A convenience function to fetch down the test results from a jenkins job by build number. @@ -85,12 +101,12 @@ def get_build( def get_artifacts( - jenkinsurl, - jobid=None, - build_no=None, - username=None, - password=None, - ssl_verify=True, + jenkinsurl: str, + jobname: str, + build_no: int, + username: str = "", + password: str = "", + ssl_verify: bool = True, ): """ Find all the artifacts for the latest build of a job. @@ -98,25 +114,26 @@ def get_artifacts( jenkinsci = Jenkins( jenkinsurl, username=username, password=password, ssl_verify=ssl_verify ) - job = jenkinsci[jobid] + job = jenkinsci[jobname] if build_no: build = job.get_build(build_no) else: build = job.get_last_good_build() artifacts = build.get_artifact_dict() log.info( - msg="Found %i artifacts in '%s'" % (len(artifacts.keys()), build_no) + msg=f"Found {len(artifacts.keys())} \ + artifacts in '{jobname}[{build_no}]" ) return artifacts def search_artifacts( - jenkinsurl, - jobid, + jenkinsurl: str, + jobname: str, artifact_ids=None, - username=None, - password=None, - ssl_verify=True, + username: str = "", + password: str = "", + ssl_verify: bool = True, ): """ Search the entire history of a jenkins job for a list of artifact names. @@ -129,8 +146,9 @@ def search_artifacts( jenkinsci = Jenkins( jenkinsurl, username=username, password=password, ssl_verify=ssl_verify ) - job = jenkinsci[jobid] + job = jenkinsci[jobname] build_ids = job.get_build_ids() + missing_artifacts = set() for build_id in build_ids: build = job.get_build(build_id) artifacts = build.get_artifact_dict() @@ -139,22 +157,22 @@ def search_artifacts( missing_artifacts = set(artifact_ids) - set(artifacts.keys()) log.debug( msg="Artifacts %s missing from %s #%i" - % (", ".join(missing_artifacts), jobid, build_id) + % (", ".join(missing_artifacts), jobname, build_id) ) - # noinspection PyUnboundLocalVariable + raise ArtifactsMissing(missing_artifacts) def grab_artifact( - jenkinsurl, - jobid, + jenkinsurl: str, + jobname: str, artifactid, - targetdir, - username=None, - password=None, - strict_validation=False, - ssl_verify=True, -): + targetdir: str, + username: str = "", + password: str = "", + strict_validation: bool = True, + ssl_verify: bool = True, +) -> None: """ Convenience method to find the latest good version of an artifact and save it to a target directory. @@ -162,7 +180,8 @@ def grab_artifact( """ artifacts = get_artifacts( jenkinsurl, - jobid, + jobname, + artifactid, username=username, password=password, ssl_verify=ssl_verify, @@ -174,15 +193,15 @@ def grab_artifact( def block_until_complete( - jenkinsurl, - jobs, - maxwait=12000, - interval=30, - raise_on_timeout=True, - username=None, - password=None, - ssl_verify=True, -): + jenkinsurl: str, + jobs: List[str], + maxwait: int = 12000, + interval: int = 30, + raise_on_timeout: bool = True, + username: str = "", + password: str = "", + ssl_verify: bool = True, +) -> None: """ Wait until all of the jobs in the list are complete. """ @@ -190,30 +209,32 @@ def block_until_complete( assert maxwait > interval assert interval > 0 - obj_jenkins = Jenkins( + report: str = "" + obj_jenkins: Jenkins = Jenkins( jenkinsurl, username=username, password=password, ssl_verify=ssl_verify ) - obj_jobs = [obj_jenkins[jid] for jid in jobs] + obj_jobs: List[Job] = [obj_jenkins[jid] for jid in jobs] for time_left in range(maxwait, 0, -interval): still_running = [j for j in obj_jobs if j.is_queued_or_running()] if not still_running: return - str_still_running = ", ".join('"%s"' % str(a) for a in still_running) + report = ", ".join('"%s"' % str(a) for a in still_running) log.warning( "Waiting for jobs %s to complete. Will wait another %is", - str_still_running, + report, time_left, ) time.sleep(interval) if raise_on_timeout: # noinspection PyUnboundLocalVariable raise TimeOut( - "Waited too long for these jobs to complete: %s" - % str_still_running + "Waited too long for these jobs to complete: %s" % report ) -def get_view_from_url(url, username=None, password=None, ssl_verify=True): +def get_view_from_url( + url: str, username: str = "", password: str = "", ssl_verify: bool = True +) -> View: """ Factory method """ @@ -228,8 +249,8 @@ def get_view_from_url(url, username=None, password=None, ssl_verify=True): def get_nested_view_from_url( - url, username=None, password=None, ssl_verify=True -): + url: str, username: str = "", password: str = "", ssl_verify: bool = True +) -> View: """ Returns View based on provided URL. Convenient for nested views. """ @@ -246,7 +267,11 @@ def get_nested_view_from_url( def install_artifacts( - artifacts, dirstruct, installdir, basestaticurl, strict_validation=False + artifacts, + dirstruct: Dict[str, str], + installdir: str, + basestaticurl: str, + strict_validation: bool = False, ): """ Install the artifacts. @@ -276,15 +301,15 @@ def install_artifacts( def search_artifact_by_regexp( - jenkinsurl, - jobid, - artifactRegExp, - username=None, - password=None, - ssl_verify=True, -): + jenkinsurl: str, + jobname: str, + artifactRegExp: re.Pattern, + username: str = "", + password: str = "", + ssl_verify: bool = True, +) -> Artifact: """ - Search the entire history of a hudson job for a build which has an + Search the entire history of a Jenkins job for a build which has an artifact whose name matches a supplied regular expression. Return only that artifact. @@ -298,7 +323,7 @@ def search_artifact_by_regexp( job = Jenkins( jenkinsurl, username=username, password=password, ssl_verify=ssl_verify ) - j = job[jobid] + j = job[jobname] build_ids = j.get_build_ids() @@ -306,9 +331,7 @@ def search_artifact_by_regexp( build = j.get_build(build_id) artifacts = build.get_artifact_dict() - it = six.iteritems(artifacts) - - for name, art in it: + for name, art in artifacts.items(): md_match = artifactRegExp.search(name) if md_match: diff --git a/jenkinsapi/artifact.py b/jenkinsapi/artifact.py index 48bbb9b9..2b2f1634 100644 --- a/jenkinsapi/artifact.py +++ b/jenkinsapi/artifact.py @@ -8,9 +8,12 @@ This module provides a class called Artifact which allows you to download objects from the server and also access them as a stream. """ +from __future__ import annotations + import os import logging import hashlib +from typing import Any, Literal from jenkinsapi.fingerprint import Fingerprint from jenkinsapi.custom_exceptions import ArtifactBroken @@ -25,13 +28,19 @@ class Artifact(object): generated as a by-product of executing a Jenkins build. """ - def __init__(self, filename, url, build, relative_path=None): - self.filename = filename - self.url = url - self.build = build - self.relative_path = relative_path - - def save(self, fspath, strict_validation=False): + def __init__( + self, + filename: str, + url: str, + build: "Build", + relative_path: str | None = None, + ) -> None: + self.filename: str = filename + self.url: str = url + self.build: "Build" = build + self.relative_path: str | None = relative_path + + def save(self, fspath: str, strict_validation: bool = False) -> str: """ Save the artifact to an explicit path. The containing directory must exist. Returns a reference to the file which has just been writen to. @@ -67,10 +76,10 @@ def save(self, fspath, strict_validation=False): self._verify_download(filepath, strict_validation) return fspath - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> Jenkins: return self.build.get_jenkins_obj() - def get_data(self): + def get_data(self) -> Any: """ Grab the text of the artifact """ @@ -79,7 +88,7 @@ def get_data(self): ) return response.content - def _do_download(self, fspath): + def _do_download(self, fspath: str) -> str: """ Download the the artifact to a path. """ @@ -91,9 +100,12 @@ def _do_download(self, fspath): out.write(chunk) return fspath - def _verify_download(self, fspath, strict_validation): + def _verify_download(self, fspath, strict_validation) -> Literal[True]: """ Verify that a downloaded object has a valid fingerprint. + + Returns True if the fingerprint is valid, raises an exception if + the fingerprint is invalid. """ local_md5 = self._md5sum(fspath) baseurl = self.build.job.jenkins.baseurl @@ -109,7 +121,7 @@ def _verify_download(self, fspath, strict_validation): ) return True - def _md5sum(self, fspath, chunksize=2**20): + def _md5sum(self, fspath: str, chunksize: int = 2**20) -> str: """ A MD5 hashing function intended to produce the same results as that used by Jenkins. @@ -123,17 +135,19 @@ def _md5sum(self, fspath, chunksize=2**20): break return md5.hexdigest() - def save_to_dir(self, dirpath, strict_validation=False): + def save_to_dir( + self, dirpath: str, strict_validation: bool = False + ) -> str: """ 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) + outputfilepath: str = os.path.join(dirpath, self.filename) return self.save(outputfilepath, strict_validation) - def __repr__(self): + def __repr__(self) -> str: """ Produce a handy repr-string. """ diff --git a/jenkinsapi/build.py b/jenkinsapi/build.py index fcdc831b..336dfafa 100644 --- a/jenkinsapi/build.py +++ b/jenkinsapi/build.py @@ -7,6 +7,7 @@ Build objects can be associated with Results and Artifacts. """ +from __future__ import annotations import time import logging @@ -14,16 +15,20 @@ import datetime from time import sleep +from typing import Iterator, List, Dict, Any + import pytz from jenkinsapi import config from jenkinsapi.artifact import Artifact + +# from jenkinsapi.job import Job from jenkinsapi.result_set import ResultSet from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.constants import STATUS_SUCCESS from jenkinsapi.custom_exceptions import NoResults from jenkinsapi.custom_exceptions import JenkinsAPIException -from six.moves.urllib.parse import quote +from urllib.parse import quote from requests import HTTPError @@ -41,7 +46,9 @@ class Build(JenkinsBase): "%s has status %s, and does not have " "any test results" ) - def __init__(self, url, buildno, job, depth=1): + def __init__( + self, url: str, buildno: int, job: "Job", depth: int = 1 + ) -> None: """ depth=1 is for backward compatibility consideration @@ -50,9 +57,8 @@ def __init__(self, url, buildno, job, depth=1): information, see https://www.jenkins.io/doc/book/using/remote-access-api/#RemoteaccessAPI-Depthcontrol """ - assert isinstance(buildno, int) - self.buildno = buildno - self.job = job + self.buildno: int = buildno + self.job: "Job" = job self.depth = depth JenkinsBase.__init__(self, url) @@ -63,39 +69,37 @@ def _poll(self, tree=None): url = self.python_api_url(self.baseurl) return self.get_data(url, params={"depth": self.depth}, tree=tree) - def __str__(self): + def __str__(self) -> str: return self._data["fullDisplayName"] @property def name(self): return str(self) - def get_description(self): + def get_description(self) -> str: return self._data["description"] - def get_number(self): + def get_number(self) -> int: return self._data["number"] - def get_status(self): + def get_status(self) -> str: return self._data["result"] - def get_slave(self): + def get_slave(self) -> str: return self._data["builtOn"] - def get_revision(self): - return getattr(self, "_get_%s_rev" % self._get_vcs(), lambda: None)() + def get_revision(self) -> str: + return getattr(self, f"_get_{self._get_vcs()}_rev", lambda: "")() - def get_revision_branch(self): + def get_revision_branch(self) -> str: return getattr( - self, "_get_%s_rev_branch" % self._get_vcs(), lambda: None + self, f"_get_{self._get_vcs()}_rev_branch", lambda: "" )() - def get_repo_url(self): - return getattr( - self, "_get_%s_repo_url" % self._get_vcs(), lambda: None - )() + def get_repo_url(self) -> str: + return getattr(self, f"_get_{self._get_vcs()}_repo_url", lambda: "")() - def get_params(self): + def get_params(self) -> dict[str, str]: """ Return a dictionary of params names and their values, or an empty dictionary if no parameters are returned. @@ -149,7 +153,7 @@ def get_changeset_items(self): return self._data["changeSets"]["items"] return [] - def _get_vcs(self): + def _get_vcs(self) -> str: """ Returns a string VCS. By default, 'git' will be used. @@ -161,17 +165,7 @@ def _get_vcs(self): vcs = self._data["changeSets"]["kind"] or "git" return vcs - def _get_svn_rev(self): - warnings.warn( - "This untested function may soon be removed from Jenkinsapi " - "(get_svn_rev)." - ) - maxRevision = 0 - for repoPathSet in self._data["changeSet"]["revisions"]: - maxRevision = max(repoPathSet["revision"], maxRevision) - return maxRevision - - def _get_git_rev(self): + def _get_git_rev(self) -> str | None: # Sometimes we have None as part of actions. Filter those actions # which have lastBuiltRevision in them _actions = [ @@ -183,21 +177,7 @@ def _get_git_rev(self): return None - def _get_hg_rev(self): - warnings.warn( - "This untested function may soon be removed from Jenkinsapi " - "(_get_hg_rev)." - ) - return [ - x["mercurialNodeName"] - for x in self._data["actions"] - if "mercurialNodeName" in x - ][0] - - def _get_svn_rev_branch(self): - raise NotImplementedError("_get_svn_rev_branch is not yet implemented") - - def _get_git_rev_branch(self): + def _get_git_rev_branch(self) -> str: # Sometimes we have None as part of actions. Filter those actions # which have lastBuiltRevision in them _actions = [ @@ -206,10 +186,7 @@ def _get_git_rev_branch(self): return _actions[0]["lastBuiltRevision"]["branch"] - def _get_hg_rev_branch(self): - raise NotImplementedError("_get_hg_rev_branch is not yet implemented") - - def _get_git_repo_url(self): + def _get_git_repo_url(self) -> str: # Sometimes we have None as part of actions. Filter those actions # which have lastBuiltRevision in them _actions = [ @@ -222,19 +199,13 @@ def _get_git_repo_url(self): result = ",".join(result) return result - def _get_svn_repo_url(self): - raise NotImplementedError("_get_svn_repo_url is not yet implemented") - - def _get_hg_repo_url(self): - raise NotImplementedError("_get_hg_repo_url is not yet implemented") - - def get_duration(self): + def get_duration(self) -> datetime.timedelta: return datetime.timedelta(milliseconds=self._data["duration"]) - def get_build_url(self): + def get_build_url(self) -> str: return self._data["url"] - def get_artifacts(self): + def get_artifacts(self) -> Iterator[Artifact]: data = self.poll(tree="artifacts[relativePath,fileName]") for afinfo in data["artifacts"]: url = "%s/artifact/%s" % ( @@ -249,10 +220,10 @@ def get_artifacts(self): ) yield af - def get_artifact_dict(self): - return dict((af.relative_path, af) for af in self.get_artifacts()) + def get_artifact_dict(self) -> dict[str, Artifact]: + return {af.relative_path: af for af in self.get_artifacts()} - def get_upstream_job_name(self): + def get_upstream_job_name(self) -> str | None: """ Get the upstream job name if it exist, None otherwise :return: String or None @@ -262,7 +233,7 @@ def get_upstream_job_name(self): except KeyError: return None - def get_upstream_job(self): + def get_upstream_job(self) -> Job | None: """ Get the upstream job object if it exist, None otherwise :return: Job or None @@ -271,7 +242,7 @@ def get_upstream_job(self): return self.get_jenkins_obj().get_job(self.get_upstream_job_name()) return None - def get_upstream_build_number(self): + def get_upstream_build_number(self) -> int | None: """ Get the upstream build number if it exist, None otherwise :return: int or None @@ -281,18 +252,18 @@ def get_upstream_build_number(self): except KeyError: return None - def get_upstream_build(self): + def get_upstream_build(self) -> "Build" | None: """ Get the upstream build if it exist, None otherwise :return Build or None """ - upstream_job = self.get_upstream_job() + upstream_job: "Job" = self.get_upstream_job() if upstream_job: return upstream_job.get_build(self.get_upstream_build_number()) return None - def get_master_job_name(self): + def get_master_job_name(self) -> str | None: """ Get the master job name if it exist, None otherwise :return: String or None @@ -302,59 +273,43 @@ def get_master_job_name(self): except KeyError: return None - def get_master_job(self): + def get_master_job(self) -> Job | None: """ Get the master job object if it exist, None otherwise :return: Job or None """ - warnings.warn( - "This untested function may soon be removed from Jenkinsapi " - "(get_master_job)." - ) if self.get_master_job_name(): return self.get_jenkins_obj().get_job(self.get_master_job_name()) return None - def get_master_build_number(self): + def get_master_build_number(self) -> int | None: """ Get the master build number if it exist, None otherwise :return: int or None """ - warnings.warn( - "This untested function may soon be removed from Jenkinsapi " - "(get_master_build_number)." - ) try: return int(self.get_actions()["parameters"][1]["value"]) except KeyError: return None - def get_master_build(self): + def get_master_build(self) -> "Build" | None: """ Get the master build if it exist, None otherwise :return Build or None """ - warnings.warn( - "This untested function may soon be removed from Jenkinsapi " - "(get_master_build)." - ) - master_job = self.get_master_job() + master_job: Job | None = self.get_master_job() if master_job: return master_job.get_build(self.get_master_build_number()) return None - def get_downstream_jobs(self): + def get_downstream_jobs(self) -> List[Job]: """ Get the downstream jobs for this build :return List of jobs or None """ - warnings.warn( - "This untested function may soon be removed from Jenkinsapi " - "(get_downstream_jobs)." - ) - downstream_jobs = [] + downstream_jobs: List[Job] = [] try: for job_name in self.get_downstream_job_names(): downstream_jobs.append( @@ -364,13 +319,13 @@ def get_downstream_jobs(self): except (IndexError, KeyError): return [] - def get_downstream_job_names(self): + def get_downstream_job_names(self) -> List[str]: """ Get the downstream job names for this build :return List of string or None """ - downstream_job_names = self.job.get_downstream_job_names() - downstream_names = [] + downstream_job_names: List[str] = self.job.get_downstream_job_names() + downstream_names: List[str] = [] try: fingerprints = self._data["fingerprint"] for fingerprint in fingerprints: @@ -381,13 +336,13 @@ def get_downstream_job_names(self): except (IndexError, KeyError): return [] - def get_downstream_builds(self): + def get_downstream_builds(self) -> List["Build"]: """ Get the downstream builds for this build :return List of Build or None """ - downstream_job_names = self.get_downstream_job_names() - downstream_builds = [] + downstream_job_names: List[str] = self.get_downstream_job_names() + downstream_builds: List[Build] = [] try: # pylint: disable=R1702 fingerprints = self._data["fingerprint"] for fingerprint in fingerprints: @@ -405,7 +360,7 @@ def get_downstream_builds(self): except (IndexError, KeyError): return [] - def get_matrix_runs(self): + def get_matrix_runs(self) -> Iterator["Build"]: """ For a matrix job, get the individual builds for each matrix configuration @@ -413,22 +368,22 @@ def get_matrix_runs(self): """ if "runs" in self._data: for rinfo in self._data["runs"]: - number = rinfo["number"] + number: int = rinfo["number"] if number == self._data["number"]: yield Build(rinfo["url"], number, self.job) - def is_running(self): + def is_running(self) -> bool: """ Return a bool if running. """ data = self.poll(tree="building") return data.get("building", False) - def block(self): + def block(self) -> None: while self.is_running(): time.sleep(1) - def is_good(self): + def is_good(self) -> bool: """ Return a bool, true if the build was good. If the build is still running, return False. @@ -437,11 +392,10 @@ def is_good(self): "result" ] == STATUS_SUCCESS - def block_until_complete(self, delay=15): - assert isinstance(delay, int) - count = 0 + def block_until_complete(self, delay: int = 15) -> None: + count: int = 0 while self.is_running(): - total_wait = delay * count + total_wait: int = delay * count log.info( msg="Waited %is for %s #%s to complete" % (total_wait, self.job.name, self.name) @@ -449,55 +403,58 @@ def block_until_complete(self, delay=15): sleep(delay) count += 1 - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.job.get_jenkins_obj() - def get_result_url(self): + def get_result_url(self) -> str: """ Return the URL for the object which provides the job's result summary. """ - url_tpl = r"%stestReport/%s" + url_tpl: str = r"%stestReport/%s" return url_tpl % (self._data["url"], config.JENKINS_API) - def get_resultset(self): + def get_resultset(self) -> ResultSet: """ Obtain detailed results for this build. + + Raises NoResults if the build has no results. + + :return: ResultSet """ - result_url = self.get_result_url() + result_url: str = self.get_result_url() if self.STR_TOTALCOUNT not in self.get_actions(): raise NoResults( "%s does not have any published results" % str(self) ) - buildstatus = self.get_status() + buildstatus: str = self.get_status() if not self.get_actions()[self.STR_TOTALCOUNT]: raise NoResults( self.STR_TPL_NOTESTS_ERR % (str(self), buildstatus) ) - obj_results = ResultSet(result_url, build=self) - return obj_results + return ResultSet(result_url, build=self) - def has_resultset(self): + def has_resultset(self) -> bool: """ Return a boolean, true if a result set is available. false if not. """ return self.STR_TOTALCOUNT in self.get_actions() - def get_actions(self): - all_actions = {} + def get_actions(self) -> Dict[str, Any]: + all_actions: Dict[str, Any] = {} for dct_action in self._data["actions"]: if dct_action is None: continue all_actions.update(dct_action) return all_actions - def get_causes(self): + def get_causes(self) -> List[str]: """ Returns a list of causes. There can be multiple causes lists and some of the can be empty. For instance, when a build is manually aborted, Jenkins could add an empty causes list to the actions dict. Empty ones are ignored. """ - all_causes = [] + all_causes: List[str] = [] for dct_action in self._data["actions"]: if dct_action is None: continue @@ -505,7 +462,7 @@ def get_causes(self): all_causes.extend(dct_action["causes"]) return all_causes - def get_timestamp(self): + def get_timestamp(self) -> datetime.datetime: """ Returns build timestamp in UTC """ @@ -515,12 +472,12 @@ def get_timestamp(self): ) return pytz.utc.localize(naive_timestamp) - def get_console(self): + def get_console(self) -> str: """ Return the current state of the text console. """ - url = "%s/consoleText" % self.baseurl - content = self.job.jenkins.requester.get_url(url).content + url: str = "%s/consoleText" % self.baseurl + content: Any = self.job.jenkins.requester.get_url(url).content # This check was made for Python 3.x # In this version content is a bytes string # By contract this function must return string @@ -531,13 +488,13 @@ def get_console(self): else: raise JenkinsAPIException("Unknown content type for console") - def stream_logs(self, interval=0): + def stream_logs(self, interval=0) -> Iterator[str]: """ Return generator which streams parts of text console. """ - url = "%s/logText/progressiveText" % self.baseurl - size = 0 - more_data = True + url: str = "%s/logText/progressiveText" % self.baseurl + size: int = 0 + more_data: bool = True while more_data: resp = self.job.jenkins.requester.get_url( url, params={"start": size} @@ -556,7 +513,7 @@ def stream_logs(self, interval=0): more_data = resp.headers.get("X-More-Data") sleep(interval) - def get_estimated_duration(self): + def get_estimated_duration(self) -> int | None: """ Return the estimated build duration (in seconds) or none. """ @@ -566,14 +523,14 @@ def get_estimated_duration(self): except KeyError: return None - def stop(self): + def stop(self) -> bool: """ Stops the build execution if it's running :return boolean True if succeded False otherwise or the build is not running """ if self.is_running(): - url = "%s/stop" % self.baseurl + url: str = "%s/stop" % self.baseurl # Starting from Jenkins 2.7 stop function sometimes breaks # on redirect to job page. Call to stop works fine, and # we don't need to have job page here. @@ -589,30 +546,30 @@ def stop(self): return True return False - def get_env_vars(self): + def get_env_vars(self) -> Dict[str, str]: """ Return the environment variables. This method is using the Environment Injector plugin: https://wiki.jenkins-ci.org/display/JENKINS/EnvInject+Plugin """ - url = self.python_api_url("%s/injectedEnvVars" % self.baseurl) + url: str = self.python_api_url("%s/injectedEnvVars" % self.baseurl) try: data = self.get_data(url, params={"depth": self.depth}) except HTTPError as ex: warnings.warn( - "Make sure the Environment Injector plugin " "is installed." + "Make sure the Environment Injector plugin is installed." ) raise ex return data["envMap"] - def toggle_keep(self): + def toggle_keep(self) -> None: """ Toggle "keep this build forever" on and off """ - url = "%s/toggleLogKeep" % self.baseurl + url: str = "%s/toggleLogKeep" % self.baseurl self.get_jenkins_obj().requester.post_and_confirm_status(url, data={}) self._data = self._poll() - def is_kept_forever(self): + def is_kept_forever(self) -> bool: return self._data["keepLog"] diff --git a/jenkinsapi/credential.py b/jenkinsapi/credential.py index 02408da1..e95b14a8 100644 --- a/jenkinsapi/credential.py +++ b/jenkinsapi/credential.py @@ -82,18 +82,21 @@ class UsernamePasswordCredential(Credential): dict """ - def __init__(self, cred_dict): - jenkins_class = "com.cloudbees.plugins.credentials.impl.UsernamePasswordCredentialsImpl" # noqa + def __init__(self, cred_dict: dict) -> None: + jenkins_class: str = ( + "com.cloudbees.plugins.credentials.impl." + "UsernamePasswordCredentialsImpl" + ) super(UsernamePasswordCredential, self).__init__( cred_dict, jenkins_class ) if "typeName" in cred_dict: - username = cred_dict["displayName"].split("/")[0] + username: str = cred_dict["displayName"].split("/")[0] else: - username = cred_dict["userName"] + username: str = cred_dict["userName"] - self.username = username - self.password = cred_dict.get("password", None) + self.username: str = username + self.password: str = cred_dict.get("password", "") def get_attributes(self): """ @@ -219,23 +222,26 @@ class SSHKeyCredential(Credential): dict """ - def __init__(self, cred_dict): - jenkins_class = "com.cloudbees.jenkins.plugins.sshcredentials.impl.BasicSSHUserPrivateKey" # noqa + def __init__(self, cred_dict: dict) -> None: + jenkins_class: str = ( + "com.cloudbees.jenkins.plugins.sshcredentials.impl." + "BasicSSHUserPrivateKey" + ) super(SSHKeyCredential, self).__init__(cred_dict, jenkins_class) if "typeName" in cred_dict: - username = cred_dict["displayName"].split(" ")[0] + username: str = cred_dict["displayName"].split(" ")[0] else: - username = cred_dict["userName"] + username: str = cred_dict["userName"] - self.username = username - self.passphrase = cred_dict.get("passphrase", "") + self.username: str = username + self.passphrase: str = cred_dict.get("passphrase", "") if "private_key" not in cred_dict or cred_dict["private_key"] is None: - self.key_type = -1 - self.key_value = None + self.key_type: int = -1 + self.key_value: str = "" elif cred_dict["private_key"].startswith("-"): - self.key_type = 0 - self.key_value = cred_dict["private_key"] + self.key_type: int = 0 + self.key_value: str = cred_dict["private_key"] else: raise ValueError("Invalid private_key value") diff --git a/jenkinsapi/credentials.py b/jenkinsapi/credentials.py index 83abe632..a87434d6 100644 --- a/jenkinsapi/credentials.py +++ b/jenkinsapi/credentials.py @@ -3,9 +3,13 @@ container-like interface for all of the Global credentials defined on a single Jenkins node. """ +from __future__ import annotations + +from typing import Iterator + import logging +from urllib.parse import urlencode -from six.moves.urllib.parse import urlencode from jenkinsapi.credential import Credential from jenkinsapi.credential import UsernamePasswordCredential from jenkinsapi.credential import SecretTextCredential @@ -13,7 +17,7 @@ from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.custom_exceptions import JenkinsAPIException -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class Credentials(JenkinsBase): @@ -24,15 +28,15 @@ class Credentials(JenkinsBase): Returns a list of Credential Objects. """ - def __init__(self, baseurl, jenkins_obj): - self.baseurl = baseurl - self.jenkins = jenkins_obj + def __init__(self, baseurl: str, jenkins_obj: "Jenkins"): + self.baseurl: str = baseurl + self.jenkins: "Jenkins" = jenkins_obj JenkinsBase.__init__(self, baseurl) self.credentials = self._data["credentials"] def _poll(self, tree=None): - url = self.python_api_url(self.baseurl) + "?depth=2" + url: str = self.python_api_url(self.baseurl) + "?depth=2" data = self.get_data(url, tree=tree) credentials = data["credentials"] for cred_id, cred_dict in credentials.items(): @@ -41,17 +45,17 @@ def _poll(self, tree=None): return data - def __str__(self): + def __str__(self) -> str: return "Global Credentials @ %s" % self.baseurl - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def __iter__(self): + def __iter__(self) -> Iterator[Credential]: for cred in self.credentials.values(): yield cred.description - def __contains__(self, description): + def __contains__(self, description: str) -> bool: return description in self.keys() def iterkeys(self): @@ -60,11 +64,11 @@ def iterkeys(self): def keys(self): return list(self.iterkeys()) - def iteritems(self): + def iteritems(self) -> Iterator[str, "Credential"]: for cred in self.credentials.values(): yield cred.description, cred - def __getitem__(self, description): + def __getitem__(self, description: str) -> "Credential": for cred in self.credentials.values(): if cred.description == description: return cred @@ -73,10 +77,10 @@ def __getitem__(self, description): 'Credential with description "%s" not found' % description ) - def __len__(self): + def __len__(self) -> int: return len(self.keys()) - def __setitem__(self, description, credential): + def __setitem__(self, description: str, credential: "Credential"): """ Creates Credential in Jenkins using username, password and description Description must be unique in Jenkins instance @@ -127,7 +131,7 @@ def __setitem__(self, description, credential): def get(self, item, default): return self[item] if item in self else default - def __delitem__(self, description): + def __delitem__(self, description: str): if description not in self: raise KeyError( 'Credential with description "%s" not found' % description diff --git a/jenkinsapi/executor.py b/jenkinsapi/executor.py index 809fdd45..23e3d077 100644 --- a/jenkinsapi/executor.py +++ b/jenkinsapi/executor.py @@ -1,6 +1,7 @@ """ Module for jenkinsapi Executer class """ +from __future__ import annotations from jenkinsapi.jenkinsbase import JenkinsBase import logging @@ -15,7 +16,9 @@ class Executor(JenkinsBase): master jenkins instance """ - def __init__(self, baseurl, nodename, jenkins_obj, number): + def __init__( + self, baseurl: str, nodename: str, jenkins_obj: "Jenkins", number: int + ) -> None: """ Init a node object by providing all relevant pointers to it :param baseurl: basic url for querying information on a node @@ -23,41 +26,41 @@ def __init__(self, baseurl, nodename, jenkins_obj, number): :param jenkins_obj: ref to the jenkins obj :return: Node obj """ - self.nodename = nodename - self.number = number - self.jenkins = jenkins_obj - self.baseurl = baseurl + self.nodename: str = nodename + self.number: int = number + self.jenkins: "Jenkins" = jenkins_obj + self.baseurl: str = baseurl JenkinsBase.__init__(self, baseurl) - def __str__(self): - return "%s %s" % (self.nodename, self.number) + def __str__(self) -> str: + return f"{self.nodename} {self.number}" - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def get_progress(self): + def get_progress(self) -> str: """Returns percentage""" return self.poll(tree="progress")["progress"] - def get_number(self): + def get_number(self) -> int: """ Get Executor number. """ return self.poll(tree="number")["number"] - def is_idle(self): + def is_idle(self) -> bool: """ Returns Boolean: whether Executor is idle or not. """ return self.poll(tree="idle")["idle"] - def likely_stuck(self): + def likely_stuck(self) -> bool: """ Returns Boolean: whether Executor is likely stuck or not. """ return self.poll(tree="likelyStuck")["likelyStuck"] - def get_current_executable(self): + def get_current_executable(self) -> str: """ Returns the current Queue.Task this executor is running. """ diff --git a/jenkinsapi/executors.py b/jenkinsapi/executors.py index c3406e0e..e66d64cc 100644 --- a/jenkinsapi/executors.py +++ b/jenkinsapi/executors.py @@ -3,11 +3,15 @@ container-like interface for all of the executors defined on a single Jenkins node. """ +from __future__ import annotations + import logging +from typing import Iterator + from jenkinsapi.executor import Executor from jenkinsapi.jenkinsbase import JenkinsBase -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class Executors(JenkinsBase): @@ -19,19 +23,21 @@ class Executors(JenkinsBase): Returns a list of Executor Objects. """ - def __init__(self, baseurl, nodename, jenkins): - self.nodename = nodename - self.jenkins = jenkins + def __init__( + self, baseurl: str, nodename: str, jenkins: "Jenkins" + ) -> None: + self.nodename: str = nodename + self.jenkins: str = jenkins JenkinsBase.__init__(self, baseurl) - self.count = self._data["numExecutors"] + self.count: int = self._data["numExecutors"] - def __str__(self): - return "Executors @ %s" % self.baseurl + def __str__(self) -> str: + return f"Executors @ {self.baseurl}" - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def __iter__(self): + def __iter__(self) -> Iterator[Executor]: for index in range(self.count): executor_url = "%s/executors/%s" % (self.baseurl, index) yield Executor(executor_url, self.nodename, self.jenkins, index) diff --git a/jenkinsapi/fingerprint.py b/jenkinsapi/fingerprint.py index 5423f68d..5ccc4cca 100644 --- a/jenkinsapi/fingerprint.py +++ b/jenkinsapi/fingerprint.py @@ -1,16 +1,18 @@ """ Module for jenkinsapi Fingerprint """ - -from jenkinsapi.jenkinsbase import JenkinsBase -from jenkinsapi.custom_exceptions import ArtifactBroken +from __future__ import annotations import re +import logging +from typing import Any + import requests -import logging +from jenkinsapi.jenkinsbase import JenkinsBase +from jenkinsapi.custom_exceptions import ArtifactBroken -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class Fingerprint(JenkinsBase): @@ -21,24 +23,24 @@ class Fingerprint(JenkinsBase): RE_MD5 = re.compile("^([0-9a-z]{32})$") - def __init__(self, baseurl, id_, jenkins_obj): - logging.basicConfig() - self.jenkins_obj = jenkins_obj + def __init__(self, baseurl: str, id_: str, jenkins_obj: "Jenkins") -> None: + self.jenkins_obj: "Jenkins" = jenkins_obj assert self.RE_MD5.search(id_), ( "%s does not look like " "a valid id" % id_ ) - url = "%s/fingerprint/%s/" % (baseurl, id_) + url: str = f"{baseurl}/fingerprint/{id_}/" + JenkinsBase.__init__(self, url, poll=False) - self.id_ = id_ - self.unknown = False # Previously uninitialized in ctor + self.id_: str = id_ + self.unknown: bool = False # Previously uninitialized in ctor - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins_obj - def __str__(self): + def __str__(self) -> str: return self.id_ - def valid(self): + def valid(self) -> bool: """ Return True / False if valid. If returns True, self.unknown is set to either True or False, and can be checked if we have @@ -56,10 +58,10 @@ def valid(self): # valid or not. # The response object is of type: requests.models.Response # extract the status code from it - response_obj = err.response + response_obj: Any = err.response if response_obj.status_code == 404: logging.warning( - "MD5 cannot be checked if fingerprints are not " "enabled" + "MD5 cannot be checked if fingerprints are not enabled" ) self.unknown = True return True @@ -68,9 +70,9 @@ def valid(self): return True - def validate_for_build(self, filename, job, build): + def validate_for_build(self, filename: str, job: str, build: int) -> bool: if not self.valid(): - log.info("Unknown to jenkins.") + log.info("Fingerprint is not known to jenkins.") return False if self.unknown: # not request error, but unknown to jenkins @@ -98,14 +100,14 @@ def validate_for_build(self, filename, job, build): return True return False - def validate(self): + def validate(self) -> bool: try: assert self.valid() - except AssertionError: + except AssertionError as ae: raise ArtifactBroken( "Artifact %s seems to be broken, check %s" % (self.id_, self.baseurl) - ) + ) from ae except requests.exceptions.HTTPError: raise ArtifactBroken( "Unable to validate artifact id %s using %s" diff --git a/jenkinsapi/jenkins.py b/jenkinsapi/jenkins.py index 126f046a..45a06473 100644 --- a/jenkinsapi/jenkins.py +++ b/jenkinsapi/jenkins.py @@ -1,15 +1,19 @@ """ Module for jenkinsapi Jenkins object """ +from __future__ import annotations + import time import logging import warnings -import six.moves.urllib.parse as urlparse -from six.moves.urllib.request import Request, HTTPRedirectHandler, build_opener -from six.moves.urllib.parse import quote as urlquote -from six.moves.urllib.parse import urlencode +from urllib.parse import urlparse +from urllib.request import Request, HTTPRedirectHandler, build_opener +from urllib.parse import quote as urlquote +from urllib.parse import urlencode + from requests import HTTPError, ConnectionError + from jenkinsapi import config from jenkinsapi.credentials import Credentials from jenkinsapi.credentials import Credentials2x @@ -19,6 +23,7 @@ from jenkinsapi.job import Job from jenkinsapi.view import View from jenkinsapi.label import Label +from jenkinsapi.node import Node from jenkinsapi.nodes import Nodes from jenkinsapi.plugins import Plugins from jenkinsapi.plugin import Plugin @@ -43,17 +48,17 @@ class Jenkins(JenkinsBase): # pylint: disable=too-many-arguments def __init__( self, - baseurl, - username=None, - password=None, + baseurl: str, + username: str = "", + password: str = "", requester=None, - lazy=False, - ssl_verify=True, + lazy: bool = False, + ssl_verify: bool = True, cert=None, - timeout=10, - use_crumb=True, + timeout: int = 10, + use_crumb: bool = True, max_retries=None, - ): + ) -> None: """ :param baseurl: baseurl for jenkins instance including port, str :param username: username for jenkins auth, str @@ -113,10 +118,6 @@ def validate_fingerprint(self, id_): obj_fingerprint.validate() log.info(msg="Jenkins says %s is valid" % id_) - # def reload(self): - # '''Try and reload the configuration from disk''' - # self.requester.get_url("%(baseurl)s/reload" % self.__dict__) - def get_artifact_data(self, id_): obj_fingerprint = Fingerprint(self.baseurl, id_, jenkins_obj=self) obj_fingerprint.validate() @@ -129,14 +130,14 @@ def validate_fingerprint_for_build(self, digest, filename, job, build): def get_jenkins_obj(self): return self - def get_jenkins_obj_from_url(self, url): + def get_jenkins_obj_from_url(self, url: str): return Jenkins(url, self.username, self.password, self.requester) - def get_create_url(self): + def get_create_url(self) -> str: # This only ever needs to work on the base object return "%s/createItem" % self.baseurl - def get_nodes_url(self): + def get_nodes_url(self) -> str: # This only ever needs to work on the base object return self.nodes.baseurl @@ -161,7 +162,7 @@ def get_jobs_info(self): for name, job in self.jobs.iteritems(): yield job.url, name - def get_job(self, jobname): + def get_job(self, jobname: str) -> Job: """ Get a job by name :param jobname: name of the job, str @@ -169,7 +170,7 @@ def get_job(self, jobname): """ return self.jobs[jobname] - def get_job_by_url(self, url, job_name): + def get_job_by_url(self, url: str, job_name: str) -> Job: """ Get a job by url :param url: jobs' url @@ -178,7 +179,7 @@ def get_job_by_url(self, url, job_name): """ return Job(url, job_name, self) - def has_job(self, jobname): + def has_job(self, jobname: str) -> bool: """ Does a job by the name specified exist :param jobname: string @@ -186,7 +187,7 @@ def has_job(self, jobname): """ return jobname in self.jobs - def create_job(self, jobname, xml): + def create_job(self, jobname: str, xml) -> Job: """ Create a job @@ -208,10 +209,10 @@ def create_multibranch_pipeline_job( jobname, xml, block, delay ) - def copy_job(self, jobname, newjobname): + def copy_job(self, jobname: str, newjobname: str) -> Job: return self.jobs.copy(jobname, newjobname) - def build_job(self, jobname, params=None): + def build_job(self, jobname: str, params=None) -> None: """ Invoke a build by job name :param jobname: name of exist job, str @@ -220,7 +221,7 @@ def build_job(self, jobname, params=None): """ self[jobname].invoke(build_params=params or {}) - def delete_job(self, jobname): + def delete_job(self, jobname: str) -> None: """ Delete a job by name :param jobname: name of a exist job, str @@ -228,7 +229,7 @@ def delete_job(self, jobname): """ del self.jobs[jobname] - def rename_job(self, jobname, newjobname): + def rename_job(self, jobname: str, newjobname: str) -> Job: """ Rename a job :param jobname: name of a exist job, str @@ -256,29 +257,29 @@ def iteritems(self): def keys(self): return self.jobs.keys() - def __str__(self): + def __str__(self) -> str: return "Jenkins server at %s" % self.baseurl @property def views(self): return Views(self) - def get_view_by_url(self, str_view_url): + def get_view_by_url(self, view_url: str): # for nested view - str_view_name = str_view_url.split("/view/")[-1].replace("/", "") - return View(str_view_url, str_view_name, jenkins_obj=self) + view_name = view_url.split("/view/")[-1].replace("/", "") + return View(view_url, view_name, jenkins_obj=self) - def delete_view_by_url(self, str_url): - url = "%s/doDelete" % str_url + def delete_view_by_url(self, viewurl: str): + url = f"{viewurl}/doDelete" self.requester.post_and_confirm_status(url, data="") self.poll() return self - def get_label(self, label_name): + def get_label(self, label_name: str) -> Label: label_url = "%s/label/%s" % (self.baseurl, label_name) return Label(label_url, label_name, jenkins_obj=self) - def __getitem__(self, jobname): + def __getitem__(self, jobname: str) -> Job: """ Get a job by name :param jobname: name of job, str @@ -286,10 +287,10 @@ def __getitem__(self, jobname): """ return self.jobs[jobname] - def __len__(self): + def __len__(self) -> int: return len(self.jobs) - def __contains__(self, jobname): + def __contains__(self, jobname: str) -> bool: """ Does a job by the name specified exist :param jobname: string @@ -297,14 +298,14 @@ def __contains__(self, jobname): """ return jobname in self.jobs - def __delitem__(self, job_name): + def __delitem__(self, job_name: str) -> None: del self.jobs[job_name] - def get_node(self, nodename): + def get_node(self, nodename: str) -> Node: """Get a node object for a specific node""" return self.nodes[nodename] - def get_node_url(self, nodename=""): + def get_node_url(self, nodename: str = "") -> str: """Return the url for nodes""" url = urlparse.urljoin( self.base_server_url(), "computer/%s" % urlquote(nodename) @@ -312,21 +313,21 @@ def get_node_url(self, nodename=""): return url def get_queue_url(self): - url = "%s/%s" % (self.base_server_url(), "queue") + url = f"{self.base_server_url()}/queue" return url - def get_queue(self): + def get_queue(self) -> Queue: queue_url = self.get_queue_url() return Queue(queue_url, self) - def get_nodes(self): + def get_nodes(self) -> Nodes: return Nodes(self.baseurl, self) @property def nodes(self): return self.get_nodes() - def has_node(self, nodename): + def has_node(self, nodename: str) -> bool: """ Does a node by the name specified exist :param nodename: string, hostname @@ -335,7 +336,7 @@ def has_node(self, nodename): self.poll() return nodename in self.nodes - def delete_node(self, nodename): + def delete_node(self, nodename: str) -> None: """ Remove a node from the managed slave list Please note that you cannot remove the master node @@ -347,13 +348,13 @@ def delete_node(self, nodename): def create_node( self, - name, - num_executors=2, - node_description=None, - remote_fs="/var/lib/jenkins", + name: str, + num_executors: int = 2, + node_description: str = "", + remote_fs: str = "/var/lib/jenkins", labels=None, - exclusive=False, - ): + exclusive: bool = False, + ) -> Node: """ Create a new JNLP slave node by name. @@ -376,7 +377,7 @@ def create_node( } return self.nodes.create_node(name, node_dict) - def create_node_with_config(self, name, config): + def create_node_with_config(self, name: str, config) -> Node | None: """ Create a new slave node with specific configuration. Config should be resemble the output of node.get_node_attributes() @@ -390,15 +391,15 @@ def create_node_with_config(self, name, config): def get_plugins_url(self, depth): # This only ever needs to work on the base object - return "%s/pluginManager/api/python?depth=%i" % (self.baseurl, depth) + return f"{self.baseurl}/pluginManager/api/python?depth={depth}" def install_plugin( self, - plugin, - restart=True, - force_restart=False, - wait_for_reboot=True, - no_reboot_warning=False, + plugin: str | Plugin, + restart: bool = True, + force_restart: bool = False, + wait_for_reboot: bool = True, + no_reboot_warning: bool = False, ): """ Install a plugin and optionally restart jenkins. @@ -423,11 +424,11 @@ def install_plugin( def install_plugins( self, plugin_list, - restart=True, - force_restart=False, - wait_for_reboot=True, - no_reboot_warning=False, - ): + restart: bool = True, + force_restart: bool = False, + wait_for_reboot: bool = True, + no_reboot_warning: bool = False, + ) -> None: """ Install a list of plugins and optionally restart jenkins. @param plugin_list: List of plugins (strings, Plugin objects or @@ -451,12 +452,12 @@ def install_plugins( def delete_plugin( self, - plugin, - restart=True, - force_restart=False, - wait_for_reboot=True, - no_reboot_warning=False, - ): + plugin: str | Plugin, + restart: bool = True, + force_restart: bool = False, + wait_for_reboot: bool = True, + no_reboot_warning: bool = False, + ) -> None: """ Delete a plugin and optionally restart jenkins. Will not delete dependencies. @@ -479,10 +480,10 @@ def delete_plugin( def delete_plugins( self, plugin_list, - restart=True, - force_restart=False, - wait_for_reboot=True, - no_reboot_warning=False, + restart: bool = True, + force_restart: bool = False, + wait_for_reboot: bool = True, + no_reboot_warning: bool = False, ): """ Delete a list of plugins and optionally restart jenkins. Will not @@ -503,7 +504,7 @@ def delete_plugins( "Please reboot manually." ) - def safe_restart(self, wait_for_reboot=True): + def safe_restart(self, wait_for_reboot: bool = True): """restarts jenkins when no jobs are running""" # NB: unlike other methods, the value of resp.status_code # here can be 503 even when everything is normal @@ -516,7 +517,7 @@ def safe_restart(self, wait_for_reboot=True): self._wait_for_reboot() return resp - def _wait_for_reboot(self): + def _wait_for_reboot(self) -> None: # We need to make sure all jobs have finished, # and that jenkins is actually restarting. # One way to be sure is to make sure jenkins is really down. @@ -561,13 +562,13 @@ def __jenkins_is_unavailable(self): # so Jenkins is likely available time.sleep(1) - def safe_exit(self, wait_for_exit=True, max_wait=360): + def safe_exit(self, wait_for_exit: bool = True, max_wait: int = 360): """ Restarts jenkins when no jobs are running, except for pipeline jobs """ # NB: unlike other methods, the value of resp.status_code # here can be 503 even when everything is normal - url = "%s/safeExit" % (self.baseurl,) + url = f"{self.baseurl}/safeExit" valid = self.requester.VALID_STATUS_CODES + [503, 500] resp = self.requester.post_and_confirm_status( url, data="", valid=valid @@ -576,12 +577,12 @@ def safe_exit(self, wait_for_exit=True, max_wait=360): self._wait_for_exit(max_wait=max_wait) return resp - def _wait_for_exit(self, max_wait=360): + def _wait_for_exit(self, max_wait: int = 360) -> None: # We need to make sure all non pipeline jobs have finished, # and that jenkins is unavailable self.__jenkins_is_unresponsive(max_wait=max_wait) - def __jenkins_is_unresponsive(self, max_wait=360): + def __jenkins_is_unresponsive(self, max_wait: int = 360): # Blocks until jenkins returns ConnectionError or JenkinsAPIException # Default wait is one hour is_alive = True @@ -634,23 +635,23 @@ def cancel_quiet_down(self): def plugins(self): return self.get_plugins() - def get_plugins(self, depth=1): + def get_plugins(self, depth: int = 1) -> Plugins: url = self.get_plugins_url(depth=depth) return Plugins(url, self) - def has_plugin(self, plugin_name): + def has_plugin(self, plugin_name: str) -> bool: return plugin_name in self.plugins - def get_executors(self, nodename): - url = "%s/computer/%s" % (self.baseurl, nodename) + def get_executors(self, nodename: str) -> Executors: + url = f"{self.baseurl}/computer/{nodename}" return Executors(url, nodename, self) def get_master_data(self): - url = "%s/computer/api/python" % self.baseurl + url = f"{self.baseurl}/computer/api/python" return self.get_data(url) @property - def version(self): + def version(self) -> str: """ Return version number of Jenkins """ @@ -667,10 +668,10 @@ def get_credentials(self, cred_class=Credentials2x): raise JenkinsAPIException("Credentials plugin not installed") if self.plugins["credentials"].version.startswith("1."): - url = "%s/credential-store/domain/_/" % self.baseurl + url = f"{self.baseurl}/credential-store/domain/_/" return Credentials(url, self) - url = "%s/credentials/store/system/domain/_/" % self.baseurl + url = f"{self.baseurl}/credentials/store/system/domain/_/" return cred_class(url, self) @property @@ -682,17 +683,17 @@ def credentials_by_id(self): return self.get_credentials(CredentialsById) @property - def is_quieting_down(self): + def is_quieting_down(self) -> bool: url = "%s/api/python?tree=quietingDown" % (self.baseurl,) data = self.get_data(url=url) return data.get("quietingDown", False) - def shutdown(self): + def shutdown(self) -> None: url = "%s/exit" % self.baseurl self.requester.post_and_confirm_status(url, data="") def generate_new_api_token( - self, new_token_name="Token By jenkinsapi python" + self, new_token_name: str = "Token By jenkinsapi python" ): subUrl = ( "/me/descriptorByName/jenkins.security." @@ -704,7 +705,7 @@ def generate_new_api_token( token = response.json()["data"]["tokenValue"] return token - def run_groovy_script(self, script): + def run_groovy_script(self, script: str) -> str: """ Runs the requested groovy script on the Jenkins server returning the result as text. @@ -718,7 +719,7 @@ def run_groovy_script(self, script): result = server.run_groovy_script(script) print(result) # will print "Hello world!" """ - url = "%s/scriptText" % self.baseurl + url = f"{self.baseurl}/scriptText" data = urlencode({"script": script}) response = self.requester.post_and_confirm_status(url, data=data) @@ -729,7 +730,7 @@ def run_groovy_script(self, script): return response.text - def use_auth_cookie(self): + def use_auth_cookie(self) -> None: assert self.username and self.baseurl, ( "Please provide jenkins url, username " "and password to get the session ID cookie." diff --git a/jenkinsapi/jenkinsbase.py b/jenkinsapi/jenkinsbase.py index 011c84ab..e79d43ae 100644 --- a/jenkinsapi/jenkinsbase.py +++ b/jenkinsapi/jenkinsbase.py @@ -1,11 +1,12 @@ """ Module for JenkinsBase class """ +from __future__ import annotations import ast import pprint import logging -from six.moves.urllib.parse import quote as urlquote +from urllib.parse import quote from jenkinsapi import config from jenkinsapi.custom_exceptions import JenkinsAPIException @@ -29,7 +30,7 @@ def __repr__(self): def __str__(self): raise NotImplementedError - def __init__(self, baseurl, poll=True): + def __init__(self, baseurl: str, poll: bool = True): """ Initialize a jenkins connection """ @@ -43,7 +44,7 @@ def get_jenkins_obj(self): "Please implement this method on %s" % self.__class__.__name__ ) - def __eq__(self, other): + def __eq__(self, other) -> bool: """ Return true if the other object represents a connection to the same server @@ -53,7 +54,7 @@ def __eq__(self, other): return other.baseurl == self.baseurl @classmethod - def strip_trailing_slash(cls, url): + def strip_trailing_slash(cls, url: str) -> str: while url.endswith("/"): url = url[:-1] return url @@ -110,23 +111,23 @@ def resolve_job_folders(self, jobs): def process_job_folder(self, folder, folder_path): logger.debug("Processing folder %s in %s", folder["name"], folder_path) - folder_path += "/job/%s" % urlquote(folder["name"]) + folder_path += "/job/%s" % quote(folder["name"]) data = self.get_data( self.python_api_url(folder_path), tree="jobs[name,color]" ) - result = [] + result = [] for job in data.get("jobs", []): if "color" not in job.keys(): result += self.process_job_folder(job, folder_path) else: - job["url"] = "%s/job/%s" % (folder_path, urlquote(job["name"])) + job["url"] = "%s/job/%s" % (folder_path, quote(job["name"])) result.append(job) return result @classmethod - def python_api_url(cls, url): + def python_api_url(cls, url: str) -> str: if url.endswith(config.JENKINS_API): return url else: diff --git a/jenkinsapi/job.py b/jenkinsapi/job.py index 7dc2af98..bf3af0f7 100644 --- a/jenkinsapi/job.py +++ b/jenkinsapi/job.py @@ -1,10 +1,12 @@ """ Module for jenkinsapi Job """ +from __future__ import annotations + import json import logging import xml.etree.ElementTree as ET -import six.moves.urllib.parse as urlparse +import urllib.parse as urlparse from collections import defaultdict from jenkinsapi.build import Build @@ -20,7 +22,6 @@ from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.mutable_jenkins_thing import MutableJenkinsThing from jenkinsapi.queue import QueueItem -from jenkinsapi.utils.compat import to_string SVN_URL = "./scm/locations/hudson.scm.SubversionSCM_-ModuleLocation/remote" @@ -40,9 +41,9 @@ class Job(JenkinsBase, MutableJenkinsThing): A job can hold N builds which are the actual execution environments """ - def __init__(self, url, name, jenkins_obj): - self.name = name - self.jenkins = jenkins_obj + def __init__(self, url: str, name: str, jenkins_obj: "Jenkins") -> None: + self.name: str = name + self.jenkins: "Jenkins" = jenkins_obj self._revmap = None self._config = None self._element_tree = None @@ -69,16 +70,16 @@ def __init__(self, url, name, jenkins_obj): "hg": self._get_hg_branch, None: lambda element_tree: [], } - self.url = url + self.url: str = url JenkinsBase.__init__(self, self.url) - def __str__(self): + def __str__(self) -> str: return self.name - def get_description(self): + def get_description(self) -> str: return self._data["description"] - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins # When the name of the hg branch used in the job is default hg branch (i.e. @@ -138,7 +139,7 @@ def _get_config_element_tree(self): self._element_tree = ET.fromstring(self._config) return self._element_tree - def get_build_triggerurl(self): + def get_build_triggerurl(self) -> str: if not self.has_params(): return "%s/build" % self.baseurl return "%s/buildWithParameters" % self.baseurl @@ -154,8 +155,7 @@ def _mk_json_from_build_parameters(build_params, file_params=None): raise ValueError("Build parameters must be a dict") build_p = [ - {"name": k, "value": to_string(v)} - for k, v in sorted(build_params.items()) + {"name": k, "value": v} for k, v in sorted(build_params.items()) ] out = {"parameter": build_p} @@ -180,13 +180,13 @@ def mk_json_from_build_parameters(build_params, file_params=None): def invoke( self, securitytoken=None, - block=False, + block: bool = False, build_params=None, cause=None, files=None, - delay=5, + delay: int = 5, quiet_period=None, - ): + ) -> QueueItem: assert isinstance(block, bool) if build_params and (not self.has_params()): raise BadParams("This job does not support parameters") diff --git a/jenkinsapi/jobs.py b/jenkinsapi/jobs.py index e2d9f803..60a612c4 100644 --- a/jenkinsapi/jobs.py +++ b/jenkinsapi/jobs.py @@ -2,8 +2,12 @@ This module implements the Jobs class, which is intended to be a container-like interface for all of the jobs defined on a single Jenkins server. """ +from __future__ import annotations + +from typing import Iterator import logging import time + from jenkinsapi.job import Job from jenkinsapi.custom_exceptions import JenkinsAPIException, UnknownJob @@ -19,11 +23,11 @@ class Jobs(object): jenkinsapi.Job objects. """ - def __init__(self, jenkins): + def __init__(self, jenkins: "Jenkins") -> None: self.jenkins = jenkins self._data = [] - def _del_data(self, job_name): + def _del_data(self, job_name: str) -> None: if not self._data: return for num, job_data in enumerate(self._data): @@ -31,13 +35,13 @@ def _del_data(self, job_name): del self._data[num] return - def __len__(self): + def __len__(self) -> int: return len(self.keys()) def poll(self, tree="jobs[name,color,url]"): return self.jenkins.poll(tree=tree) - def __delitem__(self, job_name): + def __delitem__(self, job_name: str) -> None: """ Delete a job by name :param str job_name: name of a existing job @@ -62,7 +66,7 @@ def __delitem__(self, job_name): ) self._del_data(job_name) - def __setitem__(self, key, value): + def __setitem__(self, key: str, value: str) -> "Job": """ Create Job @@ -75,7 +79,7 @@ def __setitem__(self, key, value): """ return self.create(key, value) - def __getitem__(self, job_name): + def __getitem__(self, job_name: str) -> "Job": if job_name in self: job_data = [ job_row @@ -90,7 +94,7 @@ def __getitem__(self, job_name): else: raise UnknownJob(job_name) - def iteritems(self): + def iteritems(self) -> Iterator[str, "Job"]: """ Iterate over the names & objects for all jobs """ @@ -100,13 +104,13 @@ def iteritems(self): else: yield job.name, job - def __contains__(self, job_name): + def __contains__(self, job_name: str) -> bool: """ True if job_name exists in Jenkins """ return job_name in self.keys() - def iterkeys(self): + def iterkeys(self) -> Iterator[str]: """ Iterate over the names of all available jobs """ @@ -122,7 +126,7 @@ def iterkeys(self): else: yield row["name"] - def itervalues(self): + def itervalues(self) -> Iterator["Job"]: """ Iterate over all available jobs """ @@ -131,13 +135,13 @@ def itervalues(self): for row in self._data: yield Job(row["url"], row["name"], self.jenkins) - def keys(self): + def keys(self) -> list[str]: """ Return a list of the names of all jobs """ return list(self.iterkeys()) - def create(self, job_name, config): + def create(self, job_name: str, config: str | bytes) -> "Job": """ Create a job @@ -152,14 +156,9 @@ def create(self, job_name, config): raise JenkinsAPIException("Job XML config cannot be empty") params = {"name": job_name} - try: - if isinstance( - config, unicode - ): # pylint: disable=undefined-variable - config = str(config) - except NameError: - # Python2 already a str - pass + if isinstance(config, bytes): + config = config.decode("utf-8") + self.jenkins.requester.post_xml_and_confirm_status( self.jenkins.get_create_url(), data=config, params=params ) @@ -169,8 +168,8 @@ def create(self, job_name, config): return self[job_name] def create_multibranch_pipeline( - self, job_name, config, block=True, delay=60 - ): + self, job_name: str, config: str, block: bool = True, delay: int = 60 + ) -> list["Job"]: """ Create a multibranch pipeline job @@ -184,14 +183,9 @@ def create_multibranch_pipeline( raise JenkinsAPIException("Job XML config cannot be empty") params = {"name": job_name} - try: - if isinstance( - config, unicode - ): # pylint: disable=undefined-variable - config = str(config) - except NameError: - # Python2 already a str - pass + if isinstance(config, bytes): + config = config.decode("utf-8") + self.jenkins.requester.post_xml_and_confirm_status( self.jenkins.get_create_url(), data=config, params=params ) @@ -233,7 +227,7 @@ def create_multibranch_pipeline( return jobs - def copy(self, job_name, new_job_name): + def copy(self, job_name: str, new_job_name: str) -> "Job": """ Copy a job :param str job_name: Name of an existing job @@ -250,7 +244,7 @@ def copy(self, job_name, new_job_name): return self[new_job_name] - def rename(self, job_name, new_job_name): + def rename(self, job_name: str, new_job_name: str) -> "Job": """ Rename a job @@ -268,7 +262,7 @@ def rename(self, job_name, new_job_name): return self[new_job_name] - def build(self, job_name, params=None, **kwargs): + def build(self, job_name: str, params=None, **kwargs) -> "QueueItem": """ Executes build of a job diff --git a/jenkinsapi/mutable_jenkins_thing.py b/jenkinsapi/mutable_jenkins_thing.py index f19cf9b4..471a6046 100644 --- a/jenkinsapi/mutable_jenkins_thing.py +++ b/jenkinsapi/mutable_jenkins_thing.py @@ -9,8 +9,8 @@ class MutableJenkinsThing(object): A mixin for certain mutable objects which can be renamed and deleted. """ - def get_delete_url(self): - return "%s/doDelete" % self.baseurl + def get_delete_url(self) -> str: + return f"{self.baseurl}/doDelete" - def get_rename_url(self): - return "%s/doRename" % self.baseurl + def get_rename_url(self) -> str: + return f"{self.baseurl}/doRename" diff --git a/jenkinsapi/node.py b/jenkinsapi/node.py index 9eec932c..2c64cac0 100644 --- a/jenkinsapi/node.py +++ b/jenkinsapi/node.py @@ -1,6 +1,8 @@ """ Module for jenkinsapi Node class """ +from __future__ import annotations + import json import logging @@ -10,7 +12,7 @@ from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.custom_exceptions import PostRequired, TimeOut from jenkinsapi.custom_exceptions import JenkinsAPIException -from six.moves.urllib.parse import quote as urlquote +from urllib.parse import quote as urlquote log = logging.getLogger(__name__) @@ -22,7 +24,14 @@ class Node(JenkinsBase): to the master jenkins instance """ - def __init__(self, jenkins_obj, baseurl, nodename, node_dict, poll=True): + def __init__( + self, + jenkins_obj: "Jenkins", + baseurl: str, + nodename: str, + node_dict, + poll: bool = True, + ) -> None: """ Init a node object by providing all relevant pointers to it :param jenkins_obj: ref to the jenkins obj @@ -89,17 +98,17 @@ def __init__(self, jenkins_obj, baseurl, nodename, node_dict, poll=True): :return: None :return: Node obj """ - self.name = nodename - self.jenkins = jenkins_obj + self.name: str = nodename + self.jenkins: "Jenkins" = jenkins_obj if not baseurl: poll = False - baseurl = "%s/computer/%s" % (self.jenkins.baseurl, self.name) + baseurl = f"{self.jenkins.baseurl}/computer/{self.name}" JenkinsBase.__init__(self, baseurl, poll=poll) - self.node_attributes = node_dict + self.node_attributes: dict = node_dict self._element_tree = None self._config = None - def get_node_attributes(self): + def get_node_attributes(self) -> dict: """ Gets node attributes as dict @@ -108,7 +117,7 @@ def get_node_attributes(self): :return: Node attributes dict formatted for Jenkins API request to create node """ - na = self.node_attributes + na: dict = self.node_attributes if not na.get("credential_description", False): # If credentials description is not present - we will create # JNLP node @@ -124,8 +133,12 @@ def get_node_attributes(self): " not found" % na["credential_description"] ) - retries = na["max_num_retries"] if "max_num_retries" in na else "" - re_wait = na["retry_wait_time"] if "retry_wait_time" in na else "" + retries: int = ( + na["max_num_retries"] if "max_num_retries" in na else 0 + ) + re_wait: int = ( + na["retry_wait_time"] if "retry_wait_time" in na else 0 + ) launcher = { "stapler-class": "hudson.plugins.sshslaves.SSHLauncher", "$class": "hudson.plugins.sshslaves.SSHLauncher", @@ -152,7 +165,7 @@ def get_node_attributes(self): "idleDelay": na["ondemand_idle_delay"], } - node_props = {"stapler-class-bag": "true"} + node_props: dict = {"stapler-class-bag": "true"} if "env" in na: node_props.update( { @@ -191,25 +204,25 @@ def get_node_attributes(self): return params - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def __str__(self): + def __str__(self) -> str: return self.name - def is_online(self): + def is_online(self) -> bool: return not self.poll(tree="offline")["offline"] - def is_temporarily_offline(self): + def is_temporarily_offline(self) -> bool: return self.poll(tree="temporarilyOffline")["temporarilyOffline"] - def is_jnlpagent(self): + def is_jnlpagent(self) -> bool: return self._data["jnlpAgent"] - def is_idle(self): + def is_idle(self) -> bool: return self.poll(tree="idle")["idle"] - def set_online(self): + def set_online(self) -> None: """ Set node online. Before change state verify client state: if node set 'offline' @@ -239,7 +252,7 @@ def set_online(self): % (self._data["offline"], self._data["temporarilyOffline"]) ) - def set_offline(self, message="requested from jenkinsapi"): + def set_offline(self, message="requested from jenkinsapi") -> None: """ Set node offline. If after run node state has not been changed raise AssertionError. @@ -256,7 +269,9 @@ def set_offline(self, message="requested from jenkinsapi"): % (data["offline"], data["temporarilyOffline"]) ) - def toggle_temporarily_offline(self, message="requested from jenkinsapi"): + def toggle_temporarily_offline( + self, message="requested from jenkinsapi" + ) -> None: """ Switches state of connected node (online/offline) and set 'temporarilyOffline' property (True/False) @@ -284,7 +299,7 @@ def toggle_temporarily_offline(self, message="requested from jenkinsapi"): % state ) - def update_offline_reason(self, reason): + def update_offline_reason(self, reason: str) -> None: """ Update offline reason on a temporary offline clsuter """ @@ -297,14 +312,14 @@ def update_offline_reason(self, reason): ) self.jenkins.requester.post_and_confirm_status(url, data={}) - def offline_reason(self): + def offline_reason(self) -> str: return self._data["offlineCauseReason"] @property def _et(self): return self._get_config_element_tree() - def _get_config_element_tree(self): + def _get_config_element_tree(self) -> ET.Element: """ Returns an xml element tree for the node's config.xml. The resulting tree is cached for quick lookup. @@ -316,7 +331,7 @@ def _get_config_element_tree(self): self._element_tree = ET.fromstring(self._config) return self._element_tree - def get_config(self): + def get_config(self) -> str: """ Returns the config.xml from the node. """ @@ -325,7 +340,7 @@ def get_config(self): ) return response.text - def load_config(self): + def load_config(self) -> None: """ Loads the config.xml for the node allowing it to be re-queried without generating new requests. @@ -336,7 +351,7 @@ def load_config(self): self._config = self.get_config() self._get_config_element_tree() - def upload_config(self, config_xml): + def upload_config(self, config_xml: str) -> None: """ Uploads config_xml to the config.xml for the node. """ @@ -347,20 +362,20 @@ def upload_config(self, config_xml): "%(baseurl)s/config.xml" % self.__dict__, data=config_xml ) - def get_labels(self): + def get_labels(self) -> str | None: """ Returns the labels for a slave as a string with each label separated by the ' ' character. """ return self.get_config_element("label") - def get_num_executors(self): + def get_num_executors(self) -> str: try: return self.get_config_element("numExecutors") except JenkinsAPIException: return self._data["numExecutors"] - def set_num_executors(self, value): + def set_num_executors(self, value: int | str) -> None: """ Sets number of executors for node @@ -388,7 +403,7 @@ def set_num_executors(self, value): self.poll() - def get_config_element(self, el_name): + def get_config_element(self, el_name: str) -> str: """ Returns simple config element. @@ -396,7 +411,7 @@ def get_config_element(self, el_name): """ return self._et.find(el_name).text - def set_config_element(self, el_name, value): + def set_config_element(self, el_name: str, value: str) -> None: """ Sets simple config element """ @@ -404,7 +419,7 @@ def set_config_element(self, el_name, value): xml_str = ET.tostring(self._et) self.upload_config(xml_str) - def get_monitor(self, monitor_name, poll_monitor=True): + def get_monitor(self, monitor_name: str, poll_monitor=True) -> str: """ Polls the node returning one of the monitors in the monitorData branch of the returned node api tree. @@ -422,70 +437,70 @@ def get_monitor(self, monitor_name, poll_monitor=True): return monitor_data[full_monitor_name] - def get_available_physical_memory(self): + def get_available_physical_memory(self) -> int: """ Returns the node's available physical memory in bytes. """ monitor_data = self.get_monitor("SwapSpaceMonitor") return monitor_data["availablePhysicalMemory"] - def get_available_swap_space(self): + def get_available_swap_space(self) -> int: """ Returns the node's available swap space in bytes. """ monitor_data = self.get_monitor("SwapSpaceMonitor") return monitor_data["availableSwapSpace"] - def get_total_physical_memory(self): + def get_total_physical_memory(self) -> int: """ Returns the node's total physical memory in bytes. """ monitor_data = self.get_monitor("SwapSpaceMonitor") return monitor_data["totalPhysicalMemory"] - def get_total_swap_space(self): + def get_total_swap_space(self) -> int: """ Returns the node's total swap space in bytes. """ monitor_data = self.get_monitor("SwapSpaceMonitor") return monitor_data["totalSwapSpace"] - def get_workspace_path(self): + def get_workspace_path(self) -> str: """ Returns the local path to the node's Jenkins workspace directory. """ monitor_data = self.get_monitor("DiskSpaceMonitor") return monitor_data["path"] - def get_workspace_size(self): + def get_workspace_size(self) -> int: """ Returns the size in bytes of the node's Jenkins workspace directory. """ monitor_data = self.get_monitor("DiskSpaceMonitor") return monitor_data["size"] - def get_temp_path(self): + def get_temp_path(self) -> str: """ Returns the local path to the node's temp directory. """ monitor_data = self.get_monitor("TemporarySpaceMonitor") return monitor_data["path"] - def get_temp_size(self): + def get_temp_size(self) -> int: """ Returns the size in bytes of the node's temp directory. """ monitor_data = self.get_monitor("TemporarySpaceMonitor") return monitor_data["size"] - def get_architecture(self): + def get_architecture(self) -> str: """ Returns the system architecture of the node eg. "Linux (amd64)". """ # no need to poll as the architecture will never change return self.get_monitor("ArchitectureMonitor", poll_monitor=False) - def block_until_idle(self, timeout, poll_time=5): + def block_until_idle(self, timeout: int, poll_time: int = 5) -> None: """ Blocks until the node become idle. :param timeout: Time in second when the wait is aborted. @@ -507,14 +522,14 @@ def block_until_idle(self, timeout, poll_time=5): ) ) - def get_response_time(self): + def get_response_time(self) -> int: """ Returns the node's average response time. """ monitor_data = self.get_monitor("ResponseTimeMonitor") return monitor_data["average"] - def get_clock_difference(self): + def get_clock_difference(self) -> int: """ Returns the difference between the node's clock and the master Jenkins clock. diff --git a/jenkinsapi/nodes.py b/jenkinsapi/nodes.py index a495d09b..46ed0e42 100644 --- a/jenkinsapi/nodes.py +++ b/jenkinsapi/nodes.py @@ -1,16 +1,20 @@ """ Module for jenkinsapi nodes """ +from __future__ import annotations + +from typing import Iterator + import logging -from six.moves.urllib.parse import urlencode +from urllib.parse import urlencode from jenkinsapi.node import Node from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.custom_exceptions import JenkinsAPIException from jenkinsapi.custom_exceptions import UnknownNode from jenkinsapi.custom_exceptions import PostRequired -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class Nodes(JenkinsBase): @@ -19,7 +23,7 @@ class Nodes(JenkinsBase): Class to hold information on a collection of nodes """ - def __init__(self, baseurl, jenkins_obj): + def __init__(self, baseurl: str, jenkins_obj: "Jenkins") -> None: """ Handy access to all of the nodes on your Jenkins server """ @@ -31,16 +35,16 @@ def __init__(self, baseurl, jenkins_obj): else baseurl.rstrip("/") + "/computer", ) - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def __str__(self): + def __str__(self) -> str: return "Nodes @ %s" % self.baseurl - def __contains__(self, node_name): + def __contains__(self, node_name: str) -> bool: return node_name in self.keys() - def iterkeys(self): + def iterkeys(self) -> Iterator[str]: """ Return an iterator over the container's node names. @@ -50,13 +54,13 @@ def iterkeys(self): for item in self._data["computer"]: yield item["displayName"] - def keys(self): + def keys(self) -> list[str]: """ Return a copy of the container's list of node names. """ return list(self.iterkeys()) - def _make_node(self, nodename): + def _make_node(self, nodename) -> Node: """ Creates an instance of Node for the given nodename. This function assumes the returned node exists. @@ -67,7 +71,7 @@ def _make_node(self, nodename): nodeurl = "%s/%s" % (self.baseurl, nodename) return Node(self.jenkins, nodeurl, nodename, node_dict={}) - def iteritems(self): + def iteritems(self) -> Iterator[tuple[str, Node]]: """ Return an iterator over the container's (name, node) pairs. @@ -81,13 +85,13 @@ def iteritems(self): except Exception: raise JenkinsAPIException("Unable to iterate nodes") - def items(self): + def items(self) -> list[tuple[str, Node]]: """ Return a copy of the container's list of (name, node) pairs. """ return list(self.iteritems()) - def itervalues(self): + def itervalues(self) -> Iterator[Node]: """ Return an iterator over the container's nodes. @@ -100,21 +104,21 @@ def itervalues(self): except Exception: raise JenkinsAPIException("Unable to iterate nodes") - def values(self): + def values(self) -> list[Node]: """ Return a copy of the container's list of nodes. """ return list(self.itervalues()) - def __getitem__(self, nodename): + def __getitem__(self, nodename: str) -> Node: if nodename in self: return self._make_node(nodename) raise UnknownNode(nodename) - def __len__(self): + def __len__(self) -> int: return len(self.keys()) - def __delitem__(self, item): + def __delitem__(self, item: str) -> None: if item in self and item != "Built-In Node": url = "%s/doDelete" % self[item].baseurl try: @@ -129,14 +133,14 @@ def __delitem__(self, item): log.info("Requests to remove built-in node ignored") - def __setitem__(self, name, node_dict): + def __setitem__(self, name: str, node_dict: dict): if not isinstance(node_dict, dict): raise ValueError('"node_dict" parameter must be a Node dict') if name not in self: self.create_node(name, node_dict) self.poll() - def create_node(self, name, node_dict): + def create_node(self, name: str, node_dict: dict) -> Node: """ Create a new slave node @@ -149,7 +153,7 @@ def create_node(self, name, node_dict): node = Node( jenkins_obj=self.jenkins, - baseurl=None, + baseurl="", nodename=name, node_dict=node_dict, poll=False, @@ -164,7 +168,7 @@ def create_node(self, name, node_dict): self.poll() return self[name] - def create_node_with_config(self, name, config): + def create_node_with_config(self, name: str, config: dict) -> Node | None: """ Create a new slave node with specific configuration. Config should be resemble the output of node.get_node_attributes() diff --git a/jenkinsapi/plugin.py b/jenkinsapi/plugin.py index 920e056b..0ca8b219 100644 --- a/jenkinsapi/plugin.py +++ b/jenkinsapi/plugin.py @@ -1,6 +1,9 @@ """ Module for jenkinsapi Plugin """ +from __future__ import annotations + +from typing import Union class Plugin(object): @@ -9,16 +12,18 @@ class Plugin(object): Plugin class """ - def __init__(self, plugin_dict): + def __init__(self, plugin_dict: Union[dict, str]) -> None: if isinstance(plugin_dict, dict): self.__dict__ = plugin_dict else: self.__dict__ = self.to_plugin(plugin_dict) + self.shortName: str = self.__dict__["shortName"] + self.version: str = self.__dict__.get("version", "Unknown") - def to_plugin(self, plugin_string): + def to_plugin(self, plugin_string: str) -> dict: plugin_string = str(plugin_string) if "@" not in plugin_string or len(plugin_string.split("@")) != 2: - usage_err = ( + usage_err: str = ( "plugin specification must be a string like " '"plugin-name@version", not "{0}"' ) @@ -28,20 +33,20 @@ def to_plugin(self, plugin_string): shortName, version = plugin_string.split("@") return {"shortName": shortName, "version": version} - def __eq__(self, other): + def __eq__(self, other) -> bool: return self.__dict__ == other.__dict__ - def __str__(self): + def __str__(self) -> str: return self.shortName - def __repr__(self): + def __repr__(self) -> str: return "<%s.%s %s>" % ( self.__class__.__module__, self.__class__.__name__, str(self), ) - def get_attributes(self): + def get_attributes(self) -> str: """ Used by Plugins object to install plugins in Jenkins """ @@ -50,7 +55,7 @@ def get_attributes(self): self.version, ) - def is_latest(self, update_center_dict): + def is_latest(self, update_center_dict: dict) -> bool: """ Used by Plugins object to determine if plugin can be installed through the update center (when plugin version is @@ -62,7 +67,7 @@ def is_latest(self, update_center_dict): center_plugin = update_center_dict["plugins"][self.shortName] return center_plugin["version"] == self.version - def get_download_link(self, update_center_dict): + def get_download_link(self, update_center_dict) -> str: latest_version = update_center_dict["plugins"][self.shortName][ "version" ] diff --git a/jenkinsapi/plugins.py b/jenkinsapi/plugins.py index a81e15a6..1247ad1b 100644 --- a/jenkinsapi/plugins.py +++ b/jenkinsapi/plugins.py @@ -1,30 +1,26 @@ """ jenkinsapi plugins """ -from __future__ import print_function +from __future__ import annotations +from typing import Generator import logging import time import re - -try: - from StringIO import StringIO - from urllib import urlencode -except ImportError: - # Python3 - from io import BytesIO as StringIO - from urllib.parse import urlencode +from io import BytesIO +from urllib.parse import urlencode import json + import requests from jenkinsapi.plugin import Plugin from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.custom_exceptions import UnknownPlugin from jenkinsapi.custom_exceptions import JenkinsAPIException from jenkinsapi.utils.jsonp_to_json import jsonp_to_json -from jenkinsapi.utils.manifest import read_manifest +from jenkinsapi.utils.manifest import Manifest, read_manifest -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class Plugins(JenkinsBase): @@ -33,16 +29,17 @@ class Plugins(JenkinsBase): Plugins class for jenkinsapi """ - def __init__(self, url, jenkins_obj): - self.jenkins_obj = jenkins_obj + def __init__(self, url: str, jenkins_obj: "Jenkins") -> None: + self.jenkins_obj: "Jenkins" = jenkins_obj JenkinsBase.__init__(self, url) - # print('DEBUG: Plugins._data=', self._data) - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins_obj - def check_updates_server(self): - url = "%s/pluginManager/checkUpdatesServer" % self.jenkins_obj.baseurl + def check_updates_server(self) -> None: + url: str = ( + f"{self.jenkins_obj.baseurl}/pluginManager/checkUpdatesServer" + ) self.jenkins_obj.requester.post_and_confirm_status( url, params={}, data={} ) @@ -56,35 +53,35 @@ def update_center_dict(self): def _poll(self, tree=None): return self.get_data(self.baseurl, tree=tree) - def keys(self): + def keys(self) -> list[str]: return self.get_plugins_dict().keys() __iter__ = keys - def iteritems(self): + def iteritems(self) -> Generator[str, "Plugin"]: return self._get_plugins() - def values(self): + def values(self) -> list["Plugin"]: return [a[1] for a in self.iteritems()] - def _get_plugins(self): + def _get_plugins(self) -> Generator[str, "Plugin"]: if "plugins" in self._data: for p_dict in self._data["plugins"]: yield p_dict["shortName"], Plugin(p_dict) - def get_plugins_dict(self): + def get_plugins_dict(self) -> dict[str, "Plugin"]: return dict(self._get_plugins()) - def __len__(self): + def __len__(self) -> int: return len(self.get_plugins_dict().keys()) - def __getitem__(self, plugin_name): + def __getitem__(self, plugin_name: str) -> Plugin: try: return self.get_plugins_dict()[plugin_name] except KeyError: raise UnknownPlugin(plugin_name) - def __setitem__(self, shortName, plugin): + def __setitem__(self, shortName, plugin: "Plugin") -> None: """ Installs plugin in Jenkins. @@ -103,13 +100,13 @@ def __setitem__(self, shortName, plugin): self._install_specific_version(plugin) self._wait_until_plugin_installed(plugin) - def _install_plugin_from_updatecenter(self, plugin): + def _install_plugin_from_updatecenter(self, plugin: "Plugin") -> None: """ Latest versions of plugins can be installed from the update center (and don't need a restart.) """ - xml_str = plugin.get_attributes() - url = ( + xml_str: str = plugin.get_attributes() + url: str = ( "%s/pluginManager/installNecessaryPlugins" % self.jenkins_obj.baseurl ) @@ -122,7 +119,7 @@ def update_center_install_status(self): """ Jenkins 2.x specific """ - url = "%s/updateCenter/installStatus" % self.jenkins_obj.baseurl + url: str = "%s/updateCenter/installStatus" % self.jenkins_obj.baseurl status = self.jenkins_obj.requester.get_url(url) if status.status_code == 404: raise JenkinsAPIException( @@ -141,14 +138,14 @@ def restart_required(self): return True # Jenkins 1.X has no update_center return any([job for job in jobs if job["requiresRestart"] == "true"]) - def _install_specific_version(self, plugin): + def _install_specific_version(self, plugin: "Plugin") -> None: """ Plugins that are not the latest version have to be uploaded. """ - download_link = plugin.get_download_link( + download_link: str = plugin.get_download_link( update_center_dict=self.update_center_dict ) - downloaded_plugin = self._download_plugin(download_link) + downloaded_plugin: BytesIO = self._download_plugin(download_link) plugin_dependencies = self._get_plugin_dependencies(downloaded_plugin) log.debug("Installing dependencies for plugin '%s'", plugin.shortName) self.jenkins_obj.install_plugins(plugin_dependencies) @@ -162,12 +159,14 @@ def _install_specific_version(self, plugin): params={}, ) - def _get_plugin_dependencies(self, downloaded_plugin): + def _get_plugin_dependencies( + self, downloaded_plugin: BytesIO + ) -> list["Plugin"]: """ Returns a list of all dependencies for a downloaded plugin """ plugin_dependencies = [] - manifest = read_manifest(downloaded_plugin) + manifest: Manifest = read_manifest(downloaded_plugin) manifest_dependencies = manifest.main_section.get( "Plugin-Dependencies" ) @@ -186,11 +185,11 @@ def _get_plugin_dependencies(self, downloaded_plugin): return plugin_dependencies def _download_plugin(self, download_link): - downloaded_plugin = StringIO() + downloaded_plugin = BytesIO() downloaded_plugin.write(requests.get(download_link).content) return downloaded_plugin - def _plugin_has_finished_installation(self, plugin): + def _plugin_has_finished_installation(self, plugin) -> bool: """ Return True if installation is marked as 'Success' or 'SuccessButRequiresRestart' in Jenkins' update_center, @@ -210,7 +209,7 @@ def _plugin_has_finished_installation(self, plugin): except JenkinsAPIException: return False # lack of update_center in Jenkins 1.X - def plugin_version_is_being_installed(self, plugin): + def plugin_version_is_being_installed(self, plugin) -> bool: """ Return true if plugin is currently being installed. """ @@ -227,7 +226,7 @@ def plugin_version_is_being_installed(self, plugin): ] ) - def plugin_version_already_installed(self, plugin): + def plugin_version_already_installed(self, plugin) -> bool: """ Check if plugin version is already installed """ diff --git a/jenkinsapi/queue.py b/jenkinsapi/queue.py index e0df94cd..60fc313a 100644 --- a/jenkinsapi/queue.py +++ b/jenkinsapi/queue.py @@ -1,13 +1,16 @@ """ Queue module for jenkinsapi """ +from __future__ import annotations + +from typing import Iterator, Tuple import logging import time from requests import HTTPError from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.custom_exceptions import UnknownQueueItem, NotBuiltYet -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class Queue(JenkinsBase): @@ -16,22 +19,22 @@ class Queue(JenkinsBase): Class that represents the Jenkins queue """ - def __init__(self, baseurl, jenkins_obj): + def __init__(self, baseurl: str, jenkins_obj: "Jenkins") -> None: """ Init the Jenkins queue object :param baseurl: basic url for the queue :param jenkins_obj: ref to the jenkins obj """ - self.jenkins = jenkins_obj + self.jenkins: "Jenkins" = jenkins_obj JenkinsBase.__init__(self, baseurl) - def __str__(self): + def __str__(self) -> str: return self.baseurl - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def iteritems(self): + def iteritems(self) -> Iterator[Tuple[str, "QueueItem"]]: for item in self._data["items"]: queue_id = item["id"] item_baseurl = "%s/item/%i" % (self.baseurl, queue_id) @@ -39,48 +42,48 @@ def iteritems(self): baseurl=item_baseurl, jenkins_obj=self.jenkins ) - def iterkeys(self): + def iterkeys(self) -> Iterator[str]: for item in self._data["items"]: yield item["id"] - def itervalues(self): + def itervalues(self) -> Iterator["QueueItem"]: for item in self._data["items"]: yield QueueItem(self.jenkins, **item) - def keys(self): + def keys(self) -> list[str]: return list(self.iterkeys()) - def values(self): + def values(self) -> list["QueueItem"]: return list(self.itervalues()) - def __len__(self): + def __len__(self) -> int: return len(self._data["items"]) - def __getitem__(self, item_id): + def __getitem__(self, item_id: str) -> "QueueItem": self_as_dict = dict(self.iteritems()) if item_id in self_as_dict: return self_as_dict[item_id] else: raise UnknownQueueItem(item_id) - def _get_queue_items_for_job(self, job_name): + def _get_queue_items_for_job(self, job_name: str) -> Iterator["QueueItem"]: for item in self._data["items"]: if "name" in item["task"] and item["task"]["name"] == job_name: yield QueueItem( self.get_queue_item_url(item), jenkins_obj=self.jenkins ) - def get_queue_items_for_job(self, job_name): + def get_queue_items_for_job(self, job_name: str): return list(self._get_queue_items_for_job(job_name)) - def get_queue_item_url(self, item): + def get_queue_item_url(self, item: str) -> str: return "%s/item/%i" % (self.baseurl, item["id"]) - def delete_item(self, queue_item): + def delete_item(self, queue_item: "QueueItem"): self.delete_item_by_id(queue_item.queue_id) - def delete_item_by_id(self, item_id): - deleteurl = "%s/cancelItem?id=%s" % (self.baseurl, item_id) + def delete_item_by_id(self, item_id: str): + deleteurl: str = "%s/cancelItem?id=%s" % (self.baseurl, item_id) self.get_jenkins_obj().requester.post_url(deleteurl) @@ -88,8 +91,8 @@ class QueueItem(JenkinsBase): """An individual item in the queue""" - def __init__(self, baseurl, jenkins_obj): - self.jenkins = jenkins_obj + def __init__(self, baseurl: str, jenkins_obj: "Jenkins") -> None: + self.jenkins: "Jenkins" = jenkins_obj JenkinsBase.__init__(self, baseurl) @property @@ -104,10 +107,10 @@ def name(self): def why(self): return self._data.get("why") - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins - def get_job(self): + def get_job(self) -> "Job": """ Return the job associated with this queue item """ @@ -127,17 +130,17 @@ def get_parameters(self): ) return [] - def __repr__(self): + def __repr__(self) -> str: return "<%s.%s %s>" % ( self.__class__.__module__, self.__class__.__name__, str(self), ) - def __str__(self): + def __str__(self) -> str: return "%s Queue #%i" % (self.name, self.queue_id) - def get_build(self): + def get_build(self) -> "Build": build_number = self.get_build_number() job = self.get_job() return job[build_number] @@ -159,28 +162,29 @@ def block_until_building(self, delay=5): time.sleep(delay) continue - def is_running(self): + def is_running(self) -> bool: """Return True if this queued item is running.""" try: return self.get_build().is_running() except NotBuiltYet: return False - def is_queued(self): + def is_queued(self) -> bool: """Return True if this queued item is queued.""" try: self.get_build() - return False except NotBuiltYet: return True + else: + return False - def get_build_number(self): + def get_build_number(self) -> int: try: return self._data["executable"]["number"] except (KeyError, TypeError): raise NotBuiltYet() - def get_job_name(self): + def get_job_name(self) -> str: try: return self._data["task"]["name"] except KeyError: diff --git a/jenkinsapi/result.py b/jenkinsapi/result.py index 747f8ae3..f3501325 100644 --- a/jenkinsapi/result.py +++ b/jenkinsapi/result.py @@ -13,16 +13,16 @@ def __init__(self, **kwargs): self.__dict__.update(kwargs) def __str__(self): - return "%s %s %s" % (self.className, self.name, self.status) + return f"{self.className} {self.name} {self.status}" - def __repr__(self): + def __repr__(self) -> str: module_name = self.__class__.__module__ class_name = self.__class__.__name__ self_str = str(self) return "<%s.%s %s>" % (module_name, class_name, self_str) - def identifier(self): + def identifier(self) -> str: """ Calculate an ID for this object. """ - return "%s.%s" % (self.className, self.name) + return f"{self.className}.{self.name}" diff --git a/jenkinsapi/result_set.py b/jenkinsapi/result_set.py index 19aa5566..32e9b3ac 100644 --- a/jenkinsapi/result_set.py +++ b/jenkinsapi/result_set.py @@ -1,6 +1,7 @@ """ Module for jenkinsapi ResultSet """ +from __future__ import annotations from jenkinsapi.jenkinsbase import JenkinsBase from jenkinsapi.result import Result @@ -12,26 +13,26 @@ class ResultSet(JenkinsBase): Represents a result from a completed Jenkins run. """ - def __init__(self, url, build): + def __init__(self, url: str, build: "Build") -> None: """ Init a resultset :param url: url for a build, str :param build: build obj """ - self.build = build + self.build: "Build" = build JenkinsBase.__init__(self, url) - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.build.job.get_jenkins_obj() - def __str__(self): + def __str__(self) -> str: return "Test Result for %s" % str(self.build) @property def name(self): return str(self) - def keys(self): + def keys(self) -> list[str]: return [a[0] for a in self.iteritems()] def items(self): diff --git a/jenkinsapi/utils/compat.py b/jenkinsapi/utils/compat.py deleted file mode 100644 index 72b3e752..00000000 --- a/jenkinsapi/utils/compat.py +++ /dev/null @@ -1,28 +0,0 @@ -""" -Module for Python 2 and Python 3 compatibility -""" -import six -import sys - - -if sys.version_info[0] >= 3: - unicode = str - - -def needs_encoding(data): - """ - Check whether data is Python 2 unicode variable and needs to be - encoded - """ - if six.PY2 and isinstance(data, unicode): - return True - return False - - -def to_string(data, encoding="utf-8"): - """ - Return string representation for the data. In case of Python 2 and unicode - do additional encoding before - """ - encoded_text = data.encode(encoding) if needs_encoding(data) else data - return str(encoded_text) diff --git a/jenkinsapi/utils/jenkins_launcher.py b/jenkinsapi/utils/jenkins_launcher.py index 3bc7703e..3769194f 100644 --- a/jenkinsapi/utils/jenkins_launcher.py +++ b/jenkinsapi/utils/jenkins_launcher.py @@ -6,14 +6,15 @@ import tempfile import posixpath import requests -from requests.adapters import HTTPAdapter -from urllib3 import Retry +import queue import threading import subprocess from pkg_resources import resource_stream +from urllib3 import Retry +from urllib.parse import urlparse from tarfile import TarFile -from six.moves import queue -from six.moves.urllib.parse import urlparse + +from requests.adapters import HTTPAdapter from jenkinsapi.jenkins import Jenkins from jenkinsapi.custom_exceptions import JenkinsAPIException diff --git a/jenkinsapi/utils/krb_requester.py b/jenkinsapi/utils/krb_requester.py index 5d553417..b826730f 100644 --- a/jenkinsapi/utils/krb_requester.py +++ b/jenkinsapi/utils/krb_requester.py @@ -31,7 +31,6 @@ def __init__(self, *args, **kwargs): def get_request_dict( self, params=None, data=None, files=None, headers=None, **kwargs ): - req_dict = super(KrbRequester, self).get_request_dict( params=params, data=data, files=files, headers=headers, **kwargs ) diff --git a/jenkinsapi/utils/requester.py b/jenkinsapi/utils/requester.py index 4e16c6e2..1fa030ab 100644 --- a/jenkinsapi/utils/requester.py +++ b/jenkinsapi/utils/requester.py @@ -2,7 +2,7 @@ Module for jenkinsapi requester (which is a wrapper around python-requests) """ import requests -import six.moves.urllib.parse as urlparse +import urllib.parse as urlparse from jenkinsapi.custom_exceptions import JenkinsAPIException, PostRequired @@ -43,7 +43,6 @@ class with one of your own implementation if you require some other AUTH_COOKIE = None def __init__(self, *args, **kwargs): - username = None password = None ssl_verify = True diff --git a/jenkinsapi/view.py b/jenkinsapi/view.py index d71b0430..732f3582 100644 --- a/jenkinsapi/view.py +++ b/jenkinsapi/view.py @@ -1,7 +1,10 @@ """ Module for jenkinsapi views """ -import six +from __future__ import annotations + +from typing import Iterator, Tuple + import logging from jenkinsapi.jenkinsbase import JenkinsBase @@ -9,7 +12,7 @@ from jenkinsapi.custom_exceptions import NotFound -log = logging.getLogger(__name__) +log: logging.Logger = logging.getLogger(__name__) class View(JenkinsBase): @@ -18,65 +21,65 @@ class View(JenkinsBase): View class """ - def __init__(self, url, name, jenkins_obj): - self.name = name - self.jenkins_obj = jenkins_obj + def __init__(self, url: str, name: str, jenkins_obj: "Jenkins") -> None: + self.name: str = name + self.jenkins_obj: "Jenkins" = jenkins_obj JenkinsBase.__init__(self, url) - self.deleted = False + self.deleted: bool = False - def __len__(self): + def __len__(self) -> int: return len(self.get_job_dict().keys()) - def __str__(self): + def __str__(self) -> str: return self.name - def __repr__(self): + def __repr__(self) -> str: return self.name - def __getitem__(self, job_name): + def __getitem__(self, job_name) -> Job: assert isinstance(job_name, str) api_url = self.python_api_url(self.get_job_url(job_name)) return Job(api_url, job_name, self.jenkins_obj) - def __contains__(self, job_name): + def __contains__(self, job_name: str) -> bool: """ True if view_name is the name of a defined view """ return job_name in self.keys() - def delete(self): + def delete(self) -> None: """ Remove this view object """ - url = "%s/doDelete" % self.baseurl + url: str = f"{self.baseurl}/doDelete" self.jenkins_obj.requester.post_and_confirm_status(url, data="") self.jenkins_obj.poll() self.deleted = True - def keys(self): + def keys(self) -> list[str]: return self.get_job_dict().keys() - def iteritems(self): - it = six.iteritems(self.get_job_dict()) + def iteritems(self) -> Iterator[Tuple[str, Job]]: + it = self.get_job_dict().items() for name, url in it: yield name, Job(url, name, self.jenkins_obj) - def values(self): + def values(self) -> list[Job]: return [a[1] for a in self.iteritems()] def items(self): return [a for a in self.iteritems()] - def _get_jobs(self): + def _get_jobs(self) -> Iterator[Tuple[str, str]]: if "jobs" in self._data: for viewdict in self._data["jobs"]: yield viewdict["name"], viewdict["url"] - def get_job_dict(self): + def get_job_dict(self) -> dict: return dict(self._get_jobs()) - def get_job_url(self, str_job_name): + def get_job_url(self, str_job_name: str) -> str: if str_job_name in self: return self.get_job_dict()[str_job_name] else: @@ -87,10 +90,10 @@ def get_job_url(self, str_job_name): " in view are: %s" % (str_job_name, views_jobs) ) - def get_jenkins_obj(self): + def get_jenkins_obj(self) -> "Jenkins": return self.jenkins_obj - def add_job(self, job_name, job=None): + def add_job(self, job_name: str, job=None) -> bool: """ Add job to a view @@ -136,7 +139,7 @@ def add_job(self, job_name, job=None): ) return True - def remove_job(self, job_name): + def remove_job(self, job_name: str) -> bool: """ Remove job from a view @@ -160,17 +163,17 @@ def remove_job(self, job_name): ) return True - def _get_nested_views(self): + def _get_nested_views(self) -> Iterator[Tuple[str, str]]: for viewdict in self._data.get("views", []): yield viewdict["name"], viewdict["url"] - def get_nested_view_dict(self): + def get_nested_view_dict(self) -> dict: return dict(self._get_nested_views()) - def get_config_xml_url(self): + def get_config_xml_url(self) -> str: return "%s/config.xml" % self.baseurl - def get_config(self): + def get_config(self) -> str: """ Return the config.xml from the view """ @@ -178,7 +181,7 @@ def get_config(self): response = self.get_jenkins_obj().requester.get_and_confirm_status(url) return response.text - def update_config(self, config): + def update_config(self, config: str) -> str: """ Update the config.xml to the view """ diff --git a/jenkinsapi_tests/systests/test_credentials.py b/jenkinsapi_tests/systests/test_credentials.py index 21d57d11..39d35032 100644 --- a/jenkinsapi_tests/systests/test_credentials.py +++ b/jenkinsapi_tests/systests/test_credentials.py @@ -39,7 +39,7 @@ def test_create_user_pass_credential(jenkins): cred = creds[cred_descr] assert isinstance(cred, UsernamePasswordCredential) is True - assert cred.password is None + assert cred.password == "" assert cred.description == cred_descr del creds[cred_descr] diff --git a/jenkinsapi_tests/systests/test_crumbs_requester.py b/jenkinsapi_tests/systests/test_crumbs_requester.py index 530bc94e..411f9a02 100644 --- a/jenkinsapi_tests/systests/test_crumbs_requester.py +++ b/jenkinsapi_tests/systests/test_crumbs_requester.py @@ -1,10 +1,10 @@ +import io import time import json import logging import pytest -from six import StringIO -from six.moves.urllib.parse import urljoin +from urllib.parse import urljoin from jenkinsapi.jenkins import Jenkins from jenkinsapi.utils.crumb_requester import CrumbRequester from jenkinsapi_tests.test_utils.random_strings import random_string @@ -91,7 +91,7 @@ def crumbed_jenkins(jenkins): def test_invoke_job_with_file(crumbed_jenkins): file_data = random_string() - param_file = StringIO(file_data) + param_file = io.BytesIO(file_data.encode("utf-8")) job_name = "create1_%s" % random_string() job = crumbed_jenkins.create_job(job_name, JOB_WITH_FILE) diff --git a/jenkinsapi_tests/systests/test_parameterized_builds.py b/jenkinsapi_tests/systests/test_parameterized_builds.py index 9c04a9ce..1df9942f 100644 --- a/jenkinsapi_tests/systests/test_parameterized_builds.py +++ b/jenkinsapi_tests/systests/test_parameterized_builds.py @@ -1,9 +1,9 @@ """ System tests for `jenkinsapi.jenkins` module. """ +import io import time -from six import StringIO from jenkinsapi_tests.test_utils.random_strings import random_string from jenkinsapi_tests.systests.job_configs import JOB_WITH_FILE from jenkinsapi_tests.systests.job_configs import JOB_WITH_FILE_AND_PARAMS @@ -12,7 +12,7 @@ def test_invoke_job_with_file(jenkins): file_data = random_string() - param_file = StringIO(file_data) + param_file = io.BytesIO(file_data.encode("utf-8")) job_name = "create1_%s" % random_string() job = jenkins.create_job(job_name, JOB_WITH_FILE) @@ -111,7 +111,7 @@ def test_parameterized_multiple_builds_get_the_same_queue_item(jenkins): def test_invoke_job_with_file_and_params(jenkins): file_data = random_string() param_data = random_string() - param_file = StringIO(file_data) + param_file = io.BytesIO(file_data.encode("utf-8")) job_name = "create_%s" % random_string() job = jenkins.create_job(job_name, JOB_WITH_FILE_AND_PARAMS) diff --git a/jenkinsapi_tests/unittests/configs.py b/jenkinsapi_tests/unittests/configs.py index 2783b1bc..9f604770 100644 --- a/jenkinsapi_tests/unittests/configs.py +++ b/jenkinsapi_tests/unittests/configs.py @@ -20,7 +20,7 @@ "name": "param2", "type": "StringParameterDefinition", }, - ] + ], } ], "description": "test job", @@ -83,11 +83,17 @@ "shortDescription": "Started by user anonymous", "userId": None, "userName": "anonymous", + "upstreamProject": "parentBuild", + "upstreamBuild": 1, } - ] + ], + "parameters": [ + {"name": "masterBuild", "value": "masterBuild"}, + {"name": "lastBuild", "value": 1}, + ], } ], - "artifacts": [], + "artifacts": [{"fileName": "foo.txt", "relativePath": "foo.txt"}], "building": False, "builtOn": "localhost", "changeSet": { diff --git a/jenkinsapi_tests/unittests/test_build.py b/jenkinsapi_tests/unittests/test_build.py index 2e2c2f44..a53c439c 100644 --- a/jenkinsapi_tests/unittests/test_build.py +++ b/jenkinsapi_tests/unittests/test_build.py @@ -3,8 +3,10 @@ import pytz from . import configs import datetime +from typing import List from jenkinsapi.build import Build from jenkinsapi.job import Job +from jenkinsapi.artifact import Artifact @pytest.fixture(scope="function") @@ -13,18 +15,18 @@ def jenkins(mocker): @pytest.fixture(scope="function") -def job(monkeypatch, jenkins): +def job(monkeypatch, jenkins) -> Job: def fake_poll(cls, tree=None): # pylint: disable=unused-argument return configs.JOB_DATA monkeypatch.setattr(Job, "_poll", fake_poll) - fake_job = Job("http://", "Fake_Job", jenkins) + fake_job: Job = Job("http://", "Fake_Job", jenkins) return fake_job @pytest.fixture(scope="function") -def build(job, monkeypatch): +def build(job, monkeypatch) -> Build: def fake_poll(cls, tree=None): # pylint: disable=unused-argument return configs.BUILD_DATA @@ -34,7 +36,7 @@ def fake_poll(cls, tree=None): # pylint: disable=unused-argument @pytest.fixture(scope="function") -def build_pipeline(job, monkeypatch): +def build_pipeline(job, monkeypatch) -> Build: def fake_poll(cls, tree=None): # pylint: disable=unused-argument return configs.BUILD_DATA_PIPELINE @@ -43,21 +45,23 @@ def fake_poll(cls, tree=None): # pylint: disable=unused-argument return Build("http://", 97, job) -def test_timestamp(build): +def test_timestamp(build) -> None: assert isinstance(build.get_timestamp(), datetime.datetime) - expected = pytz.utc.localize(datetime.datetime(2013, 5, 31, 23, 15, 40)) + expected: datetime.datetime = pytz.utc.localize( + datetime.datetime(2013, 5, 31, 23, 15, 40) + ) assert build.get_timestamp() == expected -def test_name(build): +def test_name(build) -> None: with pytest.raises(AttributeError): build.id() assert build.name == "foo #1" -def test_duration(build): +def test_duration(build) -> None: expected = datetime.timedelta(milliseconds=5782) assert build.get_duration() == expected assert build.get_duration().seconds == 5 @@ -65,12 +69,14 @@ def test_duration(build): assert str(build.get_duration()) == "0:00:05.782000" -def test_get_causes(build): +def test_get_causes(build) -> None: assert build.get_causes() == [ { "shortDescription": "Started by user anonymous", "userId": None, "userName": "anonymous", + "upstreamProject": "parentBuild", + "upstreamBuild": 1, } ] @@ -273,3 +279,51 @@ def fake_get_data(cls, tree=None, params=None): build.get_env_vars() assert "" == str(excinfo.value) assert len(record) == 0 + + +def test_build_get_status(build) -> None: + assert build.get_status() == "SUCCESS" + + +def test_build_get_params_return_empty_dict(build) -> None: + build._data = {"actions": []} + assert build.get_params() == {} + + +def test_build_get_changeset_empty(build) -> None: + build._data = {"changeSet": {}} + assert build.get_changeset_items() == [] + + +def test_build_get_changesets_vcs(build) -> None: + # This test shall test lines 162-164 in build.py + build._data = {"changeSets": {"kind": "git"}} + assert build._get_vcs() == "git" + + +def test_build_get_number(build) -> None: + assert build.get_number() == 1 + + +def test_build_get_artifacts(build) -> None: + afs: List[Artifact] = list(build.get_artifacts()) + assert len(afs) == 1 + assert isinstance(afs[0], Artifact) + assert list(afs)[0].filename == "foo.txt" + + +def test_build_get_upstream_job_name(build) -> None: + assert build.get_upstream_job_name() == "parentBuild" + + +def test_build_get_ustream_job_name_none(build) -> None: + build._data = {"actions": []} + assert build.get_upstream_job_name() is None + + +def test_build_get_master_job_name(build) -> None: + assert build.get_master_job_name() == "masterBuild" + + +def test_build_get_master_build_number(build) -> None: + assert build.get_master_build_number() == 1 diff --git a/jenkinsapi_tests/unittests/test_build_scm_git.py b/jenkinsapi_tests/unittests/test_build_scm_git.py index 10297a30..1cb06645 100644 --- a/jenkinsapi_tests/unittests/test_build_scm_git.py +++ b/jenkinsapi_tests/unittests/test_build_scm_git.py @@ -1,4 +1,3 @@ -import six import pytest from . import configs from jenkinsapi.build import Build @@ -35,7 +34,7 @@ def test_git_scm(build): """ Can we extract git build revision data from a build object? """ - assert isinstance(build.get_revision(), six.string_types) + assert isinstance(build.get_revision(), str) assert build.get_revision() == "7def9ed6e92580f37d00e4980c36c4d36e68f702" diff --git a/jenkinsapi_tests/unittests/test_compat.py b/jenkinsapi_tests/unittests/test_compat.py deleted file mode 100644 index fef9e429..00000000 --- a/jenkinsapi_tests/unittests/test_compat.py +++ /dev/null @@ -1,23 +0,0 @@ -# -*- coding: utf-8 -*- -import six -from jenkinsapi.utils.compat import ( - to_string, - needs_encoding, -) - - -def test_needs_encoding_py2(): - if six.PY2: - unicode_str = "юникод" - assert needs_encoding(unicode_str) - - assert not needs_encoding("string") - assert not needs_encoding(5) - assert not needs_encoding(["list", "of", "strings"]) - - -def test_to_string(): - assert isinstance(to_string(5), str) - assert isinstance(to_string("string"), str) - assert isinstance(to_string(["list", "of", "strings"]), str) - assert isinstance(to_string("unicode"), str) diff --git a/jenkinsapi_tests/unittests/test_job.py b/jenkinsapi_tests/unittests/test_job.py index 6b768b5e..068103b8 100644 --- a/jenkinsapi_tests/unittests/test_job.py +++ b/jenkinsapi_tests/unittests/test_job.py @@ -274,7 +274,7 @@ def test_get_json_for_many_params(): expected = ( '{"parameter": [{"name": "A", "value": "Boo"}, ' '{"name": "B", "value": "Honey"}, ' - '{"name": "C", "value": "2"}], ' + '{"name": "C", "value": 2}], ' '"statusCode": "303", "redirectTo": "."}' ) diff --git a/requirements.txt b/requirements.txt index 9c22e1f5..da330e69 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,2 @@ pytz>=2014.4 requests>=2.3.0 -six>=1.10.0