From 354fa358608d84d85b88d1f2fd21029e53317c17 Mon Sep 17 00:00:00 2001 From: Nick Meyer Date: Thu, 9 Jul 2026 08:42:06 -0400 Subject: [PATCH] =?UTF-8?q?AAP-80457=20-=20Fix=20credential=5Finput=5Fsour?= =?UTF-8?q?ce=20to=20handle=20state:=20absent=20when=20target=20crede?= =?UTF-8?q?=E2=80=A6=20(#16523)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix credential_input_source to handle state: absent when target credential is missing The credential_input_source module would error when trying to delete a credential input source (state: absent) if the target credential didn't exist. This breaks idempotent playbook runs where cleanup tasks assume resources may already be gone. The fix only applies to state: absent; state: present still correctly fails when the target credential doesn't exist. Co-authored-by: Liam Allen --- .../modules/credential_input_source.py | 10 +- .../test/awx/test_credential_input_source.py | 92 +++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) diff --git a/awx_collection/plugins/modules/credential_input_source.py b/awx_collection/plugins/modules/credential_input_source.py index 3cc0cc2457..c9f371aa7f 100644 --- a/awx_collection/plugins/modules/credential_input_source.py +++ b/awx_collection/plugins/modules/credential_input_source.py @@ -93,7 +93,15 @@ def main(): metadata = module.params.get('metadata') state = module.params.get('state') - target_credential_id = module.resolve_name_to_id('credentials', target_credential) + # The target credential lookup should not fail if the target credential is absent and the + # state on the credential input source is also absent. If the credential input source has a + # state of present, then this should fail as the target credential cannot be nonexistent. + target_credential_lookup = module.get_one('credentials', name_or_id=target_credential, allow_none=(state == 'absent')) + + if target_credential_lookup is None: + module.exit_json(**{'changed': False}) + else: + target_credential_id = target_credential_lookup['id'] # Attempt to look up the object based on the target credential and input field lookup_data = { diff --git a/awx_collection/test/awx/test_credential_input_source.py b/awx_collection/test/awx/test_credential_input_source.py index 5978939af9..07fd49640e 100644 --- a/awx_collection/test/awx/test_credential_input_source.py +++ b/awx_collection/test/awx/test_credential_input_source.py @@ -358,3 +358,95 @@ def test_centrify_vault_credential_source(run_module, admin_user, organization, assert cis.target_credential.name == tgt_cred.name assert cis.input_field_name == 'password' assert result['id'] == cis.pk + + +@pytest.mark.django_db +def test_credential_input_source_delete(run_module, admin_user, organization, source_cred_aim, silence_deprecation): + ct = CredentialType.defaults['ssh']() + ct.save() + tgt_cred = Credential.objects.create(name='Test Machine Credential', organization=organization, credential_type=ct, inputs={'username': 'nick'}) + + result = run_module( + 'credential_input_source', + dict( + source_credential=source_cred_aim.name, + target_credential=tgt_cred.name, + input_field_name='password', + metadata={"object_query": "Safe=SUPERSAFE;Object=MyAccount"}, + state='present', + ), + admin_user, + ) + + assert not result.get('failed', False), result.get('msg', result) + assert result.get('changed'), result + assert CredentialInputSource.objects.count() == 1 + + delete_result = run_module( + 'credential_input_source', + dict( + target_credential=tgt_cred.name, + input_field_name='password', + state='absent', + ), + admin_user, + ) + + assert not delete_result.get('failed', False), delete_result.get('msg', delete_result) + assert delete_result.get('changed'), delete_result + assert CredentialInputSource.objects.count() == 0 + + +@pytest.mark.django_db +def test_credential_input_source_delete_nonexistent(run_module, admin_user, organization, source_cred_aim, silence_deprecation): + ct = CredentialType.defaults['ssh']() + ct.save() + tgt_cred = Credential.objects.create(name='Test Machine Credential', organization=organization, credential_type=ct, inputs={'username': 'bob'}) + + result = run_module( + 'credential_input_source', + dict( + target_credential=tgt_cred.name, + input_field_name='password', + state='absent', + ), + admin_user, + ) + + assert not result.get('failed', False), result.get('msg', result) + assert not result.get('changed'), result + assert CredentialInputSource.objects.count() == 0 + + +@pytest.mark.django_db +def test_credential_input_source_delete_missing_target_credential(run_module, admin_user, organization, silence_deprecation): + result = run_module( + 'credential_input_source', + dict( + target_credential='nonexistent-credential', + input_field_name='password', + state='absent', + ), + admin_user, + ) + + assert not result.get('failed', False), result.get('msg', result) + assert not result.get('changed'), result + assert CredentialInputSource.objects.count() == 0 + + +@pytest.mark.django_db +def test_credential_input_source_create_missing_target_credential(run_module, admin_user, organization, source_cred_aim, silence_deprecation): + result = run_module( + 'credential_input_source', + dict( + source_credential=source_cred_aim.name, + target_credential='nonexistent-credential', + input_field_name='password', + state='present', + ), + admin_user, + ) + + assert result.get('failed', True), "Should fail when target credential doesn't exist with state: present" + assert CredentialInputSource.objects.count() == 0