AAP-80457 - Fix credential_input_source to handle state: absent when target crede… (#16523)

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 <lallen@redhat.com>
This commit is contained in:
Nick Meyer
2026-07-09 08:42:06 -04:00
committed by GitHub
parent d0576d7823
commit 354fa35860
2 changed files with 101 additions and 1 deletions

View File

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

View File

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