password lookup - re-attempt acquiring lock file regardless of locale (#85318) (#85349)

* Fix handling FileExistsError, instead of only handling OSError when the human-readable error message is "File exists".


(cherry picked from commit 8e9f5fb9d5)

Co-authored-by: Matt Clay <matt@mystile.com>
pull/85454/head
Sloane Hertel 5 months ago committed by GitHub
parent 21f9b1a3a4
commit 0fbbafb581
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -0,0 +1,2 @@
bugfixes:
- password lookup - fix acquiring the lock when human-readable FileExistsError error message is not English.

@ -126,6 +126,7 @@ _raw:
elements: str elements: str
""" """
import contextlib
import os import os
import string import string
import time import time
@ -269,15 +270,12 @@ def _get_lock(b_path):
b_pathdir = os.path.dirname(b_path) b_pathdir = os.path.dirname(b_path)
lockfile_name = to_bytes("%s.ansible_lockfile" % hashlib.sha1(b_path).hexdigest()) lockfile_name = to_bytes("%s.ansible_lockfile" % hashlib.sha1(b_path).hexdigest())
lockfile = os.path.join(b_pathdir, lockfile_name) lockfile = os.path.join(b_pathdir, lockfile_name)
if not os.path.exists(lockfile) and b_path != to_bytes('/dev/null'): if b_path != b'/dev/null':
try: makedirs_safe(b_pathdir, mode=0o700)
makedirs_safe(b_pathdir, mode=0o700) with contextlib.suppress(FileExistsError):
fd = os.open(lockfile, os.O_CREAT | os.O_EXCL) fd = os.open(lockfile, os.O_CREAT | os.O_EXCL)
os.close(fd) os.close(fd)
first_process = True first_process = True
except OSError as e:
if e.strerror != 'File exists':
raise
counter = 0 counter = 0
# if the lock is got by other process, wait until it's released # if the lock is got by other process, wait until it's released

@ -466,11 +466,16 @@ class TestLookupModuleWithoutPasslib(BaseTestLookupModule):
@patch('time.sleep') @patch('time.sleep')
def test_lock_been_held(self, mock_sleep): def test_lock_been_held(self, mock_sleep):
# pretend the lock file is here # pretend the lock file is here
password.os.path.exists = lambda x: True def _already_exists(*args, **kwargs):
with pytest.raises(AnsibleError): raise FileExistsError("The lock is busy, wait and try again.")
with patch.object(builtins, 'open', mock_open(read_data=b'hunter42 salt=87654321\n')) as m:
# should timeout here with (
self.password_lookup.run([u'/path/to/somewhere chars=anything'], None) pytest.raises(AnsibleError, match='^Password lookup cannot get the lock in 7 seconds.*'),
patch.object(password.os, 'open', _already_exists),
patch.object(password.os.path, 'exists', lambda *args, **kwargs: True),
):
# should timeout here
self.password_lookup.run([u'/path/to/somewhere chars=anything'], None)
def test_lock_not_been_held(self): def test_lock_not_been_held(self):
# pretend now there is password file but no lock # pretend now there is password file but no lock

Loading…
Cancel
Save