[3006.x] Fix x509_v2 state without changes can trigger file module to run in test mode

changelog/64195.fixed.md

@@ -0,0 +1 @@
+Fixed x509_v2 state without changes can trigger file module to run in test mode

salt/states/x509_v2.py

@@ -186,6 +186,7 @@
 import logging
 import os.path
 
+import salt.utils.context
 import salt.utils.files
 from salt.exceptions import CommandExecutionError, SaltInvocationError
 from salt.features import features
@@ -1578,8 +1579,24 @@ def _file_managed(name, test=None, **kwargs):
         raise SaltInvocationError("test param can only be None or True")
     # work around https://github.com/saltstack/salt/issues/62590
     test = test or __opts__["test"]
-    res = __salt__["state.single"]("file.managed", name, test=test, **kwargs)
-    return res[next(iter(res))]
+    file_managed = __states__["file.managed"]
+    if test:
+        # calls via __salt__["state.single"](..., test=test)
+        # can overwrite __opts__["test"] permanently. Workaround:
+        opts = __opts__
+        if not __opts__["test"]:
+            opts = copy.copy(__opts__)
+            opts["test"] = test
+        with salt.utils.context.func_globals_inject(file_managed, __opts__=opts):
+            # The file execution module accesses __opts__["test"] as well
+            with salt.utils.context.func_globals_inject(
+                __salt__["file.check_perms"], __opts__=opts
+            ):
+                with salt.utils.context.func_globals_inject(
+                    __salt__["file.manage_file"], __opts__=opts
+                ):
+                    return file_managed(name, **kwargs)
+    return file_managed(name, **kwargs)
 
 
 def _check_file_ret(fret, ret, current):

tests/pytests/integration/states/test_x509_v2.py

@@ -502,6 +502,17 @@ def existing_cert(x509_salt_call_cli, cert_args, ca_key, rsa_privkey, request):
     yield cert_args["name"]
 
 
+@pytest.fixture(params=[{}])
+def existing_privkey(x509_salt_call_cli, request, tmp_path):
+    pk_file = tmp_path / "priv.key"
+    pk_args = {"name": str(pk_file)}
+    pk_args.update(request.param)
+    ret = x509_salt_call_cli.run("state.single", "x509.private_key_managed", **pk_args)
+    assert ret.returncode == 0
+    assert pk_file.exists()
+    yield pk_args["name"]
+
+
 @pytest.fixture(scope="module")
 def x509_salt_run_cli(x509_salt_master):
     return x509_salt_master.salt_run_cli()
@@ -626,6 +637,56 @@ def test_privkey_new_with_prereq_pkcs12(x509_salt_call_cli, tmp_path):
     assert not _belongs_to(cert_new, pk_cur)
 
 
+def test_file_managed_does_not_run_in_test_mode_after_x509_v2_invocation_without_changes(
+    x509_salt_master, x509_salt_call_cli, tmp_path, existing_privkey
+):
+    """
+    The x509_v2 state module tries to workaround issue #62590 (Test mode does
+    not propagate to __states__ when using prereq) by invoking the ``state.single``
+    execution module with an explicit test parameter. In some cases, this seems
+    to trigger another bug: The file module always runs in test mode afterwards.
+    This seems to be the case when the x509_v2 state module does not report changes
+    after having been invoked at least once before, until another x509_v2 call results
+    in a ``file.managed`` call without test mode.
+    Issue #64195.
+    """
+    new_privkey = tmp_path / "new_privkey"
+    new_file = tmp_path / "new_file"
+    assert not new_file.exists()
+    state = f"""\
+# The result of this call is irrelevant, just that it exists
+Some private key is present:
+  x509.private_key_managed:
+    - name: {new_privkey}
+
+# This single call without changes does not trigger the bug on its own
+Another private key is (already) present:
+  x509.private_key_managed:
+    - name: {existing_privkey}
+
+Subsequent file.managed call should not run in test mode:
+  file.managed:
+    - name: {new_file}
+    - contents: foo
+    - require:
+      - Another private key is (already) present
+    """
+    with x509_salt_master.state_tree.base.temp_file("file_managed_test.sls", state):
+        ret = x509_salt_call_cli.run("state.apply", "file_managed_test")
+        assert ret.returncode == 0
+        assert ret.data
+        x509_res = next(ret.data[x] for x in ret.data if x.startswith("x509_|-Another"))
+        assert x509_res["result"] is True
+        assert not x509_res["changes"]
+        file_res = next(
+            ret.data[x] for x in ret.data if x.startswith("file_|-Subsequent")
+        )
+        assert file_res["result"] is True
+        assert file_res["changes"]
+        assert new_file.exists()
+        assert new_file.read_text() == "foo\n"
+
+
 def _belongs_to(cert_or_pubkey, privkey):
     if isinstance(cert_or_pubkey, cx509.Certificate):
         cert_or_pubkey = cert_or_pubkey.public_key()