use new-style API tokens for Jenkins >= 2.129.1

lib/charms/layer/jenkins/api.py

@@ -1,4 +1,5 @@
 import requests
+from distutils.version import LooseVersion
 from urllib.parse import urljoin, urlparse
 
 import jenkins
@@ -7,18 +8,27 @@
 from charmhelpers.core.hookenv import ERROR
 
 from charms.layer.jenkins.credentials import Credentials
+from charms.layer.jenkins.packages import Packages
 
 RETRIABLE = (
     requests.exceptions.RequestException,
     jenkins.JenkinsException,
     )
 
-GET_TOKEN_SCRIPT = """
+GET_LEGACY_TOKEN_SCRIPT = """
 user = hudson.model.User.get('{}')
 prop = user.getProperty(jenkins.security.ApiTokenProperty.class)
 println(prop.getApiToken())
 """
 
+GET_NEW_TOKEN_SCRIPT = """
+user = hudson.model.User.get('{}')
+prop = user.getProperty(jenkins.security.ApiTokenProperty.class)
+result = prop.tokenStore.generateNewToken("token-created-by-script")
+user.save()
+println(result.plainValue)
+"""
+
 # flake8: noqa
 UPDATE_PASSWORD_SCRIPT = """
 user = hudson.model.User.get('{username}')
@@ -30,6 +40,9 @@
 class Api(object):
     """Encapsulate operations on the Jenkins master."""
 
+    def __init__(self, packages=None):
+        self._packages = packages or Packages()
+
     @property
     def url(self):
         config = hookenv.config()
@@ -124,7 +137,12 @@ def _make_client(self):
         # TODO: also handle regenerated tokens
         if token is None:
             client = jenkins.Jenkins(self.url, user, creds.password())
-            token = client.run_script(GET_TOKEN_SCRIPT.format(user)).strip()
+            # If we're using Jenkins >= 2.129 we need to request a new token.
+            jenkins_version = self._packages.jenkins_version()
+            if LooseVersion(jenkins_version) >= LooseVersion('2.129'):
+                token = client.run_script(GET_NEW_TOKEN_SCRIPT.format(user)).strip()
+            else:
+                token = client.run_script(GET_LEGACY_TOKEN_SCRIPT.format(user)).strip()
             creds.token(token)
 
         client = jenkins.Jenkins(self.url, user, token)

lib/charms/layer/jenkins/packages.py

@@ -50,6 +50,9 @@ def install_jenkins(self):
             self._setup_source(release)
         self._apt.queue_install(["jenkins"])
 
+    def jenkins_version(self):
+        return self._apt.get_package_version('jenkins', full_version=True)
+
     def _install_from_bundle(self):
         """Install Jenkins from bundled package."""
         # Check bundled package exists.

requirements.txt

@@ -1,3 +1,4 @@
-flake8
 charmhelpers
+flake8
+mock
 python-jenkins

unit_tests/stubs/apt.py

@@ -8,9 +8,18 @@ class AptStub(object):
     def __init__(self):
         self.installs = []
         self.sources = []
+        self._package_versions = {
+            'jenkins': '2.150.1',
+            }
 
     def queue_install(self, packages):
         self.installs.extend(packages)
 
     def add_source(self, source, key=None):
         self.sources.append((source, key))
+
+    def get_package_version(self, package, full_version=False):
+        return self._package_versions[package]
+
+    def _set_jenkins_version(self, jenkins_version):
+        self._package_versions['jenkins'] = jenkins_version

unit_tests/test_api.py

@@ -9,24 +9,32 @@
 from states import JenkinsConfiguredAdmin
 
 from charms.layer.jenkins.api import (
-    GET_TOKEN_SCRIPT,
+    GET_LEGACY_TOKEN_SCRIPT,
+    GET_NEW_TOKEN_SCRIPT,
     UPDATE_PASSWORD_SCRIPT,
     Api,
 )
+from charms.layer.jenkins.packages import Packages
+
+from stubs.apt import AptStub
 
 
 class ApiTest(JenkinsTest):
 
     def setUp(self):
         super(ApiTest, self).setUp()
         self.useFixture(JenkinsConfiguredAdmin(self.fakes))
-        self.fakes.jenkins.scripts[GET_TOKEN_SCRIPT.format("admin")] = "abc\n"
-        self.api = Api()
+        self.fakes.jenkins.scripts[GET_LEGACY_TOKEN_SCRIPT.format("admin")] = "abc\n"
+        self.fakes.jenkins.scripts[GET_NEW_TOKEN_SCRIPT.format("admin")] = "xyz\n"
+        self.apt = AptStub()
+        self.packages = Packages(apt=self.apt)
+        self.api = Api(packages=self.packages)
 
     def test_wait_transient_failure(self):
         """
         Wait for Jenkins to be fully up, even in spite of transient failures.
         """
+        self.apt._set_jenkins_version('2.120.1')
         get_whoami = self.fakes.jenkins.get_whoami
         tries = []
 
@@ -46,6 +54,7 @@ def test_update_password(self):
         The update_password() method runs a groovy script to update the
         password for the given user.
         """
+        self.apt._set_jenkins_version('2.120.1')
         username = "joe"
         password = "new"
         script = UPDATE_PASSWORD_SCRIPT.format(
@@ -55,12 +64,18 @@ def test_update_password(self):
 
     def test_version(self):
         """The version() method returns the version of the Jenkins server."""
+        self.apt._set_jenkins_version('2.120.1')
+        self.assertEqual("2.0.0", self.api.version())
+
+    def test_new_token_script(self):
+        self.apt._set_jenkins_version('2.150.1')
         self.assertEqual("2.0.0", self.api.version())
 
     def test_add(self):
         """
         A slave node can be added by specifying executors and labels.
         """
+        self.apt._set_jenkins_version('2.120.1')
         self.api.add_node("slave-0", 1, labels=["python"])
         [node] = self.fakes.jenkins.nodes
         self.assertEqual("slave-0", node.host)
@@ -73,6 +88,7 @@ def test_add_exists(self):
         """
         If a node already exists, nothing is done.
         """
+        self.apt._set_jenkins_version('2.120.1')
         self.fakes.jenkins.create_node("slave-0", 1, "slave-0")
         self.api.add_node("slave-0", 1, labels=["python"])
         self.assertEqual(1, len(self.fakes.jenkins.nodes))
@@ -81,6 +97,7 @@ def test_add_transient_failure(self):
         """
         Transient failures get retried.
         """
+        self.apt._set_jenkins_version('2.120.1')
         create_node = self.fakes.jenkins.create_node
         tries = []
 
@@ -100,6 +117,7 @@ def test_add_retry_give_up(self):
         """
         If errors persist, we give up.
         """
+        self.apt._set_jenkins_version('2.120.1')
 
         def failure(*args, **kwargs):
             raise JenkinsException("error")
@@ -113,6 +131,7 @@ def test_add_spurious(self):
         If adding a node apparently succeeds, but actually didn't then we
         log an error.
         """
+        self.apt._set_jenkins_version('2.120.1')
         self.fakes.jenkins.create_node = lambda *args, **kwargs: None
         self.api.add_node("slave-0", 1, labels=["python"])
         self.assertEqual(
@@ -122,6 +141,7 @@ def test_deleted(self):
         """
         A slave node can be deleted by specifyng its host name.
         """
+        self.apt._set_jenkins_version('2.120.1')
         self.api.add_node("slave-0", 1, labels=["python"])
         self.api.delete_node("slave-0")
         self.assertEqual([], self.fakes.jenkins.nodes)
@@ -130,6 +150,7 @@ def test_deleted_no_present(self):
         """
         If a slave node doesn't exists, deleting it is a no-op.
         """
+        self.apt._set_jenkins_version('2.120.1')
         self.api.delete_node("slave-0")
         self.assertEqual([], self.fakes.jenkins.nodes)
 
@@ -145,6 +166,7 @@ def test_reload(self):
         The reload method POSTs a request to the '/reload' URL, expecting
         a 503 on the homepage (which happens after redirection).
         """
+        self.apt._set_jenkins_version('2.120.1')
         error = self._make_httperror(self.api.url, 503, "Service Unavailable")
         self.fakes.jenkins.responses[urljoin(self.api.url, "reload")] = error
         self.api.reload()
@@ -153,6 +175,7 @@ def test_reload_unexpected_error(self):
         """
         If the error code is not 403, the error is propagated.
         """
+        self.apt._set_jenkins_version('2.120.1')
         error = self._make_httperror(self.api.url, 403, "Forbidden")
         self.fakes.jenkins.responses[urljoin(self.api.url, "reload")] = error
         self.assertRaises(HTTPError, self.api.reload)
@@ -161,6 +184,7 @@ def test_reload_unexpected_url(self):
         """
         If the error URL is not the root, the error is propagated.
         """
+        self.apt._set_jenkins_version('2.120.1')
         error = self._make_httperror(self.api.url, 503, "Service Unavailable")
         error.response.url = urljoin(self.api.url, "/foo")
         self.fakes.jenkins.responses[urljoin(self.api.url, "reload")] = error
@@ -170,6 +194,7 @@ def test_reload_unexpected_success(self):
         """
         If the request unexpectedly succeeds, an error is raised.
         """
+        self.apt._set_jenkins_version('2.120.1')
         self.fakes.jenkins.responses[urljoin(self.api.url, "reload")] = "home"
         self.assertRaises(RuntimeError, self.api.reload)
 

unit_tests/test_packages.py

@@ -143,3 +143,9 @@ def test_install_jenkins_invalid_release(self):
                 "Release 'foo' configuration not recognised", str(error))
         finally:
             hookenv.config()["release"] = orig_release
+
+    def test_jenkins_version(self):
+        self.assertEqual(self.packages.jenkins_version(), '2.150.1')
+        # And now test older version.
+        self.apt._set_jenkins_version('2.128.1')
+        self.assertEqual(self.packages.jenkins_version(), '2.128.1')

unit_tests/test_users.py

@@ -1,3 +1,5 @@
+import mock
+
 from fixtures import MonkeyPatch
 
 from testtools.matchers import (
@@ -12,7 +14,7 @@
 from charms.layer.jenkins import paths
 from charms.layer.jenkins.users import Users
 from charms.layer.jenkins.api import (
-    GET_TOKEN_SCRIPT,
+    GET_LEGACY_TOKEN_SCRIPT,
     UPDATE_PASSWORD_SCRIPT,
 )
 
@@ -25,13 +27,15 @@ class UsersTest(JenkinsTest):
     def setUp(self):
         super(UsersTest, self).setUp()
         self.useFixture(AptInstalledJenkins(self.fakes))
-        self.fakes.jenkins.scripts[GET_TOKEN_SCRIPT.format("admin")] = "abc\n"
+        self.fakes.jenkins.scripts[GET_LEGACY_TOKEN_SCRIPT.format("admin")] = "abc\n"
         self.users = Users()
 
-    def test_configure_admin_custom_password(self):
+    @mock.patch('charms.layer.jenkins.packages.Packages.jenkins_version')
+    def test_configure_admin_custom_password(self, _jenkins_version):
         """
         If a password is provided, it's used to configure the admin user.
         """
+        _jenkins_version.return_value = '2.120.1'
         config = hookenv.config()
         orig_password = config["password"]
         try:
@@ -51,10 +55,12 @@ def test_configure_admin_custom_password(self):
         finally:
             config["password"] = orig_password
 
-    def test_configure_admin_random_password(self):
+    @mock.patch('charms.layer.jenkins.packages.Packages.jenkins_version')
+    def test_configure_admin_random_password(self, _jenkins_version):
         """
         If a password is not provided, a random one will be generated.
         """
+        _jenkins_version.return_value = '2.120.1'
         def pwgen(length):
             return "z"
         script = UPDATE_PASSWORD_SCRIPT.format(username="admin", password="z")