From 6414ae14537b4bf91382f8f5fc34bf0a4782c6af Mon Sep 17 00:00:00 2001 From: YoussefKhaildAli Date: Sun, 6 Jul 2025 05:38:59 +0300 Subject: [PATCH] Added a test and some adjustments --- plugins/modules/jenkins_plugin.py | 27 +++++---- .../plugins/modules/test_jenkins_plugin.py | 60 +++++++++++++++++++ 2 files changed, 74 insertions(+), 13 deletions(-) diff --git a/plugins/modules/jenkins_plugin.py b/plugins/modules/jenkins_plugin.py index 91f2688199..7a8c745513 100644 --- a/plugins/modules/jenkins_plugin.py +++ b/plugins/modules/jenkins_plugin.py @@ -326,6 +326,7 @@ import os import tempfile import time import base64 +from collections import OrderedDict from ansible.module_utils.basic import AnsibleModule, to_bytes from ansible.module_utils.six.moves import http_cookiejar as cookiejar @@ -353,19 +354,19 @@ class JenkinsPlugin(object): # Authentication for non-Jenkins calls self.updates_url_credentials = {} - if self.params['updates_url_username'] and self.params['updates_url_password']: - auth = f"{self.params['updates_url_username']}:{self.params['updates_url_password']}".encode("utf-8") + if self.params.get('updates_url_username') and self.params.get('updates_url_password'): + auth = "{}:{}".format(self.params['updates_url_username'], self.params['updates_url_password']).encode("utf-8") b64_auth = base64.b64encode(auth).decode("ascii") - self.updates_url_credentials["Authorization"] = f"Basic {b64_auth}" + self.updates_url_credentials["Authorization"] = "Basic {}".format(b64_auth) # Crumb self.crumb = {} # Authentication for Jenkins calls - if self.params['url_username'] and self.params['url_password']: - auth = f"{self.params['url_username']}:{self.params['url_password']}".encode("utf-8") + if self.params.get('url_username') and self.params.get('url_password'): + auth = "{}:{}".format(self.params['url_username'], self.params['url_password']).encode("utf-8") b64_auth = base64.b64encode(auth).decode("ascii") - self.crumb["Authorization"] = f"Basic {b64_auth}" + self.crumb["Authorization"] = "Basic {}".format(b64_auth) # Cookie jar for crumb session self.cookies = None @@ -457,11 +458,11 @@ class JenkinsPlugin(object): if response.getcode() != 200: if dont_fail: - raise FailedInstallingWithPluginManager(f"HTTP {response.getcode()}") + raise FailedInstallingWithPluginManager("HTTP {}".format(response.getcode())) else: self.module.fail_json( msg=msg_status, - details=f"Received status code {response.getcode()} from {url}" + details="Received status code {} from {}".format(response.getcode(), url) ) except Exception as e: if dont_fail: @@ -693,13 +694,13 @@ class JenkinsPlugin(object): else: raise FileNotFoundError("Cache file is outdated.") except Exception: - resp, info = fetch_url(self.module, "https://updates.jenkins.io/current/plugin-versions.json") # Get list of plugins and their dependencies + response = open_url("https://updates.jenkins.io/current/plugin-versions.json") # Get list of plugins and their dependencies - if info.get("status") != 200: + if response.getcode() != 200: self.module.fail_json(msg="Failed to fetch plugin-versions.json", details=info) try: - plugin_data = json.loads(to_native(resp.read())) + plugin_data = json.loads(to_native(response.read()), object_pairs_hook=OrderedDict) # Save it to file for next time with open(cache_path, "w") as f: @@ -711,9 +712,9 @@ class JenkinsPlugin(object): if not plugin_versions: self.module.fail_json(msg="Plugin '{}' not found.".format(name)) - sorted_versions = dict(reversed(list(plugin_versions.items()))) + sorted_versions = list(reversed(plugin_versions.items())) - for idx, (version_title, version_info) in enumerate(sorted_versions.items()): + for idx, (version_title, version_info) in enumerate(sorted_versions): required_core = version_info.get("requiredCore", "0.0") if self.parse_version(required_core) <= self.jenkins_version: return 'latest' if idx == 0 else version_title diff --git a/tests/unit/plugins/modules/test_jenkins_plugin.py b/tests/unit/plugins/modules/test_jenkins_plugin.py index 194cc2d724..7161ef4b37 100644 --- a/tests/unit/plugins/modules/test_jenkins_plugin.py +++ b/tests/unit/plugins/modules/test_jenkins_plugin.py @@ -6,9 +6,16 @@ from __future__ import (absolute_import, division, print_function) __metaclass__ = type from io import BytesIO +import json +import socket +from collections import OrderedDict from ansible_collections.community.general.plugins.modules.jenkins_plugin import JenkinsPlugin from ansible.module_utils.common._collections_compat import Mapping +from ansible_collections.community.internal_test_tools.tests.unit.compat.mock import ( + MagicMock, + patch, +) def pass_function(*args, **kwargs): @@ -190,3 +197,56 @@ def isInList(l, i): if item == i: return True return False + + +@patch("ansible_collections.community.general.plugins.modules.jenkins_plugin.open_url") +@patch("ansible_collections.community.general.plugins.modules.jenkins_plugin.fetch_url") +def test__get_latest_compatible_plugin_version(fetch_mock, open_mock, mocker): + "test the latest compatible plugin version retrieval" + + params = { + "url": "http://fake.jenkins.server", + "timeout": 30, + "name": "git", + "version": "latest", + "updates_url": ["https://some.base.url"], + "latest_plugins_url_segments": ["test_latest"], + "jenkins_home": "/var/lib/jenkins", + } + module = mocker.Mock() + module.params = params + + mock_response = MagicMock() + mock_response.read.return_value = b"" + fetch_mock.return_value = (mock_response, {"x-jenkins": "2.263.1"}) + + try: + socket.gethostbyname("updates.jenkins.io") + online = True + except socket.gaierror: + online = False + + # Mock the open_url to simulate the response from Jenkins update center if tests are run offline + if not online: + plugin_data = { + "plugins": { + "git": OrderedDict([ + ("4.8.2", {"requiredCore": "2.263.1"}), + ("4.8.3", {"requiredCore": "2.263.1"}), + ("4.9.0", {"requiredCore": "2.289.1"}), + ("4.9.1", {"requiredCore": "2.289.1"}), + ]) + } + } + mock_open_resp = MagicMock() + mock_open_resp.getcode.return_value = 200 + mock_open_resp.read.return_value = json.dumps(plugin_data).encode("utf-8") + open_mock.return_value = mock_open_resp + + JenkinsPlugin._csrf_enabled = pass_function + JenkinsPlugin._get_installed_plugins = pass_function + + jenkins_plugin = JenkinsPlugin(module) + + latest_version = jenkins_plugin._get_latest_compatible_plugin_version() + assert latest_version == '4.8.3'