fix(helm): support Helm v4 force/server-side apply flags (#1164)

* fix(helm): support Helm v4 force/server-side apply flags

Helm v4 deprecated --force and enables server-side apply by default, so
playbooks relying on force=true broke with "Flag --force has been
deprecated" and "cannot use server-side apply and force replace together".
Because force always meant a client-side delete/recreate replacement, the
module now reproduces that behaviour on v4 (--server-side=false
--force-replace) so existing playbooks keep working unchanged across both
Helm versions.

To let users opt into Helm v4's server-side world explicitly, add
server_side and force_conflicts options. These are mutually exclusive with
force (a client-side replacement cannot coexist with a server-side mode),
keeping the surface unambiguous and surfacing bad combinations up front
instead of as cryptic Helm errors.

AI-assisted commit with Claude Opus

* fix(linter) reformat tests/unit/modules/test_module_helm.py with black
This commit is contained in:
Yuriy Novostavskiy
2026-07-01 18:24:55 +02:00
committed by GitHub
parent ceeafa8c6d
commit 3634105366
3 changed files with 237 additions and 1 deletions

View File

@@ -10,6 +10,9 @@ import unittest
from unittest.mock import MagicMock, patch
from ansible.module_utils import basic
from ansible_collections.kubernetes.core.plugins.module_utils.helm import (
AnsibleHelmModule,
)
from ansible_collections.kubernetes.core.plugins.modules import helm
from ansible_collections.kubernetes.core.tests.unit.utils.ansible_module_mock import (
AnsibleExitJson,
@@ -548,3 +551,162 @@ class TestDependencyUpdateWithChartRefIsUrl(unittest.TestCase):
result.exception.args[0]["command"]
== "/usr/bin/helm install --dependency-update --replace test 'http://repo.example/charts/application.tgz'"
)
class TestForceFlagHelmVersion(unittest.TestCase):
def setUp(self):
self.mock_module_helper = patch.multiple(
basic.AnsibleModule,
exit_json=exit_json,
fail_json=fail_json,
get_bin_path=get_bin_path,
)
self.mock_module_helper.start()
self.addCleanup(self.mock_module_helper.stop)
self.chart_info = {
"apiVersion": "v2",
"appVersion": "default",
"description": "A chart used in molecule tests",
"name": "test-chart",
"type": "application",
"version": "0.1.0",
}
def _deploy_command(self, params, helm_version, release_status=None):
"""Run helm.main() with mocked dependencies and return the helm command."""
set_module_args(
{
"release_name": "test",
"release_namespace": "test",
"chart_ref": "/tmp/path",
**params,
}
)
helm.get_release_status = MagicMock(return_value=release_status)
helm.fetch_chart_info = MagicMock(return_value=self.chart_info)
# Force the default (non helm-diff) idempotency check path for upgrades.
helm.get_plugin_version = MagicMock(return_value=None)
helm.default_check = MagicMock(return_value=False)
with patch.object(
AnsibleHelmModule, "get_helm_version", return_value=helm_version
):
with patch.object(basic.AnsibleModule, "run_command") as mock_run_command:
mock_run_command.return_value = (0, "deployed", "")
try:
helm.main()
except AnsibleExitJson:
pass
return mock_run_command.call_args_list[-1][0][0]
# --- force flag (existing release, so the upgrade path is exercised) ---
def test_force_uses_force_on_helm_v3(self):
cmd = self._deploy_command(
{"force": True}, "3.10.0", release_status={"status": "deployed"}
)
assert " --force" in cmd
assert "--force-replace" not in cmd
def test_force_uses_force_replace_on_helm_v4(self):
cmd = self._deploy_command(
{"force": True}, "4.0.0", release_status={"status": "deployed"}
)
assert " --server-side=false --force-replace" in cmd
assert " --force " not in cmd
# --- server_side option (fresh install path) ---
def test_server_side_on_helm_v4(self):
cmd = self._deploy_command({"server_side": "true"}, "4.0.0")
assert " --server-side=true" in cmd
def test_server_side_requires_helm_v4(self):
set_module_args(
{
"release_name": "test",
"release_namespace": "test",
"chart_ref": "/tmp/path",
"server_side": "true",
}
)
helm.get_release_status = MagicMock(return_value=None)
helm.fetch_chart_info = MagicMock(return_value=self.chart_info)
with patch.object(AnsibleHelmModule, "get_helm_version", return_value="3.10.0"):
with patch.object(basic.AnsibleModule, "run_command") as mock_run_command:
mock_run_command.return_value = (0, "deployed", "")
with self.assertRaises(AnsibleFailJson):
helm.main()
# --- force_conflicts option (fresh install path) ---
def test_force_conflicts_on_helm_v4(self):
cmd = self._deploy_command({"force_conflicts": True}, "4.0.0")
assert " --force-conflicts" in cmd
def test_force_conflicts_with_server_side(self):
cmd = self._deploy_command(
{"force_conflicts": True, "server_side": "auto"}, "4.0.0"
)
assert " --server-side=auto --force-conflicts" in cmd
def test_force_conflicts_requires_helm_v4(self):
set_module_args(
{
"release_name": "test",
"release_namespace": "test",
"chart_ref": "/tmp/path",
"force_conflicts": True,
}
)
helm.get_release_status = MagicMock(return_value=None)
helm.fetch_chart_info = MagicMock(return_value=self.chart_info)
with patch.object(AnsibleHelmModule, "get_helm_version", return_value="3.10.0"):
with patch.object(basic.AnsibleModule, "run_command") as mock_run_command:
mock_run_command.return_value = (0, "deployed", "")
with self.assertRaises(AnsibleFailJson):
helm.main()
def test_force_conflicts_conflicts_with_server_side_false(self):
set_module_args(
{
"release_name": "test",
"release_namespace": "test",
"chart_ref": "/tmp/path",
"force_conflicts": True,
"server_side": "false",
}
)
helm.get_release_status = MagicMock(return_value=None)
helm.fetch_chart_info = MagicMock(return_value=self.chart_info)
with patch.object(AnsibleHelmModule, "get_helm_version", return_value="4.0.0"):
with patch.object(basic.AnsibleModule, "run_command") as mock_run_command:
mock_run_command.return_value = (0, "deployed", "")
with self.assertRaises(AnsibleFailJson):
helm.main()
# --- mutually exclusive combinations (rejected at module init) ---
def test_force_and_server_side_are_mutually_exclusive(self):
with self.assertRaises(AnsibleFailJson):
set_module_args(
{
"release_name": "test",
"chart_ref": "/tmp/path",
"force": True,
"server_side": "true",
}
)
helm.main()
def test_force_and_force_conflicts_are_mutually_exclusive(self):
with self.assertRaises(AnsibleFailJson):
set_module_args(
{
"release_name": "test",
"chart_ref": "/tmp/path",
"force": True,
"force_conflicts": True,
}
)
helm.main()