Added a test and some adjustments

This commit is contained in:
YoussefKhaildAli 2025-07-06 05:38:59 +03:00
commit 6414ae1453
2 changed files with 74 additions and 13 deletions

View file

@ -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

View file

@ -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'